testing-patterns

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

# FastStream Testing Patterns

## Run only the test files the change touches

Name the specific test files:

```bash
uv run pytest tests/brokers/kafka/test_misconfigure.py tests/asyncapi/kafka/v3_0_0/test_address.py -m "not connected"
```

The set is the files that could see the edit, and it stops there. A directory
or a bare `tests/` is CI's shape: CI runs everything anyway, so a local run
exists to answer whether *this* edit works, and every test beyond that is time
spent not finding out.

To find the files: grep the symbol you changed across `tests/`, and follow the
naming — a change to `faststream/<broker>/<endpoint>/<thing>.py` is usually
covered by `tests/brokers/<broker>/test_<thing>.py` and
`tests/asyncapi/<broker>/v*/test_<thing>.py`.

A shared testcase in `tests/brokers/base/` or `tests/asyncapi/base/` is
inherited by every broker, so editing one does widen the set — but widen it by
naming the inheriting files, not by running their directories.

**Reach for `connected` only when the change reaches the wire** — subscribing,
publishing, acks, bindings, reconnects. A change to a specification, a config,
or a declaration-time check is decided in memory, in seconds. Broker-backed
runs take minutes and share state between runs.

When a `connected` run does fail, **re-run the failures alone before believing
them**. The brokers are shared and accumulate topics, queues and consumer
groups; a failure that passes in isolation is the container, not the code. To
tell them apart, run the same set against the committed tree (`git stash`) and
compare counts.

## Commands

Run pytest directly or via just — **never through the rtk proxy**.

Direct pytest needs no container, which is why named files run there. The
`just test*` recipes run whole suites inside the dev container
(`docker compose exec faststream`, so `just up` first):

- `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.
- `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).
- `just test-all` — everything (`-m "all"`).

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

Global pytest timeout is 30s per test; the suite runs parallel — keep tests independent and use the `queue` fixture for unique names.

## Markers — strict

`--strict-markers` is enabled; the allowed set is defined in `pyproject.toml` (`kafka`, `confluent`, `rabbit`, `nats`, `redis`, `redis_cluster`, `mqtt`, `slow`, `connected`, `all`, `benchmark`).

- Broker-specific test → its broker mark: `@pytest.mark.kafka()`.
- Talks to a real broker over the network → add `@pytest.mark.connected()` (excluded by `just test`; bare pytest excludes only `slow` by default).
- Slow test → `@pytest.mark.slow()` (also excluded by default).
- Async test → `@pytest.mark.asyncio()`.

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

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

## Shared base testcases

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

Each broker defines its config in `tests/brokers/<broker>/basic.py`:

```python
class KafkaTestcaseConfig(BaseTestcaseConfig):
    def get_broker(self, apply_types: bool = False, **kwargs: Any) -> KafkaBroker:
        return KafkaBroker(apply_types=apply_types, **kwargs)

    def get_router(self, **kwargs: Any) -> KafkaRouter:
        return KafkaRouter(**kwargs)


class KafkaMemoryTestcaseConfig(KafkaTestcaseConfig):
    def patch_broker(self, *brokers: KafkaBroker, **kwargs: Any) -> TestKafkaBroker:
        return TestKafkaBroker(*brokers, **kwargs)
```

Test classes multiply-inherit config + behavior suite:

```python
@pytest.mark.kafka()
class TestKafkaCodec(KafkaMemoryTestcaseConfig, CodecTestcase): ...


@pytest.mark.connected()
@pytest.mark.kafka()
class TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ...
```

**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>/`.

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

```python
# subscribe to "queue.{level}"
subscriber = self.declare_subscriber(
    broker,
    f"{queue}{self.separator}{{level}}",
    queue,
)
```

## Add a test only when nothing else already breaks on it

Before writing a new test, ask what already goes red if the change is wrong: another test
in the suite, or `mypy` running over library code that already exercises the path (a
registrator signature checked through `faststream/<broker>/testing.py`, say). If something
already catches it, don't add another one — a ticket asking for "a test per case" doesn't
override this; say which existing check covers the rest instead of writing one that just
restates it. Never pin language or stdlib behaviour FastStream doesn't own (a `NamedTuple`
unpacks, `==` on tuples).

The name is the behaviour, so a test carries no docstring by default, and an assertion
that needs explaining gets a single `#` comment directly over it. Two cases earn one:

- the regression pattern below — the docstring is the issue URL and nothing else;
- the rare test that is unreadable without it, because what it guards is invisible from
  the body. `tests/utils/test_lazy_imports.py` runs an import in a subprocess: only the
  docstring can say what that protects, how it broke before, and why nothing else goes
  red. If a better name or one comment would do, the test is not this case.

## One equality per behaviour

When a test checks one value from several angles — a tuple's fields, a few keys of a
dict, a length and an element — build the expected shape from `dirty-equals` matchers
and compare once. The failure then prints the whole shape, and the test reads as a single
statement of the behaviour:

```python
# Claimed entries come first, with their previous deliveries and idle time
assert received[:2] == [
    ("pending_message", IsInt(ge=1), IsInt(ge=100)),
    ("new_message", 0, 0),
]

assert snapshot == IsPartialDict({
    "delivery_counts": HasLen(size),
    "idle_times": HasLen(size),
})
```

A chain of `assert x[0] ...`, `assert x[1] ...`, or a loop carrying a `found` flag, is this
shape spelled out one field at a time: collapse it into the one equality. `IsPartialDict`
takes a dict literal, so dotted keys and enum values read the same as the config they
mirror. An equality that already fails on a missing delivery stands alone; the
`assert event.is_set()` in front of it says nothing more.

## Tests are type checked

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

- **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.
- **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`).
- **The raw client comes from the public API**: `client = await br.connect()`, never `br._connection`, which is `None`-able.
- **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.
- **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.
- **`# 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`).

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

## Regression tests

A test defending a fixed bug names the issue by **full URL**, so the case it pins is one click away:

```python
@pytest.mark.xfail(reason="https://github.com/ag2ai/faststream/issues/2513")
async def test_publisher_without_destination(self) -> None:
    """Fixes https://github.com/ag2ai/faststream/issues/2513."""
```

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

A test written after the fix earns its place by going **red** on the old code — revert the fix, run, restore:

```bash
git show <fix-commit> -- faststream/ | git apply -R -
uv run pytest tests/... -m "not slow and not connected"
git checkout -- faststream/
```

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

## In-memory vs real broker

- Default to the in-memory `TestBroker` (`faststream/<broker>/testing.py`) via a `*MemoryTestcaseConfig` — fast, runs everywhere, no `connected` mark.
- 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`.

## Fixtures & utilities

- Global fixtures (`tests/conftest.py`): `queue` (unique uuid string), `event` (`asyncio.Event`), `mock` / `async_mock` (function-scoped, reset via teardown), `context`, `runner` (CLI).
- `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`.
- `tests/tools.py`: `spy_decorator` — wraps a real method with a mock spy (call assertions via `.mock`) while preserving behavior.
- `tests/mocks.py`: `mock_pydantic_settings_env` for env-driven settings tests.
- `freezegun` is available as a test dep.

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

## Related skills

- **dev-workflow** — docker broker management and the full just recipe matrix.
- **code-architecture** — where the code under test lives and how it's shaped.
- **documentation-writing** — docs snippets get tests under `tests/docs/`.

More agent context in ag2ai/faststream

3 other files this repository gives its agents.