diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index e9e1037..70da00f 100644 --- a/faststream_outbox/testing.py +++ b/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 diff --git a/tests/test_client_contract.py b/tests/test_client_contract.py index b0d9601..d258cfd 100644 --- a/tests/test_client_contract.py +++ b/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