refactor(testing): name the lease-token guard; fill contract coverage

faststream_outbox/testing.py

@@ -69,6 +69,19 @@ def _claim_fake_row(row: _FakeRow, *, now: _dt.datetime, token: uuid.UUID) -> No
     row.deliveries_count += 1
 
 
+def _lease_token_matches(row_token: "uuid.UUID | None", claim_token: "uuid.UUID | None") -> bool:
+    """Mirror SQL's ``WHERE acquired_token = :token`` guard, ``NULL = NULL`` included.
+
+    ``NULL = NULL`` is NULL (no match) in SQL, so a ``None`` claim token must never match —
+    not even a row whose own token is ``None``. A ``None`` claim short-circuits to ``False``;
+    otherwise the row token must equal it (``None == uuid`` is ``False``). Names the rule the
+    real client's terminal writes encode in SQL; parity with that SQL is pinned by
+    ``tests/test_client_contract.py`` (delete / mark_pending token-match scenarios), not by
+    these comments.
+    """
+    return claim_token is not None and row_token == claim_token
+
+
 class FakeOutboxClient(AbstractOutboxClient):
     """In-memory ``OutboxClient`` substitute. Same surface, list-of-rows storage."""
 
@@ -170,11 +183,7 @@ async def delete_with_lease(
         dlq_payload: "typing.Mapping[str, typing.Any] | None" = None,
     ) -> bool:
         for i, row in enumerate(self._rows):
-            # ``acquired_token is not None`` mirrors SQL's ``WHERE acquired_token = :token``:
-            # ``NULL = NULL`` is NULL (no match), so a None token must never match — even a
-            # row whose own token is None. Without this, the fake's ``None == None`` would
-            # delete where the real client no-ops.
-            if row.id == message_id and acquired_token is not None and row.acquired_token == acquired_token:
+            if row.id == message_id and _lease_token_matches(row.acquired_token, acquired_token):
                 if dlq_payload is not None:
                     # Mirror the real CTE side-effect: DLQ row materializes in the same
                     # call as the DELETE, before the row is removed. Built from the shared
@@ -196,6 +205,9 @@ async def delete_batch_with_lease(
         conn: typing.Any,  # noqa: ARG002
         pairs: "Sequence[tuple[int, uuid.UUID]]",
     ) -> set[int]:
+        # Set-based form of the same rule as `_lease_token_matches`: the `ptok is not None`
+        # filter is the claim-side NULL guard, `row.acquired_token is not None` the row-side —
+        # so a NULL token on either side never matches (SQL `NULL = NULL`).
         wanted = {(pid, ptok) for pid, ptok in pairs if ptok is not None}
         deleted: set[int] = set()
         # Iterate a copy of indices high-to-low so deletions don't shift unscanned rows.
@@ -218,8 +230,7 @@ async def mark_pending_with_lease(
         last_attempt_at: _dt.datetime,
     ) -> bool:
         for row in self._rows:
-            # Mirror SQL ``NULL = NULL`` semantics — see ``delete_with_lease``.
-            if row.id == message_id and acquired_token is not None and row.acquired_token == acquired_token:
+            if row.id == message_id and _lease_token_matches(row.acquired_token, acquired_token):
                 row.next_attempt_at = utcnow() + _dt.timedelta(seconds=max(0.0, delay_seconds))
                 row.attempts_count = attempts_count
                 row.first_attempt_at = first_attempt_at

tests/test_client_contract.py

@@ -323,6 +323,13 @@ async def test_fetch_filters_by_queue(contract: _Harness) -> None:
     assert {m.queue for m in msgs} == {"orders"}
 
 
+async def test_fetch_empty_queues_returns_empty(contract: _Harness) -> None:
+    # Both adapters short-circuit an empty queue list before touching storage.
+    await contract.seed(queue="orders")
+    msgs = await contract.fetch([], limit=10, lease_ttl_seconds=60.0)
+    assert msgs == []
+
+
 # --- delete_with_lease -----------------------------------------------------
 
 
@@ -439,3 +446,23 @@ async def test_mark_pending_noop_on_token_mismatch(contract: _Harness) -> None:
     row = await contract.get_row(rid)
     assert row is not None
     assert row["acquired_token"] == token
+
+
+async def test_mark_pending_noop_on_unleased_row(contract: _Harness) -> None:
+    # The NULL = NULL token guard on the retry path: an unleased row (acquired_token
+    # None) is never reschedulable by any claim token. Parallels the delete twin.
+    now = utcnow()
+    rid = await contract.seed()  # never leased -> acquired_token is None
+    ok = await contract.mark_pending(
+        rid,
+        uuid.uuid4(),
+        delay_seconds=60.0,
+        attempts_count=1,
+        first_attempt_at=now,
+        last_attempt_at=now,
+    )
+    assert ok is False
+    row = await contract.get_row(rid)
+    assert row is not None
+    assert row["acquired_token"] is None
+    assert row["attempts_count"] == 0