diff --git a/docs/usage/testing.md b/docs/usage/testing.md index e9a55de..58c28f3 100644 --- a/docs/usage/testing.md +++ b/docs/usage/testing.md @@ -134,7 +134,8 @@ registered handlers are not started, matching production. `run_loops=True` if you need scheduled delivery to actually wait. - `cancel_timer` and `fetch_unprocessed` run against the fake client (the test broker swaps its client in, so `broker.` reaches it). - The `session` argument is still required but is ignored in tests. + The `session` argument must still be an `AsyncSession` or `AsyncConnection` + (`fake_session()` works), as in production, but the fake never touches it. - Publishing runs the production producer, so encoding, the `published` metric and publish middlewares (Prometheus, OpenTelemetry or your own `publish_scope`) behave as in production. diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index fe4d3c8..3da9685 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -25,7 +25,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from faststream_outbox._scheduling import resolve_next_attempt_client_side -from faststream_outbox._session import SessionOrConnection +from faststream_outbox._session import SessionOrConnection, require_session_or_connection from faststream_outbox._time import utcnow from faststream_outbox.broker import OutboxBroker from faststream_outbox.client import _NO_DLQ_TABLE_MSG, AbstractOutboxClient @@ -344,7 +344,7 @@ async def mark_pending_with_lease( async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool: """Mirror :meth:`OutboxBroker.cancel_timer` — drop a not-yet-leased timer row.""" - del session # ignored (no real DB), matching the fake publish contract + require_session_or_connection(session, "FakeOutboxClient.cancel_timer") for i, row in enumerate(self._rows): if row.queue == queue and row.timer_id == timer_id and row.acquired_token is None: del self._rows[i] @@ -359,7 +359,7 @@ async def fetch_unprocessed( limit: int = 1000, ) -> list[OutboxInnerMessage]: """Mirror :meth:`OutboxBroker.fetch_unprocessed` — read rows from the in-memory store.""" - del session # ignored (no real DB), matching the fake publish contract + require_session_or_connection(session, "FakeOutboxClient.fetch_unprocessed") if limit < 1: # Mirror the real client's validation so a non-positive limit can't # silently mis-slice (rows[:-1]) under the test broker (F4-04). diff --git a/tests/test_client_contract.py b/tests/test_client_contract.py index c7b8c62..ac93ab5 100644 --- a/tests/test_client_contract.py +++ b/tests/test_client_contract.py @@ -9,9 +9,9 @@ Scope is exactly the shared ``AbstractOutboxClient`` surface: ``insert`` / ``insert_batch`` (+ ``timer_id`` dedup and headers JSONB rejects), ``fetch``, -``delete_with_lease``, ``mark_pending_with_lease`` (+ the DLQ side-effect). -``cancel_timer`` lives on the broker, so it is covered by ``test_integration.py`` / -``test_fake.py``, not here. +``delete_with_lease``, ``mark_pending_with_lease`` (+ the DLQ side-effect), and the +session type check on ``cancel_timer`` / ``fetch_unprocessed``. Their row behaviour +is covered through the broker in ``test_integration.py`` / ``test_fake.py``. A per-adapter harness hides the substrate differences (how a row is seeded, which connection a terminal write needs) so the scenarios read adapter-agnostically. @@ -169,6 +169,9 @@ async def get_row(self, message_id: int) -> dict[str, object] | None: async def dlq_rows(self) -> list[dict[str, object]]: return [dict(r) for r in self._client.dlq_rows] + async def call(self, method: str, **kwargs: object) -> object: + return await getattr(self._client, method)(**kwargs) + class _RealHarness: """Adapter-agnostic harness backed by a real ``OutboxClient`` against Postgres.""" @@ -307,6 +310,9 @@ async def dlq_rows(self) -> list[dict[str, object]]: row["payload"] = bytes(row["payload"]) return rows + async def call(self, method: str, **kwargs: object) -> object: + return await getattr(self._client, method)(**kwargs) + _Harness = _FakeHarness | _RealHarness @@ -351,6 +357,30 @@ async def contract_without_dlq(request: pytest.FixtureRequest) -> AsyncIterator[ yield harness +# --- session type check ---------------------------------------------------- + + +@pytest.mark.parametrize( + ("method", "kwargs"), + [ + ("cancel_timer", {"queue": "q", "timer_id": "t"}), + ("fetch_unprocessed", {}), + ], +) +@pytest.mark.parametrize("session", [None, object()]) +async def test_session_methods_reject_a_non_session( + contract: _Harness, + method: str, + kwargs: dict[str, object], + session: object, +) -> None: + with pytest.raises( + TypeError, + match=rf"OutboxClient\.{method} requires an sqlalchemy\.ext\.asyncio\.AsyncSession or AsyncConnection", + ): + await contract.call(method, session=session, **kwargs) + + # --- insert ---------------------------------------------------------------- diff --git a/tests/test_fake.py b/tests/test_fake.py index df72ba6..7f33d39 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -432,7 +432,7 @@ async def test_fake_broker_cancel_timer_removes_row() -> None: async with test_broker: await broker.publish("x", queue="timers", timer_id="email-1", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 - cancelled = await broker.cancel_timer(queue="timers", timer_id="email-1", session=object()) # ty: ignore[invalid-argument-type] + cancelled = await broker.cancel_timer(queue="timers", timer_id="email-1", session=fake_session()) assert cancelled is True assert test_broker.fake_client.rows == [] @@ -446,10 +446,10 @@ async def test_fake_broker_fetch_unprocessed_reads_fake_client() -> None: await broker.publish("a", queue="q1", session=fake_session()) await broker.publish("b", queue="q2", session=fake_session()) - all_rows = await broker.fetch_unprocessed(session=object()) # ty: ignore[invalid-argument-type] + all_rows = await broker.fetch_unprocessed(session=fake_session()) assert [r.queue for r in all_rows] == ["q1", "q2"] - q1_only = await broker.fetch_unprocessed(queue="q1", session=object()) # ty: ignore[invalid-argument-type] + q1_only = await broker.fetch_unprocessed(queue="q1", session=fake_session()) assert [r.queue for r in q1_only] == ["q1"] @@ -460,9 +460,9 @@ async def test_fake_broker_fetch_unprocessed_respects_limit() -> None: async with test_broker: for i in range(5): await broker.publish(str(i), queue="q1", session=fake_session()) - limited = await broker.fetch_unprocessed(limit=2, session=object()) # ty: ignore[invalid-argument-type] + limited = await broker.fetch_unprocessed(limit=2, session=fake_session()) assert len(limited) == 2 - all_rows = await broker.fetch_unprocessed(session=object()) # ty: ignore[invalid-argument-type] + all_rows = await broker.fetch_unprocessed(session=fake_session()) assert len(all_rows) == 5 @@ -473,7 +473,7 @@ async def test_fake_broker_fetch_unprocessed_rejects_non_positive_limit() -> Non async with test_broker: for bad in (0, -1): with pytest.raises(ValueError, match="limit"): - await broker.fetch_unprocessed(limit=bad, session=object()) # ty: ignore[invalid-argument-type] + await broker.fetch_unprocessed(limit=bad, session=fake_session()) async def test_fake_broker_router_subscriber_receives_publish() -> None: @@ -1289,7 +1289,7 @@ async def test_fake_client_feed_timer_id_different_queues_allowed() -> None: async def test_fake_client_cancel_timer_removes_unleased_row() -> None: fake = FakeOutboxClient() fake.feed(queue="q", payload=b"x", timer_id="email-1") - assert await fake.cancel_timer(queue="q", timer_id="email-1", session=object()) is True # ty: ignore[invalid-argument-type] + assert await fake.cancel_timer(queue="q", timer_id="email-1", session=fake_session()) is True assert fake.rows == [] @@ -1396,7 +1396,7 @@ async def test_fake_headers_not_shared_by_reference() -> None: async def test_fake_client_cancel_timer_unknown_returns_false() -> None: fake = FakeOutboxClient() - assert await fake.cancel_timer(queue="q", timer_id="never-existed", session=object()) is False # ty: ignore[invalid-argument-type] + assert await fake.cancel_timer(queue="q", timer_id="never-existed", session=fake_session()) is False async def test_fake_client_cancel_timer_skips_leased_row() -> None: @@ -1404,7 +1404,7 @@ async def test_fake_client_cancel_timer_skips_leased_row() -> None: fake.feed(queue="q", payload=b"x", timer_id="email-1") fake.rows[0].acquired_token = uuid.uuid4() fake.rows[0].acquired_at = dt.datetime.now(tz=dt.UTC) - assert await fake.cancel_timer(queue="q", timer_id="email-1", session=object()) is False # ty: ignore[invalid-argument-type] + assert await fake.cancel_timer(queue="q", timer_id="email-1", session=fake_session()) is False assert len(fake.rows) == 1