summaryrefslogtreecommitdiff
path: root/linuxnamespaces
diff options
context:
space:
mode:
authorHelmut Grohne <helmut@subdivi.de>2024-11-16 07:58:25 +0100
committerHelmut Grohne <helmut@subdivi.de>2024-11-16 08:01:37 +0100
commit09c8ee5649ddda8c3af120de5ed0ecdefe332b3b (patch)
tree9410f6a2e774de11da9fb8ed3c97ce80900f2463 /linuxnamespaces
parent49e73d4b1f69e53dd2ab2c4b7ac0f462f96a5e53 (diff)
downloadpython-linuxnamespaces-09c8ee5649ddda8c3af120de5ed0ecdefe332b3b.tar.gz
systemd: improve dbus type guessingHEADmain
The PIDs property receives integers. Since the guessing code did not handle integers, it would fail immediately.
Diffstat (limited to 'linuxnamespaces')
-rw-r--r--linuxnamespaces/systemd/__init__.py53
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