diff --git a/AGENTS.md b/AGENTS.md index 83d624a..3382753 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,8 +41,10 @@ will **not** tell you: seams, deliberately. Each fires for events the other physically cannot observe — `fetched` has no `StreamMessage`, `lease_lost` fires after `consume_scope` exits. **Don't collapse them.** -`TestOutboxBroker` swaps in `FakeOutboxClient`. Sync mode is the default; `run_loops=True` runs the -real fetch and worker loops against the fake, which retry, lease-expiry, and scheduling tests need. +`TestOutboxBroker` swaps in `FakeOutboxClient` and nothing else on the publish side: the real producer +and publish middlewares run, and a thin wrapper dispatches what the fake stored once the producer +returns. Sync mode is the default; `run_loops=True` runs the real fetch and worker loops against the +fake, which retry, lease-expiry, and scheduling tests need. ## Workflow diff --git a/docs/usage/dlq.md b/docs/usage/dlq.md index 854fc51..07b246f 100644 --- a/docs/usage/dlq.md +++ b/docs/usage/dlq.md @@ -209,6 +209,7 @@ audit dict is appended to `fake_client.dlq_rows` in the same call. from sqlalchemy import MetaData from faststream_outbox import NoRetry, OutboxBroker, TestOutboxBroker, make_dlq_table, make_outbox_table +from faststream_outbox.testing import fake_session metadata = MetaData() @@ -224,7 +225,7 @@ async def handle(body: dict) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] assert len(test_broker.fake_client.dlq_rows) == 1 diff --git a/docs/usage/observability.md b/docs/usage/observability.md index 99e3472..c09b179 100644 --- a/docs/usage/observability.md +++ b/docs/usage/observability.md @@ -152,15 +152,9 @@ dlq_written](./dlq.md#metric-dlq_written). ## Test broker note -`TestOutboxBroker` replaces `broker.publish` with a patched version that -skips the middleware publish path, so middleware-registered -publish-scope metrics do not fire in test mode. Middleware -consume-scope metrics still fire, because handlers are still consumed -through the normal middleware stack. - -The recorder-seam `published` event still fires in test mode, with the -same tags as in production: `status="error"` and `exception_type` when -encoding fails, and a measured `duration_seconds` for the in-memory -insert. - -`TestKafkaBroker` and `TestRabbitBroker` behave the same way, for the same reason. +`TestOutboxBroker` publishes through the production producer and +middleware stack, so publish-scope middleware metrics and the recorder +`published` event fire in test mode with the same tags as in production. +Consume-scope metrics fire too. In sync mode the handler runs inside +`publish`, so a publish-scope duration includes the handler's run time; +the recorder's `duration_seconds` covers only the in-memory insert. diff --git a/docs/usage/testing.md b/docs/usage/testing.md index 55d68d8..e9a55de 100644 --- a/docs/usage/testing.md +++ b/docs/usage/testing.md @@ -2,7 +2,8 @@ `faststream-outbox` ships `TestOutboxBroker`, a test context manager that swaps the SQLAlchemy-backed client for an in-memory `FakeOutboxClient` so -unit tests don't need Postgres. +unit tests don't need Postgres. Only the storage is fake: publishing runs +the broker's real producer and publish middlewares, as in production. By default it dispatches handlers synchronously inside `publish`, matching `TestKafkaBroker` / `TestRabbitBroker`, so tests need no `_wait_until` or @@ -16,7 +17,7 @@ bottom), so async test functions need no `@pytest.mark.asyncio` marker. ```python from faststream_outbox import OutboxBroker, make_outbox_table -from faststream_outbox.testing import TestOutboxBroker +from faststream_outbox.testing import TestOutboxBroker, fake_session from sqlalchemy import MetaData @@ -31,16 +32,17 @@ async def test_handler() -> None: received.append(order_id) async with TestOutboxBroker(broker): - await broker.publish(1, queue="orders") + await broker.publish(1, queue="orders", session=fake_session()) # Handler has already run. assert received == [1] ``` -On `broker.publish` (and `publish_batch`), `session=` is optional in tests: -the test broker patches those methods to ignore it. This does not extend -to `broker.publisher("q").publish(...)`, which still requires a `session` -(see [Testing publishers](#testing-publishers) below). +Every publish needs a `session` in tests, as in production. +`fake_session()` returns a mock `AsyncSession` that passes the session check +and reports an open transaction, so an `OutboxResponse` accepts it too. Its +methods do nothing: the fake store keeps the row whether you commit or roll +back. The fake client keeps an in-memory list of rows you can inspect via `fake_client.rows`. `fake_client` is an attribute of the @@ -52,7 +54,7 @@ async with tb: # "orders" has a subscriber, so in sync mode the row is dispatched and # acked (deleted) before publish returns. Publish to a queue with no # subscriber to inspect a row that survives. - await broker.publish(1, queue="audit") + await broker.publish(1, queue="audit", session=fake_session()) assert len(tb.fake_client.rows) == 1 ``` @@ -64,11 +66,6 @@ An unconsumed queue is used deliberately: a row published to a queue that ## Testing publishers ```python -from unittest.mock import AsyncMock - -from sqlalchemy.ext.asyncio import AsyncSession - - async def test_publisher() -> None: metadata = MetaData() outbox_table = make_outbox_table(metadata, table_name="outbox") @@ -82,21 +79,13 @@ async def test_publisher() -> None: pub = broker.publisher("orders") async with TestOutboxBroker(broker): - # The publisher path builds its publish command (which validates the - # session type) before the fake producer is reached, so session= is - # required and must be an AsyncSession. The fake never touches it — a - # spec'd mock satisfies the check. - await pub.publish({"order_id": 1}, session=AsyncMock(spec=AsyncSession)) + await pub.publish({"order_id": 1}, session=fake_session()) assert received == [{"order_id": 1}] ``` -`broker.publisher("q").publish(...)` lands rows in the same fake store as -`broker.publish(queue="q", ...)`, because the test broker swaps in a fake -producer that both paths go through. The one -difference in tests is the `session`: `broker.publish` is patched to make it -optional, but the publisher path is not, so pass a (mock) `AsyncSession` as -shown. +`broker.publisher("q").publish(...)` and `broker.publish(queue="q", ...)` +go through the same producer and land rows in the same fake store. ## Loop-driven mode @@ -146,8 +135,12 @@ registered handlers are not started, matching production. - `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 fake producer uses the same envelope format as the real one, so - all serialization paths are exercised. +- Publishing runs the production producer, so encoding, the `published` + metric and publish middlewares (Prometheus, OpenTelemetry or your own + `publish_scope`) behave as in production. +- The fake store ignores transactions. A row published in a transaction + that rolls back is still stored and delivered. Test rollback behaviour + against a real database. - `lease_ttl_seconds` and re-delivery are not simulated in sync mode. Handlers that exceed the configured TTL in production may be re-delivered to another worker, but tests will only invoke the handler once. @@ -166,7 +159,7 @@ registered handlers are not started, matching production. fake subscribers that `TestKafkaBroker` and friends attach to each publisher). Those spies mock the registered handler to forward `publisher.publish()` calls, which would conflict with the outbox's real -dispatch path: the fake producer already lands rows in the fake client +dispatch path: the test broker already lands rows in the fake client *and* drives the real handler. Publisher-mock assertions such as `publisher.mock.assert_called_once_with(...)` are therefore not available; assert on handler side effects or `fake_client.rows` instead. diff --git a/faststream_outbox/_scheduling.py b/faststream_outbox/_scheduling.py index 353a366..0e40124 100644 --- a/faststream_outbox/_scheduling.py +++ b/faststream_outbox/_scheduling.py @@ -1,12 +1,12 @@ -"""Pure activate-args resolution + validation, shared by the real and fake publish paths. +"""Pure activate-args resolution, shared by the real and fake insert paths. ``activate_in`` / ``activate_at`` are the user-facing scheduling knobs. Turning them into a single ``next_attempt_at`` (client clock) and deciding whether a row is future-dated (so NOTIFY can be skipped) are the same rules everywhere, so they live here as pure functions a caller invokes with its own ``now``. -Leaf module: depends only on the stdlib so ``producer``, ``broker`` and ``testing`` -can all import *from* it without a cycle. The single-publish real path computes +Leaf module: depends only on the stdlib so ``client`` and ``testing`` +can both import *from* it without a cycle. The single-publish real path computes ``next_attempt_at`` server-side via ``make_interval`` (clock-skew-safe) and does not use ``resolve_next_attempt_client_side`` — only the batch and fake paths do. """ @@ -36,17 +36,3 @@ def resolve_next_attempt_client_side( if activate_in is not None: return now + activate_in return activate_at - - -def validate_activate_args( - method_name: str, - activate_in: dt.timedelta | None, - activate_at: dt.datetime | None, -) -> None: - """Mutex + tz-aware checks shared by the test fakes. Real broker delegates to ``OutboxPublishCommand``.""" - if activate_in is not None and activate_at is not None: - msg = f"{method_name} accepts at most one of activate_in / activate_at" - raise ValueError(msg) - if activate_at is not None and activate_at.tzinfo is None: - msg = f"{method_name} requires activate_at to be timezone-aware" - raise ValueError(msg) diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index df4d6e8..339f392 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -1,7 +1,9 @@ """Test broker with an in-memory ``OutboxClient`` substitute. ``TestOutboxBroker`` wraps an ``OutboxBroker`` and swaps in a ``FakeOutboxClient`` -backed by a list of ``_FakeRow`` records. Defaults to **sync dispatch**: ``await broker.publish(...)`` +backed by a list of ``_FakeRow`` records. Publishing runs the broker's real producer +and publish middlewares; only the storage is fake, so ``broker.publish`` needs a +session here too (``fake_session()`` returns a mock one). Defaults to **sync dispatch**: ``await broker.publish(...)`` finds the matching subscriber and awaits its consume pipeline before returning, the same model as ``TestKafkaBroker`` / ``TestRabbitBroker``. Timers fire immediately in sync mode — ``activate_in`` / ``activate_at`` are recorded on the fake row but not @@ -13,7 +15,6 @@ import datetime as dt import json -import time import typing import uuid from contextlib import contextmanager @@ -21,14 +22,13 @@ from unittest import mock from faststream._internal.testing.broker import TestBroker, patch_broker_calls +from sqlalchemy.ext.asyncio import AsyncSession -from faststream_outbox._scheduling import resolve_next_attempt_client_side, validate_activate_args +from faststream_outbox._scheduling import resolve_next_attempt_client_side from faststream_outbox._time import utcnow from faststream_outbox.broker import OutboxBroker from faststream_outbox.client import _NO_DLQ_TABLE_MSG, AbstractOutboxClient -from faststream_outbox.envelope import _encode_payload from faststream_outbox.message import OutboxInnerMessage -from faststream_outbox.metrics import _safe_emit from faststream_outbox.response import _MAX_QUEUE_LENGTH, _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand from faststream_outbox.schema import _DLQ_INJECTED_COLUMNS, _DLQ_PROJECTION @@ -36,7 +36,7 @@ if typing.TYPE_CHECKING: from collections.abc import Generator, Sequence - from sqlalchemy.ext.asyncio import AsyncSession + from faststream._internal.producer import ProducerProto from faststream_outbox.subscriber.usecase import OutboxSubscriber @@ -114,6 +114,7 @@ def __init__(self, *, dlq_enabled: bool = True) -> None: # the real client's CTE side-effect: outbox row gone + DLQ row created in # the same call. Tests assert against ``test_broker.fake_client.dlq_rows``. self._dlq_rows: list[dict[str, typing.Any]] = [] + self._dispatchable: list[_FakeRow] = [] def feed( self, @@ -156,7 +157,10 @@ async def insert( next_attempt_at=resolve_next_attempt_client_side(activate_in, activate_at, utcnow()), timer_id=timer_id, ) - return None if row is None else row.id + if row is None: + return None + self._dispatchable.append(row) + return row.id async def insert_batch( self, @@ -172,7 +176,14 @@ async def insert_batch( _validate_like_postgres(queue=queue, timer_id=None, headers=headers) next_at = resolve_next_attempt_client_side(activate_in, activate_at, utcnow()) for payload, headers in rows: - self._store(queue=queue, payload=payload, headers=headers, next_attempt_at=next_at, timer_id=None) + self._dispatchable.append( + self._append(queue=queue, payload=payload, headers=headers, next_attempt_at=next_at, timer_id=None) + ) + + def _take_dispatchable(self) -> list[_FakeRow]: + """Return and forget the rows ``insert`` / ``insert_batch`` stored since the last call.""" + rows, self._dispatchable = self._dispatchable, [] + return rows def _store( self, @@ -187,6 +198,19 @@ def _store( # re-feeding a timer that already exists is a no-op. if timer_id is not None and any(r.queue == queue and r.timer_id == timer_id for r in self._rows): return None + return self._append( + queue=queue, payload=payload, headers=headers, next_attempt_at=next_attempt_at, timer_id=timer_id + ) + + def _append( + self, + *, + queue: str, + payload: bytes, + headers: dict[str, str] | None, + next_attempt_at: dt.datetime | None, + timer_id: str | None, + ) -> _FakeRow: row = _FakeRow( id=self._next_id, queue=queue, @@ -413,28 +437,6 @@ async def _sync_dispatch(fake_client: FakeOutboxClient, broker: OutboxBroker, qu await subscriber.dispatch_one(_to_inner(fake_row)) -def _emit_published( - broker: OutboxBroker, - queue: str, - *, - count: int, - size_bytes: int, - start_perf: float, - exc: Exception | None = None, -) -> None: - """Emit the producer-side ``published`` metric with the same tags as ``OutboxProducer``.""" - tags: dict[str, typing.Any] = { - "queue": queue, - "status": "success" if exc is None else "error", - "count": count, - "size_bytes": size_bytes, - "duration_seconds": time.perf_counter() - start_perf, - } - if exc is not None: - tags["exception_type"] = type(exc).__name__ - _safe_emit(broker.config.broker_config.metrics_recorder, "published", tags) - - def _notify_subscriber(broker: OutboxBroker, queue: str) -> None: """P30: wake the matching subscriber's fetch loop immediately, mirroring production NOTIFY. @@ -446,221 +448,39 @@ def _notify_subscriber(broker: OutboxBroker, queue: str) -> None: subscriber._notify_event.set() # noqa: SLF001 -async def _fake_publish_one( - fake_client: FakeOutboxClient, - broker: OutboxBroker, - serializer: typing.Any, - *, - body: typing.Any, - queue: str, - headers: dict[str, str] | None, - correlation_id: str | None, - next_at: "dt.datetime | None", - timer_id: str | None, - run_loops: bool, -) -> int | None: - """Shared single-row insert for ``FakeOutboxProducer.publish`` and the ``broker.publish`` patch (P29).""" - start_perf = time.perf_counter() - size_bytes = 0 - try: - payload, hdrs = _encode_payload(body, headers=headers, correlation_id=correlation_id, serializer=serializer) - size_bytes = len(payload) - row_id = fake_client.feed( - queue=queue, payload=payload, headers=hdrs, next_attempt_at=next_at, timer_id=timer_id - ) - except Exception as exc: - _emit_published(broker, queue, count=0, size_bytes=size_bytes, start_perf=start_perf, exc=exc) - raise - _emit_published(broker, queue, count=0 if row_id is None else 1, size_bytes=size_bytes, start_perf=start_perf) - if run_loops: - _notify_subscriber(broker, queue) # P30: the loops own dispatch; just wake them - elif row_id is not None: - # Sync dispatch ignores next_attempt_at (timers fire immediately in test mode); - # skip only when the insert was a timer-dedup no-op. - await _sync_dispatch(fake_client, broker, queue, row_id) - return row_id - - -async def _fake_publish_many( - fake_client: FakeOutboxClient, - broker: OutboxBroker, - serializer: typing.Any, - *, - bodies: "Sequence[typing.Any]", - queue: str, - headers: dict[str, str] | None, - next_at: "dt.datetime | None", - run_loops: bool, -) -> None: - """Shared batch insert for both batch paths (P29). - - S5: insert the WHOLE batch, emit ``published``, then dispatch — mirroring production - (atomic batch INSERT -> published -> subscriber fetch), so a handler never observes a - half-inserted batch and the event order isn't inverted. - """ - start_perf = time.perf_counter() - total_size = 0 - encoded: list[tuple[bytes, dict[str, str] | None]] = [] - try: - for body in bodies: - payload, hdrs = _encode_payload(body, headers=headers, serializer=serializer) - total_size += len(payload) - _validate_like_postgres(queue=queue, timer_id=None, headers=hdrs) - encoded.append((payload, hdrs)) - except Exception as exc: - _emit_published(broker, queue, count=0, size_bytes=total_size, start_perf=start_perf, exc=exc) - raise - landed_ids: list[int] = [] - for payload, hdrs in encoded: - row_id = fake_client.feed(queue=queue, payload=payload, headers=hdrs, next_attempt_at=next_at) - if row_id is not None: - landed_ids.append(row_id) - _emit_published(broker, queue, count=len(landed_ids), size_bytes=total_size, start_perf=start_perf) - if run_loops: - _notify_subscriber(broker, queue) - else: - for row_id in landed_ids: - await _sync_dispatch(fake_client, broker, queue, row_id) - - -class FakeOutboxProducer: - """In-memory ``OutboxProducer`` substitute routing inserts through ``FakeOutboxClient``. - - Used by ``TestOutboxBroker`` so ``broker.publisher(queue).publish(body, session=...)`` - drives the same in-memory fake store as ``broker.publish(body, session=...)``. The - *session* on the command is ignored — the fake client has no transaction. - - In sync mode (``run_loops=False``), each successful insert short-circuits into - ``_sync_dispatch`` so handlers run before ``publish`` returns — matches the - ``broker.publish`` patch in ``_build_fake_publish``. - """ - - _parser: typing.Any = None - _decoder: typing.Any = None - # ProducerProto[0.7] requires `codec`. The fake producer ignores it at - # runtime, same as OutboxProducer. - codec: typing.Any = None +class _DispatchingProducer: + """Runs the broker's real producer, then hands the rows it stored to the test broker's dispatch.""" - def __init__( - self, - fake_client: FakeOutboxClient, - broker: OutboxBroker, - serializer: typing.Any, - *, - run_loops: bool, - ) -> None: - self._fake_client = fake_client - self._broker = broker - self._serializer = serializer - self._run_loops = run_loops + def __init__(self, producer: "ProducerProto[OutboxPublishCommand]", test_broker: "TestOutboxBroker") -> None: + self._producer = producer + self._test_broker = test_broker + self._parser = producer._parser # noqa: SLF001 # ProducerProto declares it private + self._decoder = producer._decoder # noqa: SLF001 # ProducerProto declares it private + self.codec = producer.codec async def publish(self, cmd: OutboxPublishCommand) -> int | None: - validate_activate_args("broker.publish", cmd.activate_in, cmd.activate_at) - next_at = resolve_next_attempt_client_side(cmd.activate_in, cmd.activate_at, utcnow()) - return await _fake_publish_one( - self._fake_client, - self._broker, - self._serializer, - body=cmd.body, - queue=cmd.queue, - headers=cmd.headers, - correlation_id=cmd.correlation_id, - next_at=next_at, - timer_id=cmd.timer_id, - run_loops=self._run_loops, - ) + row_id = await self._producer.publish(cmd) + await self._test_broker._dispatch_stored() # noqa: SLF001 # internal hook between two testing.py classes + return typing.cast("int | None", row_id) async def publish_batch(self, cmd: OutboxPublishCommand) -> None: - validate_activate_args("broker.publish_batch", cmd.activate_in, cmd.activate_at) - next_at = resolve_next_attempt_client_side(cmd.activate_in, cmd.activate_at, utcnow()) - await _fake_publish_many( - self._fake_client, - self._broker, - self._serializer, - bodies=cmd.batch_bodies, - queue=cmd.queue, - headers=cmd.headers, - next_at=next_at, - run_loops=self._run_loops, - ) + await self._producer.publish_batch(cmd) + await self._test_broker._dispatch_stored() # noqa: SLF001 # internal hook between two testing.py classes async def request(self, cmd: OutboxPublishCommand) -> typing.NoReturn: raise NotImplementedError(_REQUEST_UNSUPPORTED_MSG) -def _build_fake_publish( - fake_client: FakeOutboxClient, - broker: OutboxBroker, - serializer: typing.Any, - *, - run_loops: bool, -) -> typing.Callable[..., typing.Awaitable[int | None]]: - async def fake_publish( - body: typing.Any, - *, - queue: str, - session: typing.Any = None, - headers: dict[str, str] | None = None, - correlation_id: str | None = None, - activate_in: dt.timedelta | None = None, - activate_at: dt.datetime | None = None, - timer_id: str | None = None, - ) -> int | None: - # P33: test mode is intentionally lenient about ``session`` — the fake client has - # no transaction, so any value (including None) is accepted here. This DIVERGES from - # production and from ``publisher.publish()`` / ``OutboxResponse``, which require a - # real AsyncSession; tests that assert that contract must use those paths. - del session - validate_activate_args("broker.publish", activate_in, activate_at) - next_at = resolve_next_attempt_client_side(activate_in, activate_at, utcnow()) - return await _fake_publish_one( - fake_client, - broker, - serializer, - body=body, - queue=queue, - headers=headers, - correlation_id=correlation_id, - next_at=next_at, - timer_id=timer_id, - run_loops=run_loops, - ) - - return fake_publish - - -def _build_fake_publish_batch( - fake_client: FakeOutboxClient, - broker: OutboxBroker, - serializer: typing.Any, - *, - run_loops: bool, -) -> typing.Callable[..., typing.Awaitable[None]]: - async def fake_publish_batch( - *bodies: typing.Any, - queue: str, - session: typing.Any = None, - headers: dict[str, str] | None = None, - activate_in: dt.timedelta | None = None, - activate_at: dt.datetime | None = None, - ) -> None: - del session - validate_activate_args("broker.publish_batch", activate_in, activate_at) - if not bodies: - return - next_at = resolve_next_attempt_client_side(activate_in, activate_at, utcnow()) - await _fake_publish_many( - fake_client, - broker, - serializer, - bodies=bodies, - queue=queue, - headers=headers, - next_at=next_at, - run_loops=run_loops, - ) +def fake_session() -> "AsyncSession": + """Return a mock ``AsyncSession`` for publishing under ``TestOutboxBroker``. - return fake_publish_batch + It passes the session type check and reports an open transaction, so + ``OutboxResponse`` accepts it too. Every method is a no-op: ``commit`` and + ``rollback`` change nothing, and the fake store keeps rows either way. + """ + session = mock.AsyncMock(spec=AsyncSession) + session.in_transaction.return_value = True + return session class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker], broker=OutboxBroker): @@ -680,13 +500,11 @@ class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker], broker=OutboxBrok """ fake_client: FakeOutboxClient - fake_producer: "FakeOutboxProducer | None" run_loops: bool def __init__(self, broker: OutboxBroker, *, run_loops: bool = False, **kwargs: typing.Any) -> None: super().__init__(broker, **kwargs) self.fake_client = FakeOutboxClient(dlq_enabled=broker.config.broker_config.dlq_table is not None) - self.fake_producer = None # set when the producer slot is swapped (see _patch_producer) self.run_loops = run_loops self._outbox_broker = broker # for feed()'s P30 wakeup # Guards against the upstream harness spawning the loops twice (B14); reset @@ -716,47 +534,35 @@ def feed( @contextmanager def _patch_producer(self, broker: OutboxBroker) -> "Generator[None]": - # Swap the broker's producer slot for one that routes inserts through - # the in-memory fake client. ``OutboxPublisher.publish`` flows through - # ``_basic_publish(cmd, producer=...)``, so replacing the producer is - # how publisher.publish() lands rows in the fake store. ``broker.publish`` - # is patched separately to bypass the producer — that path is the - # canonical sync-dispatch entry point and matches existing tests. - serializer = broker.config.broker_config.fd_config._serializer # noqa: SLF001 - fake_producer = FakeOutboxProducer( - self.fake_client, - broker, - serializer, - run_loops=self.run_loops, - ) original_producer = broker.config.broker_config.producer - broker.config.broker_config.producer = fake_producer - # Expose for identity assertions, symmetric with ``self.fake_client``. - self.fake_producer = fake_producer + broker.config.broker_config.producer = _DispatchingProducer(original_producer, self) try: yield finally: broker.config.broker_config.producer = original_producer + self.fake_client._take_dispatchable() # noqa: SLF001 # drop rows no publish dispatched @contextmanager def _patch_broker(self, broker: OutboxBroker) -> "Generator[None]": original_client = broker.config.broker_config.client broker.config.broker_config.client = self.fake_client - # Mirror real publish's serializer wiring so pydantic / dataclass bodies - # encode identically in tests. - serializer = broker.config.broker_config.fd_config._serializer # noqa: SLF001 - fake_publish = _build_fake_publish(self.fake_client, broker, serializer, run_loops=self.run_loops) - fake_publish_batch = _build_fake_publish_batch(self.fake_client, broker, serializer, run_loops=self.run_loops) try: - with ( - mock.patch.object(broker, "publish", new=fake_publish), - mock.patch.object(broker, "publish_batch", new=fake_publish_batch), - super()._patch_broker(broker), - ): + with super()._patch_broker(broker): yield finally: broker.config.broker_config.client = original_client + async def _dispatch_stored(self) -> None: + """Deliver what the last publish stored: run each row's handler (sync mode) or wake the fetch loops.""" + rows = self.fake_client._take_dispatchable() # noqa: SLF001 # internal hook between two testing.py classes + if self.run_loops: + for queue in dict.fromkeys(row.queue for row in rows): + _notify_subscriber(self._outbox_broker, queue) # P30: the loops own dispatch; just wake them + return + for row in rows: + # Sync dispatch ignores next_attempt_at (timers fire immediately in test mode). + await _sync_dispatch(self.fake_client, self._outbox_broker, row.queue, row.id) + def create_publisher_fake_subscriber( # pragma: no cover - TestBroker abstract method; _fake_start skips its caller self, broker: OutboxBroker, @@ -764,11 +570,11 @@ def create_publisher_fake_subscriber( # pragma: no cover - TestBroker abstract ) -> tuple["OutboxSubscriber", bool]: # Required by FastStream's TestBroker abstract base, but never called — # we skip the publisher fake-subscriber loop in ``_fake_start`` because - # ``FakeOutboxProducer`` already lands rows in the fake client AND drives - # the real subscriber via ``_sync_dispatch``. The FastStream + # the real producer already lands rows in the fake client AND ``_dispatch_stored`` + # drives the real subscriber. The FastStream # publisher-spy infrastructure would mock the real handler and break that. del self, broker, publisher - msg = "TestOutboxBroker handles publisher dispatch via FakeOutboxProducer; this is unreachable." + msg = "TestOutboxBroker dispatches publisher rows itself; this is unreachable." raise NotImplementedError(msg) def _fake_start(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing.Any) -> None: diff --git a/tests/test_fake.py b/tests/test_fake.py index e9602fa..d7abde2 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -44,19 +44,10 @@ from faststream_outbox.client import AbstractOutboxClient from faststream_outbox.configs import OutboxBrokerConfig from faststream_outbox.envelope import _encode_payload as encode_payload +from faststream_outbox.response import OutboxPublishCommand from faststream_outbox.router import OutboxRoute from faststream_outbox.subscriber.config import OutboxSubscriberConfig -from faststream_outbox.testing import FakeOutboxClient, FakeOutboxProducer, _FakeRow, _to_inner - - -def _fake_session() -> AsyncMock: - """Build an ``AsyncMock(spec=AsyncSession)`` for tests where the session is ignored. - - ``OutboxPublishCommand`` requires an ``AsyncSession``; the fake producer doesn't - touch it. The mock passes ``isinstance`` and lets publisher tests focus on the - fake-store side effects. - """ - return AsyncMock(spec=AsyncSession) +from faststream_outbox.testing import FakeOutboxClient, _FakeRow, _to_inner, fake_session def _make_broker() -> OutboxBroker: @@ -76,14 +67,21 @@ async def _wait_until(predicate: Callable[[], object], *, timeout: float = 2.0) raise AssertionError(msg) # pragma: no cover - only when the predicate times out, failing the test -def test_fake_outbox_producer_satisfies_producer_proto() -> None: - """FakeOutboxProducer satisfies ProducerProto: codec attribute present, no missing structural members.""" +async def test_test_broker_producer_satisfies_producer_proto() -> None: + """The producer the test broker swaps in satisfies ProducerProto: no missing structural members.""" + broker = _make_broker() + async with TestOutboxBroker(broker): + producer = broker.config.broker_config.producer + missing = get_protocol_members(ProducerProto) - set(dir(producer)) + assert not missing, f"test-broker producer missing ProducerProto attrs: {missing}" + + +async def test_test_broker_producer_refuses_request() -> None: broker = _make_broker() - fc = FakeOutboxClient() - fp = FakeOutboxProducer(fc, broker, serializer=None, run_loops=False) - missing = get_protocol_members(ProducerProto) - set(dir(fp)) - assert not missing, f"FakeOutboxProducer missing ProducerProto attrs: {missing}" - assert fp.codec is None + async with TestOutboxBroker(broker): + producer = broker.config.broker_config.producer + with pytest.raises(NotImplementedError, match="request-reply"): + await producer.request(OutboxPublishCommand(b"x", queue="orders", session=fake_session())) # --- Sync-mode tests (default TestOutboxBroker) --------------------------------------- @@ -106,7 +104,7 @@ async def handle(body: dict) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert received == [{"order_id": 1}] assert test_broker.fake_client.rows == [] # row deleted after ack @@ -132,7 +130,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish_batch("a", "b", "c", queue="orders") # ty: ignore[missing-argument] + await broker.publish_batch("a", "b", "c", queue="orders", session=fake_session()) # The whole 3-row batch was present before the first handler ran (the old per-feed # dispatch would have shown only 1), and ``published`` was emitted before delivery. @@ -152,7 +150,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish_batch("a", "b", "c", queue="orders") # ty: ignore[missing-argument] + await broker.publish_batch("a", "b", "c", queue="orders", session=fake_session()) assert received == ["a", "b", "c"] assert test_broker.fake_client.rows == [] @@ -168,8 +166,8 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("o-1", queue="orders") # ty: ignore[missing-argument] - await broker.publish("s-1", queue="shipments") # ty: ignore[missing-argument] + await broker.publish("o-1", queue="orders", session=fake_session()) + await broker.publish("s-1", queue="shipments", session=fake_session()) assert seen == ["o-1", "s-1"] @@ -183,7 +181,7 @@ async def handle(body: str) -> None: ... test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("not-mine", queue="other-queue") # ty: ignore[missing-argument] + await broker.publish("not-mine", queue="other-queue", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 assert test_broker.fake_client.rows[0].queue == "other-queue" @@ -200,7 +198,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] @@ -217,7 +215,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 row = test_broker.fake_client.rows[0] @@ -238,7 +236,7 @@ async def handle(body: dict, correlation_id: str = Context("message.correlation_ test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders", correlation_id="trace-xyz") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", correlation_id="trace-xyz", session=fake_session()) assert seen == ["trace-xyz"] @@ -248,7 +246,7 @@ async def test_fake_broker_no_handler_no_dispatch() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: # No subscribers registered — publishing should leave the row in the fake client. - await broker.publish("nope", queue="orders") # ty: ignore[missing-argument] + await broker.publish("nope", queue="orders", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 @@ -265,8 +263,8 @@ async def test_fake_broker_publish_with_timer_id_dedups() -> None: # No subscriber for "timers": rows persist so we can observe dedup behavior. test_broker = TestOutboxBroker(broker) async with test_broker: - first = await broker.publish("x", queue="timers", timer_id="email-1") # ty: ignore[missing-argument] - second = await broker.publish("y", queue="timers", timer_id="email-1") # ty: ignore[missing-argument] + first = await broker.publish("x", queue="timers", timer_id="email-1", session=fake_session()) + second = await broker.publish("y", queue="timers", timer_id="email-1", session=fake_session()) assert first is not None assert second is None assert len(test_broker.fake_client.rows) == 1 @@ -283,10 +281,11 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish( # ty: ignore[missing-argument] + await broker.publish( "delayed", queue="orders", activate_in=dt.timedelta(seconds=5), + session=fake_session(), ) assert received == ["delayed"] @@ -304,7 +303,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: future = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=5) - await broker.publish("at-future", queue="orders", activate_at=future) # ty: ignore[missing-argument] + await broker.publish("at-future", queue="orders", activate_at=future, session=fake_session()) assert received == ["at-future"] assert test_broker.fake_client.rows == [] @@ -320,11 +319,12 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish_batch( # ty: ignore[missing-argument] + await broker.publish_batch( "a", "b", queue="orders", activate_in=dt.timedelta(seconds=5), + session=fake_session(), ) assert received == ["a", "b"] @@ -342,7 +342,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: future = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=5) - await broker.publish_batch("a", "b", queue="orders", activate_at=future) # ty: ignore[missing-argument] + await broker.publish_batch("a", "b", queue="orders", activate_at=future, session=fake_session()) assert received == ["a", "b"] assert test_broker.fake_client.rows == [] @@ -353,11 +353,12 @@ async def test_fake_broker_publish_rejects_both_activate_in_and_activate_at() -> test_broker = TestOutboxBroker(broker) async with test_broker: with pytest.raises(ValueError, match="at most one of activate_in / activate_at"): - await broker.publish( # ty: ignore[missing-argument] + await broker.publish( "x", queue="orders", activate_in=dt.timedelta(seconds=1), activate_at=dt.datetime.now(tz=dt.UTC), + session=fake_session(), ) @@ -366,11 +367,12 @@ async def test_fake_broker_publish_batch_rejects_both_activate_in_and_activate_a test_broker = TestOutboxBroker(broker) async with test_broker: with pytest.raises(ValueError, match="at most one of activate_in / activate_at"): - await broker.publish_batch( # ty: ignore[missing-argument] + await broker.publish_batch( "x", queue="orders", activate_in=dt.timedelta(seconds=1), activate_at=dt.datetime.now(tz=dt.UTC), + session=fake_session(), ) @@ -379,11 +381,12 @@ async def test_fake_broker_publish_rejects_naive_activate_at() -> None: broker = _make_broker() test_broker = TestOutboxBroker(broker) async with test_broker: - with pytest.raises(ValueError, match=r"broker\.publish requires activate_at to be timezone-aware"): - await broker.publish( # ty: ignore[missing-argument] + with pytest.raises(ValueError, match="requires activate_at to be timezone-aware"): + await broker.publish( "x", queue="orders", activate_at=dt.datetime.now(), # noqa: DTZ005 + session=fake_session(), ) @@ -391,11 +394,12 @@ async def test_fake_broker_publish_batch_rejects_naive_activate_at() -> None: broker = _make_broker() test_broker = TestOutboxBroker(broker) async with test_broker: - with pytest.raises(ValueError, match=r"broker\.publish_batch requires activate_at to be timezone-aware"): - await broker.publish_batch( # ty: ignore[missing-argument] + with pytest.raises(ValueError, match="requires activate_at to be timezone-aware"): + await broker.publish_batch( "x", queue="orders", activate_at=dt.datetime.now(), # noqa: DTZ005 + session=fake_session(), ) @@ -403,7 +407,7 @@ async def test_fake_broker_publish_batch_empty_bodies_is_noop() -> None: broker = _make_broker() test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish_batch(queue="orders") # ty: ignore[missing-argument] + await broker.publish_batch(queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] @@ -412,7 +416,7 @@ async def test_fake_broker_cancel_timer_removes_row() -> None: test_broker = TestOutboxBroker(broker) # No handler for "timers" — row persists after publish so we can cancel it. async with test_broker: - await broker.publish("x", queue="timers", timer_id="email-1") # ty: ignore[missing-argument] + 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] assert cancelled is True @@ -425,8 +429,8 @@ async def test_fake_broker_fetch_unprocessed_reads_fake_client() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: # No subscriber — rows stay in fake client. - await broker.publish("a", queue="q1") # ty: ignore[missing-argument] - await broker.publish("b", queue="q2") # ty: ignore[missing-argument] + 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] assert [r.queue for r in all_rows] == ["q1", "q2"] @@ -441,7 +445,7 @@ async def test_fake_broker_fetch_unprocessed_respects_limit() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: for i in range(5): - await broker.publish(str(i), queue="q1") # ty: ignore[missing-argument] + await broker.publish(str(i), queue="q1", session=fake_session()) limited = await broker.fetch_unprocessed(limit=2, session=object()) # ty: ignore[invalid-argument-type] assert len(limited) == 2 all_rows = await broker.fetch_unprocessed(session=object()) # ty: ignore[invalid-argument-type] @@ -471,7 +475,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("via-router", queue="orders") # ty: ignore[missing-argument] + await broker.publish("via-router", queue="orders", session=fake_session()) assert received == ["via-router"] @@ -495,7 +499,7 @@ async def handle(body: str) -> None: # ack_policy through the full dispatch path (not just the stored value). test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("payload", queue="orders") # ty: ignore[missing-argument] + await broker.publish("payload", queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] # handler succeeded → row deleted @@ -521,7 +525,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) test_broker.fake_client = LeaseLostClient() async with test_broker: - await broker.publish("lease-lost", queue="orders") # ty: ignore[missing-argument] + await broker.publish("lease-lost", queue="orders", session=fake_session()) assert received == ["lease-lost"] # P17 + lease-lost: the failed delete emits lease_lost(phase=terminal); the paired @@ -554,7 +558,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) test_broker.fake_client = LeaseLostRetryClient() async with test_broker: - await broker.publish("never-cleared", queue="orders") # ty: ignore[missing-argument] + await broker.publish("never-cleared", queue="orders", session=fake_session()) assert attempts == ["never-cleared"] # The failed retry-update emits lease_lost(phase=retry); the paired ``nacked_retried`` @@ -596,8 +600,8 @@ async def handle(body: str) -> None: raising = RaisingDeleteClient() test_broker.fake_client = raising async with test_broker: - await broker.publish("first-fails-on-delete", queue="orders") # ty: ignore[missing-argument] - await broker.publish("second-ok", queue="orders") # ty: ignore[missing-argument] + await broker.publish("first-fails-on-delete", queue="orders", session=fake_session()) + await broker.publish("second-ok", queue="orders", session=fake_session()) assert raising.calls >= 2 assert received == ["first-fails-on-delete", "second-ok"] @@ -629,7 +633,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("payload", queue="orders") # ty: ignore[missing-argument] + await broker.publish("payload", queue="orders", session=fake_session()) assert len(seen_exceptions) == 1 exc = seen_exceptions[0] @@ -733,7 +737,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker, run_loops=True) async with test_broker: await asyncio.sleep(0.05) - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) await asyncio.wait_for(handled.wait(), timeout=2.0) @@ -752,7 +756,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker, run_loops=True) async with test_broker: await asyncio.sleep(0.05) - await broker.publish_batch("a", "b", queue="orders") # ty: ignore[missing-argument] + await broker.publish_batch("a", "b", queue="orders", session=fake_session()) await asyncio.wait_for(done.wait(), timeout=2.0) @@ -1220,7 +1224,7 @@ async def test_sync_publish_skips_callless_subscriber() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("ignored", queue="orphans") # ty: ignore[missing-argument] + await broker.publish("ignored", queue="orphans", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 @@ -1308,9 +1312,9 @@ def recorder(event: str, tags: Mapping[str, typing.Any]) -> None: unencodable = object() async with TestOutboxBroker(broker): publish = ( - broker.publish_batch("ok", unencodable, queue="orders") # ty: ignore[missing-argument] + broker.publish_batch("ok", unencodable, queue="orders", session=fake_session()) if batch - else broker.publish(unencodable, queue="orders") # ty: ignore[missing-argument] + else broker.publish(unencodable, queue="orders", session=fake_session()) ) with pytest.raises(Exception): # noqa: B017, PT011 # whatever the serializer raises, the metric must record it await publish @@ -1331,8 +1335,8 @@ def recorder(event: str, tags: Mapping[str, typing.Any]) -> None: broker = OutboxBroker(outbox_table=make_outbox_table(MetaData()), metrics_recorder=recorder) async with TestOutboxBroker(broker): - await broker.publish("one", queue="orders") # ty: ignore[missing-argument] - await broker.publish_batch("a", "b", queue="orders") # ty: ignore[missing-argument] + await broker.publish("one", queue="orders", session=fake_session()) + await broker.publish_batch("a", "b", queue="orders", session=fake_session()) assert len(durations) == 2 assert all(d > 0.0 for d in durations) @@ -1432,7 +1436,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) # Handler ran once, then row was deleted — no retry despite default ExponentialRetry. assert attempts == ["x"] @@ -1451,7 +1455,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) # Row rescheduled, not deleted — same as the no-ack_policy default. assert len(test_broker.fake_client.rows) == 1 @@ -1472,7 +1476,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) # The row survived: the default ExponentialRetry rescheduled it instead of # the destructive reject fallback deleting it. @@ -1493,7 +1497,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) # The kwargs no longer TypeError inside the ack middleware -> no reject fallback. assert len(test_broker.fake_client.rows) == 1 @@ -1517,7 +1521,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) # Rejected → row DELETEd. The mutation (no assert_state_set) leaves the row in place # (and emits a false ``acked``), so it would redeliver forever. @@ -1527,8 +1531,8 @@ async def handle(body: str) -> None: # --- Publisher tests -------------------------------------------------------------------- -async def test_publisher_publish_inserts_row_via_fake_producer() -> None: - """``publisher.publish`` routes through the fake producer → fake client store.""" +async def test_publisher_publish_inserts_row_into_fake_store() -> None: + """``publisher.publish`` routes through the real producer into the fake client store.""" broker = _make_broker() received: list[dict] = [] @@ -1539,7 +1543,7 @@ async def handle(body: dict) -> None: pub = broker.publisher("orders") test_broker = TestOutboxBroker(broker) async with test_broker: - result = await pub.publish({"id": 1}, session=_fake_session()) + result = await pub.publish({"id": 1}, session=fake_session()) assert isinstance(result, int) assert received == [{"id": 1}] @@ -1552,7 +1556,7 @@ async def test_publisher_static_headers_merge_with_per_call() -> None: pub = broker.publisher("orders", headers={"source": "default", "trace": "abc"}) test_broker = TestOutboxBroker(broker) async with test_broker: - await pub.publish(b"x", session=_fake_session(), headers={"source": "override"}) + await pub.publish(b"x", session=fake_session(), headers={"source": "override"}) assert len(test_broker.fake_client.rows) == 1 row_headers = test_broker.fake_client.rows[0].headers @@ -1570,8 +1574,8 @@ async def test_publisher_with_timer_id_dedups() -> None: pub = broker.publisher("orders") test_broker = TestOutboxBroker(broker) async with test_broker: - first = await pub.publish(b"hello", session=_fake_session(), timer_id="dedup-key") - second = await pub.publish(b"hello", session=_fake_session(), timer_id="dedup-key") + first = await pub.publish(b"hello", session=fake_session(), timer_id="dedup-key") + second = await pub.publish(b"hello", session=fake_session(), timer_id="dedup-key") assert first is not None assert second is None # dedup'd @@ -1588,7 +1592,7 @@ async def test_publisher_with_activate_in_records_next_attempt() -> None: test_broker = TestOutboxBroker(broker) before = dt.datetime.now(tz=dt.UTC) async with test_broker: - await pub.publish(b"x", session=_fake_session(), activate_in=dt.timedelta(seconds=30)) + await pub.publish(b"x", session=fake_session(), activate_in=dt.timedelta(seconds=30)) assert len(test_broker.fake_client.rows) == 1 assert test_broker.fake_client.rows[0].next_attempt_at > before @@ -1617,7 +1621,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish_batch("a", "b", queue="orders") # ty: ignore[missing-argument] + await broker.publish_batch("a", "b", queue="orders", session=fake_session()) assert received == ["a", "b"] @@ -1654,7 +1658,7 @@ async def handle(body: dict) -> OutboxResponse: return OutboxResponse( body={"echoed": body["x"]}, queue="downstream", - session=_fake_session(), + session=fake_session(), ) @broker.subscriber("downstream") @@ -1663,7 +1667,7 @@ async def handle_next(body: dict) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": "hi"}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": "hi"}, queue="orders", session=fake_session()) assert next_received == [{"echoed": "hi"}] assert test_broker.fake_client.rows == [] @@ -1686,7 +1690,7 @@ async def handle(body: str) -> OutboxResponse: return OutboxResponse( body={"echoed": body}, queue="downstream", - session=_fake_session(), + session=fake_session(), ) @broker.subscriber("downstream") @@ -1695,7 +1699,7 @@ async def handle_next(body: dict) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("hi", queue="orders") # ty: ignore[missing-argument] + await broker.publish("hi", queue="orders", session=fake_session()) assert next_received == [{"echoed": "hi"}] assert test_broker.fake_client.rows == [] @@ -1718,7 +1722,7 @@ async def handle(body: dict) -> OutboxResponse: return OutboxResponse( body={"v": 1}, queue="downstream", - session=_fake_session(), + session=fake_session(), correlation_id="custom-cid", ) @@ -1732,7 +1736,7 @@ async def handle_next( test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders", correlation_id="inbound-cid") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", correlation_id="inbound-cid", session=fake_session()) assert downstream_cors == ["custom-cid"] @@ -1745,7 +1749,7 @@ async def test_handler_returning_outbox_response_inherits_correlation_id() -> No @broker.subscriber("orders") async def handle(body: dict) -> OutboxResponse: del body - return OutboxResponse(body={"v": 1}, queue="downstream", session=_fake_session()) + return OutboxResponse(body={"v": 1}, queue="downstream", session=fake_session()) @broker.subscriber("downstream") async def handle_next( @@ -1757,7 +1761,7 @@ async def handle_next( test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders", correlation_id="trace-xyz") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", correlation_id="trace-xyz", session=fake_session()) assert downstream_cors == ["trace-xyz"] @@ -1772,13 +1776,13 @@ async def handle(body: dict) -> OutboxResponse: return OutboxResponse( body="delayed", queue="downstream", # no subscriber — row persists for inspection - session=_fake_session(), + session=fake_session(), activate_in=dt.timedelta(seconds=60), ) test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", session=fake_session()) downstream = [r for r in test_broker.fake_client.rows if r.queue == "downstream"] assert len(downstream) == 1 @@ -1795,14 +1799,14 @@ async def handle(body: dict) -> OutboxResponse: return OutboxResponse( body="once", queue="downstream", - session=_fake_session(), + session=fake_session(), timer_id="dedupe-key", ) test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] - await broker.publish({"x": 2}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", session=fake_session()) + await broker.publish({"x": 2}, queue="orders", session=fake_session()) downstream = [r for r in test_broker.fake_client.rows if r.queue == "downstream"] assert len(downstream) == 1 @@ -1819,7 +1823,7 @@ async def handle(body: dict) -> dict: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", session=fake_session()) # Inbound row deleted on ack; no extra rows from the dict return. assert test_broker.fake_client.rows == [] @@ -1833,14 +1837,14 @@ async def test_handler_raising_before_returning_outbox_response_does_not_publish async def handle(body: dict) -> OutboxResponse: # Build the response then raise — process_message exits on exception # before iterating the response publishers, so nothing lands downstream. - _unused = OutboxResponse(body="never", queue="downstream", session=_fake_session()) + _unused = OutboxResponse(body="never", queue="downstream", session=fake_session()) del _unused, body msg = "boom" raise RuntimeError(msg) test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", session=fake_session()) # Inbound deleted by NoRetry; downstream untouched. assert test_broker.fake_client.rows == [] @@ -1867,14 +1871,12 @@ async def handle( test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + swapped_producer = broker.config.broker_config.producer + await broker.publish({"x": 1}, queue="orders", session=fake_session()) assert isinstance(captured["msg"], OutboxMessage) assert captured["broker"] is broker - # In test mode the producer slot is swapped to FakeOutboxProducer during the - # ``async with`` block; identity-check would compare against the restored - # post-exit value, so check by type instead. - assert isinstance(captured["producer"], FakeOutboxProducer) + assert captured["producer"] is swapped_producer assert isinstance(captured["client"], AbstractOutboxClient) assert captured["client"] is test_broker.fake_client @@ -1927,7 +1929,7 @@ async def test_fake_dlq_captures_retry_terminal_failure() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("audit-me", queue="orders") # ty: ignore[missing-argument] + await broker.publish("audit-me", queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] assert len(test_broker.fake_client.dlq_rows) == 1 @@ -1952,7 +1954,7 @@ async def test_fake_dlq_redacts_last_exception_with_renderer() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("audit-me", queue="orders") # ty: ignore[missing-argument] + await broker.publish("audit-me", queue="orders", session=fake_session()) row = test_broker.fake_client.dlq_rows[0] assert row["last_exception"] == "RuntimeError" # redacted: no message, no payload @@ -1969,7 +1971,7 @@ async def test_fake_dlq_drops_last_exception_when_renderer_returns_none() -> Non test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("audit-me", queue="orders") # ty: ignore[missing-argument] + await broker.publish("audit-me", queue="orders", session=fake_session()) assert test_broker.fake_client.dlq_rows[0]["last_exception"] is None @@ -1981,7 +1983,7 @@ async def test_fake_dlq_captures_rejected_failure() -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("poison", queue="orders") # ty: ignore[missing-argument] + await broker.publish("poison", queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] assert len(test_broker.fake_client.dlq_rows) == 1 @@ -2023,7 +2025,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] assert test_broker.fake_client.dlq_rows == [] @@ -2040,7 +2042,7 @@ def recorder(event: str, tags: typing.Any) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("x", queue="orders") # ty: ignore[missing-argument] + await broker.publish("x", queue="orders", session=fake_session()) dlq_events = [t for e, t in events if e == "dlq_written"] assert len(dlq_events) == 1 @@ -2059,7 +2061,7 @@ def recorder(event: str, tags: typing.Any) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish("happy", queue="orders") # ty: ignore[missing-argument] + await broker.publish("happy", queue="orders", session=fake_session()) assert not any(e == "dlq_written" for e, _ in events) assert test_broker.fake_client.dlq_rows == [] @@ -2119,7 +2121,7 @@ async def handle(body: dict) -> None: ... test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 assert test_broker.fake_client.rows[0].deliveries_count == 1 @@ -2135,7 +2137,7 @@ async def handle(body: dict) -> None: ... test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert len(test_broker.fake_client.rows) == 1 @@ -2161,7 +2163,7 @@ async def handle(body: dict) -> None: ... test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert test_broker.fake_client.rows == [] @@ -2178,7 +2180,7 @@ async def handle(body: dict) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert not sub.running assert len(test_broker.fake_client.rows) == 1 @@ -2198,7 +2200,7 @@ async def handle(body: dict) -> None: test_broker = TestOutboxBroker(broker) with patch.object(app, "exit") as app_exit: async with test_broker: - await broker.publish({"order_id": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"order_id": 1}, queue="orders", session=fake_session()) assert not sub.running app_exit.assert_called_once_with() @@ -2362,7 +2364,7 @@ async def handle(body: str) -> None: raise StopConsume async with TestOutboxBroker(broker): - await broker.publish("stop", queue="orders") # ty: ignore[missing-argument] + await broker.publish("stop", queue="orders", session=fake_session()) handled.clear() test_broker = TestOutboxBroker(broker, run_loops=True) @@ -2439,7 +2441,7 @@ async def test_handler_returning_outbox_response_takes_correlation_id_from_heade async def handle(body: dict) -> OutboxResponse: del body return OutboxResponse( - body={"v": 1}, queue="downstream", session=_fake_session(), headers={"correlation_id": "from-header"} + body={"v": 1}, queue="downstream", session=fake_session(), headers={"correlation_id": "from-header"} ) @broker.subscriber("downstream") @@ -2448,14 +2450,14 @@ async def handle_next(body: dict, correlation_id: str = _Context("message.correl downstream_cors.append(correlation_id) async with TestOutboxBroker(broker): - await broker.publish({"x": 1}, queue="orders", correlation_id="inbound") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", correlation_id="inbound", session=fake_session()) assert downstream_cors == ["from-header"] async def test_outbox_response_on_a_session_without_a_transaction_retries_the_inbound_row() -> None: broker = _make_broker() - closed_session = _fake_session() + closed_session = fake_session() closed_session.in_transaction = MagicMock(return_value=False) @broker.subscriber("orders") @@ -2465,7 +2467,7 @@ async def handle(body: dict) -> OutboxResponse: test_broker = TestOutboxBroker(broker) async with test_broker: with pytest.raises(RuntimeError, match="transaction"): - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", session=fake_session()) assert [(r.queue, r.attempts_count, r.acquired_token) for r in test_broker.fake_client.rows] == [ ("orders", 1, None) @@ -2474,7 +2476,7 @@ async def handle(body: dict) -> OutboxResponse: async def test_config_error_under_manual_ack_retries_the_row() -> None: broker = _make_broker() - closed_session = _fake_session() + closed_session = fake_session() closed_session.in_transaction = MagicMock(return_value=False) @broker.subscriber("orders", ack_policy=AckPolicy.MANUAL) @@ -2484,8 +2486,126 @@ async def handle(body: dict) -> OutboxResponse: test_broker = TestOutboxBroker(broker) async with test_broker: with pytest.raises(RuntimeError, match="transaction"): - await broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await broker.publish({"x": 1}, queue="orders", session=fake_session()) assert [(r.queue, r.attempts_count, r.acquired_token) for r in test_broker.fake_client.rows] == [ ("orders", 1, None) ] + + +_MIDDLEWARE_PUBLISHES: list[str] = [] + + +class _PublishSpyMiddleware(BaseMiddleware): + async def publish_scope( + self, + call_next: Callable[[typing.Any], Awaitable[typing.Any]], + cmd: typing.Any, + ) -> typing.Any: + _MIDDLEWARE_PUBLISHES.append(cmd.destination) + return await call_next(cmd) + + +@pytest.mark.parametrize("path", ["broker", "publisher", "response"]) +async def test_publish_middleware_runs_under_test_broker(path: str) -> None: + """#219: every publish path runs the broker's publish middlewares, as in production.""" + _MIDDLEWARE_PUBLISHES.clear() + broker = OutboxBroker(outbox_table=make_outbox_table(MetaData()), middlewares=[_PublishSpyMiddleware]) + delivered: list[str] = [] + + @broker.subscriber("orders") + async def handle(body: str) -> OutboxResponse | None: + delivered.append(body) + if path == "response": + return OutboxResponse(body="next", queue="downstream", session=fake_session()) + return None + + @broker.subscriber("downstream") + async def downstream(body: str) -> None: + delivered.append(body) + + async with TestOutboxBroker(broker): + if path == "publisher": + await broker.publisher("orders").publish("first", session=fake_session()) + else: + await broker.publish("first", queue="orders", session=fake_session()) + + assert delivered == (["first", "next"] if path == "response" else ["first"]) + assert (["orders", "downstream"] if path == "response" else ["orders"]) == _MIDDLEWARE_PUBLISHES + + +@pytest.mark.parametrize("batch", [False, True]) +async def test_publish_without_session_raises_under_test_broker(batch: bool) -> None: + broker = _make_broker() + + @broker.subscriber("orders") + async def handle(body: str) -> None: ... + + publish = broker.publish_batch if batch else broker.publish + test_broker = TestOutboxBroker(broker) + async with test_broker: + with pytest.raises(TypeError, match="AsyncSession"): + await publish("a", queue="orders", session=None) # ty: ignore[invalid-argument-type] + assert test_broker.fake_client.rows == [] + + +async def test_sync_publish_emits_published_before_the_handler_runs() -> None: + events: list[tuple[str, dict[str, typing.Any]]] = [] + + def _recorder(event: str, tags: Mapping[str, typing.Any]) -> None: + events.append((event, dict(tags))) + + broker = OutboxBroker(outbox_table=make_outbox_table(MetaData()), metrics_recorder=_recorder) + + @broker.subscriber("orders") + async def handle(body: str) -> None: + del body + events.append(("handler", {})) + + async with TestOutboxBroker(broker): + await broker.publish("a", queue="orders", session=fake_session()) + + names = [name for name, _ in events] + assert names.index("published") < names.index("handler") + assert events[names.index("published")][1]["status"] == "success" + + +async def test_config_error_raised_by_dispatch_is_not_a_failed_publish() -> None: + """A guard re-raised by sync dispatch surfaces from publish, but the row was stored, so ``published`` succeeded.""" + events: list[tuple[str, dict[str, typing.Any]]] = [] + + def _recorder(event: str, tags: Mapping[str, typing.Any]) -> None: + events.append((event, dict(tags))) + + broker = OutboxBroker(outbox_table=make_outbox_table(MetaData()), metrics_recorder=_recorder) + closed = AsyncMock(spec=AsyncSession) + closed.in_transaction.return_value = False + + @broker.subscriber("orders") + async def handle(body: str) -> OutboxResponse: + del body + return OutboxResponse(body="next", queue="downstream", session=closed) + + async with TestOutboxBroker(broker): + with pytest.raises(RuntimeError, match="no open transaction"): + await broker.publish("a", queue="orders", session=fake_session()) + + published = [tags for name, tags in events if name == "published"] + assert [(tags["queue"], tags["status"]) for tags in published] == [("orders", "success")] + + +async def test_rows_inserted_without_a_publish_are_not_dispatched_after_reentry() -> None: + broker = _make_broker() + received: list[bytes] = [] + + @broker.subscriber("orders") + async def handle(body: bytes) -> None: + received.append(body) + + test_broker = TestOutboxBroker(broker) + async with test_broker: + await test_broker.fake_client.insert(fake_session(), queue="orders", payload=b"direct", headers=None) + async with test_broker: + await broker.publish(b"published", queue="orders", session=fake_session()) + + assert received == [b"published"] diff --git a/tests/test_fastapi.py b/tests/test_fastapi.py index e97258d..76d85fd 100644 --- a/tests/test_fastapi.py +++ b/tests/test_fastapi.py @@ -37,7 +37,7 @@ from faststream_outbox.fastapi import ( OutboxRouter, ) -from faststream_outbox.testing import TestOutboxBroker +from faststream_outbox.testing import TestOutboxBroker, fake_session def _make_outbox_table() -> Any: @@ -72,7 +72,7 @@ async def handle(body: dict) -> None: app = _make_app_with_router(router) with TestClient(app): - await router.broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await router.broker.publish({"x": 1}, queue="orders", session=fake_session()) assert received == [{"x": 1}] @@ -121,7 +121,7 @@ async def handle( app = _make_app_with_router(router) with TestClient(app): - await router.broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await router.broker.publish({"x": 1}, queue="orders", session=fake_session()) assert seen_session == [sentinel_session] @@ -148,7 +148,7 @@ async def handle(body: str) -> None: assert resp.json() == {"ok": True} # Subscriber works through the same router - await router.broker.publish("via-subscriber", queue="orders") # ty: ignore[missing-argument] + await router.broker.publish("via-subscriber", queue="orders", session=fake_session()) assert subscriber_fired == ["via-subscriber"] @@ -174,24 +174,26 @@ async def handle( captured["client"] = client test_broker = TestOutboxBroker(router.broker) + swapped: dict[str, object] = {} @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None]: del app async with test_broker: + swapped["producer"] = router.broker.config.broker_config.producer yield app = FastAPI(lifespan=lifespan) app.include_router(router) with TestClient(app): - await router.broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await router.broker.publish({"x": 1}, queue="orders", session=fake_session()) assert isinstance(captured["msg"], OutboxMessage) assert captured["broker"] is router.broker # F7-10: identity-check the injected producer/client against the swapped fakes — the # ``Context("broker.config.broker_config.client")`` annotation *string* is the load-bearing # artifact, and an isinstance check would pass even if it resolved to a different instance. - assert captured["producer"] is test_broker.fake_producer + assert captured["producer"] is swapped["producer"] assert captured["client"] is test_broker.fake_client @@ -244,8 +246,8 @@ async def handle_downstream(body: dict) -> None: app = _make_app_with_router(router) with TestClient(app): - await router.broker.publish({"id": 1}, queue="orders") # ty: ignore[missing-argument] - await router.broker.publish({"id": 2}, queue="orders") # ty: ignore[missing-argument] + await router.broker.publish({"id": 1}, queue="orders", session=fake_session()) + await router.broker.publish({"id": 2}, queue="orders", session=fake_session()) # OutboxResponse chaining works through the FastAPI wrapper: the bridged Depends session is # the one the follow-on row is published with. @@ -285,7 +287,7 @@ async def handle(body: dict) -> None: # ...and the forwarded recorder actually fires on the inner broker's dispatch path. app = _make_app_with_router(router) with TestClient(app): - await router.broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await router.broker.publish({"x": 1}, queue="orders", session=fake_session()) assert events # the recorder seam is live under the FastAPI router @@ -315,7 +317,7 @@ async def handle(body: dict, tenant: str = tenant_dep) -> None: app = _make_app_with_router(router) with TestClient(app) as client: - await router.broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + await router.broker.publish({"x": 1}, queue="orders", session=fake_session()) schema = client.get("/asyncapi.json").json() assert seen == [({"x": 1}, "tenant-a")] diff --git a/tests/test_middleware_opentelemetry.py b/tests/test_middleware_opentelemetry.py index 1131e6e..a3a621e 100644 --- a/tests/test_middleware_opentelemetry.py +++ b/tests/test_middleware_opentelemetry.py @@ -206,15 +206,9 @@ def test_outbox_telemetry_provider_publish_destination_name() -> None: assert provider.get_publish_destination_name(cmd) == "dst-q" -async def test_outbox_telemetry_middleware_publish_scope_does_not_fire_under_test_broker() -> None: - """``TestOutboxBroker`` patches ``broker.publish`` directly, bypassing ``_basic_publish``. - - The middleware's ``publish_scope`` therefore must not fire. The recorder seam - (via ``FakeOutboxProducer`` / ``_build_fake_publish``) is the publish-side - metrics path in test mode. Negative-assert here so a future refactor that - routes test-broker publishes through ``_basic_publish`` (and thus through - the publish-scope middleware) trips this guardrail. - """ +@pytest.mark.parametrize("path", ["broker", "publisher"]) +async def test_outbox_telemetry_middleware_records_publishes_under_test_broker(path: str) -> None: + """``TestOutboxBroker`` publishes through ``_basic_publish``, so the middleware's ``publish_scope`` records them.""" reader = InMemoryMetricReader() broker = _make_broker(reader) @@ -223,13 +217,13 @@ async def handle(body: dict) -> None: pass async with TestOutboxBroker(broker): - await broker.publish({"x": 1}, queue="orders", session=_session_mock()) + if path == "publisher": + await broker.publisher("orders").publish({"x": 1}, session=_session_mock()) + else: + await broker.publish({"x": 1}, queue="orders", session=_session_mock()) - instruments = _instruments(reader) - # Publish-scope middleware would create messaging.publish.duration; under - # the test broker the publish path bypasses _basic_publish, so the instrument - # must be absent. - assert "messaging.publish.duration" not in instruments + duration = _instruments(reader)["messaging.publish.duration"] + assert [point.count for point in duration.data_points] == [1] def test_outbox_telemetry_middleware_raises_friendly_error_when_extra_missing() -> None: @@ -266,7 +260,12 @@ async def handle(body: dict) -> None: await broker.publish({"x": 1}, queue="orders", session=_session_mock()) spans = span_exporter.get_finished_spans() - outbox_spans = [sp for sp in spans if dict(sp.attributes or {}).get("messaging.system") == "outbox"] + outbox_spans = [ + sp + for sp in spans + if dict(sp.attributes or {}).get("messaging.system") == "outbox" + and dict(sp.attributes or {}).get("messaging.operation") == "process" + ] assert outbox_spans, f"no outbox consume span emitted; spans={[sp.name for sp in spans]}" attrs = dict(outbox_spans[0].attributes or {}) assert attrs["messaging.system"] == "outbox" diff --git a/tests/test_middleware_prometheus.py b/tests/test_middleware_prometheus.py index c0a3131..9d80280 100644 --- a/tests/test_middleware_prometheus.py +++ b/tests/test_middleware_prometheus.py @@ -231,16 +231,9 @@ def _acked_from(reg: CollectorRegistry) -> float: assert _acked_from(recorder_reg) == 1.0 -async def test_outbox_prometheus_middleware_publish_scope_does_not_fire_under_test_broker() -> None: - """``TestOutboxBroker`` patches ``broker.publish`` directly, bypassing ``_basic_publish``. - - The middleware's ``publish_scope`` therefore must not fire. The recorder seam - (via ``FakeOutboxProducer`` / ``_build_fake_publish``) is the publish-side - metrics path in test mode. Negative-assert here so a future refactor that - routes test-broker publishes through ``_basic_publish`` (and thus through - the publish-scope middleware) trips this guardrail instead of silently - double-counting in production users that test through ``TestOutboxBroker``. - """ +@pytest.mark.parametrize("path", ["broker", "publisher"]) +async def test_outbox_prometheus_middleware_counts_publishes_under_test_broker(path: str) -> None: + """``TestOutboxBroker`` publishes through ``_basic_publish``, so the middleware's ``publish_scope`` counts them.""" reg = CollectorRegistry() broker = _make_broker(reg) @@ -249,20 +242,21 @@ async def handle(body: dict) -> None: pass async with TestOutboxBroker(broker): - await broker.publish({"x": 1}, queue="orders", session=_session_mock()) + if path == "publisher": + await broker.publisher("orders").publish({"x": 1}, session=_session_mock()) + else: + await broker.publish({"x": 1}, queue="orders", session=_session_mock()) - # A Counter that's never been ``.labels(...).inc()``-ed produces zero samples; - # any non-zero sample on the publish-side series means middleware fired. - non_zero_publish_samples = [ - (sample.name, dict(sample.labels), sample.value) + published = [ + (dict(sample.labels), sample.value) for metric in reg.collect() if metric.name == "faststream_published_messages" for sample in metric.samples - if sample.value > 0.0 + if sample.name.endswith("_total") + ] + assert [(labels["destination"], labels["status"], value) for labels, value in published] == [ + ("orders", "success", 1.0) ] - assert non_zero_publish_samples == [], ( - f"publish_scope middleware fired under TestOutboxBroker: {non_zero_publish_samples}" - ) def test_outbox_prometheus_middleware_raises_friendly_error_when_extra_missing() -> None: diff --git a/tests/test_relay.py b/tests/test_relay.py index b21c417..a04b137 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -9,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from faststream_outbox import OutboxBroker, OutboxResponse, OutboxRouter, make_outbox_table -from faststream_outbox.testing import TestOutboxBroker +from faststream_outbox.testing import TestOutboxBroker, fake_session pytestmark = pytest.mark.asyncio @@ -38,7 +38,7 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: # must be wired before our subscriber starts, because the outbox's start() probes # foreign producers and warns about any whose broker isn't started yet. async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: - await outbox.publish({"hello": "world"}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + await outbox.publish({"hello": "world"}, queue="relay_queue", session=fake_session()) publisher_kafka.mock.assert_called_once_with({"hello": "world"}) @@ -63,7 +63,7 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: broker_kafka.include_router(kafka_router) async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: - await outbox.publish({"router": True}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + await outbox.publish({"router": True}, queue="relay_queue", session=fake_session()) publisher_kafka.mock.assert_called_once_with({"router": True}) @@ -88,7 +88,7 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: broker_outbox.include_router(outbox_router) async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: - await outbox.publish({"outbox_router": True}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + await outbox.publish({"outbox_router": True}, queue="relay_queue", session=fake_session()) publisher_kafka.mock.assert_called_once_with({"outbox_router": True}) @@ -120,7 +120,7 @@ async def capture_publish(cmd: Any, **kwargs: Any) -> Any: await outbox.publish( {"hi": 1}, queue="relay_queue", - session=None, # ty: ignore[invalid-argument-type] + session=fake_session(), headers={"x-trace-id": "abc123", "content-type": "application/json"}, ) @@ -157,7 +157,7 @@ async def capture_publish(cmd: Any, **kwargs: Any) -> Any: await outbox.publish( {"hi": 1}, queue="relay_queue", - session=None, # ty: ignore[invalid-argument-type] + session=fake_session(), headers={"x-trace-id": "should-be-dropped"}, ) @@ -194,7 +194,7 @@ async def capture_publish(cmd: Any, **kwargs: Any) -> Any: await outbox.publish( {"hi": 1}, queue="relay_queue", - session=None, # ty: ignore[invalid-argument-type] + session=fake_session(), headers={"x-trace-id": "should-not-clobber"}, ) @@ -224,7 +224,7 @@ async def relay(body: dict[str, Any]) -> OutboxResponse: async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: with pytest.raises(RuntimeError, match="OutboxResponse"): - await outbox.publish({"x": 1}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + await outbox.publish({"x": 1}, queue="relay_queue", session=fake_session()) async def test_outbox_response_with_foreign_publisher_fires_guard_before_side_effects() -> None: @@ -248,7 +248,7 @@ async def relay(body: dict[str, Any]) -> OutboxResponse: test_outbox = TestOutboxBroker(broker_outbox, run_loops=False) async with TestKafkaBroker(broker_kafka), test_outbox as outbox: with pytest.raises(RuntimeError, match="OutboxResponse"): - await outbox.publish({"x": 1}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + await outbox.publish({"x": 1}, queue="relay_queue", session=fake_session()) # Guard fired before the chain → the foreign Kafka publish never happened… publisher_kafka.mock.assert_not_called() # …and no follow-on outbox row was inserted for the relayed body. @@ -275,7 +275,7 @@ async def relay(body: dict[str, Any]) -> OutboxResponse: test_outbox = TestOutboxBroker(broker_outbox, run_loops=False) async with TestKafkaBroker(broker_kafka), test_outbox as outbox: with pytest.raises(RuntimeError, match="OutboxResponse"): - await outbox.publish({"x": 1}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + await outbox.publish({"x": 1}, queue="relay_queue", session=fake_session()) rows = [(r.queue, r.attempts_count, r.acquired_token) for r in test_outbox.fake_client.rows] assert rows == [("relay_queue", 1, None)] diff --git a/tests/test_unit.py b/tests/test_unit.py index d45b5e2..59e6f48 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -72,7 +72,7 @@ _OutboxConfigError, _PendingFlush, ) -from faststream_outbox.testing import FakeOutboxClient, FakeOutboxProducer +from faststream_outbox.testing import FakeOutboxClient def test_outbox_broker_registered_in_test_broker_registry() -> None: @@ -1339,49 +1339,6 @@ def test_publisher_spec_get_schema_returns_publisher_spec() -> None: assert entry.operation.message.title == "orders:Publisher:Message" -# --- FakeOutboxProducer direct coverage --- - - -async def test_fake_outbox_producer_publish_batch_inserts_rows() -> None: - broker = _make_broker() - fake_client = FakeOutboxClient() - producer = FakeOutboxProducer(fake_client, broker, serializer=None, run_loops=False) - cmd = OutboxPublishCommand(b"a", b"b", queue="orders", session=_make_session_mock()) - await producer.publish_batch(cmd) - assert len(fake_client.rows) == 2 - assert {r.payload for r in fake_client.rows} == {b"a", b"b"} - - -async def test_fake_outbox_producer_publish_batch_with_none_body_inserts_row() -> None: - """B8: a sole None body inserts one b"" row via the fake producer too.""" - broker = _make_broker() - fake_client = FakeOutboxClient() - producer = FakeOutboxProducer(fake_client, broker, serializer=None, run_loops=False) - cmd = OutboxPublishCommand(None, queue="orders", session=_make_session_mock()) - await producer.publish_batch(cmd) - assert len(fake_client.rows) == 1 - assert fake_client.rows[0].payload == b"" - - -async def test_fake_outbox_producer_publish_batch_leading_none_inserts_all() -> None: - """B8: (None, b"x") inserts 2 rows — the leading None is not dropped.""" - broker = _make_broker() - fake_client = FakeOutboxClient() - producer = FakeOutboxProducer(fake_client, broker, serializer=None, run_loops=False) - cmd = OutboxPublishCommand(None, b"x", queue="orders", session=_make_session_mock()) - await producer.publish_batch(cmd) - assert len(fake_client.rows) == 2 - assert {r.payload for r in fake_client.rows} == {b"", b"x"} - - -async def test_fake_outbox_producer_request_raises() -> None: - broker = _make_broker() - producer = FakeOutboxProducer(FakeOutboxClient(), broker, serializer=None, run_loops=False) - cmd = OutboxPublishCommand(b"x", queue="orders", session=_make_session_mock()) - with pytest.raises(NotImplementedError, match="request-reply"): - await producer.request(cmd) - - def test_broker_client_property_raises_without_engine() -> None: metadata = MetaData() t = make_outbox_table(metadata) @@ -3650,8 +3607,7 @@ async def test_metrics_published_batch_emits_count_equal_to_batch_size() -> None await broker.publish_batch({"x": 1}, {"y": 2}, {"z": 3}, queue="orders", session=session) pub = [t for e, t in events if e == "published"] - # ``_build_fake_publish_batch`` aggregates: one event per ``publish_batch`` call - # with ``count == landed`` (mirrors the real producer's batch contract). + # One event per ``publish_batch`` call, with ``count`` equal to the rows inserted. assert len(pub) == 1 assert pub[0]["count"] == 3