diff options
| -rw-r--r-- | linuxnamespaces/systemd/__init__.py | 53 |
1 files changed, 47 insertions, 6 deletions
diff --git a/linuxnamespaces/systemd/__init__.py b/linuxnamespaces/systemd/__init__.py index d8e7f86..84cb135 100644 --- a/linuxnamespaces/systemd/__init__.py +++ b/linuxnamespaces/systemd/__init__.py @@ -8,6 +8,48 @@ import sys import typing +_DBUS_INTEGER_BOUNDS = ( + ("q", 0, 1 << 16), + ("n", -(1 << 15), 1 << 15), + ("u", 0, 1 << 32), + ("i", -(1 << 31), 1 << 31), + ("t", 0, 1 << 64), + ("x", -(1 << 63), 1 << 63), +) + + +def _guess_dbus_type(value: typing.Any) -> typing.Iterator[str]: + """Guess the type of a Python value in dbus. May yield multiple candidates. + """ + if isinstance(value, bool): + yield "b" + elif isinstance(value, str): + yield "s" + elif isinstance(value, int): + found = False + for guess, low, high in _DBUS_INTEGER_BOUNDS: + if low <= value < high: + found = True + yield guess + if not found: + raise ValueError("integer out of bounds for dbus") + elif isinstance(value, float): + yield "d" + elif isinstance(value, list): + if not value: + raise ValueError("cannot guess dbus type for empty list") + types = [list(_guess_dbus_type(v)) for v in value] + found = False + for guess in types[0]: + if all(guess in guesses for guesses in types): + found = True + yield "a" + guess + if not found: + raise ValueError("could not determine homogeneous type of list") + else: + raise ValueError("failed to guess dbus type") + + async def start_transient_unit( unitname: str, pids: list[int] | None = None, @@ -20,14 +62,13 @@ async def start_transient_unit( pids = [os.getpid()] dbus_properties.append(("PIDs", ("au", pids))) for key, value in ({} if properties is None else properties).items(): - if isinstance(value, bool): - dbus_properties.append((key, ("b", value))) - elif isinstance(value, str): - dbus_properties.append((key, ("s", value))) - else: + try: + guess = next(_guess_dbus_type(value)) + except ValueError as err: raise ValueError( f"cannot infer dbus type for property {key} value" - ) + ) from err + dbus_properties.append((key, (guess, value))) if dbusdriver in ("auto", "jeepney"): try: from .jeepney import start_transient_unit as jeepney_impl |