fix: adapt to faststream 0.7.6, and make the AsyncAPI document non-blank

CONTEXT.md

@@ -24,8 +24,9 @@ The named channel a Timer is Scheduled on and that a subscriber consumes from. T
 derived per Topic; a Timer only ever exists on one.
 _Avoid_: queue — a Topic is not a work queue, and the difference matters: Timers on a Topic are
 ordered by Activation time, not arrival, and a Claimed Timer stays on the Topic rather than leaving
-it. `channel` survives only where FastStream owns the spelling: the `raw_message` field and the
-log-context key. In our own prose it is Topic.
+it. `channel` survives only where FastStream owns the spelling: the `raw_message` field, the
+log-context key, and `channel_labels` with the AsyncAPI document it feeds, where a Topic is a
+channel — say *AsyncAPI channel* there. In our own prose it is Topic.
 
 **Activation time**:
 The single absolute UTC instant at which a Timer becomes Due. Exactly one of `activate_in` (a

faststream_redis_timers/__init__.py

@@ -13,23 +13,3 @@
     "TimersRoutePublisher",
     "TimersRouter",
 ]
-
-try:
-    import functools
-    import typing
-
-    import faststream.asgi.factories.asyncapi.try_it_out
-    from faststream._internal.broker import BrokerUsecase
-    from faststream._internal.testing.broker import TestBroker
-
-    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]]
-    ]:
-        return {**original_get_broker_registry(), TimersBroker: TestTimersBroker}
-
-    faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry = get_broker_registry  # noqa: SLF001
-except Exception:  # noqa: BLE001, S110  # pragma: no cover
-    pass

faststream_redis_timers/broker.py

@@ -19,6 +19,7 @@
 from faststream.specification.schema import BrokerSpec
 from faststream.specification.schema.extra import Tag, TagDict
 from redis.asyncio.cluster import RedisCluster
+from redis.asyncio.connection import SSLConnection
 from typing_extensions import override
 
 from faststream_redis_timers.configs import ConnectionState, RedisClient, TimersBrokerConfig
@@ -41,6 +42,29 @@ def _require_topic(topic: str) -> None:
         raise ValueError(msg)
 
 
+def _spec_url(client: "RedisClient | None", timeline_key: str) -> list[str]:
+    """AsyncAPI server URL(s) for the broker spec.
+
+    **Must be non-empty.** Upstream collects channels only for brokers that reach its
+    ``broker_servers`` mapping, which it fills inside ``for url in specification.url``;
+    an empty list renders a document with no servers, channels or operations while every
+    per-endpoint ``get_schema()`` stays correct. Rebuilt from the pool's connection
+    parameters rather than the caller's DSN, so no credential can reach the document.
+    """
+    if client is None:
+        return [f"redis://timers/{timeline_key}"]
+
+    connection_kwargs = client.connection_pool.connection_kwargs
+    if socket_path := connection_kwargs.get("path"):
+        return [f"unix://{socket_path}"]
+
+    scheme = "rediss" if issubclass(client.connection_pool.connection_class, SSLConnection) else "redis"
+    host = connection_kwargs.get("host", "localhost")
+    port = connection_kwargs.get("port", 6379)
+    db = connection_kwargs.get("db", 0)
+    return [f"{scheme}://{host}:{port}/{db}"]
+
+
 class TimersParamsStorage(DefaultLoggerStorage):
     __max_msg_id_ln = -1
     _max_channel_name = 7
@@ -97,7 +121,7 @@ def __init__(  # noqa: PLR0913
         start_timeout: float = 3.0,
         decoder: CustomCallable | None = None,
         parser: CustomCallable | None = None,
-        dependencies: Iterable[Dependant] = (),
+        dependencies: Sequence[Dependant] = (),
         middlewares: Sequence[type[BaseMiddleware] | BrokerMiddleware[TimerMessage]] = (),
         graceful_timeout: float | None = 15.0,
         routers: Sequence[Registrator[TimerMessage]] = (),
@@ -142,7 +166,7 @@ def __init__(  # noqa: PLR0913
             ),
         )
         specification = BrokerSpec(
-            url=[],
+            url=_spec_url(client, timeline_key),
             protocol="redis",
             protocol_version="5.0",
             description=description,

faststream_redis_timers/publisher/usecase.py

@@ -22,13 +22,18 @@
 
 class TimersPublisherSpecification(PublisherSpecification["TimersBrokerConfig", TimersPublisherSpecificationConfig]):  # ty: ignore[unresolved-reference]
     @property
-    def name(self) -> str:
+    def full_topic(self) -> str:
         prefix = getattr(self._outer_config, "prefix", "")
-        return f"{prefix}{self.config.topic}:Publisher"
+        return f"{prefix}{self.config.topic}"
+
+    @property
+    def name(self) -> str:
+        return f"{self.full_topic}:Publisher"
 
     def get_schema(self) -> dict[str, PublisherSpec]:
         return {
             self.name: PublisherSpec(
+                address=self.full_topic,
                 description=self.config.description_,
                 operation=Operation(
                     message=Message(

faststream_redis_timers/registrator.py

@@ -1,5 +1,5 @@
 import warnings
-from collections.abc import Iterable
+from collections.abc import Sequence
 from typing import Any
 
 from fast_depends.dependencies import Dependant
@@ -24,7 +24,7 @@ def subscriber(  # ty: ignore[invalid-method-override]
         max_polling_interval: float = 5.0,
         max_concurrent: int = 5,
         lease_ttl: int = 30,
-        dependencies: Iterable[Dependant] = (),
+        dependencies: Sequence[Dependant] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,
         title_: str | None = None,

faststream_redis_timers/router.py

@@ -45,7 +45,7 @@ def __init__(  # noqa: PLR0913
         max_concurrent: int = 5,
         lease_ttl: int = 30,
         publishers: Iterable[TimersRoutePublisher] = (),
-        dependencies: Iterable[Dependant] = (),
+        dependencies: Sequence[Dependant] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,
         title_: str | None = None,
@@ -77,7 +77,7 @@ def __init__(  # noqa: PLR0913
         prefix: str = "",
         handlers: Iterable[TimersRoute] = (),
         *,
-        dependencies: Iterable[Dependant] = (),
+        dependencies: Sequence[Dependant] = (),
         middlewares: Sequence[BrokerMiddleware[TimerMessage]] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,

faststream_redis_timers/subscriber/usecase.py

@@ -29,13 +29,18 @@
 
 class TimersSubscriberSpecification(SubscriberSpecification["TimersBrokerConfig", TimersSubscriberSpecificationConfig]):
     @property
-    def name(self) -> str:
+    def full_topic(self) -> str:
         prefix = getattr(self._outer_config, "prefix", "")
-        return f"{prefix}{self.config.topic}:{self.call_name}"
+        return f"{prefix}{self.config.topic}"
+
+    @property
+    def channel_labels(self) -> list[str]:
+        return [self.full_topic]
 
     def get_schema(self) -> dict[str, SubscriberSpec]:
         return {
             self.name: SubscriberSpec(
+                address=self.full_topic,
                 description=self.description,
                 operation=Operation(
                     message=Message(
@@ -75,7 +80,8 @@ async def start(self) -> None:
 
         start_signal = anyio.Event()
         if self.calls:
-            consume_task = self.add_task(self._consume, (self._client,), {"start_signal": start_signal})
+            self.add_task(self._consume, (self._client,), {"start_signal": start_signal})
+            consume_task = self.tasks[-1]
             try:
                 with anyio.fail_after(self._outer_config.start_timeout):
                     await start_signal.wait()

faststream_redis_timers/testing.py

@@ -31,7 +31,7 @@ class ScheduledTimer:
     headers: dict[str, typing.Any] | None = None
 
 
-class TestTimersBroker(TestBroker[TimersBroker, TimersBroker]):
+class TestTimersBroker(TestBroker[TimersBroker, TimersBroker], broker=TimersBroker):
     scheduled_timers: list[ScheduledTimer]
 
     def __init__(self, broker: TimersBroker, **kwargs: typing.Any) -> None:

pyproject.toml

@@ -21,8 +21,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
 ]
 dependencies = [
-    # 0.7.6 makes SubscriberUsecase.add_task return None and drops _get_broker_registry; lift once this adapts.
-    "faststream>=0.7.1,<0.7.6",
+    "faststream>=0.7.6,<0.8",
     "redis>=5.0",
     "typing-extensions>=4.12.0",
 ]

tests/test_unit.py

@@ -6,10 +6,12 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
-import faststream.asgi.factories.asyncapi.try_it_out
 import pytest
 from faststream._internal.parser import DefaultCodec
+from faststream._internal.testing.broker import find_test_broker
 from faststream.exceptions import IncorrectState
+from faststream.specification import AsyncAPI
+from redis.asyncio import Redis
 from redis.asyncio.cluster import RedisCluster
 from redis.exceptions import NoScriptError
 
@@ -31,9 +33,15 @@
 # --- AsyncAPI try_it_out registry ---
 
 
-def test_timers_broker_registered_in_try_it_out_registry() -> None:
-    registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry()  # noqa: SLF001
-    assert registry[TimersBroker] is TestTimersBroker
+def test_timers_broker_registered_in_test_broker_registry() -> None:
+    """INVARIANT: a `TimersBroker` resolves to `TestTimersBroker` through FastStream's own registry.
+
+    The AsyncAPI "try it out" page looks the test broker up by broker class, so a `TimersBroker`
+    that is absent from the registry silently loses that page. Registration is declarative — the
+    `broker=` argument on the class statement — so deleting it breaks this with nothing else to
+    notice, and a registry populated by hand goes stale against whatever FastStream does next.
+    """
+    assert find_test_broker(TimersBroker()) is TestTimersBroker
 
 
 # --- ConnectionState ---
@@ -174,30 +182,111 @@ async def test_publisher_request_raises() -> None:
         await pub.request("x")
 
 
+# --- AsyncAPI document assembly ---
+
+
+async def test_asyncapi_document_carries_the_declared_channels() -> None:
+    """INVARIANT: a broker's endpoints reach the assembled AsyncAPI document, not just `get_schema()`.
+
+    Upstream collects channels only for brokers that reach `broker_servers`, and it fills that
+    mapping inside `for url in specification.url`. A broker whose spec carries an empty `url` is
+    therefore skipped outright and renders a structurally blank document — no servers, channels or
+    operations — while every per-endpoint `get_schema()` stays correct, so specification-level tests
+    cannot see it. Emptying `BrokerSpec.url` breaks this again, silently.
+    """
+    broker = TimersBroker(Redis.from_url("redis://cache.example:6399/3"))
+    sub = broker.subscriber("reminders")
+
+    @sub
+    async def handle(body: str) -> None: ...
+
+    broker.publisher("reminders")
+
+    async with TestTimersBroker(broker):
+        spec = AsyncAPI(broker).to_specification().to_jsonable()
+
+    assert sorted(spec["channels"]) == ["reminders:Handle", "reminders:Publisher"]
+    assert [server["host"] for server in spec["servers"].values()] == ["cache.example:6399"]
+
+
+def test_specification_url_carries_no_credentials() -> None:
+    """INVARIANT: the AsyncAPI server URL is rebuilt from connection parameters, never the DSN.
+
+    The document is published — FastStream serves it from the ASGI app — so a URL echoed back from
+    `Redis.from_url("redis://user:secret@...")` would publish the password. Reading the pool's
+    `username`/`password` into the URL, or passing the caller's DSN through, breaks it.
+    """
+    broker = TimersBroker(Redis.from_url("redis://user:secret@cache.example:6399/3"))
+
+    assert broker.specification.url == ["redis://cache.example:6399/3"]
+
+
+@pytest.mark.parametrize(
+    ("client", "expected"),
+    [
+        (None, "redis://timers/tl"),
+        (Redis.from_url("unix:///tmp/redis.sock?db=2"), "unix:///tmp/redis.sock"),
+        (Redis.from_url("rediss://cache.example:6379/0"), "rediss://cache.example:6379/0"),
+    ],
+    ids=["no-client", "unix-socket", "tls"],
+)
+def test_specification_url_shapes(client: Redis | None, expected: str) -> None:
+    assert TimersBroker(client, timeline_key="tl").specification.url == [expected]
+
+
 # --- TimersSubscriberSpecification.name / get_schema ---
 
 
-def test_subscriber_specification_name_and_schema() -> None:
+async def test_subscriber_specification_addresses_the_channel_by_its_full_topic() -> None:
+    """INVARIANT: the channel a subscriber emits is addressed by the topic it actually polls.
+
+    The channel key carries the handler name with it, so it names nothing subscribable on its own;
+    `address` is the only field a reader can act on. Dropping the prefix breaks it just as surely
+    as omitting the field: a broker-prefixed deployment would publish an address no timer lands on.
+    """
+    broker = TimersBroker()
+    router = TimersRouter(prefix="app:")
+    sub = router.subscriber("my-topic")
+
+    @sub
+    async def handle(body: str) -> None: ...
+
+    broker.include_router(router)
+    async with TestTimersBroker(broker):
+        schema = sub.specification.get_schema()
+
+    assert {key: spec.address for key, spec in schema.items()} == {"app:my-topic:Handle": "app:my-topic"}
+
+
+async def test_subscriber_title_names_the_channel_not_only_the_operation() -> None:
+    """`title_` names the channel, matching the publisher and every built-in broker."""
     broker = TimersBroker()
-    sub = broker.subscriber("my-topic")
-    spec = sub.specification
-    name = spec.name
-    assert "my-topic" in name
-    schema = spec.get_schema()
-    assert schema  # non-empty dict
+    sub = broker.subscriber("my-topic", title_="TimerIngest")
+
+    @sub
+    async def handle(body: str) -> None: ...
+
+    async with TestTimersBroker(broker):
+        assert list(sub.specification.get_schema()) == ["TimerIngest"]
 
 
 # --- TimersPublisherSpecification.name / get_schema ---
 
 
-def test_publisher_specification_name_and_schema() -> None:
+def test_publisher_specification_addresses_the_channel_by_its_full_topic() -> None:
+    """INVARIANT: the channel a publisher emits is addressed by the topic it actually writes to.
+
+    Same reason as the subscriber: the key is `topic:Publisher`, which no client can publish to,
+    and an address that drops the broker prefix points at a topic nothing reads.
+    """
     broker = TimersBroker()
-    pub = broker.publisher("my-topic")
-    spec = pub.specification
-    name = spec.name
-    assert "my-topic" in name
-    schema = spec.get_schema()
-    assert schema  # non-empty dict
+    router = TimersRouter(prefix="app:")
+    pub = router.publisher("my-topic")
+
+    broker.include_router(router)
+    schema = pub.specification.get_schema()
+
+    assert {key: spec.address for key, spec in schema.items()} == {"app:my-topic:Publisher": "app:my-topic"}
 
 
 # --- TimerStreamMessage.nack ---