config logger wo/breaking cli tests

Author: avaldebe

src/pms/cli.py

@@ -34,6 +34,20 @@ def version_callback(value: bool):
     raise typer.Exit()
 
 
+def configure_logger(debug: bool):
+    logger.enable("pms")
+    if debug:
+        return
+
+    # replace default logger handler
+    try:
+        logger.remove(0)
+    except ValueError as e:
+        logger.debug(f"no default logger handler {e}")
+    else:
+        logger.add(sys.stderr, level="INFO", format="<level>{message}</level>")
+
+
 @main.callback()
 def callback(
     ctx: typer.Context,
@@ -47,23 +61,13 @@ def callback(
     samples: Annotated[
         Optional[int], typer.Option("--samples", "-n", min=1, help="stop after N samples")
     ] = None,
-    debug: Annotated[bool, typer.Option("--debug", help="print DEBUG/logging messages")] = False,
+    debug: Annotated[
+        bool,
+        typer.Option("--debug", help="print DEBUG/logging messages", callback=configure_logger),
+    ] = False,
     version: Annotated[bool, typer.Option("--version", "-V", callback=version_callback)] = False,
 ):
     """Data acquisition and logging for Air Quality Sensors with UART interface"""
-    if debug:
-        logger.enable("pms")
-    else:
-        logger.configure(
-            handlers=[
-                {
-                    "sink": sys.stderr,
-                    "format": "<level>{message}</level>",
-                    "level": "INFO",
-                },
-            ],
-        )
-
     logger.debug(f"PyPMS v{__version__}")
     obj = ctx.ensure_object(dict)
     obj.update(reader=SensorReader(model, port, seconds, samples))

tests/conftest.py

@@ -112,7 +112,7 @@ def msg_hex(self) -> Iterator[str]:
     def samples(self, command: str) -> int:
         return len(self.value) - (command == "mqtt")
 
-    def options(self, command: str) -> list[str]:
+    def options(self, command: str, *, debug: bool = False) -> list[str]:
         capture = f"-m {self.name} -n {self.samples(command)} -i 0"
         cmd = dict(
             serial_csv="serial -f csv",
@@ -121,7 +121,9 @@ def options(self, command: str) -> list[str]:
             capture=f"csv --overwrite --capture {self}_pypms.csv",
             decode=f"serial -f csv --decode {self}_pypms.csv",
         ).get(command, command)
-        return f"--debug {capture} {cmd}".split()
+        if debug:
+            return f"--debug {capture} {cmd}".split()
+        return f"{capture} {cmd}".split()
 
     def output(self, ending: str | None) -> str:
         if ending is None:

tests/test_cli.py

@@ -14,22 +14,24 @@
 
 
 @pytest.mark.parametrize("cmd", ("--version", "-V"))
-def test_version(cmd: str):
+def test_version(cmd: str, logot: Logot):
     result = runner.invoke(main, cmd)
     assert result.exit_code == 0
     assert result.stdout.strip() == f"PyPMS v{__version__}"
+    logot.assert_not_logged(logged.debug(f"PyPMS v{__version__}"))
 
 
-def test_info(capture):
+def test_info(capture, logot: Logot):
     result = runner.invoke(main, capture.options("info"))
     assert result.exit_code == 0
     assert dedent(result.stdout).strip() == capture.output("info").strip()
+    logot.assert_logged(logged.debug(f"PyPMS v{__version__}"))
 
 
 @pytest.mark.parametrize("format", (None, "csv", "hexdump"))
 def test_serial(capture, format: str | None, logot: Logot):
     cmd = "serial" if format is None else f"serial_{format}"
-    result = runner.invoke(main, capture.options(cmd))
+    result = runner.invoke(main, capture.options(cmd, debug=True))
     assert result.exit_code == 0
     assert result.stdout == capture.output(format)
 
@@ -38,7 +40,7 @@ def test_serial(capture, format: str | None, logot: Logot):
 
 
 def test_csv(capture, logot: Logot):
-    result = runner.invoke(main, capture.options("csv"))
+    result = runner.invoke(main, capture.options("csv", debug=True))
     assert result.exit_code == 0
 
     csv = Path(capture.options("csv")[-1])
@@ -53,10 +55,12 @@ def test_csv(capture, logot: Logot):
 def test_capture_decode(capture, logot: Logot):
     result = runner.invoke(main, capture.options("capture"))
     assert result.exit_code == 0
-
     csv = Path(capture.options("capture")[-1])
     assert csv.exists()
 
+    for msg in capture.debug_messages("capture"):
+        logot.assert_logged(logged.debug(msg))
+
     result = runner.invoke(main, capture.options("decode"))
     assert result.exit_code == 0
     csv.unlink()