{"owner":"ag2ai","repo":"faststream","hasSkills":true,"totalSkillsCount":4,"totalTokensCount":6891,"categories":["anthropic-skill"],"hasMcp":false,"mcpConfig":null,"found":[".agents/skills/documentation-writing/SKILL.md",".agents/skills/dev-workflow/SKILL.md",".agents/skills/code-architecture/SKILL.md",".agents/skills/testing-patterns/SKILL.md"],"skills":{".agents/skills/documentation-writing/SKILL.md":"---\nname: documentation-writing\ndescription: Use when writing or editing documentation under docs/ — page layout, docs_src snippet embedding, snippet tests, link conventions, and FastStream example-code rules.\n---\n\n# FastStream Documentation Writing\n\n## Layout\n\n- Pages: `docs/docs/en/...` (markdown, mkdocs-material). `docs/docs/en/` is the mkdocs `docs_dir`, so images live in `docs/docs/en/assets/`.\n- Runnable code snippets: `docs/docs_src/<topic>/...` (fully excluded from ruff — see `exclude` in `ruff.toml`).\n- Navigation: `docs/docs/navigation_template.txt` — new pages need an entry there. It is rendered into the gitignored `docs/docs/en/SUMMARY.md` (literate-nav).\n- API reference pages are generated by `docs/create_api_docs.py` — never hand-edit them.\n\n## Code snippets\n\nNever inline non-trivial code in markdown. Put a runnable Python file in `docs/docs_src/` and embed it with `mdx_include`:\n\n```python linenums=\"1\" hl_lines=\"10\"\n{!> docs_src/getting_started/publishing/kafka/broker.py !}\n```\n\nNew `docs_src` snippets should get a test under `tests/docs/<topic>/test_<feature>.py` — one test file covers all broker variants via require marks (e.g. `@require_aiokafka`), not one file per broker subdir (see the **testing-patterns** skill). Some examples are intentionally untested; the exemptions are listed in the coverage `omit` list in `pyproject.toml`. `tests/docs/test_snippets_are_tested.py` holds the line: a `docs_src` file that no test imports (directly, or through another tested snippet) fails it unless it is in `omit`. Its `NOT_TESTED_YET` set is the backlog from before the check — take an entry out when you test it, never add one.\n\nA snippet test imports the runnable objects from the module and exercises them in-memory — never a bare `import module` (it tests nothing and trips ruff `F401`):\n\n```python\nimport pytest\n\nfrom faststream.nats import TestApp, TestNatsBroker\n\n\n@pytest.mark.nats()\n@pytest.mark.asyncio()\nasync def test_basic() -> None:\n    from docs.docs_src.nats.js.pull_sub import app, broker, handle\n\n    async with TestNatsBroker(broker), TestApp(app):\n        await broker.publish(\"Hi!\", \"test\")\n        handle.mock.assert_called_once_with(\"Hi!\")\n```\n\n`just mypy` checks `docs/docs_src/` too. The annotation checks are off there (a handler may omit `-> None`, a bare `dict` is fine), every other check is on: a snippet that calls the public API wrongly fails CI. A snippet that is right while mypy disagrees points at an annotation in `faststream/` — fix it there, in its own PR with a case in `tests/mypy/`. A snippet that is wrong on purpose gets a commented per-module `[[tool.mypy.overrides]]` in `pyproject.toml`, not a `# type: ignore` the reader would see.\n\nThe test imports via the `docs.docs_src.…` package path, while the markdown embeds it via the `docs_src/…` path (mdx_include `base_path` is `docs/`). Either way the snippet MUST live under `docs/docs_src/`, not repo-root `docs_src/` — otherwise the embed silently resolves to nothing and the import fails.\n\n## Markdown conventions\n\n- External links: `[text](https://...){.external-link target=\"_blank\"}`; internal links: `{.internal-link}`.\n- Multi-broker pages use pymdownx tabs with one snippet per broker:\n\n  ```markdown\n  === \"AIOKafka\"\n      ...embedded kafka snippet...\n\n  === \"Confluent\"\n      ...embedded confluent snippet...\n\n  === \"RabbitMQ\"\n      ...embedded rabbit snippet...\n  ```\n\n- Admonitions (`!!! note`, `!!! tip`, `!!! warning`) are available and encouraged for caveats.\n\n## Example-code rules (FastStream usage)\n\nAll example code must follow idiomatic FastStream usage:\n\n- Create the broker outside the app and pass it positionally: `app = FastStream(broker)`. Don't use `app.broker`.\n- Prefer the `faststream.Logger` annotation in broker handlers over `print()` (some HTTP-integration examples use `print()` for brevity).\n- Register subscribers/publishers only via `@broker.subscriber(...)` / `@broker.publisher(...)` (or the router equivalents).\n- Fully annotate handler parameters; handlers return `None` unless their return value is published.\n- Don't add decorator options the example doesn't need.\n- Use `Annotated[str, Path()]` for subject path params and `Annotated[..., Context()]` for context access (built-ins: `broker`, `context`, `logger`, `message`).\n- Import only from public packages — never `faststream._internal`.\n\n## Build & serve\n\n- `just docs-serve` — live dev server on port 8000.\n- `just docs-build` — build the static site.\n- `just docs-build-api` — regenerate the API reference.\n- `just docs-check` — the PR gate: strict build of the guides (broken links and anchors fail) plus `docs/check_site.py` (unique titles, canonicals, sitemap, JSON-LD, image `alt`), about 10 s. Run it before pushing a docs change.\n\n## Related skills\n\n- **testing-patterns** — how to mark and run the snippet tests in `tests/docs/`.\n- **code-architecture** — source conventions example code must follow (public API only, no `_internal` imports).\n- **dev-workflow** — environment setup for the docs toolchain.\n",".agents/skills/dev-workflow/SKILL.md":"---\nname: dev-workflow\ndescription: Use when setting up the FastStream dev environment, running tests/linters/static analysis, managing docker brokers, or preparing a PR.\n---\n\n# FastStream Dev Workflow\n\n## Toolchain\n\n`uv` for Python deps + `just` as the task runner + docker compose for brokers. Never use bare `pip`. Run `just` with no args to list all recipes.\n\n- `just init [python-version]` — build the dev environment (default Python 3.10).\n- `uv.lock` travels with `pyproject.toml`: a PR that changes dependencies or the version runs `uv lock` and commits the result. `just lock-check` (a pre-commit hook, so CI too) fails otherwise — CI installs without the lock, so nothing else would notice.\n\n## Docker brokers\n\n- `just up` / `just stop` / `just down` — start / stop / remove all containers (brokers + the `faststream` dev container the test recipes exec into).\n- Per broker: `just kafka-up`, `rabbit-up`, `nats-up`, `redis-up`, `redis-cluster-up` (+ matching `-stop` and `-logs` recipes). MQTT has no local just recipes — it is tested in CI only.\n\n## Test matrix\n\nAll `just test*` recipes execute inside the dev container (`docker compose exec faststream`) — run `just up` first.\n\n| Command | Selection |\n|---|---|\n| `just test [path]` | fast: `not slow and not connected`, parallel `-n auto` |\n| `just test-all [path]` | full suite (`-m \"all\"`) — all brokers must be up |\n| `just test-<broker> [path]` | broker-marked tests excluding `connected` and `slow` (kafka, confluent, rabbit, redis, redis-cluster, nats) |\n| `just test-<broker>-all [path]` | every broker-marked test incl. slow/connected — needs that broker up |\n| `just test-coverage [path]` / `test-coverage-all` | with coverage |\n\nExtra pytest args pass through: `just test tests/brokers/kafka -vv`. Run pytest directly when needed (`uv run pytest ...` — no container required, but bare pytest excludes only `slow` by default) — never via the rtk proxy.\n\n## Lint & static analysis (run before any PR, in this order)\n\n1. `just linter` — runs `ruff format` (rewrites files in place), then `ruff check --exit-non-zero-on-fix` (reports fixable issues without applying them), then codespell (alias: `just lint`). Expect formatting changes in your working tree after running it.\n2. `just mypy` — strict mode over `faststream/` and all of `tests/`; `docs/docs_src/` with the annotation checks relaxed.\n3. `just static-analysis` — mypy + pyright + pyrefly + bandit + semgrep + import-linter + slotscheck (pyright and pyrefly read `tests/mypy` only: the public API has to type check under every checker a user may run; `just import-linter` alone for the import contracts; slotscheck checks that declared `__slots__` take effect); `just zizmor` (security) and `just actionlint` (syntax, expressions, shellcheck over `run:` scripts) separately for GitHub Actions workflows.\n4. `just pre-commit` — pre-commit hooks on modified files (`just pre-commit-all` for the whole tree).\n\n## Docs recipes\n\n- `just docs-serve` — live server; `just docs-build` — static build; `just docs-check` — the docs PR gate (details in the **documentation-writing** skill).\n\n## CI expectations\n\n- CI (`.github/workflows/pr_tests.yaml`): core jobs run everything except `connected` (`-m \"(slow and not connected) or not connected\"`); per-broker jobs run `<broker> and not connected`, plus dedicated jobs with real broker services for `connected` tests.\n- 30s per-test timeout; xdist parallelism — tests must be order-independent.\n- Coverage sources include `faststream/`, `tests/`, `docs/docs_src/`, and `examples/`.\n- `coverage-combine` runs `diff-cover` on PRs: 90% of the lines a PR changes under `faststream/` must be executed by some job of the matrix. The report lands in the job summary.\n\n## Related skills\n\n- **testing-patterns** — markers, base testcases, fixtures.\n- **code-architecture** — source conventions the linters enforce.\n- **documentation-writing** — docs authoring and snippet testing.\n",".agents/skills/code-architecture/SKILL.md":"---\nname: code-architecture\ndescription: Use when writing or modifying FastStream library source code under faststream/ — package layout, broker package anatomy, typing rules, configs, and public API conventions.\n---\n\n# FastStream Code Architecture\n\n## Public vs internal split\n\n- `faststream/_internal/` holds shared machinery: `broker/` (abstract `BrokerUsecase`, registrator, router), `endpoint/`, `di/` (fast-depends integration), `context/`, `configs/`, `logger/`, `testing/`, `cli/`, `fastapi/`, `utils/`.\n- Broker packages (`faststream/kafka/`, `rabbit/`, `nats/`, `redis/`, `confluent/`, `mqtt/`) are thin public layers over `_internal`.\n- Cross-broker public packages: `faststream/middlewares/`, `params/`, `response/`, `specification/`, `message/`, `asgi/`, `opentelemetry/`, `prometheus/`.\n\n**Rule:** implement shared behavior in `_internal/`, expose it through broker packages. User-facing code (docs, examples, error messages) must never import from `faststream._internal`.\n\n## Broker package anatomy\n\nEvery broker package mirrors the same layout. Canonical reference: `faststream/kafka/`.\n\n```\nfaststream/<broker>/\n├── __init__.py        # public exports with explicit __all__\n├── annotations.py     # broker-specific Annotated type aliases\n├── broker/            # broker.py (BrokerUsecase subclass), router.py, registrator.py, logging.py\n├── configs/           # @dataclass(kw_only=True) configs inheriting BrokerConfig\n├── message.py         # StreamMessage subclass\n├── parser.py          # message parser\n├── publisher/         # publisher endpoint + producer.py\n├── subscriber/        # subscriber endpoint (usecase.py; nats/redis split into usecases/)\n├── response.py        # PublishCommand subclasses\n├── security.py        # auth/security helpers\n├── testing.py         # in-memory TestBroker\n└── exceptions.py      # broker-specific exceptions\n```\n\nBrokers also carry optional integration subpackages where supported — kafka has `fastapi/`, `helpers/`, `opentelemetry/`, `prometheus/`, and `schemas/` — follow kafka's structure when adding these to another broker.\n\nA broker package is closed over its driver: `faststream.<broker>` never imports another broker package, nothing outside it imports it, and its driver (`aiokafka`, `confluent_kafka`, `aio_pika`, `nats`, `redis`, `zmqtt`) is imported nowhere else. `just import-linter` checks all three; the contracts live in `[tool.importlinter]` in `pyproject.toml`.\n\n## Feature mirroring\n\nAll brokers expose the same surface: `publish()`, `request()`, `ping()`, `start()`, `stop()`, routers, publishers, message/response types. When adding a feature:\n\n1. Find the closest analogue in another broker (kafka is usually the most complete) and follow its shape and naming.\n2. Keep the public API identical across brokers unless the feature is inherently broker-specific.\n3. Broker-specific features stay in the broker package — don't leak them into `_internal/`.\n\n## Typing\n\n- mypy runs with `strict = true` (see `[tool.mypy]` in `pyproject.toml`): every function fully annotated, no implicit `Optional`, decorators typed. Checked paths: `faststream/`, all of `tests/` and `docs/docs_src/`.\n- Generics are used for broker abstractions: `BrokerUsecase[MsgType, ConnectionType, BrokerConfigType]` (see `faststream/_internal/broker/broker.py`), `BaseMiddleware[PublishCommandType, AnyMsg]`.\n- Import `Callable`, `Awaitable`, `Sequence`, `Mapping` from `collections.abc`; newer typing features (`Self`, `ParamSpec`, `TypedDict`, ...) from `typing_extensions`.\n- Connection kwargs use `TypedDict` (e.g. `KafkaInitKwargs` in `faststream/kafka/broker/broker.py`).\n- Pydantic v1/v2 and Python-version differences go through `faststream/_internal/_compat.py` — never inline version checks elsewhere.\n\n## Configs\n\nConfig classes are `@dataclass(kw_only=True)` inheriting `BrokerConfig` (base in `faststream/_internal/configs/`). Example: `faststream/kafka/configs/broker.py`.\n\n## Public API\n\n- Every `__init__.py` declares `__all__` explicitly.\n- Optional dependencies are guarded with try/except raising an `ImportError` that tells the user which extra to install — see `faststream/kafka/__init__.py`.\n\n## Style\n\n- ruff uses `select = [\"ALL\"]` with curated ignores in `ruff.toml` — don't assume a rule is disabled; run `just linter` to check.\n- Line length 90, double quotes, Google-style docstrings.\n- `just mypy` must pass before a PR.\n- A class that declares `__slots__` needs slotted bases all the way up (`__slots__ = ()` on a `Protocol` or mixin), otherwise instances keep a `__dict__` and the slots do nothing; `just slotscheck` checks it.\n\n### Comments\n\nA comment carries the **why** — the constraint, the surprise, the reason this line is not the obvious one. What the code does is already on the screen.\n\n- **Two lines, maximum.** A caveat needing more than that is usually a docstring, an ADR, or a sign the code should be clearer.\n- **Directly above the line it explains**, so the reader meets the explanation at the moment the code surprises them. A caveat about one assertion belongs over that assertion, not in the docstring header several screens up.\n- **Name the issue** when the code exists because of a reported bug: `# ... which no broker can express (see issue #3056).` The next reader gets the whole investigation for free.\n\n## Related skills\n\n- **testing-patterns** — every source change needs tests following the base-testcase model.\n- **dev-workflow** — full command reference (lint, mypy, docker brokers).\n- **documentation-writing** — user-facing features need docs with tested snippets.\n",".agents/skills/testing-patterns/SKILL.md":"---\nname: testing-patterns\ndescription: Choosing which tests to run after a change, or writing tests under tests/ — base testcases, markers, in-memory vs connected brokers.\n---\n\n# FastStream Testing Patterns\n\n## Run only the test files the change touches\n\nName the specific test files:\n\n```bash\nuv run pytest tests/brokers/kafka/test_misconfigure.py tests/asyncapi/kafka/v3_0_0/test_address.py -m \"not connected\"\n```\n\nThe set is the files that could see the edit, and it stops there. A directory\nor a bare `tests/` is CI's shape: CI runs everything anyway, so a local run\nexists to answer whether *this* edit works, and every test beyond that is time\nspent not finding out.\n\nTo find the files: grep the symbol you changed across `tests/`, and follow the\nnaming — a change to `faststream/<broker>/<endpoint>/<thing>.py` is usually\ncovered by `tests/brokers/<broker>/test_<thing>.py` and\n`tests/asyncapi/<broker>/v*/test_<thing>.py`.\n\nA shared testcase in `tests/brokers/base/` or `tests/asyncapi/base/` is\ninherited by every broker, so editing one does widen the set — but widen it by\nnaming the inheriting files, not by running their directories.\n\n**Reach for `connected` only when the change reaches the wire** — subscribing,\npublishing, acks, bindings, reconnects. A change to a specification, a config,\nor a declaration-time check is decided in memory, in seconds. Broker-backed\nruns take minutes and share state between runs.\n\nWhen a `connected` run does fail, **re-run the failures alone before believing\nthem**. The brokers are shared and accumulate topics, queues and consumer\ngroups; a failure that passes in isolation is the container, not the code. To\ntell them apart, run the same set against the committed tree (`git stash`) and\ncompare counts.\n\n## Commands\n\nRun pytest directly or via just — **never through the rtk proxy**.\n\nDirect pytest needs no container, which is why named files run there. The\n`just test*` recipes run whole suites inside the dev container\n(`docker compose exec faststream`, so `just up` first):\n\n- `just test [path]` — fast selection: `-m \"not slow and not connected\"`, parallel `-n auto`. Takes a path, so it can be pointed at named files.\n- `just test-kafka` / `test-rabbit` / `test-nats` / `test-redis` / `test-redis-cluster` / `test-confluent` — a whole broker, excluding `connected` and `slow`; the `-all` variants add the slow and connected ones (that broker must be up).\n- `just test-all` — everything (`-m \"all\"`).\n\nHeads-up: the pyproject default addopts exclude only `slow` (`-m 'not slow'`) — bare pytest WILL collect `connected` tests, so pass `-m \"not slow and not connected\"` explicitly when no broker is running.\n\nGlobal pytest timeout is 30s per test; the suite runs parallel — keep tests independent and use the `queue` fixture for unique names.\n\n## Markers — strict\n\n`--strict-markers` is enabled; the allowed set is defined in `pyproject.toml` (`kafka`, `confluent`, `rabbit`, `nats`, `redis`, `redis_cluster`, `mqtt`, `slow`, `connected`, `all`, `benchmark`).\n\n- Broker-specific test → its broker mark: `@pytest.mark.kafka()`.\n- Talks to a real broker over the network → add `@pytest.mark.connected()` (excluded by `just test`; bare pytest excludes only `slow` by default).\n- Slow test → `@pytest.mark.slow()` (also excluded by default).\n- Async test → `@pytest.mark.asyncio()`.\n\nMarks pick the CI job, so a wrong one drops a test silently. `just misplaced-marks` (a CI step too) fails on the three shapes that did it before:\n\n- a test under a `<broker>/` directory anywhere in `tests/` without the `<broker>` mark (`redis_cluster` under `redis/cluster/`). A test imported from `docs_src` or `examples` is marked through `pytestmark` in the importing module.\n- a `connected` test whose module is built on one broker package, without that broker's mark. An unmarked test still runs in `test-basic`; a `connected` one runs only in its broker's job, so there it never runs at all.\n- `connected` on a whole `*MemoryTestcaseConfig` class (or its module). When an in-memory class inherits a test that does open a connection, mark that test — in the base testcase if it is inherited.\n\n## Shared base testcases\n\nCross-broker behavior is specified ONCE in `tests/brokers/base/` (`basic.py`, `consume.py`, `publish.py`, `router.py`, `codec.py`, `middlewares.py`, `parser.py`, `requests.py`, `connection.py`, `fastapi.py`, `testclient.py`, ...) and inherited by every broker.\n\nEach broker defines its config in `tests/brokers/<broker>/basic.py`:\n\n```python\nclass KafkaTestcaseConfig(BaseTestcaseConfig):\n    def get_broker(self, apply_types: bool = False, **kwargs: Any) -> KafkaBroker:\n        return KafkaBroker(apply_types=apply_types, **kwargs)\n\n    def get_router(self, **kwargs: Any) -> KafkaRouter:\n        return KafkaRouter(**kwargs)\n\n\nclass KafkaMemoryTestcaseConfig(KafkaTestcaseConfig):\n    def patch_broker(self, *brokers: KafkaBroker, **kwargs: Any) -> TestKafkaBroker:\n        return TestKafkaBroker(*brokers, **kwargs)\n```\n\nTest classes multiply-inherit config + behavior suite:\n\n```python\n@pytest.mark.kafka()\nclass TestKafkaCodec(KafkaMemoryTestcaseConfig, CodecTestcase): ...\n\n\n@pytest.mark.connected()\n@pytest.mark.kafka()\nclass TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ...\n```\n\n**Rule:** new cross-broker behavior goes into a base class in `tests/brokers/base/` so every broker inherits the test. Broker-specific behavior is tested directly in `tests/brokers/<broker>/`.\n\n**Hook surface:** a member of a base testcase earns its place by being overridden in `tests/brokers/<broker>/` — `separator`, `declare_subscriber` and `publish` in `base/address.py` are hooks because MQTT, Kafka and RabbitMQ respell them, and each docstring names who does. What no broker overrides goes inline in the test body, beside the assertion it feeds, with a comment carrying the value it compiles to:\n\n```python\n# subscribe to \"queue.{level}\"\nsubscriber = self.declare_subscriber(\n    broker,\n    f\"{queue}{self.separator}{{level}}\",\n    queue,\n)\n```\n\n## Add a test only when nothing else already breaks on it\n\nBefore writing a new test, ask what already goes red if the change is wrong: another test\nin the suite, or `mypy` running over library code that already exercises the path (a\nregistrator signature checked through `faststream/<broker>/testing.py`, say). If something\nalready catches it, don't add another one — a ticket asking for \"a test per case\" doesn't\noverride this; say which existing check covers the rest instead of writing one that just\nrestates it. Never pin language or stdlib behaviour FastStream doesn't own (a `NamedTuple`\nunpacks, `==` on tuples).\n\nThe name is the behaviour, so a test carries no docstring by default, and an assertion\nthat needs explaining gets a single `#` comment directly over it. Two cases earn one:\n\n- the regression pattern below — the docstring is the issue URL and nothing else;\n- the rare test that is unreadable without it, because what it guards is invisible from\n  the body. `tests/utils/test_lazy_imports.py` runs an import in a subprocess: only the\n  docstring can say what that protects, how it broke before, and why nothing else goes\n  red. If a better name or one comment would do, the test is not this case.\n\n## One equality per behaviour\n\nWhen a test checks one value from several angles — a tuple's fields, a few keys of a\ndict, a length and an element — build the expected shape from `dirty-equals` matchers\nand compare once. The failure then prints the whole shape, and the test reads as a single\nstatement of the behaviour:\n\n```python\n# Claimed entries come first, with their previous deliveries and idle time\nassert received[:2] == [\n    (\"pending_message\", IsInt(ge=1), IsInt(ge=100)),\n    (\"new_message\", 0, 0),\n]\n\nassert snapshot == IsPartialDict({\n    \"delivery_counts\": HasLen(size),\n    \"idle_times\": HasLen(size),\n})\n```\n\nA chain of `assert x[0] ...`, `assert x[1] ...`, or a loop carrying a `found` flag, is this\nshape spelled out one field at a time: collapse it into the one equality. `IsPartialDict`\ntakes a dict literal, so dotted keys and enum values read the same as the config they\nmirror. An equality that already fails on a missing delivery stands alone; the\n`assert event.is_set()` in front of it says nothing more.\n\n## Tests are type checked\n\n`just mypy` runs the same strict config over all of `tests/`, so a test is annotated like library code: every function, fixture and handler has its parameters and return typed.\n\n- **Cross-broker testcases hold the broker as `Any`.** A base class in `tests/brokers/base/`, `tests/asyncapi/base/` and the like subclasses `BaseTestcaseConfig[Any]` and types `broker_class`, `router_class` and friends as `Any` — each broker spells `subscriber()` differently. Such a module also joins the `disallow_untyped_decorators = false` override in `pyproject.toml`; broker-specific tests keep the check.\n- **A value that is wrong on purpose goes through `Any`**, not an ignore: `router: Any = NatsRouter()` before handing it to a Kafka broker, `channel_manager: Any = FakeChannelManager(mock)` for a stand-in. The same goes for private internals (`producer: Any = broker._producer`).\n- **The raw client comes from the public API**: `client = await br.connect()`, never `br._connection`, which is `None`-able.\n- **Narrow with an assertion the test already implies** — `assert message`, `assert isinstance(point, HistogramDataPoint)` — and put a repeated one in a small module-level helper.\n- **An awaitable dropped on purpose is written `_ = ...`**: `_ = tg.start_soon(app.run)`, `_ = await br.publish(..., no_confirm=True)`. A bare statement reads as a forgotten `await`, which is what `unused-awaitable` reports.\n- **`# type: ignore[code]` is for what nothing else expresses**, always with its code and before any `# noqa`: `subscriber(*args, **kwargs)` from `get_subscriber_params()` resolves to `Any` (`untyped-decorator`), a test overriding a base test with other fixtures (`override`), a name redefined on purpose (`no-redef`), a call missing a required argument to prove it raises (`call-arg`).\n\nA typing problem that turns out to live in `faststream/` is fixed there, in its own PR with a case in `tests/mypy/`, not papered over in the test.\n\n## Regression tests\n\nA test defending a fixed bug names the issue by **full URL**, so the case it pins is one click away:\n\n```python\n@pytest.mark.xfail(reason=\"https://github.com/ag2ai/faststream/issues/2513\")\nasync def test_publisher_without_destination(self) -> None:\n    \"\"\"Fixes https://github.com/ag2ai/faststream/issues/2513.\"\"\"\n```\n\nThe URL is the whole docstring — no explanation of the behavior underneath it. `xfail`/`skip` reasons take the same URL. A comment inside the test body follows the **code-architecture** rule: one line, directly over the assertion it explains.\n\nA test written after the fix earns its place by going **red** on the old code — revert the fix, run, restore:\n\n```bash\ngit show <fix-commit> -- faststream/ | git apply -R -\nuv run pytest tests/... -m \"not slow and not connected\"\ngit checkout -- faststream/\n```\n\nThe same run grades the tests already there, and it is how a suite shrinks. Two tests red for one reason are one test: keep the one whose declaration carries more (an escaped brace *beside* a Path parameter over an escaped brace alone), delete the other, and check what a broker already gets from `test_router.py` or `test_path.py` before keeping a third.\n\n## In-memory vs real broker\n\n- Default to the in-memory `TestBroker` (`faststream/<broker>/testing.py`) via a `*MemoryTestcaseConfig` — fast, runs everywhere, no `connected` mark.\n- Use a real broker (plain `*TestcaseConfig` + `@pytest.mark.connected()`) when the behavior depends on actual broker semantics (acks, consumer groups, reconnects). Connection settings come from the `Settings` dataclass in `tests/brokers/<broker>/conftest.py`.\n\n## Fixtures & utilities\n\n- Global fixtures (`tests/conftest.py`): `queue` (unique uuid string), `event` (`asyncio.Event`), `mock` / `async_mock` (function-scoped, reset via teardown), `context`, `runner` (CLI).\n- `tests/marks.py`: conditional skips — `skip_windows`, `skip_macos`, `pydantic_v1`/`pydantic_v2`, `require_aiokafka`, `require_confluent`, `require_aiopika`, `require_redis`, `require_nats`, `require_mqtt`.\n- `tests/tools.py`: `spy_decorator` — wraps a real method with a mock spy (call assertions via `.mock`) while preserving behavior.\n- `tests/mocks.py`: `mock_pydantic_settings_env` for env-driven settings tests.\n- `freezegun` is available as a test dep.\n\n**Never import from a `conftest.py`.** pytest loads conftest modules specially (their fixtures are injected into the collected files), so importing from one — `from .conftest import Settings` or `from tests.brokers.redis.conftest import ...` — can produce a duplicated/mismatched module and confusing collection errors. When conftest and a test file need the same object, declare it in a plain helper module next to them (e.g. `tests/brokers/redis/settings.py`, `basic.py`) and import it from both.\n\n## Related skills\n\n- **dev-workflow** — docker broker management and the full just recipe matrix.\n- **code-architecture** — where the code under test lives and how it's shaped.\n- **documentation-writing** — docs snippets get tests under `tests/docs/`.\n"},"files":{".agents/skills/documentation-writing/SKILL.md":"---\nname: documentation-writing\ndescription: Use when writing or editing documentation under docs/ — page layout, docs_src snippet embedding, snippet tests, link conventions, and FastStream example-code rules.\n---\n\n# FastStream Documentation Writing\n\n## Layout\n\n- Pages: `docs/docs/en/...` (markdown, mkdocs-material). `docs/docs/en/` is the mkdocs `docs_dir`, so images live in `docs/docs/en/assets/`.\n- Runnable code snippets: `docs/docs_src/<topic>/...` (fully excluded from ruff — see `exclude` in `ruff.toml`).\n- Navigation: `docs/docs/navigation_template.txt` — new pages need an entry there. It is rendered into the gitignored `docs/docs/en/SUMMARY.md` (literate-nav).\n- API reference pages are generated by `docs/create_api_docs.py` — never hand-edit them.\n\n## Code snippets\n\nNever inline non-trivial code in markdown. Put a runnable Python file in `docs/docs_src/` and embed it with `mdx_include`:\n\n```python linenums=\"1\" hl_lines=\"10\"\n{!> docs_src/getting_started/publishing/kafka/broker.py !}\n```\n\nNew `docs_src` snippets should get a test under `tests/docs/<topic>/test_<feature>.py` — one test file covers all broker variants via require marks (e.g. `@require_aiokafka`), not one file per broker subdir (see the **testing-patterns** skill). Some examples are intentionally untested; the exemptions are listed in the coverage `omit` list in `pyproject.toml`. `tests/docs/test_snippets_are_tested.py` holds the line: a `docs_src` file that no test imports (directly, or through another tested snippet) fails it unless it is in `omit`. Its `NOT_TESTED_YET` set is the backlog from before the check — take an entry out when you test it, never add one.\n\nA snippet test imports the runnable objects from the module and exercises them in-memory — never a bare `import module` (it tests nothing and trips ruff `F401`):\n\n```python\nimport pytest\n\nfrom faststream.nats import TestApp, TestNatsBroker\n\n\n@pytest.mark.nats()\n@pytest.mark.asyncio()\nasync def test_basic() -> None:\n    from docs.docs_src.nats.js.pull_sub import app, broker, handle\n\n    async with TestNatsBroker(broker), TestApp(app):\n        await broker.publish(\"Hi!\", \"test\")\n        handle.mock.assert_called_once_with(\"Hi!\")\n```\n\n`just mypy` checks `docs/docs_src/` too. The annotation checks are off there (a handler may omit `-> None`, a bare `dict` is fine), every other check is on: a snippet that calls the public API wrongly fails CI. A snippet that is right while mypy disagrees points at an annotation in `faststream/` — fix it there, in its own PR with a case in `tests/mypy/`. A snippet that is wrong on purpose gets a commented per-module `[[tool.mypy.overrides]]` in `pyproject.toml`, not a `# type: ignore` the reader would see.\n\nThe test imports via the `docs.docs_src.…` package path, while the markdown embeds it via the `docs_src/…` path (mdx_include `base_path` is `docs/`). Either way the snippet MUST live under `docs/docs_src/`, not repo-root `docs_src/` — otherwise the embed silently resolves to nothing and the import fails.\n\n## Markdown conventions\n\n- External links: `[text](https://...){.external-link target=\"_blank\"}`; internal links: `{.internal-link}`.\n- Multi-broker pages use pymdownx tabs with one snippet per broker:\n\n  ```markdown\n  === \"AIOKafka\"\n      ...embedded kafka snippet...\n\n  === \"Confluent\"\n      ...embedded confluent snippet...\n\n  === \"RabbitMQ\"\n      ...embedded rabbit snippet...\n  ```\n\n- Admonitions (`!!! note`, `!!! tip`, `!!! warning`) are available and encouraged for caveats.\n\n## Example-code rules (FastStream usage)\n\nAll example code must follow idiomatic FastStream usage:\n\n- Create the broker outside the app and pass it positionally: `app = FastStream(broker)`. Don't use `app.broker`.\n- Prefer the `faststream.Logger` annotation in broker handlers over `print()` (some HTTP-integration examples use `print()` for brevity).\n- Register subscribers/publishers only via `@broker.subscriber(...)` / `@broker.publisher(...)` (or the router equivalents).\n- Fully annotate handler parameters; handlers return `None` unless their return value is published.\n- Don't add decorator options the example doesn't need.\n- Use `Annotated[str, Path()]` for subject path params and `Annotated[..., Context()]` for context access (built-ins: `broker`, `context`, `logger`, `message`).\n- Import only from public packages — never `faststream._internal`.\n\n## Build & serve\n\n- `just docs-serve` — live dev server on port 8000.\n- `just docs-build` — build the static site.\n- `just docs-build-api` — regenerate the API reference.\n- `just docs-check` — the PR gate: strict build of the guides (broken links and anchors fail) plus `docs/check_site.py` (unique titles, canonicals, sitemap, JSON-LD, image `alt`), about 10 s. Run it before pushing a docs change.\n\n## Related skills\n\n- **testing-patterns** — how to mark and run the snippet tests in `tests/docs/`.\n- **code-architecture** — source conventions example code must follow (public API only, no `_internal` imports).\n- **dev-workflow** — environment setup for the docs toolchain.\n",".agents/skills/dev-workflow/SKILL.md":"---\nname: dev-workflow\ndescription: Use when setting up the FastStream dev environment, running tests/linters/static analysis, managing docker brokers, or preparing a PR.\n---\n\n# FastStream Dev Workflow\n\n## Toolchain\n\n`uv` for Python deps + `just` as the task runner + docker compose for brokers. Never use bare `pip`. Run `just` with no args to list all recipes.\n\n- `just init [python-version]` — build the dev environment (default Python 3.10).\n- `uv.lock` travels with `pyproject.toml`: a PR that changes dependencies or the version runs `uv lock` and commits the result. `just lock-check` (a pre-commit hook, so CI too) fails otherwise — CI installs without the lock, so nothing else would notice.\n\n## Docker brokers\n\n- `just up` / `just stop` / `just down` — start / stop / remove all containers (brokers + the `faststream` dev container the test recipes exec into).\n- Per broker: `just kafka-up`, `rabbit-up`, `nats-up`, `redis-up`, `redis-cluster-up` (+ matching `-stop` and `-logs` recipes). MQTT has no local just recipes — it is tested in CI only.\n\n## Test matrix\n\nAll `just test*` recipes execute inside the dev container (`docker compose exec faststream`) — run `just up` first.\n\n| Command | Selection |\n|---|---|\n| `just test [path]` | fast: `not slow and not connected`, parallel `-n auto` |\n| `just test-all [path]` | full suite (`-m \"all\"`) — all brokers must be up |\n| `just test-<broker> [path]` | broker-marked tests excluding `connected` and `slow` (kafka, confluent, rabbit, redis, redis-cluster, nats) |\n| `just test-<broker>-all [path]` | every broker-marked test incl. slow/connected — needs that broker up |\n| `just test-coverage [path]` / `test-coverage-all` | with coverage |\n\nExtra pytest args pass through: `just test tests/brokers/kafka -vv`. Run pytest directly when needed (`uv run pytest ...` — no container required, but bare pytest excludes only `slow` by default) — never via the rtk proxy.\n\n## Lint & static analysis (run before any PR, in this order)\n\n1. `just linter` — runs `ruff format` (rewrites files in place), then `ruff check --exit-non-zero-on-fix` (reports fixable issues without applying them), then codespell (alias: `just lint`). Expect formatting changes in your working tree after running it.\n2. `just mypy` — strict mode over `faststream/` and all of `tests/`; `docs/docs_src/` with the annotation checks relaxed.\n3. `just static-analysis` — mypy + pyright + pyrefly + bandit + semgrep + import-linter + slotscheck (pyright and pyrefly read `tests/mypy` only: the public API has to type check under every checker a user may run; `just import-linter` alone for the import contracts; slotscheck checks that declared `__slots__` take effect); `just zizmor` (security) and `just actionlint` (syntax, expressions, shellcheck over `run:` scripts) separately for GitHub Actions workflows.\n4. `just pre-commit` — pre-commit hooks on modified files (`just pre-commit-all` for the whole tree).\n\n## Docs recipes\n\n- `just docs-serve` — live server; `just docs-build` — static build; `just docs-check` — the docs PR gate (details in the **documentation-writing** skill).\n\n## CI expectations\n\n- CI (`.github/workflows/pr_tests.yaml`): core jobs run everything except `connected` (`-m \"(slow and not connected) or not connected\"`); per-broker jobs run `<broker> and not connected`, plus dedicated jobs with real broker services for `connected` tests.\n- 30s per-test timeout; xdist parallelism — tests must be order-independent.\n- Coverage sources include `faststream/`, `tests/`, `docs/docs_src/`, and `examples/`.\n- `coverage-combine` runs `diff-cover` on PRs: 90% of the lines a PR changes under `faststream/` must be executed by some job of the matrix. The report lands in the job summary.\n\n## Related skills\n\n- **testing-patterns** — markers, base testcases, fixtures.\n- **code-architecture** — source conventions the linters enforce.\n- **documentation-writing** — docs authoring and snippet testing.\n",".agents/skills/code-architecture/SKILL.md":"---\nname: code-architecture\ndescription: Use when writing or modifying FastStream library source code under faststream/ — package layout, broker package anatomy, typing rules, configs, and public API conventions.\n---\n\n# FastStream Code Architecture\n\n## Public vs internal split\n\n- `faststream/_internal/` holds shared machinery: `broker/` (abstract `BrokerUsecase`, registrator, router), `endpoint/`, `di/` (fast-depends integration), `context/`, `configs/`, `logger/`, `testing/`, `cli/`, `fastapi/`, `utils/`.\n- Broker packages (`faststream/kafka/`, `rabbit/`, `nats/`, `redis/`, `confluent/`, `mqtt/`) are thin public layers over `_internal`.\n- Cross-broker public packages: `faststream/middlewares/`, `params/`, `response/`, `specification/`, `message/`, `asgi/`, `opentelemetry/`, `prometheus/`.\n\n**Rule:** implement shared behavior in `_internal/`, expose it through broker packages. User-facing code (docs, examples, error messages) must never import from `faststream._internal`.\n\n## Broker package anatomy\n\nEvery broker package mirrors the same layout. Canonical reference: `faststream/kafka/`.\n\n```\nfaststream/<broker>/\n├── __init__.py        # public exports with explicit __all__\n├── annotations.py     # broker-specific Annotated type aliases\n├── broker/            # broker.py (BrokerUsecase subclass), router.py, registrator.py, logging.py\n├── configs/           # @dataclass(kw_only=True) configs inheriting BrokerConfig\n├── message.py         # StreamMessage subclass\n├── parser.py          # message parser\n├── publisher/         # publisher endpoint + producer.py\n├── subscriber/        # subscriber endpoint (usecase.py; nats/redis split into usecases/)\n├── response.py        # PublishCommand subclasses\n├── security.py        # auth/security helpers\n├── testing.py         # in-memory TestBroker\n└── exceptions.py      # broker-specific exceptions\n```\n\nBrokers also carry optional integration subpackages where supported — kafka has `fastapi/`, `helpers/`, `opentelemetry/`, `prometheus/`, and `schemas/` — follow kafka's structure when adding these to another broker.\n\nA broker package is closed over its driver: `faststream.<broker>` never imports another broker package, nothing outside it imports it, and its driver (`aiokafka`, `confluent_kafka`, `aio_pika`, `nats`, `redis`, `zmqtt`) is imported nowhere else. `just import-linter` checks all three; the contracts live in `[tool.importlinter]` in `pyproject.toml`.\n\n## Feature mirroring\n\nAll brokers expose the same surface: `publish()`, `request()`, `ping()`, `start()`, `stop()`, routers, publishers, message/response types. When adding a feature:\n\n1. Find the closest analogue in another broker (kafka is usually the most complete) and follow its shape and naming.\n2. Keep the public API identical across brokers unless the feature is inherently broker-specific.\n3. Broker-specific features stay in the broker package — don't leak them into `_internal/`.\n\n## Typing\n\n- mypy runs with `strict = true` (see `[tool.mypy]` in `pyproject.toml`): every function fully annotated, no implicit `Optional`, decorators typed. Checked paths: `faststream/`, all of `tests/` and `docs/docs_src/`.\n- Generics are used for broker abstractions: `BrokerUsecase[MsgType, ConnectionType, BrokerConfigType]` (see `faststream/_internal/broker/broker.py`), `BaseMiddleware[PublishCommandType, AnyMsg]`.\n- Import `Callable`, `Awaitable`, `Sequence`, `Mapping` from `collections.abc`; newer typing features (`Self`, `ParamSpec`, `TypedDict`, ...) from `typing_extensions`.\n- Connection kwargs use `TypedDict` (e.g. `KafkaInitKwargs` in `faststream/kafka/broker/broker.py`).\n- Pydantic v1/v2 and Python-version differences go through `faststream/_internal/_compat.py` — never inline version checks elsewhere.\n\n## Configs\n\nConfig classes are `@dataclass(kw_only=True)` inheriting `BrokerConfig` (base in `faststream/_internal/configs/`). Example: `faststream/kafka/configs/broker.py`.\n\n## Public API\n\n- Every `__init__.py` declares `__all__` explicitly.\n- Optional dependencies are guarded with try/except raising an `ImportError` that tells the user which extra to install — see `faststream/kafka/__init__.py`.\n\n## Style\n\n- ruff uses `select = [\"ALL\"]` with curated ignores in `ruff.toml` — don't assume a rule is disabled; run `just linter` to check.\n- Line length 90, double quotes, Google-style docstrings.\n- `just mypy` must pass before a PR.\n- A class that declares `__slots__` needs slotted bases all the way up (`__slots__ = ()` on a `Protocol` or mixin), otherwise instances keep a `__dict__` and the slots do nothing; `just slotscheck` checks it.\n\n### Comments\n\nA comment carries the **why** — the constraint, the surprise, the reason this line is not the obvious one. What the code does is already on the screen.\n\n- **Two lines, maximum.** A caveat needing more than that is usually a docstring, an ADR, or a sign the code should be clearer.\n- **Directly above the line it explains**, so the reader meets the explanation at the moment the code surprises them. A caveat about one assertion belongs over that assertion, not in the docstring header several screens up.\n- **Name the issue** when the code exists because of a reported bug: `# ... which no broker can express (see issue #3056).` The next reader gets the whole investigation for free.\n\n## Related skills\n\n- **testing-patterns** — every source change needs tests following the base-testcase model.\n- **dev-workflow** — full command reference (lint, mypy, docker brokers).\n- **documentation-writing** — user-facing features need docs with tested snippets.\n",".agents/skills/testing-patterns/SKILL.md":"---\nname: testing-patterns\ndescription: Choosing which tests to run after a change, or writing tests under tests/ — base testcases, markers, in-memory vs connected brokers.\n---\n\n# FastStream Testing Patterns\n\n## Run only the test files the change touches\n\nName the specific test files:\n\n```bash\nuv run pytest tests/brokers/kafka/test_misconfigure.py tests/asyncapi/kafka/v3_0_0/test_address.py -m \"not connected\"\n```\n\nThe set is the files that could see the edit, and it stops there. A directory\nor a bare `tests/` is CI's shape: CI runs everything anyway, so a local run\nexists to answer whether *this* edit works, and every test beyond that is time\nspent not finding out.\n\nTo find the files: grep the symbol you changed across `tests/`, and follow the\nnaming — a change to `faststream/<broker>/<endpoint>/<thing>.py` is usually\ncovered by `tests/brokers/<broker>/test_<thing>.py` and\n`tests/asyncapi/<broker>/v*/test_<thing>.py`.\n\nA shared testcase in `tests/brokers/base/` or `tests/asyncapi/base/` is\ninherited by every broker, so editing one does widen the set — but widen it by\nnaming the inheriting files, not by running their directories.\n\n**Reach for `connected` only when the change reaches the wire** — subscribing,\npublishing, acks, bindings, reconnects. A change to a specification, a config,\nor a declaration-time check is decided in memory, in seconds. Broker-backed\nruns take minutes and share state between runs.\n\nWhen a `connected` run does fail, **re-run the failures alone before believing\nthem**. The brokers are shared and accumulate topics, queues and consumer\ngroups; a failure that passes in isolation is the container, not the code. To\ntell them apart, run the same set against the committed tree (`git stash`) and\ncompare counts.\n\n## Commands\n\nRun pytest directly or via just — **never through the rtk proxy**.\n\nDirect pytest needs no container, which is why named files run there. The\n`just test*` recipes run whole suites inside the dev container\n(`docker compose exec faststream`, so `just up` first):\n\n- `just test [path]` — fast selection: `-m \"not slow and not connected\"`, parallel `-n auto`. Takes a path, so it can be pointed at named files.\n- `just test-kafka` / `test-rabbit` / `test-nats` / `test-redis` / `test-redis-cluster` / `test-confluent` — a whole broker, excluding `connected` and `slow`; the `-all` variants add the slow and connected ones (that broker must be up).\n- `just test-all` — everything (`-m \"all\"`).\n\nHeads-up: the pyproject default addopts exclude only `slow` (`-m 'not slow'`) — bare pytest WILL collect `connected` tests, so pass `-m \"not slow and not connected\"` explicitly when no broker is running.\n\nGlobal pytest timeout is 30s per test; the suite runs parallel — keep tests independent and use the `queue` fixture for unique names.\n\n## Markers — strict\n\n`--strict-markers` is enabled; the allowed set is defined in `pyproject.toml` (`kafka`, `confluent`, `rabbit`, `nats`, `redis`, `redis_cluster`, `mqtt`, `slow`, `connected`, `all`, `benchmark`).\n\n- Broker-specific test → its broker mark: `@pytest.mark.kafka()`.\n- Talks to a real broker over the network → add `@pytest.mark.connected()` (excluded by `just test`; bare pytest excludes only `slow` by default).\n- Slow test → `@pytest.mark.slow()` (also excluded by default).\n- Async test → `@pytest.mark.asyncio()`.\n\nMarks pick the CI job, so a wrong one drops a test silently. `just misplaced-marks` (a CI step too) fails on the three shapes that did it before:\n\n- a test under a `<broker>/` directory anywhere in `tests/` without the `<broker>` mark (`redis_cluster` under `redis/cluster/`). A test imported from `docs_src` or `examples` is marked through `pytestmark` in the importing module.\n- a `connected` test whose module is built on one broker package, without that broker's mark. An unmarked test still runs in `test-basic`; a `connected` one runs only in its broker's job, so there it never runs at all.\n- `connected` on a whole `*MemoryTestcaseConfig` class (or its module). When an in-memory class inherits a test that does open a connection, mark that test — in the base testcase if it is inherited.\n\n## Shared base testcases\n\nCross-broker behavior is specified ONCE in `tests/brokers/base/` (`basic.py`, `consume.py`, `publish.py`, `router.py`, `codec.py`, `middlewares.py`, `parser.py`, `requests.py`, `connection.py`, `fastapi.py`, `testclient.py`, ...) and inherited by every broker.\n\nEach broker defines its config in `tests/brokers/<broker>/basic.py`:\n\n```python\nclass KafkaTestcaseConfig(BaseTestcaseConfig):\n    def get_broker(self, apply_types: bool = False, **kwargs: Any) -> KafkaBroker:\n        return KafkaBroker(apply_types=apply_types, **kwargs)\n\n    def get_router(self, **kwargs: Any) -> KafkaRouter:\n        return KafkaRouter(**kwargs)\n\n\nclass KafkaMemoryTestcaseConfig(KafkaTestcaseConfig):\n    def patch_broker(self, *brokers: KafkaBroker, **kwargs: Any) -> TestKafkaBroker:\n        return TestKafkaBroker(*brokers, **kwargs)\n```\n\nTest classes multiply-inherit config + behavior suite:\n\n```python\n@pytest.mark.kafka()\nclass TestKafkaCodec(KafkaMemoryTestcaseConfig, CodecTestcase): ...\n\n\n@pytest.mark.connected()\n@pytest.mark.kafka()\nclass TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ...\n```\n\n**Rule:** new cross-broker behavior goes into a base class in `tests/brokers/base/` so every broker inherits the test. Broker-specific behavior is tested directly in `tests/brokers/<broker>/`.\n\n**Hook surface:** a member of a base testcase earns its place by being overridden in `tests/brokers/<broker>/` — `separator`, `declare_subscriber` and `publish` in `base/address.py` are hooks because MQTT, Kafka and RabbitMQ respell them, and each docstring names who does. What no broker overrides goes inline in the test body, beside the assertion it feeds, with a comment carrying the value it compiles to:\n\n```python\n# subscribe to \"queue.{level}\"\nsubscriber = self.declare_subscriber(\n    broker,\n    f\"{queue}{self.separator}{{level}}\",\n    queue,\n)\n```\n\n## Add a test only when nothing else already breaks on it\n\nBefore writing a new test, ask what already goes red if the change is wrong: another test\nin the suite, or `mypy` running over library code that already exercises the path (a\nregistrator signature checked through `faststream/<broker>/testing.py`, say). If something\nalready catches it, don't add another one — a ticket asking for \"a test per case\" doesn't\noverride this; say which existing check covers the rest instead of writing one that just\nrestates it. Never pin language or stdlib behaviour FastStream doesn't own (a `NamedTuple`\nunpacks, `==` on tuples).\n\nThe name is the behaviour, so a test carries no docstring by default, and an assertion\nthat needs explaining gets a single `#` comment directly over it. Two cases earn one:\n\n- the regression pattern below — the docstring is the issue URL and nothing else;\n- the rare test that is unreadable without it, because what it guards is invisible from\n  the body. `tests/utils/test_lazy_imports.py` runs an import in a subprocess: only the\n  docstring can say what that protects, how it broke before, and why nothing else goes\n  red. If a better name or one comment would do, the test is not this case.\n\n## One equality per behaviour\n\nWhen a test checks one value from several angles — a tuple's fields, a few keys of a\ndict, a length and an element — build the expected shape from `dirty-equals` matchers\nand compare once. The failure then prints the whole shape, and the test reads as a single\nstatement of the behaviour:\n\n```python\n# Claimed entries come first, with their previous deliveries and idle time\nassert received[:2] == [\n    (\"pending_message\", IsInt(ge=1), IsInt(ge=100)),\n    (\"new_message\", 0, 0),\n]\n\nassert snapshot == IsPartialDict({\n    \"delivery_counts\": HasLen(size),\n    \"idle_times\": HasLen(size),\n})\n```\n\nA chain of `assert x[0] ...`, `assert x[1] ...`, or a loop carrying a `found` flag, is this\nshape spelled out one field at a time: collapse it into the one equality. `IsPartialDict`\ntakes a dict literal, so dotted keys and enum values read the same as the config they\nmirror. An equality that already fails on a missing delivery stands alone; the\n`assert event.is_set()` in front of it says nothing more.\n\n## Tests are type checked\n\n`just mypy` runs the same strict config over all of `tests/`, so a test is annotated like library code: every function, fixture and handler has its parameters and return typed.\n\n- **Cross-broker testcases hold the broker as `Any`.** A base class in `tests/brokers/base/`, `tests/asyncapi/base/` and the like subclasses `BaseTestcaseConfig[Any]` and types `broker_class`, `router_class` and friends as `Any` — each broker spells `subscriber()` differently. Such a module also joins the `disallow_untyped_decorators = false` override in `pyproject.toml`; broker-specific tests keep the check.\n- **A value that is wrong on purpose goes through `Any`**, not an ignore: `router: Any = NatsRouter()` before handing it to a Kafka broker, `channel_manager: Any = FakeChannelManager(mock)` for a stand-in. The same goes for private internals (`producer: Any = broker._producer`).\n- **The raw client comes from the public API**: `client = await br.connect()`, never `br._connection`, which is `None`-able.\n- **Narrow with an assertion the test already implies** — `assert message`, `assert isinstance(point, HistogramDataPoint)` — and put a repeated one in a small module-level helper.\n- **An awaitable dropped on purpose is written `_ = ...`**: `_ = tg.start_soon(app.run)`, `_ = await br.publish(..., no_confirm=True)`. A bare statement reads as a forgotten `await`, which is what `unused-awaitable` reports.\n- **`# type: ignore[code]` is for what nothing else expresses**, always with its code and before any `# noqa`: `subscriber(*args, **kwargs)` from `get_subscriber_params()` resolves to `Any` (`untyped-decorator`), a test overriding a base test with other fixtures (`override`), a name redefined on purpose (`no-redef`), a call missing a required argument to prove it raises (`call-arg`).\n\nA typing problem that turns out to live in `faststream/` is fixed there, in its own PR with a case in `tests/mypy/`, not papered over in the test.\n\n## Regression tests\n\nA test defending a fixed bug names the issue by **full URL**, so the case it pins is one click away:\n\n```python\n@pytest.mark.xfail(reason=\"https://github.com/ag2ai/faststream/issues/2513\")\nasync def test_publisher_without_destination(self) -> None:\n    \"\"\"Fixes https://github.com/ag2ai/faststream/issues/2513.\"\"\"\n```\n\nThe URL is the whole docstring — no explanation of the behavior underneath it. `xfail`/`skip` reasons take the same URL. A comment inside the test body follows the **code-architecture** rule: one line, directly over the assertion it explains.\n\nA test written after the fix earns its place by going **red** on the old code — revert the fix, run, restore:\n\n```bash\ngit show <fix-commit> -- faststream/ | git apply -R -\nuv run pytest tests/... -m \"not slow and not connected\"\ngit checkout -- faststream/\n```\n\nThe same run grades the tests already there, and it is how a suite shrinks. Two tests red for one reason are one test: keep the one whose declaration carries more (an escaped brace *beside* a Path parameter over an escaped brace alone), delete the other, and check what a broker already gets from `test_router.py` or `test_path.py` before keeping a third.\n\n## In-memory vs real broker\n\n- Default to the in-memory `TestBroker` (`faststream/<broker>/testing.py`) via a `*MemoryTestcaseConfig` — fast, runs everywhere, no `connected` mark.\n- Use a real broker (plain `*TestcaseConfig` + `@pytest.mark.connected()`) when the behavior depends on actual broker semantics (acks, consumer groups, reconnects). Connection settings come from the `Settings` dataclass in `tests/brokers/<broker>/conftest.py`.\n\n## Fixtures & utilities\n\n- Global fixtures (`tests/conftest.py`): `queue` (unique uuid string), `event` (`asyncio.Event`), `mock` / `async_mock` (function-scoped, reset via teardown), `context`, `runner` (CLI).\n- `tests/marks.py`: conditional skips — `skip_windows`, `skip_macos`, `pydantic_v1`/`pydantic_v2`, `require_aiokafka`, `require_confluent`, `require_aiopika`, `require_redis`, `require_nats`, `require_mqtt`.\n- `tests/tools.py`: `spy_decorator` — wraps a real method with a mock spy (call assertions via `.mock`) while preserving behavior.\n- `tests/mocks.py`: `mock_pydantic_settings_env` for env-driven settings tests.\n- `freezegun` is available as a test dep.\n\n**Never import from a `conftest.py`.** pytest loads conftest modules specially (their fixtures are injected into the collected files), so importing from one — `from .conftest import Settings` or `from tests.brokers.redis.conftest import ...` — can produce a duplicated/mismatched module and confusing collection errors. When conftest and a test file need the same object, declare it in a plain helper module next to them (e.g. `tests/brokers/redis/settings.py`, `basic.py`) and import it from both.\n\n## Related skills\n\n- **dev-workflow** — docker broker management and the full just recipe matrix.\n- **code-architecture** — where the code under test lives and how it's shaped.\n- **documentation-writing** — docs snippets get tests under `tests/docs/`.\n"},"items":[{"name":"SKILL.md","path":".agents/skills/code-architecture/SKILL.md","rawUrl":"https://raw.githubusercontent.com/ag2ai/faststream/HEAD/.agents/skills/code-architecture/SKILL.md","title":"code-architecture","category":"anthropic-skill","format":"markdown","content":"---\nname: code-architecture\ndescription: Use when writing or modifying FastStream library source code under faststream/ — package layout, broker package anatomy, typing rules, configs, and public API conventions.\n---\n\n# FastStream Code Architecture\n\n## Public vs internal split\n\n- `faststream/_internal/` holds shared machinery: `broker/` (abstract `BrokerUsecase`, registrator, router), `endpoint/`, `di/` (fast-depends integration), `context/`, `configs/`, `logger/`, `testing/`, `cli/`, `fastapi/`, `utils/`.\n- Broker packages (`faststream/kafka/`, `rabbit/`, `nats/`, `redis/`, `confluent/`, `mqtt/`) are thin public layers over `_internal`.\n- Cross-broker public packages: `faststream/middlewares/`, `params/`, `response/`, `specification/`, `message/`, `asgi/`, `opentelemetry/`, `prometheus/`.\n\n**Rule:** implement shared behavior in `_internal/`, expose it through broker packages. User-facing code (docs, examples, error messages) must never import from `faststream._internal`.\n\n## Broker package anatomy\n\nEvery broker package mirrors the same layout. Canonical reference: `faststream/kafka/`.\n\n```\nfaststream/<broker>/\n├── __init__.py        # public exports with explicit __all__\n├── annotations.py     # broker-specific Annotated type aliases\n├── broker/            # broker.py (BrokerUsecase subclass), router.py, registrator.py, logging.py\n├── configs/           # @dataclass(kw_only=True) configs inheriting BrokerConfig\n├── message.py         # StreamMessage subclass\n├── parser.py          # message parser\n├── publisher/         # publisher endpoint + producer.py\n├── subscriber/        # subscriber endpoint (usecase.py; nats/redis split into usecases/)\n├── response.py        # PublishCommand subclasses\n├── security.py        # auth/security helpers\n├── testing.py         # in-memory TestBroker\n└── exceptions.py      # broker-specific exceptions\n```\n\nBrokers also carry optional integration subpackages where supported — kafka has `fastapi/`, `helpers/`, `opentelemetry/`, `prometheus/`, and `schemas/` — follow kafka's structure when adding these to another broker.\n\nA broker package is closed over its driver: `faststream.<broker>` never imports another broker package, nothing outside it imports it, and its driver (`aiokafka`, `confluent_kafka`, `aio_pika`, `nats`, `redis`, `zmqtt`) is imported nowhere else. `just import-linter` checks all three; the contracts live in `[tool.importlinter]` in `pyproject.toml`.\n\n## Feature mirroring\n\nAll brokers expose the same surface: `publish()`, `request()`, `ping()`, `start()`, `stop()`, routers, publishers, message/response types. When adding a feature:\n\n1. Find the closest analogue in another broker (kafka is usually the most complete) and follow its shape and naming.\n2. Keep the public API identical across brokers unless the feature is inherently broker-specific.\n3. Broker-specific features stay in the broker package — don't leak them into `_internal/`.\n\n## Typing\n\n- mypy runs with `strict = true` (see `[tool.mypy]` in `pyproject.toml`): every function fully annotated, no implicit `Optional`, decorators typed. Checked paths: `faststream/`, all of `tests/` and `docs/docs_src/`.\n- Generics are used for broker abstractions: `BrokerUsecase[MsgType, ConnectionType, BrokerConfigType]` (see `faststream/_internal/broker/broker.py`), `BaseMiddleware[PublishCommandType, AnyMsg]`.\n- Import `Callable`, `Awaitable`, `Sequence`, `Mapping` from `collections.abc`; newer typing features (`Self`, `ParamSpec`, `TypedDict`, ...) from `typing_extensions`.\n- Connection kwargs use `TypedDict` (e.g. `KafkaInitKwargs` in `faststream/kafka/broker/broker.py`).\n- Pydantic v1/v2 and Python-version differences go through `faststream/_internal/_compat.py` — never inline version checks elsewhere.\n\n## Configs\n\nConfig classes are `@dataclass(kw_only=True)` inheriting `BrokerConfig` (base in `faststream/_internal/configs/`). Example: `faststream/kafka/configs/broker.py`.\n\n## Public API\n\n- Every `__init__.py` declares `__all__` explicitly.\n- Optional dependencies are guarded with try/except raising an `ImportError` that tells the user which extra to install — see `faststream/kafka/__init__.py`.\n\n## Style\n\n- ruff uses `select = [\"ALL\"]` with curated ignores in `ruff.toml` — don't assume a rule is disabled; run `just linter` to check.\n- Line length 90, double quotes, Google-style docstrings.\n- `just mypy` must pass before a PR.\n- A class that declares `__slots__` needs slotted bases all the way up (`__slots__ = ()` on a `Protocol` or mixin), otherwise instances keep a `__dict__` and the slots do nothing; `just slotscheck` checks it.\n\n### Comments\n\nA comment carries the **why** — the constraint, the surprise, the reason this line is not the obvious one. What the code does is already on the screen.\n\n- **Two lines, maximum.** A caveat needing more than that is usually a docstring, an ADR, or a sign the code should be clearer.\n- **Directly above the line it explains**, so the reader meets the explanation at the moment the code surprises them. A caveat about one assertion belongs over that assertion, not in the docstring header several screens up.\n- **Name the issue** when the code exists because of a reported bug: `# ... which no broker can express (see issue #3056).` The next reader gets the whole investigation for free.\n\n## Related skills\n\n- **testing-patterns** — every source change needs tests following the base-testcase model.\n- **dev-workflow** — full command reference (lint, mypy, docker brokers).\n- **documentation-writing** — user-facing features need docs with tested snippets.\n","frontmatter":{"name":"code-architecture","description":"Use when writing or modifying FastStream library source code under faststream/ — package layout, broker package anatomy, typing rules, configs, and public API conventions."},"isInternal":false,"tokens":1374,"sizeBytes":5671},{"name":"SKILL.md","path":".agents/skills/dev-workflow/SKILL.md","rawUrl":"https://raw.githubusercontent.com/ag2ai/faststream/HEAD/.agents/skills/dev-workflow/SKILL.md","title":"dev-workflow","category":"anthropic-skill","format":"markdown","content":"---\nname: dev-workflow\ndescription: Use when setting up the FastStream dev environment, running tests/linters/static analysis, managing docker brokers, or preparing a PR.\n---\n\n# FastStream Dev Workflow\n\n## Toolchain\n\n`uv` for Python deps + `just` as the task runner + docker compose for brokers. Never use bare `pip`. Run `just` with no args to list all recipes.\n\n- `just init [python-version]` — build the dev environment (default Python 3.10).\n- `uv.lock` travels with `pyproject.toml`: a PR that changes dependencies or the version runs `uv lock` and commits the result. `just lock-check` (a pre-commit hook, so CI too) fails otherwise — CI installs without the lock, so nothing else would notice.\n\n## Docker brokers\n\n- `just up` / `just stop` / `just down` — start / stop / remove all containers (brokers + the `faststream` dev container the test recipes exec into).\n- Per broker: `just kafka-up`, `rabbit-up`, `nats-up`, `redis-up`, `redis-cluster-up` (+ matching `-stop` and `-logs` recipes). MQTT has no local just recipes — it is tested in CI only.\n\n## Test matrix\n\nAll `just test*` recipes execute inside the dev container (`docker compose exec faststream`) — run `just up` first.\n\n| Command | Selection |\n|---|---|\n| `just test [path]` | fast: `not slow and not connected`, parallel `-n auto` |\n| `just test-all [path]` | full suite (`-m \"all\"`) — all brokers must be up |\n| `just test-<broker> [path]` | broker-marked tests excluding `connected` and `slow` (kafka, confluent, rabbit, redis, redis-cluster, nats) |\n| `just test-<broker>-all [path]` | every broker-marked test incl. slow/connected — needs that broker up |\n| `just test-coverage [path]` / `test-coverage-all` | with coverage |\n\nExtra pytest args pass through: `just test tests/brokers/kafka -vv`. Run pytest directly when needed (`uv run pytest ...` — no container required, but bare pytest excludes only `slow` by default) — never via the rtk proxy.\n\n## Lint & static analysis (run before any PR, in this order)\n\n1. `just linter` — runs `ruff format` (rewrites files in place), then `ruff check --exit-non-zero-on-fix` (reports fixable issues without applying them), then codespell (alias: `just lint`). Expect formatting changes in your working tree after running it.\n2. `just mypy` — strict mode over `faststream/` and all of `tests/`; `docs/docs_src/` with the annotation checks relaxed.\n3. `just static-analysis` — mypy + pyright + pyrefly + bandit + semgrep + import-linter + slotscheck (pyright and pyrefly read `tests/mypy` only: the public API has to type check under every checker a user may run; `just import-linter` alone for the import contracts; slotscheck checks that declared `__slots__` take effect); `just zizmor` (security) and `just actionlint` (syntax, expressions, shellcheck over `run:` scripts) separately for GitHub Actions workflows.\n4. `just pre-commit` — pre-commit hooks on modified files (`just pre-commit-all` for the whole tree).\n\n## Docs recipes\n\n- `just docs-serve` — live server; `just docs-build` — static build; `just docs-check` — the docs PR gate (details in the **documentation-writing** skill).\n\n## CI expectations\n\n- CI (`.github/workflows/pr_tests.yaml`): core jobs run everything except `connected` (`-m \"(slow and not connected) or not connected\"`); per-broker jobs run `<broker> and not connected`, plus dedicated jobs with real broker services for `connected` tests.\n- 30s per-test timeout; xdist parallelism — tests must be order-independent.\n- Coverage sources include `faststream/`, `tests/`, `docs/docs_src/`, and `examples/`.\n- `coverage-combine` runs `diff-cover` on PRs: 90% of the lines a PR changes under `faststream/` must be executed by some job of the matrix. The report lands in the job summary.\n\n## Related skills\n\n- **testing-patterns** — markers, base testcases, fixtures.\n- **code-architecture** — source conventions the linters enforce.\n- **documentation-writing** — docs authoring and snippet testing.\n","frontmatter":{"name":"dev-workflow","description":"Use when setting up the FastStream dev environment, running tests/linters/static analysis, managing docker brokers, or preparing a PR."},"isInternal":false,"tokens":1003,"sizeBytes":3983},{"name":"SKILL.md","path":".agents/skills/documentation-writing/SKILL.md","rawUrl":"https://raw.githubusercontent.com/ag2ai/faststream/HEAD/.agents/skills/documentation-writing/SKILL.md","title":"documentation-writing","category":"anthropic-skill","format":"markdown","content":"---\nname: documentation-writing\ndescription: Use when writing or editing documentation under docs/ — page layout, docs_src snippet embedding, snippet tests, link conventions, and FastStream example-code rules.\n---\n\n# FastStream Documentation Writing\n\n## Layout\n\n- Pages: `docs/docs/en/...` (markdown, mkdocs-material). `docs/docs/en/` is the mkdocs `docs_dir`, so images live in `docs/docs/en/assets/`.\n- Runnable code snippets: `docs/docs_src/<topic>/...` (fully excluded from ruff — see `exclude` in `ruff.toml`).\n- Navigation: `docs/docs/navigation_template.txt` — new pages need an entry there. It is rendered into the gitignored `docs/docs/en/SUMMARY.md` (literate-nav).\n- API reference pages are generated by `docs/create_api_docs.py` — never hand-edit them.\n\n## Code snippets\n\nNever inline non-trivial code in markdown. Put a runnable Python file in `docs/docs_src/` and embed it with `mdx_include`:\n\n```python linenums=\"1\" hl_lines=\"10\"\n{!> docs_src/getting_started/publishing/kafka/broker.py !}\n```\n\nNew `docs_src` snippets should get a test under `tests/docs/<topic>/test_<feature>.py` — one test file covers all broker variants via require marks (e.g. `@require_aiokafka`), not one file per broker subdir (see the **testing-patterns** skill). Some examples are intentionally untested; the exemptions are listed in the coverage `omit` list in `pyproject.toml`. `tests/docs/test_snippets_are_tested.py` holds the line: a `docs_src` file that no test imports (directly, or through another tested snippet) fails it unless it is in `omit`. Its `NOT_TESTED_YET` set is the backlog from before the check — take an entry out when you test it, never add one.\n\nA snippet test imports the runnable objects from the module and exercises them in-memory — never a bare `import module` (it tests nothing and trips ruff `F401`):\n\n```python\nimport pytest\n\nfrom faststream.nats import TestApp, TestNatsBroker\n\n\n@pytest.mark.nats()\n@pytest.mark.asyncio()\nasync def test_basic() -> None:\n    from docs.docs_src.nats.js.pull_sub import app, broker, handle\n\n    async with TestNatsBroker(broker), TestApp(app):\n        await broker.publish(\"Hi!\", \"test\")\n        handle.mock.assert_called_once_with(\"Hi!\")\n```\n\n`just mypy` checks `docs/docs_src/` too. The annotation checks are off there (a handler may omit `-> None`, a bare `dict` is fine), every other check is on: a snippet that calls the public API wrongly fails CI. A snippet that is right while mypy disagrees points at an annotation in `faststream/` — fix it there, in its own PR with a case in `tests/mypy/`. A snippet that is wrong on purpose gets a commented per-module `[[tool.mypy.overrides]]` in `pyproject.toml`, not a `# type: ignore` the reader would see.\n\nThe test imports via the `docs.docs_src.…` package path, while the markdown embeds it via the `docs_src/…` path (mdx_include `base_path` is `docs/`). Either way the snippet MUST live under `docs/docs_src/`, not repo-root `docs_src/` — otherwise the embed silently resolves to nothing and the import fails.\n\n## Markdown conventions\n\n- External links: `[text](https://...){.external-link target=\"_blank\"}`; internal links: `{.internal-link}`.\n- Multi-broker pages use pymdownx tabs with one snippet per broker:\n\n  ```markdown\n  === \"AIOKafka\"\n      ...embedded kafka snippet...\n\n  === \"Confluent\"\n      ...embedded confluent snippet...\n\n  === \"RabbitMQ\"\n      ...embedded rabbit snippet...\n  ```\n\n- Admonitions (`!!! note`, `!!! tip`, `!!! warning`) are available and encouraged for caveats.\n\n## Example-code rules (FastStream usage)\n\nAll example code must follow idiomatic FastStream usage:\n\n- Create the broker outside the app and pass it positionally: `app = FastStream(broker)`. Don't use `app.broker`.\n- Prefer the `faststream.Logger` annotation in broker handlers over `print()` (some HTTP-integration examples use `print()` for brevity).\n- Register subscribers/publishers only via `@broker.subscriber(...)` / `@broker.publisher(...)` (or the router equivalents).\n- Fully annotate handler parameters; handlers return `None` unless their return value is published.\n- Don't add decorator options the example doesn't need.\n- Use `Annotated[str, Path()]` for subject path params and `Annotated[..., Context()]` for context access (built-ins: `broker`, `context`, `logger`, `message`).\n- Import only from public packages — never `faststream._internal`.\n\n## Build & serve\n\n- `just docs-serve` — live dev server on port 8000.\n- `just docs-build` — build the static site.\n- `just docs-build-api` — regenerate the API reference.\n- `just docs-check` — the PR gate: strict build of the guides (broken links and anchors fail) plus `docs/check_site.py` (unique titles, canonicals, sitemap, JSON-LD, image `alt`), about 10 s. Run it before pushing a docs change.\n\n## Related skills\n\n- **testing-patterns** — how to mark and run the snippet tests in `tests/docs/`.\n- **code-architecture** — source conventions example code must follow (public API only, no `_internal` imports).\n- **dev-workflow** — environment setup for the docs toolchain.\n","frontmatter":{"name":"documentation-writing","description":"Use when writing or editing documentation under docs/ — page layout, docs_src snippet embedding, snippet tests, link conventions, and FastStream example-code rules."},"isInternal":false,"tokens":1236,"sizeBytes":5081},{"name":"SKILL.md","path":".agents/skills/testing-patterns/SKILL.md","rawUrl":"https://raw.githubusercontent.com/ag2ai/faststream/HEAD/.agents/skills/testing-patterns/SKILL.md","title":"testing-patterns","category":"anthropic-skill","format":"markdown","content":"---\nname: testing-patterns\ndescription: Choosing which tests to run after a change, or writing tests under tests/ — base testcases, markers, in-memory vs connected brokers.\n---\n\n# FastStream Testing Patterns\n\n## Run only the test files the change touches\n\nName the specific test files:\n\n```bash\nuv run pytest tests/brokers/kafka/test_misconfigure.py tests/asyncapi/kafka/v3_0_0/test_address.py -m \"not connected\"\n```\n\nThe set is the files that could see the edit, and it stops there. A directory\nor a bare `tests/` is CI's shape: CI runs everything anyway, so a local run\nexists to answer whether *this* edit works, and every test beyond that is time\nspent not finding out.\n\nTo find the files: grep the symbol you changed across `tests/`, and follow the\nnaming — a change to `faststream/<broker>/<endpoint>/<thing>.py` is usually\ncovered by `tests/brokers/<broker>/test_<thing>.py` and\n`tests/asyncapi/<broker>/v*/test_<thing>.py`.\n\nA shared testcase in `tests/brokers/base/` or `tests/asyncapi/base/` is\ninherited by every broker, so editing one does widen the set — but widen it by\nnaming the inheriting files, not by running their directories.\n\n**Reach for `connected` only when the change reaches the wire** — subscribing,\npublishing, acks, bindings, reconnects. A change to a specification, a config,\nor a declaration-time check is decided in memory, in seconds. Broker-backed\nruns take minutes and share state between runs.\n\nWhen a `connected` run does fail, **re-run the failures alone before believing\nthem**. The brokers are shared and accumulate topics, queues and consumer\ngroups; a failure that passes in isolation is the container, not the code. To\ntell them apart, run the same set against the committed tree (`git stash`) and\ncompare counts.\n\n## Commands\n\nRun pytest directly or via just — **never through the rtk proxy**.\n\nDirect pytest needs no container, which is why named files run there. The\n`just test*` recipes run whole suites inside the dev container\n(`docker compose exec faststream`, so `just up` first):\n\n- `just test [path]` — fast selection: `-m \"not slow and not connected\"`, parallel `-n auto`. Takes a path, so it can be pointed at named files.\n- `just test-kafka` / `test-rabbit` / `test-nats` / `test-redis` / `test-redis-cluster` / `test-confluent` — a whole broker, excluding `connected` and `slow`; the `-all` variants add the slow and connected ones (that broker must be up).\n- `just test-all` — everything (`-m \"all\"`).\n\nHeads-up: the pyproject default addopts exclude only `slow` (`-m 'not slow'`) — bare pytest WILL collect `connected` tests, so pass `-m \"not slow and not connected\"` explicitly when no broker is running.\n\nGlobal pytest timeout is 30s per test; the suite runs parallel — keep tests independent and use the `queue` fixture for unique names.\n\n## Markers — strict\n\n`--strict-markers` is enabled; the allowed set is defined in `pyproject.toml` (`kafka`, `confluent`, `rabbit`, `nats`, `redis`, `redis_cluster`, `mqtt`, `slow`, `connected`, `all`, `benchmark`).\n\n- Broker-specific test → its broker mark: `@pytest.mark.kafka()`.\n- Talks to a real broker over the network → add `@pytest.mark.connected()` (excluded by `just test`; bare pytest excludes only `slow` by default).\n- Slow test → `@pytest.mark.slow()` (also excluded by default).\n- Async test → `@pytest.mark.asyncio()`.\n\nMarks pick the CI job, so a wrong one drops a test silently. `just misplaced-marks` (a CI step too) fails on the three shapes that did it before:\n\n- a test under a `<broker>/` directory anywhere in `tests/` without the `<broker>` mark (`redis_cluster` under `redis/cluster/`). A test imported from `docs_src` or `examples` is marked through `pytestmark` in the importing module.\n- a `connected` test whose module is built on one broker package, without that broker's mark. An unmarked test still runs in `test-basic`; a `connected` one runs only in its broker's job, so there it never runs at all.\n- `connected` on a whole `*MemoryTestcaseConfig` class (or its module). When an in-memory class inherits a test that does open a connection, mark that test — in the base testcase if it is inherited.\n\n## Shared base testcases\n\nCross-broker behavior is specified ONCE in `tests/brokers/base/` (`basic.py`, `consume.py`, `publish.py`, `router.py`, `codec.py`, `middlewares.py`, `parser.py`, `requests.py`, `connection.py`, `fastapi.py`, `testclient.py`, ...) and inherited by every broker.\n\nEach broker defines its config in `tests/brokers/<broker>/basic.py`:\n\n```python\nclass KafkaTestcaseConfig(BaseTestcaseConfig):\n    def get_broker(self, apply_types: bool = False, **kwargs: Any) -> KafkaBroker:\n        return KafkaBroker(apply_types=apply_types, **kwargs)\n\n    def get_router(self, **kwargs: Any) -> KafkaRouter:\n        return KafkaRouter(**kwargs)\n\n\nclass KafkaMemoryTestcaseConfig(KafkaTestcaseConfig):\n    def patch_broker(self, *brokers: KafkaBroker, **kwargs: Any) -> TestKafkaBroker:\n        return TestKafkaBroker(*brokers, **kwargs)\n```\n\nTest classes multiply-inherit config + behavior suite:\n\n```python\n@pytest.mark.kafka()\nclass TestKafkaCodec(KafkaMemoryTestcaseConfig, CodecTestcase): ...\n\n\n@pytest.mark.connected()\n@pytest.mark.kafka()\nclass TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ...\n```\n\n**Rule:** new cross-broker behavior goes into a base class in `tests/brokers/base/` so every broker inherits the test. Broker-specific behavior is tested directly in `tests/brokers/<broker>/`.\n\n**Hook surface:** a member of a base testcase earns its place by being overridden in `tests/brokers/<broker>/` — `separator`, `declare_subscriber` and `publish` in `base/address.py` are hooks because MQTT, Kafka and RabbitMQ respell them, and each docstring names who does. What no broker overrides goes inline in the test body, beside the assertion it feeds, with a comment carrying the value it compiles to:\n\n```python\n# subscribe to \"queue.{level}\"\nsubscriber = self.declare_subscriber(\n    broker,\n    f\"{queue}{self.separator}{{level}}\",\n    queue,\n)\n```\n\n## Add a test only when nothing else already breaks on it\n\nBefore writing a new test, ask what already goes red if the change is wrong: another test\nin the suite, or `mypy` running over library code that already exercises the path (a\nregistrator signature checked through `faststream/<broker>/testing.py`, say). If something\nalready catches it, don't add another one — a ticket asking for \"a test per case\" doesn't\noverride this; say which existing check covers the rest instead of writing one that just\nrestates it. Never pin language or stdlib behaviour FastStream doesn't own (a `NamedTuple`\nunpacks, `==` on tuples).\n\nThe name is the behaviour, so a test carries no docstring by default, and an assertion\nthat needs explaining gets a single `#` comment directly over it. Two cases earn one:\n\n- the regression pattern below — the docstring is the issue URL and nothing else;\n- the rare test that is unreadable without it, because what it guards is invisible from\n  the body. `tests/utils/test_lazy_imports.py` runs an import in a subprocess: only the\n  docstring can say what that protects, how it broke before, and why nothing else goes\n  red. If a better name or one comment would do, the test is not this case.\n\n## One equality per behaviour\n\nWhen a test checks one value from several angles — a tuple's fields, a few keys of a\ndict, a length and an element — build the expected shape from `dirty-equals` matchers\nand compare once. The failure then prints the whole shape, and the test reads as a single\nstatement of the behaviour:\n\n```python\n# Claimed entries come first, with their previous deliveries and idle time\nassert received[:2] == [\n    (\"pending_message\", IsInt(ge=1), IsInt(ge=100)),\n    (\"new_message\", 0, 0),\n]\n\nassert snapshot == IsPartialDict({\n    \"delivery_counts\": HasLen(size),\n    \"idle_times\": HasLen(size),\n})\n```\n\nA chain of `assert x[0] ...`, `assert x[1] ...`, or a loop carrying a `found` flag, is this\nshape spelled out one field at a time: collapse it into the one equality. `IsPartialDict`\ntakes a dict literal, so dotted keys and enum values read the same as the config they\nmirror. An equality that already fails on a missing delivery stands alone; the\n`assert event.is_set()` in front of it says nothing more.\n\n## Tests are type checked\n\n`just mypy` runs the same strict config over all of `tests/`, so a test is annotated like library code: every function, fixture and handler has its parameters and return typed.\n\n- **Cross-broker testcases hold the broker as `Any`.** A base class in `tests/brokers/base/`, `tests/asyncapi/base/` and the like subclasses `BaseTestcaseConfig[Any]` and types `broker_class`, `router_class` and friends as `Any` — each broker spells `subscriber()` differently. Such a module also joins the `disallow_untyped_decorators = false` override in `pyproject.toml`; broker-specific tests keep the check.\n- **A value that is wrong on purpose goes through `Any`**, not an ignore: `router: Any = NatsRouter()` before handing it to a Kafka broker, `channel_manager: Any = FakeChannelManager(mock)` for a stand-in. The same goes for private internals (`producer: Any = broker._producer`).\n- **The raw client comes from the public API**: `client = await br.connect()`, never `br._connection`, which is `None`-able.\n- **Narrow with an assertion the test already implies** — `assert message`, `assert isinstance(point, HistogramDataPoint)` — and put a repeated one in a small module-level helper.\n- **An awaitable dropped on purpose is written `_ = ...`**: `_ = tg.start_soon(app.run)`, `_ = await br.publish(..., no_confirm=True)`. A bare statement reads as a forgotten `await`, which is what `unused-awaitable` reports.\n- **`# type: ignore[code]` is for what nothing else expresses**, always with its code and before any `# noqa`: `subscriber(*args, **kwargs)` from `get_subscriber_params()` resolves to `Any` (`untyped-decorator`), a test overriding a base test with other fixtures (`override`), a name redefined on purpose (`no-redef`), a call missing a required argument to prove it raises (`call-arg`).\n\nA typing problem that turns out to live in `faststream/` is fixed there, in its own PR with a case in `tests/mypy/`, not papered over in the test.\n\n## Regression tests\n\nA test defending a fixed bug names the issue by **full URL**, so the case it pins is one click away:\n\n```python\n@pytest.mark.xfail(reason=\"https://github.com/ag2ai/faststream/issues/2513\")\nasync def test_publisher_without_destination(self) -> None:\n    \"\"\"Fixes https://github.com/ag2ai/faststream/issues/2513.\"\"\"\n```\n\nThe URL is the whole docstring — no explanation of the behavior underneath it. `xfail`/`skip` reasons take the same URL. A comment inside the test body follows the **code-architecture** rule: one line, directly over the assertion it explains.\n\nA test written after the fix earns its place by going **red** on the old code — revert the fix, run, restore:\n\n```bash\ngit show <fix-commit> -- faststream/ | git apply -R -\nuv run pytest tests/... -m \"not slow and not connected\"\ngit checkout -- faststream/\n```\n\nThe same run grades the tests already there, and it is how a suite shrinks. Two tests red for one reason are one test: keep the one whose declaration carries more (an escaped brace *beside* a Path parameter over an escaped brace alone), delete the other, and check what a broker already gets from `test_router.py` or `test_path.py` before keeping a third.\n\n## In-memory vs real broker\n\n- Default to the in-memory `TestBroker` (`faststream/<broker>/testing.py`) via a `*MemoryTestcaseConfig` — fast, runs everywhere, no `connected` mark.\n- Use a real broker (plain `*TestcaseConfig` + `@pytest.mark.connected()`) when the behavior depends on actual broker semantics (acks, consumer groups, reconnects). Connection settings come from the `Settings` dataclass in `tests/brokers/<broker>/conftest.py`.\n\n## Fixtures & utilities\n\n- Global fixtures (`tests/conftest.py`): `queue` (unique uuid string), `event` (`asyncio.Event`), `mock` / `async_mock` (function-scoped, reset via teardown), `context`, `runner` (CLI).\n- `tests/marks.py`: conditional skips — `skip_windows`, `skip_macos`, `pydantic_v1`/`pydantic_v2`, `require_aiokafka`, `require_confluent`, `require_aiopika`, `require_redis`, `require_nats`, `require_mqtt`.\n- `tests/tools.py`: `spy_decorator` — wraps a real method with a mock spy (call assertions via `.mock`) while preserving behavior.\n- `tests/mocks.py`: `mock_pydantic_settings_env` for env-driven settings tests.\n- `freezegun` is available as a test dep.\n\n**Never import from a `conftest.py`.** pytest loads conftest modules specially (their fixtures are injected into the collected files), so importing from one — `from .conftest import Settings` or `from tests.brokers.redis.conftest import ...` — can produce a duplicated/mismatched module and confusing collection errors. When conftest and a test file need the same object, declare it in a plain helper module next to them (e.g. `tests/brokers/redis/settings.py`, `basic.py`) and import it from both.\n\n## Related skills\n\n- **dev-workflow** — docker broker management and the full just recipe matrix.\n- **code-architecture** — where the code under test lives and how it's shaped.\n- **documentation-writing** — docs snippets get tests under `tests/docs/`.\n","frontmatter":{"name":"testing-patterns","description":"Choosing which tests to run after a change, or writing tests under tests/ — base testcases, markers, in-memory vs connected brokers."},"isInternal":false,"tokens":3278,"sizeBytes":13372}],"systemPromptSnippet":"<yakaai_skills repo=\"ag2ai/faststream\">\n<!-- File: .agents/skills/code-architecture/SKILL.md (Tokens: ~1374 | Category: anthropic-skill) -->\n---\nname: code-architecture\ndescription: Use when writing or modifying FastStream library source code under faststream/ — package layout, broker package anatomy, typing rules, configs, and public API conventions.\n---\n\n# FastStream Code Architecture\n\n## Public vs internal split\n\n- `faststream/_internal/` holds shared machinery: `broker/` (abstract `BrokerUsecase`, registrator, router), `endpoint/`, `di/` (fast-depends integration), `context/`, `configs/`, `logger/`, `testing/`, `cli/`, `fastapi/`, `utils/`.\n- Broker packages (`faststream/kafka/`, `rabbit/`, `nats/`, `redis/`, `confluent/`, `mqtt/`) are thin public layers over `_internal`.\n- Cross-broker public packages: `faststream/middlewares/`, `params/`, `response/`, `specification/`, `message/`, `asgi/`, `opentelemetry/`, `prometheus/`.\n\n**Rule:** implement shared behavior in `_internal/`, expose it through broker packages. User-facing code (docs, examples, error messages) must never import from `faststream._internal`.\n\n## Broker package anatomy\n\nEvery broker package mirrors the same layout. Canonical reference: `faststream/kafka/`.\n\n```\nfaststream/<broker>/\n├── __init__.py        # public exports with explicit __all__\n├── annotations.py     # broker-specific Annotated type aliases\n├── broker/            # broker.py (BrokerUsecase subclass), router.py, registrator.py, logging.py\n├── configs/           # @dataclass(kw_only=True) configs inheriting BrokerConfig\n├── message.py         # StreamMessage subclass\n├── parser.py          # message parser\n├── publisher/         # publisher endpoint + producer.py\n├── subscriber/        # subscriber endpoint (usecase.py; nats/redis split into usecases/)\n├── response.py        # PublishCommand subclasses\n├── security.py        # auth/security helpers\n├── testing.py         # in-memory TestBroker\n└── exceptions.py      # broker-specific exceptions\n```\n\nBrokers also carry optional integration subpackages where supported — kafka has `fastapi/`, `helpers/`, `opentelemetry/`, `prometheus/`, and `schemas/` — follow kafka's structure when adding these to another broker.\n\nA broker package is closed over its driver: `faststream.<broker>` never imports another broker package, nothing outside it imports it, and its driver (`aiokafka`, `confluent_kafka`, `aio_pika`, `nats`, `redis`, `zmqtt`) is imported nowhere else. `just import-linter` checks all three; the contracts live in `[tool.importlinter]` in `pyproject.toml`.\n\n## Feature mirroring\n\nAll brokers expose the same surface: `publish()`, `request()`, `ping()`, `start()`, `stop()`, routers, publishers, message/response types. When adding a feature:\n\n1. Find the closest analogue in another broker (kafka is usually the most complete) and follow its shape and naming.\n2. Keep the public API identical across brokers unless the feature is inherently broker-specific.\n3. Broker-specific features stay in the broker package — don't leak them into `_internal/`.\n\n## Typing\n\n- mypy runs with `strict = true` (see `[tool.mypy]` in `pyproject.toml`): every function fully annotated, no implicit `Optional`, decorators typed. Checked paths: `faststream/`, all of `tests/` and `docs/docs_src/`.\n- Generics are used for broker abstractions: `BrokerUsecase[MsgType, ConnectionType, BrokerConfigType]` (see `faststream/_internal/broker/broker.py`), `BaseMiddleware[PublishCommandType, AnyMsg]`.\n- Import `Callable`, `Awaitable`, `Sequence`, `Mapping` from `collections.abc`; newer typing features (`Self`, `ParamSpec`, `TypedDict`, ...) from `typing_extensions`.\n- Connection kwargs use `TypedDict` (e.g. `KafkaInitKwargs` in `faststream/kafka/broker/broker.py`).\n- Pydantic v1/v2 and Python-version differences go through `faststream/_internal/_compat.py` — never inline version checks elsewhere.\n\n## Configs\n\nConfig classes are `@dataclass(kw_only=True)` inheriting `BrokerConfig` (base in `faststream/_internal/configs/`). Example: `faststream/kafka/configs/broker.py`.\n\n## Public API\n\n- Every `__init__.py` declares `__all__` explicitly.\n- Optional dependencies are guarded with try/except raising an `ImportError` that tells the user which extra to install — see `faststream/kafka/__init__.py`.\n\n## Style\n\n- ruff uses `select = [\"ALL\"]` with curated ignores in `ruff.toml` — don't assume a rule is disabled; run `just linter` to check.\n- Line length 90, double quotes, Google-style docstrings.\n- `just mypy` must pass before a PR.\n- A class that declares `__slots__` needs slotted bases all the way up (`__slots__ = ()` on a `Protocol` or mixin), otherwise instances keep a `__dict__` and the slots do nothing; `just slotscheck` checks it.\n\n### Comments\n\nA comment carries the **why** — the constraint, the surprise, the reason this line is not the obvious one. What the code does is already on the screen.\n\n- **Two lines, maximum.** A caveat needing more than that is usually a docstring, an ADR, or a sign the code should be clearer.\n- **Directly above the line it explains**, so the reader meets the explanation at the moment the code surprises them. A caveat about one assertion belongs over that assertion, not in the docstring header several screens up.\n- **Name the issue** when the code exists because of a reported bug: `# ... which no broker can express (see issue #3056).` The next reader gets the whole investigation for free.\n\n## Related skills\n\n- **testing-patterns** — every source change needs tests following the base-testcase model.\n- **dev-workflow** — full command reference (lint, mypy, docker brokers).\n- **documentation-writing** — user-facing features need docs with tested snippets.\n\n<!-- File: .agents/skills/dev-workflow/SKILL.md (Tokens: ~1003 | Category: anthropic-skill) -->\n---\nname: dev-workflow\ndescription: Use when setting up the FastStream dev environment, running tests/linters/static analysis, managing docker brokers, or preparing a PR.\n---\n\n# FastStream Dev Workflow\n\n## Toolchain\n\n`uv` for Python deps + `just` as the task runner + docker compose for brokers. Never use bare `pip`. Run `just` with no args to list all recipes.\n\n- `just init [python-version]` — build the dev environment (default Python 3.10).\n- `uv.lock` travels with `pyproject.toml`: a PR that changes dependencies or the version runs `uv lock` and commits the result. `just lock-check` (a pre-commit hook, so CI too) fails otherwise — CI installs without the lock, so nothing else would notice.\n\n## Docker brokers\n\n- `just up` / `just stop` / `just down` — start / stop / remove all containers (brokers + the `faststream` dev container the test recipes exec into).\n- Per broker: `just kafka-up`, `rabbit-up`, `nats-up`, `redis-up`, `redis-cluster-up` (+ matching `-stop` and `-logs` recipes). MQTT has no local just recipes — it is tested in CI only.\n\n## Test matrix\n\nAll `just test*` recipes execute inside the dev container (`docker compose exec faststream`) — run `just up` first.\n\n| Command | Selection |\n|---|---|\n| `just test [path]` | fast: `not slow and not connected`, parallel `-n auto` |\n| `just test-all [path]` | full suite (`-m \"all\"`) — all brokers must be up |\n| `just test-<broker> [path]` | broker-marked tests excluding `connected` and `slow` (kafka, confluent, rabbit, redis, redis-cluster, nats) |\n| `just test-<broker>-all [path]` | every broker-marked test incl. slow/connected — needs that broker up |\n| `just test-coverage [path]` / `test-coverage-all` | with coverage |\n\nExtra pytest args pass through: `just test tests/brokers/kafka -vv`. Run pytest directly when needed (`uv run pytest ...` — no container required, but bare pytest excludes only `slow` by default) — never via the rtk proxy.\n\n## Lint & static analysis (run before any PR, in this order)\n\n1. `just linter` — runs `ruff format` (rewrites files in place), then `ruff check --exit-non-zero-on-fix` (reports fixable issues without applying them), then codespell (alias: `just lint`). Expect formatting changes in your working tree after running it.\n2. `just mypy` — strict mode over `faststream/` and all of `tests/`; `docs/docs_src/` with the annotation checks relaxed.\n3. `just static-analysis` — mypy + pyright + pyrefly + bandit + semgrep + import-linter + slotscheck (pyright and pyrefly read `tests/mypy` only: the public API has to type check under every checker a user may run; `just import-linter` alone for the import contracts; slotscheck checks that declared `__slots__` take effect); `just zizmor` (security) and `just actionlint` (syntax, expressions, shellcheck over `run:` scripts) separately for GitHub Actions workflows.\n4. `just pre-commit` — pre-commit hooks on modified files (`just pre-commit-all` for the whole tree).\n\n## Docs recipes\n\n- `just docs-serve` — live server; `just docs-build` — static build; `just docs-check` — the docs PR gate (details in the **documentation-writing** skill).\n\n## CI expectations\n\n- CI (`.github/workflows/pr_tests.yaml`): core jobs run everything except `connected` (`-m \"(slow and not connected) or not connected\"`); per-broker jobs run `<broker> and not connected`, plus dedicated jobs with real broker services for `connected` tests.\n- 30s per-test timeout; xdist parallelism — tests must be order-independent.\n- Coverage sources include `faststream/`, `tests/`, `docs/docs_src/`, and `examples/`.\n- `coverage-combine` runs `diff-cover` on PRs: 90% of the lines a PR changes under `faststream/` must be executed by some job of the matrix. The report lands in the job summary.\n\n## Related skills\n\n- **testing-patterns** — markers, base testcases, fixtures.\n- **code-architecture** — source conventions the linters enforce.\n- **documentation-writing** — docs authoring and snippet testing.\n\n<!-- File: .agents/skills/documentation-writing/SKILL.md (Tokens: ~1236 | Category: anthropic-skill) -->\n---\nname: documentation-writing\ndescription: Use when writing or editing documentation under docs/ — page layout, docs_src snippet embedding, snippet tests, link conventions, and FastStream example-code rules.\n---\n\n# FastStream Documentation Writing\n\n## Layout\n\n- Pages: `docs/docs/en/...` (markdown, mkdocs-material). `docs/docs/en/` is the mkdocs `docs_dir`, so images live in `docs/docs/en/assets/`.\n- Runnable code snippets: `docs/docs_src/<topic>/...` (fully excluded from ruff — see `exclude` in `ruff.toml`).\n- Navigation: `docs/docs/navigation_template.txt` — new pages need an entry there. It is rendered into the gitignored `docs/docs/en/SUMMARY.md` (literate-nav).\n- API reference pages are generated by `docs/create_api_docs.py` — never hand-edit them.\n\n## Code snippets\n\nNever inline non-trivial code in markdown. Put a runnable Python file in `docs/docs_src/` and embed it with `mdx_include`:\n\n```python linenums=\"1\" hl_lines=\"10\"\n{!> docs_src/getting_started/publishing/kafka/broker.py !}\n```\n\nNew `docs_src` snippets should get a test under `tests/docs/<topic>/test_<feature>.py` — one test file covers all broker variants via require marks (e.g. `@require_aiokafka`), not one file per broker subdir (see the **testing-patterns** skill). Some examples are intentionally untested; the exemptions are listed in the coverage `omit` list in `pyproject.toml`. `tests/docs/test_snippets_are_tested.py` holds the line: a `docs_src` file that no test imports (directly, or through another tested snippet) fails it unless it is in `omit`. Its `NOT_TESTED_YET` set is the backlog from before the check — take an entry out when you test it, never add one.\n\nA snippet test imports the runnable objects from the module and exercises them in-memory — never a bare `import module` (it tests nothing and trips ruff `F401`):\n\n```python\nimport pytest\n\nfrom faststream.nats import TestApp, TestNatsBroker\n\n\n@pytest.mark.nats()\n@pytest.mark.asyncio()\nasync def test_basic() -> None:\n    from docs.docs_src.nats.js.pull_sub import app, broker, handle\n\n    async with TestNatsBroker(broker), TestApp(app):\n        await broker.publish(\"Hi!\", \"test\")\n        handle.mock.assert_called_once_with(\"Hi!\")\n```\n\n`just mypy` checks `docs/docs_src/` too. The annotation checks are off there (a handler may omit `-> None`, a bare `dict` is fine), every other check is on: a snippet that calls the public API wrongly fails CI. A snippet that is right while mypy disagrees points at an annotation in `faststream/` — fix it there, in its own PR with a case in `tests/mypy/`. A snippet that is wrong on purpose gets a commented per-module `[[tool.mypy.overrides]]` in `pyproject.toml`, not a `# type: ignore` the reader would see.\n\nThe test imports via the `docs.docs_src.…` package path, while the markdown embeds it via the `docs_src/…` path (mdx_include `base_path` is `docs/`). Either way the snippet MUST live under `docs/docs_src/`, not repo-root `docs_src/` — otherwise the embed silently resolves to nothing and the import fails.\n\n## Markdown conventions\n\n- External links: `[text](https://...){.external-link target=\"_blank\"}`; internal links: `{.internal-link}`.\n- Multi-broker pages use pymdownx tabs with one snippet per broker:\n\n  ```markdown\n  === \"AIOKafka\"\n      ...embedded kafka snippet...\n\n  === \"Confluent\"\n      ...embedded confluent snippet...\n\n  === \"RabbitMQ\"\n      ...embedded rabbit snippet...\n  ```\n\n- Admonitions (`!!! note`, `!!! tip`, `!!! warning`) are available and encouraged for caveats.\n\n## Example-code rules (FastStream usage)\n\nAll example code must follow idiomatic FastStream usage:\n\n- Create the broker outside the app and pass it positionally: `app = FastStream(broker)`. Don't use `app.broker`.\n- Prefer the `faststream.Logger` annotation in broker handlers over `print()` (some HTTP-integration examples use `print()` for brevity).\n- Register subscribers/publishers only via `@broker.subscriber(...)` / `@broker.publisher(...)` (or the router equivalents).\n- Fully annotate handler parameters; handlers return `None` unless their return value is published.\n- Don't add decorator options the example doesn't need.\n- Use `Annotated[str, Path()]` for subject path params and `Annotated[..., Context()]` for context access (built-ins: `broker`, `context`, `logger`, `message`).\n- Import only from public packages — never `faststream._internal`.\n\n## Build & serve\n\n- `just docs-serve` — live dev server on port 8000.\n- `just docs-build` — build the static site.\n- `just docs-build-api` — regenerate the API reference.\n- `just docs-check` — the PR gate: strict build of the guides (broken links and anchors fail) plus `docs/check_site.py` (unique titles, canonicals, sitemap, JSON-LD, image `alt`), about 10 s. Run it before pushing a docs change.\n\n## Related skills\n\n- **testing-patterns** — how to mark and run the snippet tests in `tests/docs/`.\n- **code-architecture** — source conventions example code must follow (public API only, no `_internal` imports).\n- **dev-workflow** — environment setup for the docs toolchain.\n\n<!-- File: .agents/skills/testing-patterns/SKILL.md (Tokens: ~3278 | Category: anthropic-skill) -->\n---\nname: testing-patterns\ndescription: Choosing which tests to run after a change, or writing tests under tests/ — base testcases, markers, in-memory vs connected brokers.\n---\n\n# FastStream Testing Patterns\n\n## Run only the test files the change touches\n\nName the specific test files:\n\n```bash\nuv run pytest tests/brokers/kafka/test_misconfigure.py tests/asyncapi/kafka/v3_0_0/test_address.py -m \"not connected\"\n```\n\nThe set is the files that could see the edit, and it stops there. A directory\nor a bare `tests/` is CI's shape: CI runs everything anyway, so a local run\nexists to answer whether *this* edit works, and every test beyond that is time\nspent not finding out.\n\nTo find the files: grep the symbol you changed across `tests/`, and follow the\nnaming — a change to `faststream/<broker>/<endpoint>/<thing>.py` is usually\ncovered by `tests/brokers/<broker>/test_<thing>.py` and\n`tests/asyncapi/<broker>/v*/test_<thing>.py`.\n\nA shared testcase in `tests/brokers/base/` or `tests/asyncapi/base/` is\ninherited by every broker, so editing one does widen the set — but widen it by\nnaming the inheriting files, not by running their directories.\n\n**Reach for `connected` only when the change reaches the wire** — subscribing,\npublishing, acks, bindings, reconnects. A change to a specification, a config,\nor a declaration-time check is decided in memory, in seconds. Broker-backed\nruns take minutes and share state between runs.\n\nWhen a `connected` run does fail, **re-run the failures alone before believing\nthem**. The brokers are shared and accumulate topics, queues and consumer\ngroups; a failure that passes in isolation is the container, not the code. To\ntell them apart, run the same set against the committed tree (`git stash`) and\ncompare counts.\n\n## Commands\n\nRun pytest directly or via just — **never through the rtk proxy**.\n\nDirect pytest needs no container, which is why named files run there. The\n`just test*` recipes run whole suites inside the dev container\n(`docker compose exec faststream`, so `just up` first):\n\n- `just test [path]` — fast selection: `-m \"not slow and not connected\"`, parallel `-n auto`. Takes a path, so it can be pointed at named files.\n- `just test-kafka` / `test-rabbit` / `test-nats` / `test-redis` / `test-redis-cluster` / `test-confluent` — a whole broker, excluding `connected` and `slow`; the `-all` variants add the slow and connected ones (that broker must be up).\n- `just test-all` — everything (`-m \"all\"`).\n\nHeads-up: the pyproject default addopts exclude only `slow` (`-m 'not slow'`) — bare pytest WILL collect `connected` tests, so pass `-m \"not slow and not connected\"` explicitly when no broker is running.\n\nGlobal pytest timeout is 30s per test; the suite runs parallel — keep tests independent and use the `queue` fixture for unique names.\n\n## Markers — strict\n\n`--strict-markers` is enabled; the allowed set is defined in `pyproject.toml` (`kafka`, `confluent`, `rabbit`, `nats`, `redis`, `redis_cluster`, `mqtt`, `slow`, `connected`, `all`, `benchmark`).\n\n- Broker-specific test → its broker mark: `@pytest.mark.kafka()`.\n- Talks to a real broker over the network → add `@pytest.mark.connected()` (excluded by `just test`; bare pytest excludes only `slow` by default).\n- Slow test → `@pytest.mark.slow()` (also excluded by default).\n- Async test → `@pytest.mark.asyncio()`.\n\nMarks pick the CI job, so a wrong one drops a test silently. `just misplaced-marks` (a CI step too) fails on the three shapes that did it before:\n\n- a test under a `<broker>/` directory anywhere in `tests/` without the `<broker>` mark (`redis_cluster` under `redis/cluster/`). A test imported from `docs_src` or `examples` is marked through `pytestmark` in the importing module.\n- a `connected` test whose module is built on one broker package, without that broker's mark. An unmarked test still runs in `test-basic`; a `connected` one runs only in its broker's job, so there it never runs at all.\n- `connected` on a whole `*MemoryTestcaseConfig` class (or its module). When an in-memory class inherits a test that does open a connection, mark that test — in the base testcase if it is inherited.\n\n## Shared base testcases\n\nCross-broker behavior is specified ONCE in `tests/brokers/base/` (`basic.py`, `consume.py`, `publish.py`, `router.py`, `codec.py`, `middlewares.py`, `parser.py`, `requests.py`, `connection.py`, `fastapi.py`, `testclient.py`, ...) and inherited by every broker.\n\nEach broker defines its config in `tests/brokers/<broker>/basic.py`:\n\n```python\nclass KafkaTestcaseConfig(BaseTestcaseConfig):\n    def get_broker(self, apply_types: bool = False, **kwargs: Any) -> KafkaBroker:\n        return KafkaBroker(apply_types=apply_types, **kwargs)\n\n    def get_router(self, **kwargs: Any) -> KafkaRouter:\n        return KafkaRouter(**kwargs)\n\n\nclass KafkaMemoryTestcaseConfig(KafkaTestcaseConfig):\n    def patch_broker(self, *brokers: KafkaBroker, **kwargs: Any) -> TestKafkaBroker:\n        return TestKafkaBroker(*brokers, **kwargs)\n```\n\nTest classes multiply-inherit config + behavior suite:\n\n```python\n@pytest.mark.kafka()\nclass TestKafkaCodec(KafkaMemoryTestcaseConfig, CodecTestcase): ...\n\n\n@pytest.mark.connected()\n@pytest.mark.kafka()\nclass TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ...\n```\n\n**Rule:** new cross-broker behavior goes into a base class in `tests/brokers/base/` so every broker inherits the test. Broker-specific behavior is tested directly in `tests/brokers/<broker>/`.\n\n**Hook surface:** a member of a base testcase earns its place by being overridden in `tests/brokers/<broker>/` — `separator`, `declare_subscriber` and `publish` in `base/address.py` are hooks because MQTT, Kafka and RabbitMQ respell them, and each docstring names who does. What no broker overrides goes inline in the test body, beside the assertion it feeds, with a comment carrying the value it compiles to:\n\n```python\n# subscribe to \"queue.{level}\"\nsubscriber = self.declare_subscriber(\n    broker,\n    f\"{queue}{self.separator}{{level}}\",\n    queue,\n)\n```\n\n## Add a test only when nothing else already breaks on it\n\nBefore writing a new test, ask what already goes red if the change is wrong: another test\nin the suite, or `mypy` running over library code that already exercises the path (a\nregistrator signature checked through `faststream/<broker>/testing.py`, say). If something\nalready catches it, don't add another one — a ticket asking for \"a test per case\" doesn't\noverride this; say which existing check covers the rest instead of writing one that just\nrestates it. Never pin language or stdlib behaviour FastStream doesn't own (a `NamedTuple`\nunpacks, `==` on tuples).\n\nThe name is the behaviour, so a test carries no docstring by default, and an assertion\nthat needs explaining gets a single `#` comment directly over it. Two cases earn one:\n\n- the regression pattern below — the docstring is the issue URL and nothing else;\n- the rare test that is unreadable without it, because what it guards is invisible from\n  the body. `tests/utils/test_lazy_imports.py` runs an import in a subprocess: only the\n  docstring can say what that protects, how it broke before, and why nothing else goes\n  red. If a better name or one comment would do, the test is not this case.\n\n## One equality per behaviour\n\nWhen a test checks one value from several angles — a tuple's fields, a few keys of a\ndict, a length and an element — build the expected shape from `dirty-equals` matchers\nand compare once. The failure then prints the whole shape, and the test reads as a single\nstatement of the behaviour:\n\n```python\n# Claimed entries come first, with their previous deliveries and idle time\nassert received[:2] == [\n    (\"pending_message\", IsInt(ge=1), IsInt(ge=100)),\n    (\"new_message\", 0, 0),\n]\n\nassert snapshot == IsPartialDict({\n    \"delivery_counts\": HasLen(size),\n    \"idle_times\": HasLen(size),\n})\n```\n\nA chain of `assert x[0] ...`, `assert x[1] ...`, or a loop carrying a `found` flag, is this\nshape spelled out one field at a time: collapse it into the one equality. `IsPartialDict`\ntakes a dict literal, so dotted keys and enum values read the same as the config they\nmirror. An equality that already fails on a missing delivery stands alone; the\n`assert event.is_set()` in front of it says nothing more.\n\n## Tests are type checked\n\n`just mypy` runs the same strict config over all of `tests/`, so a test is annotated like library code: every function, fixture and handler has its parameters and return typed.\n\n- **Cross-broker testcases hold the broker as `Any`.** A base class in `tests/brokers/base/`, `tests/asyncapi/base/` and the like subclasses `BaseTestcaseConfig[Any]` and types `broker_class`, `router_class` and friends as `Any` — each broker spells `subscriber()` differently. Such a module also joins the `disallow_untyped_decorators = false` override in `pyproject.toml`; broker-specific tests keep the check.\n- **A value that is wrong on purpose goes through `Any`**, not an ignore: `router: Any = NatsRouter()` before handing it to a Kafka broker, `channel_manager: Any = FakeChannelManager(mock)` for a stand-in. The same goes for private internals (`producer: Any = broker._producer`).\n- **The raw client comes from the public API**: `client = await br.connect()`, never `br._connection`, which is `None`-able.\n- **Narrow with an assertion the test already implies** — `assert message`, `assert isinstance(point, HistogramDataPoint)` — and put a repeated one in a small module-level helper.\n- **An awaitable dropped on purpose is written `_ = ...`**: `_ = tg.start_soon(app.run)`, `_ = await br.publish(..., no_confirm=True)`. A bare statement reads as a forgotten `await`, which is what `unused-awaitable` reports.\n- **`# type: ignore[code]` is for what nothing else expresses**, always with its code and before any `# noqa`: `subscriber(*args, **kwargs)` from `get_subscriber_params()` resolves to `Any` (`untyped-decorator`), a test overriding a base test with other fixtures (`override`), a name redefined on purpose (`no-redef`), a call missing a required argument to prove it raises (`call-arg`).\n\nA typing problem that turns out to live in `faststream/` is fixed there, in its own PR with a case in `tests/mypy/`, not papered over in the test.\n\n## Regression tests\n\nA test defending a fixed bug names the issue by **full URL**, so the case it pins is one click away:\n\n```python\n@pytest.mark.xfail(reason=\"https://github.com/ag2ai/faststream/issues/2513\")\nasync def test_publisher_without_destination(self) -> None:\n    \"\"\"Fixes https://github.com/ag2ai/faststream/issues/2513.\"\"\"\n```\n\nThe URL is the whole docstring — no explanation of the behavior underneath it. `xfail`/`skip` reasons take the same URL. A comment inside the test body follows the **code-architecture** rule: one line, directly over the assertion it explains.\n\nA test written after the fix earns its place by going **red** on the old code — revert the fix, run, restore:\n\n```bash\ngit show <fix-commit> -- faststream/ | git apply -R -\nuv run pytest tests/... -m \"not slow and not connected\"\ngit checkout -- faststream/\n```\n\nThe same run grades the tests already there, and it is how a suite shrinks. Two tests red for one reason are one test: keep the one whose declaration carries more (an escaped brace *beside* a Path parameter over an escaped brace alone), delete the other, and check what a broker already gets from `test_router.py` or `test_path.py` before keeping a third.\n\n## In-memory vs real broker\n\n- Default to the in-memory `TestBroker` (`faststream/<broker>/testing.py`) via a `*MemoryTestcaseConfig` — fast, runs everywhere, no `connected` mark.\n- Use a real broker (plain `*TestcaseConfig` + `@pytest.mark.connected()`) when the behavior depends on actual broker semantics (acks, consumer groups, reconnects). Connection settings come from the `Settings` dataclass in `tests/brokers/<broker>/conftest.py`.\n\n## Fixtures & utilities\n\n- Global fixtures (`tests/conftest.py`): `queue` (unique uuid string), `event` (`asyncio.Event`), `mock` / `async_mock` (function-scoped, reset via teardown), `context`, `runner` (CLI).\n- `tests/marks.py`: conditional skips — `skip_windows`, `skip_macos`, `pydantic_v1`/`pydantic_v2`, `require_aiokafka`, `require_confluent`, `require_aiopika`, `require_redis`, `require_nats`, `require_mqtt`.\n- `tests/tools.py`: `spy_decorator` — wraps a real method with a mock spy (call assertions via `.mock`) while preserving behavior.\n- `tests/mocks.py`: `mock_pydantic_settings_env` for env-driven settings tests.\n- `freezegun` is available as a test dep.\n\n**Never import from a `conftest.py`.** pytest loads conftest modules specially (their fixtures are injected into the collected files), so importing from one — `from .conftest import Settings` or `from tests.brokers.redis.conftest import ...` — can produce a duplicated/mismatched module and confusing collection errors. When conftest and a test file need the same object, declare it in a plain helper module next to them (e.g. `tests/brokers/redis/settings.py`, `basic.py`) and import it from both.\n\n## Related skills\n\n- **dev-workflow** — docker broker management and the full just recipe matrix.\n- **code-architecture** — where the code under test lives and how it's shaped.\n- **documentation-writing** — docs snippets get tests under `tests/docs/`.\n</yakaai_skills>"}