chore(deps): adapt to faststream 0.7.6 and raise its floor

faststream_outbox/__init__.py

@@ -1,9 +1,3 @@
-import functools
-import typing
-
-from faststream._internal.broker import BrokerUsecase
-from faststream._internal.testing.broker import TestBroker
-
 from faststream_outbox.autovacuum import outbox_autovacuum_ddl
 from faststream_outbox.broker import OutboxBroker
 from faststream_outbox.message import OutboxMessage
@@ -39,26 +33,3 @@
     "make_outbox_table",
     "outbox_autovacuum_ddl",
 ]
-
-try:
-    # S4: import inside the guard too — if upstream moves/removes the module, this
-    # raises ImportError here and is tolerated, instead of breaking ``import
-    # faststream_outbox`` from an unguarded top-level import.
-    import faststream.asgi.factories.asyncapi.try_it_out
-
-    original_get_broker_registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry  # noqa: SLF001
-
-    @functools.lru_cache(maxsize=1)
-    def get_broker_registry() -> dict[
-        type[BrokerUsecase[typing.Any, typing.Any]],
-        type[TestBroker[typing.Any, typing.Any]],
-    ]:
-        # BrokerUsecase is invariant on its config type, so OutboxBrokerConfig won't unify with BrokerConfig.
-        return {**original_get_broker_registry(), OutboxBroker: TestOutboxBroker}  # ty: ignore[invalid-return-type]
-
-    faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry = get_broker_registry  # noqa: SLF001
-except (AttributeError, ImportError):  # pragma: no cover
-    # FastStream's private ASGI try-it-out registry is best-effort wiring;
-    # tolerate breakage if upstream renames/moves the symbol but surface other
-    # errors (config, type) loudly so we notice them in CI.
-    pass

faststream_outbox/broker.py

@@ -134,7 +134,7 @@ def __init__(  # noqa: PLR0913
         dlq_table: "Table | None" = None,
         decoder: CustomCallable | None = None,
         parser: CustomCallable | None = None,
-        dependencies: Iterable["Dependant"] = (),
+        dependencies: Sequence["Dependant"] = (),
         middlewares: Sequence[type[BaseMiddleware] | BrokerMiddleware[OutboxInnerMessage]] = (),
         graceful_timeout: float | None = 15.0,
         routers: Sequence[Registrator[OutboxInnerMessage]] = (),

faststream_outbox/publisher/specification.py

@@ -21,6 +21,7 @@ def get_schema(self) -> dict[str, PublisherSpec]:
         payloads = self.get_payloads()
         return {
             self.name: PublisherSpec(
+                address=self.config.queue,
                 description=self.config.description_,
                 operation=Operation(
                     message=Message(

faststream_outbox/registrator.py

@@ -1,5 +1,5 @@
 import warnings
-from collections.abc import Iterable
+from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
 from faststream._internal.broker.registrator import Registrator
@@ -54,7 +54,7 @@ def subscriber(  # ty: ignore[invalid-method-override]
         terminal_flush_batch_size: int = 1,
         ack_policy: AckPolicy | None = None,
         propagate_inbound_headers: bool = False,
-        dependencies: Iterable["Dependant"] = (),
+        dependencies: Sequence["Dependant"] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,
         title_: str | None = None,

faststream_outbox/router.py

@@ -35,7 +35,7 @@ def __init__(  # noqa: PLR0913
         terminal_flush_batch_size: int = 1,
         ack_policy: AckPolicy | None = None,
         propagate_inbound_headers: bool = False,
-        dependencies: Iterable["Dependant"] = (),
+        dependencies: Sequence["Dependant"] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,
         title_: str | None = None,
@@ -77,7 +77,7 @@ def __init__(
         self,
         handlers: Iterable[OutboxRoute] = (),
         *,
-        dependencies: Iterable["Dependant"] = (),
+        dependencies: Sequence["Dependant"] = (),
         middlewares: Sequence[BrokerMiddleware[OutboxInnerMessage]] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,

faststream_outbox/subscriber/usecase.py

@@ -159,14 +159,19 @@ def _render_last_exception(
 
 
 class OutboxSubscriberSpecification(SubscriberSpecification["OutboxBrokerConfig", OutboxSubscriberSpecificationConfig]):
+    @property
+    def channel_labels(self) -> list[str]:
+        """One channel, keyed by every queue this subscriber drains."""
+        return [",".join(self.config.queues)]
+
     @property
     def name(self) -> str:
-        joined = ",".join(self.config.queues)
-        return f"{joined}:{self.call_name}"
+        return f"{self.channel_labels[0]}:{self.call_name}"
 
     def get_schema(self) -> dict[str, SubscriberSpec]:
         return {
             self.name: SubscriberSpec(
+                address=self.channel_labels[0],
                 description=self.description,
                 operation=Operation(
                     message=Message(
@@ -923,13 +928,12 @@ async def get_one(self, *, timeout: float = 5.0) -> typing.NoReturn:
         raise NotImplementedError(_UNSUPPORTED_PEEK_MSG)
 
     @override
-    async def __aiter__(self) -> AsyncIterator["StreamMessage[OutboxInnerMessage]"]:
+    def __aiter__(self) -> AsyncIterator["StreamMessage[OutboxInnerMessage]"]:
         # Native FakeStream subscribers (e.g. redis ListSubscriber.__aiter__) implement
         # this against a blocking pop; for the outbox, a true peek would acquire a lease
         # and bump deliveries_count — surprising semantics for a "look but don't touch"
         # API. Route operators at ``broker.fetch_unprocessed`` instead, which is
-        # lease-free and doesn't mutate row state. Matches the base's no-yield shape so
-        # the override stays a coroutine returning AsyncIterator (not an async generator).
+        # lease-free and doesn't mutate row state.
         raise NotImplementedError(_UNSUPPORTED_PEEK_MSG)
 
     @override

faststream_outbox/testing.py

@@ -568,8 +568,7 @@ async def fake_publish_batch(
     return fake_publish_batch
 
 
-# BrokerUsecase is invariant on its config type, so OutboxBrokerConfig won't unify with BrokerConfig.
-class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker]):  # ty: ignore[invalid-type-arguments]
+class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker], broker=OutboxBroker):
     """Test harness for ``OutboxBroker``. Two dispatch modes.
 
     Default (``run_loops=False``): ``broker.publish`` synchronously drives the matching
@@ -682,8 +681,7 @@ def _fake_start(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing.
         # Skip the parent's publisher iteration — see ``create_publisher_fake_subscriber``
         # for why. We still need to fan out ``_post_start`` on subscribers so their
         # call models build (matches what TestBroker._fake_start does last).
-        # BrokerUsecase is invariant on its config type; this only iterates broker.subscribers.
-        patch_broker_calls(broker)  # ty: ignore[invalid-argument-type]
+        patch_broker_calls(broker)
         for subscriber in broker.subscribers:
             subscriber._post_start()  # noqa: SLF001
         broker._warn_on_unstarted_foreign_publishers()  # noqa: SLF001

pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
 ]
 dependencies = [
-    "faststream>=0.7.1,<0.8",
+    "faststream>=0.7.6,<0.8",
     "sqlalchemy[asyncio]>=2.0",
     "typing-extensions>=4.12.0",
 ]
@@ -46,7 +46,7 @@ dev = [
     "asyncpg>=0.29",
     "alembic>=1.13",
     "fastapi>=0.95,<0.140", # see the cap note on the fastapi extra
-    "faststream[kafka]>=0.7.1,<0.8",
+    "faststream[kafka]>=0.7.6,<0.8",
     "httpx2>=2.2",
     "prometheus-client>=0.19",
     "opentelemetry-api>=1.20",

tests/test_integration.py

@@ -1947,8 +1947,10 @@ def spy_log(*args: Any, **kwargs: Any) -> Any:
                 await broker_outbox.publish({"x": 1}, queue="relay_queue", session=session)
             await _wait_until(lambda: bool(errors), timeout=10.0)
 
-    # Guard fired before the chain → the foreign Kafka publish never happened.
-    publisher_kafka.mock.assert_not_called()
+        # Guard fired before the chain → the foreign Kafka publish never happened.
+        # Asserted inside the TestKafkaBroker context: outside it, `.mock` raises SetupError.
+        publisher_kafka.mock.assert_not_called()
+
     # Row left in place (lease held, not deleted) for lease-expiry retry.
     assert await _row_count(pg_engine, outbox_table) == 1, "config-error row must be left for lease-expiry, not deleted"
     # And the _OutboxConfigError was logged at ERROR by the worker loop.

tests/test_unit.py

@@ -11,10 +11,10 @@
 from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
 
-import faststream.asgi.factories.asyncapi.try_it_out
 import pytest
 from faststream._internal.parser import DefaultCodec
 from faststream._internal.producer import ProducerProto
+from faststream._internal.testing.broker import find_test_broker
 from faststream.exceptions import IncorrectState
 from faststream.middlewares import AckPolicy
 from faststream.response.publish_type import PublishType
@@ -72,9 +72,10 @@
 from faststream_outbox.testing import FakeOutboxClient, FakeOutboxProducer
 
 
-def test_outbox_broker_registered_in_try_it_out_registry() -> None:
-    registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry()  # noqa: SLF001
-    assert registry[OutboxBroker] is TestOutboxBroker  # ty: ignore[invalid-argument-type]
+def test_outbox_broker_registered_in_test_broker_registry() -> None:
+    metadata = MetaData()
+    broker = OutboxBroker(outbox_table=make_outbox_table(metadata))
+    assert find_test_broker(broker) is TestOutboxBroker
 
 
 def _make_broker(engine: object | None = None, table_name: str = "outbox") -> OutboxBroker:
@@ -1831,7 +1832,7 @@ async def handle(body: str) -> None: ...
 
     # __aiter__ is also unsupported (was silently abstract-inherited before B6).
     with pytest.raises(NotImplementedError, match="fetch_unprocessed"):
-        await sub.__aiter__()
+        sub.__aiter__()
 
     # _make_response_publisher returns an OutboxFakePublisher wired to the producer
     # so handlers can ``return OutboxResponse(...)``.
@@ -4179,7 +4180,7 @@ async def handle(body: dict) -> None: ...  # registered for the spec; never invo
 
     broker.publisher("events")
 
-    spec = AsyncAPI(broker).to_specification().to_jsonable()  # ty: ignore[invalid-argument-type]  # BrokerUsecase invariance
+    spec = AsyncAPI(broker).to_specification().to_jsonable()
     assert spec["servers"], "AsyncAPI servers must not be empty (url=[] regression)"
     channel_keys = " ".join(spec["channels"])
     assert "orders" in channel_keys, f"subscriber channel missing: {list(spec['channels'])}"
@@ -4198,7 +4199,7 @@ async def handle(body: dict) -> None: ...  # registered for the spec; never invo
 
     broker.publisher("hidden_events", include_in_schema=False)
 
-    spec = AsyncAPI(broker).to_specification().to_jsonable()  # ty: ignore[invalid-argument-type]  # BrokerUsecase invariance
+    spec = AsyncAPI(broker).to_specification().to_jsonable()
     channel_keys = " ".join(spec["channels"])
     assert "orders" in channel_keys  # the included subscriber is present…
     assert "hidden_events" not in channel_keys  # …the excluded publisher is not