Testing Patterns

FastStream Testing Patterns

Run only the test files the change touches

Name the specific test files:

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

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:

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:

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

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

Test functions and classes carry no docstring — the name is the behaviour. The one exception is the regression pattern below, whose docstring is the issue URL and nothing else. When an assertion needs explaining, a single # comment sits directly over it, not prose in a docstring.

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:

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

Regression tests

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

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

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