Improve CLI coverage for Redis 6.2 runs

Author: zzstoatzz

src/docket/worker.py

@@ -298,7 +298,7 @@ async def get_new_deliveries(redis: Redis) -> RedisReadGroupResponse:
                 else int(self.minimum_check_interval.total_seconds() * 1000),
                 count=available_slots,
             )
-            if is_memory and not result:
+            if is_memory and not result:  # pragma: no branch - memory backend throttle
                 await asyncio.sleep(self.minimum_check_interval.total_seconds())
             return result
 

tests/cli/test_snapshot.py

@@ -1,13 +1,16 @@
 import asyncio
 from datetime import datetime, timedelta, timezone
+from types import TracebackType
 
 import pytest
 from pytest import MonkeyPatch
+from rich.table import Table
 from typer.testing import CliRunner
 
 from docket import tasks
-from docket.cli import app, relative_time
-from docket.docket import Docket
+from docket.cli import app, relative_time, snapshot as snapshot_command
+from docket.docket import Docket, DocketSnapshot, RunningExecution, WorkerInfo
+from docket.execution import Execution
 from docket.worker import Worker
 
 
@@ -249,6 +252,97 @@ async def test_snapshot_with_stats_flag_mixed_tasks(docket: Docket, runner: CliR
         await worker_running
 
 
+def test_snapshot_cli_renders_stats_without_backend(monkeypatch: MonkeyPatch):
+    """snapshot CLI should render tables even when the docket interaction is stubbed."""
+    now = datetime(2023, 1, 1, 12, 0, tzinfo=timezone.utc)
+
+    running_base = Execution(
+        function=tasks.sleep,
+        args=(1,),
+        kwargs={},
+        when=now,
+        key="running-task",
+        attempt=1,
+    )
+    running = RunningExecution(
+        execution=running_base,
+        worker="worker-1",
+        started=now - timedelta(seconds=5),
+    )
+
+    future_execution = Execution(
+        function=tasks.trace,
+        args=("hello!",),
+        kwargs={},
+        when=now + timedelta(seconds=10),
+        key="future-task",
+        attempt=1,
+    )
+
+    snapshot_obj = DocketSnapshot(
+        taken=now,
+        total_tasks=2,
+        future=[future_execution],
+        running=[running],
+        workers=[
+            WorkerInfo(
+                name="worker-1",
+                last_seen=now - timedelta(seconds=2),
+                tasks={"sleep", "trace"},
+            )
+        ],
+    )
+
+    registered: list[str] = []
+
+    class FakeDocket:
+        def __init__(self, name: str, url: str) -> None:
+            self.name = name
+            self.url = url
+
+        async def __aenter__(self) -> "FakeDocket":
+            return self
+
+        async def __aexit__(
+            self,
+            exc_type: type[BaseException] | None,
+            exc: BaseException | None,
+            tb: TracebackType | None,
+        ) -> bool:
+            return False
+
+        def register_collection(self, task_path: str) -> None:
+            registered.append(task_path)
+
+        async def snapshot(self) -> DocketSnapshot:
+            return snapshot_obj
+
+    monkeypatch.setattr("docket.cli.Docket", FakeDocket)
+
+    class RecordingConsole:
+        def __init__(self) -> None:
+            self.calls: list[tuple[tuple[object, ...], dict[str, object]]] = []
+
+        def print(self, *args: object, **kwargs: object) -> None:
+            self.calls.append((args, kwargs))
+
+    recorder = RecordingConsole()
+    monkeypatch.setattr("docket.cli.Console", lambda: recorder)
+
+    snapshot_command(stats=True)
+
+    # The default task collection should be registered and both tables should be printed.
+    assert registered == ["docket.tasks:standard_tasks"]
+    tables = [args[0] for args, _ in recorder.calls if args]
+    assert any(
+        isinstance(obj, Table) and "Docket:" in (obj.title or "") for obj in tables
+    )
+    assert any(
+        isinstance(obj, Table) and obj.title == "Task Count Statistics by Function"
+        for obj in tables
+    )
+
+
 async def test_snapshot_with_stats_shows_timestamp_columns(
     docket: Docket, runner: CliRunner
 ):

tests/cli/test_workers.py

@@ -1,10 +1,12 @@
 import asyncio
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
+from types import TracebackType
 
+from pytest import MonkeyPatch
 from typer.testing import CliRunner
 
-from docket.cli import app
-from docket.docket import Docket
+from docket.cli import app, list_workers, print_workers, workers_for_task
+from docket.docket import Docket, WorkerInfo
 from docket.worker import Worker
 
 
@@ -63,3 +65,108 @@ async def test_list_workers_for_task(docket: Docket, runner: CliRunner):
 
         assert "worker-1" in result.output
         assert "worker-2" in result.output
+
+
+def test_print_workers_highlight(monkeypatch: MonkeyPatch):
+    """print_workers should render a table for the supplied workers."""
+    now = datetime.now(timezone.utc)
+    workers = [
+        WorkerInfo(name="worker-1", last_seen=now, tasks={"trace", "sleep"}),
+    ]
+
+    class RecordingConsole:
+        def __init__(self) -> None:
+            self.objects: list[object] = []
+
+        def print(self, obj: object) -> None:
+            self.objects.append(obj)
+
+    recorder = RecordingConsole()
+    monkeypatch.setattr("docket.cli.Console", lambda: recorder)
+
+    print_workers("demo-docket", workers, highlight_task="sleep")
+
+    assert recorder.objects, "Console.print should be invoked"
+    table = recorder.objects[0]
+    assert getattr(table, "title", "") == "Workers in Docket: demo-docket"
+
+
+def test_list_workers_delegates_to_print(monkeypatch: MonkeyPatch):
+    """list_workers should fetch workers and delegate to print_workers."""
+    now = datetime.now(timezone.utc)
+    workers = [WorkerInfo(name="worker-1", last_seen=now, tasks={"trace"})]
+    captured: list[tuple[str, list[WorkerInfo], str | None]] = []
+
+    def fake_print(
+        docket_name: str,
+        worker_list: list[WorkerInfo],
+        highlight_task: str | None = None,
+    ) -> None:
+        captured.append((docket_name, worker_list, highlight_task))
+
+    class FakeDocket:
+        def __init__(self, name: str, url: str) -> None:
+            self.name = name
+            self.url = url
+
+        async def __aenter__(self) -> "FakeDocket":
+            return self
+
+        async def __aexit__(
+            self,
+            exc_type: type[BaseException] | None,
+            exc: BaseException | None,
+            tb: TracebackType | None,
+        ) -> bool:
+            return False
+
+        async def workers(self) -> list[WorkerInfo]:
+            return workers
+
+    monkeypatch.setattr("docket.cli.print_workers", fake_print)
+    monkeypatch.setattr("docket.cli.Docket", FakeDocket)
+
+    list_workers(docket_="demo", url="redis://localhost:6379/0")
+
+    assert captured == [("demo", workers, None)]
+
+
+def test_workers_for_task_delegates_to_print(monkeypatch: MonkeyPatch):
+    """workers_for_task should fetch filtered workers and highlight the task."""
+    now = datetime.now(timezone.utc)
+    workers = [WorkerInfo(name="worker-2", last_seen=now, tasks={"trace"})]
+    captured: list[tuple[str, list[WorkerInfo], str | None]] = []
+
+    def fake_print(
+        docket_name: str,
+        worker_list: list[WorkerInfo],
+        highlight_task: str | None = None,
+    ) -> None:
+        captured.append((docket_name, worker_list, highlight_task))
+
+    class FakeDocket:
+        def __init__(self, name: str, url: str) -> None:
+            self.name = name
+            self.url = url
+
+        async def __aenter__(self) -> "FakeDocket":
+            return self
+
+        async def __aexit__(
+            self,
+            exc_type: type[BaseException] | None,
+            exc: BaseException | None,
+            tb: TracebackType | None,
+        ) -> bool:
+            return False
+
+        async def task_workers(self, task: str) -> list[WorkerInfo]:
+            assert task == "trace"
+            return workers
+
+    monkeypatch.setattr("docket.cli.print_workers", fake_print)
+    monkeypatch.setattr("docket.cli.Docket", FakeDocket)
+
+    workers_for_task("trace", docket_="demo", url="redis://localhost:6379/0")
+
+    assert captured == [("demo", workers, "trace")]