Client connects to system bus by default

Author: jfhbrook

crystalfontz/dbus/bus.py

@@ -1,48 +1,19 @@
-import functools
 import logging
-from typing import Optional
 
-import click
 from sdbus import (  # pyright: ignore [reportMissingModuleSource]
     sd_bus_open_system,
     sd_bus_open_user,
     set_default_bus,
 )
 
-from crystalfontz.cli import SyncCommand
-
 logger = logging.getLogger(__name__)
 
-BusType = Optional[bool]
-BUS_TYPE = click.BOOL
-
-DEFAULT_BUS = None
-USER_BUS = True
-SYSTEM_BUS = False
-
-
-def bus_type_option(fn: SyncCommand) -> SyncCommand:
-    @click.option(
-        "--user/--system",
-        type=BUS_TYPE,
-        default=DEFAULT_BUS,
-        help="Connect to either the user or system bus",
-    )
-    @functools.wraps(fn)
-    def wrapper(*args, **kwargs) -> None:
-        user = kwargs.pop("user")
-        kwargs["bus_type"] = user
-        return fn(*args, **kwargs)
 
-    return wrapper
+def select_session_bus() -> None:
+    logger.debug("Connecting to the user session bus")
+    set_default_bus(sd_bus_open_user())
 
 
-def configure_bus(bus_type: BusType) -> None:
-    if bus_type is USER_BUS:
-        logger.info("Connecting to the user session bus")
-        set_default_bus(sd_bus_open_user())
-    elif bus_type is SYSTEM_BUS:
-        logger.info("Connecting to the system bus")
-        set_default_bus(sd_bus_open_system())
-    else:
-        logger.info("Connecting to the default bus")
+def select_system_bus() -> None:
+    logger.debug("Connecting to the system bus")
+    set_default_bus(sd_bus_open_system())

crystalfontz/dbus/client/cli.py

@@ -29,7 +29,7 @@
     OutputMode,
     WATCHDOG_SETTING,
 )
-from crystalfontz.dbus.bus import bus_type_option, BusType, configure_bus
+from crystalfontz.dbus.bus import select_session_bus, select_system_bus
 from crystalfontz.dbus.client import DbusClient
 from crystalfontz.dbus.config import StagedConfig
 from crystalfontz.dbus.domain import (
@@ -173,15 +173,20 @@ def warn_dirty() -> None:
     envvar="CRYSTALFONTZ_RETRY_TIMES",
     help="How many times to retry a command if a response times out",
 )
-@bus_type_option
+@click.option(
+    "--user/--default",
+    type=click.BOOL,
+    default=None,
+    help="Connect to either the user or default bus",
+)
 @click.pass_context
 def main(
     ctx: click.Context,
     log_level: LogLevel,
     output: OutputMode,
     timeout: Optional[float],
     retry_times: Optional[int],
-    bus_type: BusType,
+    user: Optional[bool],
 ) -> None:
     """
     Control your Crystalfontz device.
@@ -193,7 +198,10 @@ def main(
     echo.mode = output
 
     async def load() -> None:
-        configure_bus(bus_type)
+        if user:
+            select_session_bus()
+        elif user is None:
+            select_system_bus()
 
         client = DbusClient()
         ctx.obj = Obj(

crystalfontz/dbus/service/__init__.py

@@ -10,7 +10,10 @@
 
 from crystalfontz.cli import LogLevel
 from crystalfontz.config import GLOBAL_FILE
-from crystalfontz.dbus.bus import bus_type_option, BusType, configure_bus
+from crystalfontz.dbus.bus import (
+    select_session_bus,
+    select_system_bus,
+)
 from crystalfontz.dbus.error import handle_dbus_error
 from crystalfontz.dbus.interface import DBUS_NAME, DbusInterface, load_client
 
@@ -71,9 +74,14 @@ async def serve(config_file: Optional[str] = None) -> None:
     default="INFO",
     help="Set the log level",
 )
-@bus_type_option
+@click.option(
+    "--user/--system",
+    type=click.BOOL,
+    default=None,
+    help="Connect to either the user or system bus",
+)
 def main(
-    global_: bool, config_file: str, log_level: LogLevel, bus_type: BusType
+    global_: bool, config_file: str, log_level: LogLevel, user: Optional[bool]
 ) -> None:
     """
     Expose the Crystalfontz device as a DBus service.
@@ -91,6 +99,9 @@ def main(
     elif global_:
         file = GLOBAL_FILE
 
-    configure_bus(bus_type)
+    if user:
+        select_session_bus()
+    elif user is False:
+        select_system_bus()
 
     asyncio.run(serve(file))

docs/dbus/client.md

@@ -16,7 +16,7 @@ Options:
                                   device before timing out
   --retry-times INTEGER           How many times to retry a command if a
                                   response times out
-  --user / --system               Connect to either the user or system bus
+  --user / --default              Connect to either the user or default bus
   --help                          Show this message and exit.
 
 Commands:

tests/conftest.py

@@ -58,7 +58,7 @@ def cli(cli_env: EnvFactory) -> Cli:
 def dbus_service(
     cli_env: EnvFactory, request: pytest.FixtureRequest
 ) -> Generator[None, None, None]:
-    cli = Cli(["python3", "-m", "crystalfontz.dbus.service"], env=cli_env())
+    cli = Cli(["python3", "-m", "crystalfontz.dbus.service", "--user"], env=cli_env())
 
     if request.config.getoption("--system"):
         logger.info("Connecting to system bus")
@@ -73,7 +73,12 @@ def dbus_service(
 def dbus_cli(
     cli_env: EnvFactory, dbus_service: None, request: pytest.FixtureRequest
 ) -> Cli:
-    argv: List[str] = ["python3", "-m", "crystalfontz.dbus.client"]
+    argv: List[str] = [
+        "python3",
+        "-m",
+        "crystalfontz.dbus.client",
+        "--system" if request.config.getoption("--system") else "--user",
+    ]
 
     if not request.config.getoption("--system"):
         argv.append("--user")