Verify live_write notification teardown by end state

tests/live_write/_teardown.py

@@ -0,0 +1,124 @@
+"""Fixture-free teardown helpers for the live-write suite.
+
+Kept separate from ``conftest.py`` for the same reason as ``_gates.py``: offline
+unit tests can import and exercise the teardown logic without importing the
+conftest module, which would register its session-scoped autouse fixtures and
+skip the whole run.
+"""
+
+from __future__ import annotations
+
+from typing import cast
+
+from tests.live_write._gates import MCPTEST_PREFIX
+from unraid_mcp.clients.unraid import MUTATION_DELETE_NOTIFICATION, UnraidClient
+from unraid_mcp.config import UnraidConfig, UnraidMode
+
+NOTIFICATION_BINS = ("UNREAD", "ARCHIVE")
+"""Both notification bins — a fixture can be orphaned in either one."""
+
+PAGE_SIZE = 200
+"""Entries per notification page. The archive bin grows with every session."""
+
+
+def build_raw_client() -> UnraidClient:
+    """A fresh raw GraphQL client for operations with no MCP tool (e.g. seeding)."""
+    cfg = UnraidConfig(unraid_mode=UnraidMode.READWRITE)
+    if cfg.unraid_api_key is None:
+        raise RuntimeError("UNRAID_API_KEY must be set for live_write seeding")
+    return UnraidClient(
+        graphql_url=cfg.graphql_url,
+        api_key=cfg.unraid_api_key,
+        verify_ssl=cfg.unraid_verify_ssl,
+        timeout=cfg.unraid_request_timeout,
+        max_retries=cfg.unraid_max_retries,
+    )
+
+
+async def delete_notification_any_state(notification_id: str) -> None:
+    """Delete from both unread and archive lists (state is unknown at teardown).
+
+    At most one bin holds the notification — and the test may have deleted it
+    already — so some of the two mutations are expected to be no-ops. The API is
+    free to report a no-op either as success or as an error, which makes the
+    number of failed mutations useless as a leak signal: two failures can mean
+    the entry was already gone, and one failure can hide a real error. So the
+    end state is read back instead, and the entry counts as orphaned only if a
+    bin still holds it after both attempts.
+
+    Args:
+        notification_id: Id of the seeded notification to remove.
+
+    Raises:
+        ExceptionGroup: The entry is still present, or the read-back could not
+            be performed. Carries every delete error collected along the way, so
+            ``run_cleanup`` logs the real cause rather than whichever error came
+            first.
+        RuntimeError: The entry is still present although every mutation
+            reported success.
+    """
+    client = build_raw_client()
+    failures: list[Exception] = []
+    try:
+        for ntype in NOTIFICATION_BINS:
+            try:
+                await client.mutate(MUTATION_DELETE_NOTIFICATION, variables={"id": notification_id, "type": ntype})
+            except Exception as exc:
+                failures.append(exc)
+        try:
+            still_present = await _notification_present(client, notification_id)
+        except Exception as exc:
+            raise ExceptionGroup(
+                f"could not verify deletion of notification {notification_id}",
+                [*failures, exc],
+            ) from None
+        if still_present:
+            msg = f"notification {notification_id} still present after ARCHIVE and UNREAD deletes"
+            if failures:
+                raise ExceptionGroup(msg, failures)
+            raise RuntimeError(msg)
+    finally:
+        await client.close()
+
+
+async def _notification_present(client: UnraidClient, notification_id: str) -> bool:
+    """Whether either bin still holds ``notification_id``.
+
+    Both bins are paged to the end rather than sampled: ``QUERY_NOTIFICATIONS``
+    requests no sort order, so a surviving entry can sit anywhere in a bin that
+    the archive tests keep growing.
+    """
+    for ntype in NOTIFICATION_BINS:
+        offset = 0
+        while True:
+            entries = await client.list_notifications(notification_type=ntype, limit=PAGE_SIZE, offset=offset)
+            if any(n.id == notification_id for n in entries):
+                return True
+            if len(entries) < PAGE_SIZE:
+                break
+            offset += PAGE_SIZE
+    return False
+
+
+def select_mcptest_orphans(payload: object) -> list[dict[str, object]]:
+    """Pick the ``mcptest_*`` entries out of one ``unraid_list_notifications`` result.
+
+    The tool's ``structured_content`` wraps its list in a ``{"result": [...]}``
+    envelope — pinned by ``test_orphan_scan_reads_the_live_payload_shape``,
+    because an unrecognized payload silently empties the session-end backstop.
+
+    Args:
+        payload: Structured content from one ``unraid_list_notifications`` call.
+
+    Returns:
+        Entries whose title starts with the ``mcptest`` prefix, or an empty list
+        when the payload holds no recognizable listing.
+    """
+    entries = payload.get("result") if isinstance(payload, dict) else None
+    if not isinstance(entries, list):
+        return []
+    return [
+        cast("dict[str, object]", entry)
+        for entry in entries
+        if isinstance(entry, dict) and str(entry.get("title", "")).lower().startswith(MCPTEST_PREFIX)
+    ]

tests/live_write/conftest.py

@@ -21,13 +21,19 @@
 from tests.live_write._gates import MCPTEST_PREFIX as _MCPTEST_PREFIX
 from tests.live_write._gates import assert_mcptest as _assert_mcptest
 from tests.live_write._gates import require_writes_enabled
-from unraid_mcp.clients.unraid import MUTATION_DELETE_NOTIFICATION, UnraidClient
+from tests.live_write._teardown import NOTIFICATION_BINS as _NOTIFICATION_BINS
+from tests.live_write._teardown import PAGE_SIZE as _PAGE_SIZE
+from tests.live_write._teardown import build_raw_client as _build_raw_client
+from tests.live_write._teardown import delete_notification_any_state as _delete_notification_any_state
+from tests.live_write._teardown import select_mcptest_orphans as _select_mcptest_orphans
 from unraid_mcp.config import UnraidConfig, UnraidMode
 from unraid_mcp.server import create_server
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
 
+    from unraid_mcp.clients.unraid import UnraidClient
+
 _MUTATION_CREATE_NOTIFICATION = """
 mutation CreateNotification($input: NotificationData!) {
     createNotification(input: $input) { id title type }
@@ -84,43 +90,6 @@ async def _do() -> None:
     run_cleanup(label, _do)
 
 
-def _build_raw_client() -> UnraidClient:
-    """A fresh raw GraphQL client for operations with no MCP tool (e.g. seeding)."""
-    cfg = UnraidConfig(unraid_mode=UnraidMode.READWRITE)
-    if cfg.unraid_api_key is None:
-        raise RuntimeError("UNRAID_API_KEY must be set for live_write seeding")
-    return UnraidClient(
-        graphql_url=cfg.graphql_url,
-        api_key=cfg.unraid_api_key,
-        verify_ssl=cfg.unraid_verify_ssl,
-        timeout=cfg.unraid_request_timeout,
-        max_retries=cfg.unraid_max_retries,
-    )
-
-
-async def _delete_notification_any_state(notification_id: str) -> None:
-    """Delete from both unread and archive lists (state is unknown at teardown).
-
-    The notification exists in exactly one state, so one of the two mutations is
-    expected to fail — that failure is suppressed. But if *both* fail the
-    notification is orphaned on the live tower for a real reason (auth revoked,
-    network down, server error), so re-raise to let ``run_cleanup`` surface it
-    instead of silently leaking the fixture.
-    """
-    client = _build_raw_client()
-    failures: list[Exception] = []
-    try:
-        for ntype in ("ARCHIVE", "UNREAD"):
-            try:
-                await client.mutate(MUTATION_DELETE_NOTIFICATION, variables={"id": notification_id, "type": ntype})
-            except Exception as exc:
-                failures.append(exc)
-    finally:
-        await client.close()
-    if len(failures) == 2:
-        raise failures[0]
-
-
 async def _resolve_seeded_id(client: UnraidClient, title: str, before: set[str]) -> str:
     """Resolve the persisted notification id for a just-created ``title``.
 
@@ -295,11 +264,16 @@ def _orphan_scan() -> Iterator[None]:
     """
 
     async def _run_scan() -> None:
+        notif_orphans: list[dict[str, object]] = []
         try:
             cfg = UnraidConfig(unraid_mode=UnraidMode.READWRITE)
             server = create_server(cfg)
             async with Client(server) as scan_client:
-                notifs = (await scan_client.call_tool("unraid_list_notifications", {})).structured_content
+                for ntype in _NOTIFICATION_BINS:
+                    listing = await scan_client.call_tool(
+                        "unraid_list_notifications", {"notification_type": ntype, "limit": _PAGE_SIZE}
+                    )
+                    notif_orphans.extend(_select_mcptest_orphans(listing.structured_content))
         except Exception as exc:
             log.exception("orphan scan failed")
             banner = (
@@ -315,13 +289,8 @@ async def _run_scan() -> None:
             )
             sys.stderr.write(banner)
             sys.stderr.flush()
-            return
-
-        if not isinstance(notifs, list):
-            return
-        notif_orphans = [
-            n for n in notifs if isinstance(n, dict) and str(n.get("title", "")).lower().startswith(_MCPTEST_PREFIX)
-        ]
+            # Fall through: a bin that was scanned before the failure may
+            # already have found orphans worth naming.
 
         if notif_orphans:
             msg_lines = ["\n" + "=" * 72, "ORPHAN mcptest_* ASSETS DETECTED — clean up manually:"]