fix: type-check session in fake cancel_timer and fetch_unprocessed

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.<method>` 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.

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).

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 ----------------------------------------------------------------
 
 

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,15 +1396,15 @@ 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:
     fake = FakeOutboxClient()
     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