FastStream 测试模式实战精准测试选择、严格标记体系与跨 Broker 共享测试基类【免费下载链接】faststreamAsynchronous Python framework for event-driven services. A thin client for Kafka, RabbitMQ, NATS, Redis and MQTT with full access to native broker features, plus AsyncAPI docs, in-memory tests and observability out of the box.项目地址: https://gitcode.com/GitHub_Trending/fa/faststreamFastStream 是一个面向 Kafka、RabbitMQ、NATS、Redis 与 MQTT 的异步事件驱动框架其tests/目录下沉淀了一套成熟的测试工程实践从改一处代码只跑相关文件的精准选择策略到--strict-markers严格标记体系再到通过tests/brokers/base/共享基类实现跨 Broker 行为只写一次的测试架构。本文以仓库内.agents/skills/testing-patterns/SKILL.md为核心结合pyproject.toml、justfile与真实测试文件系统讲解这套测试模式的选型逻辑、命令矩阵、标记语义与编写规范读完你既能快速定位并运行与改动相关的测试也能按既有架构规范写出跨 Broker 可继承、可回归的高质量测试。只运行变更触及的测试文件FastStream 测试工程的第一条铁律是本地跑测试的目的是回答这次改动是否可用而不是替 CI 把全量测试再跑一遍。目录或裸tests/是 CI 的形状——CI 反正会运行一切本地运行的价值在于以最快速度验证当前这次编辑超出范围的每个测试都是在浪费时间。因此改动后应当显式命名具体测试文件而不是给 pytest 传一个目录uv run pytest tests/brokers/kafka/test_misconfigure.py tests/asyncapi/kafka/v3_0_0/test_address.py -m not connected这一组文件就是可能受这次编辑影响的完整集合到此为止不再扩大。上面这条命令中同时出现的tests/brokers/kafka/test_misconfigure.py与tests/asyncapi/kafka/v3_0_0/test_address.py是仓库中真实存在的文件分别覆盖 Kafka 配置错误场景与 AsyncAPI v3.0.0 地址描述场景可以作为一个改动往往横跨 broker 行为与 AsyncAPI 描述两层的典型例子。如何找到应该运行的文件grep 你改动的符号在tests/全目录中搜索你修改的函数、类或参数名找到所有引用点。遵循命名规律对faststream/broker/endpoint/thing.py的改动通常由tests/brokers/broker/test_thing.py与tests/asyncapi/broker/v*/test_thing.py覆盖。也就是说一个功能的 broker 行为测试在tests/brokers/broker/下其 AsyncAPI 文档描述测试在tests/asyncapi/broker/下。共享 testcase 会扩大影响面——但用命名来扩大tests/brokers/base/与tests/asyncapi/base/中的共享测试类会被每个 Broker 继承因此修改这些基类确实会扩大受影响的文件集合。但扩大的方式仍然是命名继承它的具体文件而不是运行整个目录——你依然要逐一列出那些继承该基类的测试文件。只在改动触及网络时使用 connectedconnected标记是需要真实 Broker 在网络上运行的标志。只有当改动真正触及线路subscribing、publishing、acks、bindings、reconnects时才需要它。如果改动的是 specification、config 或声明期检查declaration-time check这类行为在内存中、几秒钟内就能判定完全不需要拉起真实 Broker。Broker-backed 的运行需要几分钟并且会在多次运行之间共享状态topics、queues、consumer groups 会累积成本高、隔离性差。connected 失败时先单独重跑失败项当一次 connected 运行失败时在相信它之前先单独重跑这些失败项。因为 Broker 是共享的会累积 topics、queues 和 consumer groups——一个在隔离环境下能通过的失败很可能是容器环境的问题而不是代码的问题。要区分这两者可以对已提交的代码树运行同一组测试并对比失败数量git stash # 暂存改动回到已提交树 uv run pytest 同一组文件 -m not slow and not connected git stash pop # 恢复改动如果两个树的失败数量一致问题在环境而非你的改动。命令体系直接 pytest 与 just 容器套件FastStream 的测试命令分两条路径对应两种场景直接运行 pytest如上面的uv run pytest ...不需要任何容器这也是命名文件的测试在本地直跑的原因——它们根本不依赖 Broker。just test*系列配方在开发容器内运行整套套件。容器通过docker compose exec faststream执行所以先要just up启动容器。注意测试永远不要通过 rtk 代理运行直接使用 pytest 或 just。just 配方矩阵配方等价命令在容器内适用场景just test [path]pytest [path] -m not slow and not connected -n auto快速选择可指向命名文件-n auto并行just test-kafka/test-rabbit/test-nats/test-redis/test-redis-cluster/test-confluent对应-m broker and not connected and not slow跑整个 Broker 的快速套件just test-kafka-all等-all变体对应-m broker or (broker and slow)追加 slow 与 connected 测试要求该 Broker 已启动just test-allpytest [path] -m all -n auto运行一切从 justfile 源码可以看到这些配方的实际实现test配方为docker compose exec faststream uv run pytest {{param}} -m not slow and not connected -n autotest-kafka为-m kafka and not connected and not slowtest-all为-m all -n auto与文档描述完全一致。just updocker compose up -d用于拉起全部基础设施容器仓库根目录的 docker-compose.yaml 定义了这些服务。pyproject 默认 addopts 的陷阱pyproject.toml 中的 pytest 配置是一个关键注意点[tool.pytest.ini_options] minversion 7.0 timeout 30 timeout_func_only true addopts --strict-markers --strict-config -q -m not slow testpaths [tests]默认 addopts只排除了slow-m not slow这意味着裸pytest会收集并尝试运行connected测试。所以在没有 Broker 运行的情况下必须显式传入-m not slow and not connected。这正是just test配方显式写明该表达式的原因。另外两条全局约束每个测试的全局 pytest 超时为 30 秒timeout 30且timeout_func_only true让超时只作用于测试函数体per phase从而让flaky重试拥有自己的计时器——这是为pytest-timeout与pytest-rerunfailures协作而设计的见 pyproject.toml 中该配置的注释。套件并行运行-n auto由 pytest-xdist 提供因此测试必须相互独立并通过queuefixture 获取唯一名称避免命名冲突。严格标记体系Markers项目启用了--strict-markers只允许使用 pyproject.toml 中显式声明的标记写错或遗漏声明都会直接报错。在 pyproject.toml 中声明的完整标记集合为rabbit, kafka, confluent, nats, redis, redis_cluster, redis_sentinel, mqtt, slow, connected, all, benchmark标记使用规则Broker 专属测试→ 打上对应 Broker 标记pytest.mark.kafka()、pytest.mark.rabbit()等。通过网络与真实 Broker 通信→ 追加pytest.mark.connected()被just test排除裸 pytest 默认只排除slow。慢测试→pytest.mark.slow()默认也被排除。异步测试→pytest.mark.asyncio()pytest-asyncio 提供pyproject.toml 中asyncio_default_fixture_loop_scope function。一个值得注意的源码细节是all标记它并非手动标注在测试上而是在 tests/conftest.py 中通过钩子自动添加——pytest_collection_modifyitems对收集到的每个 item 执行item.add_marker(all)。这正是just test-all使用-m all就能选中全部测试的机制。共享基础测试用例跨 Broker 行为只写一次FastStream 测试架构的核心思想是跨 Broker 的相同行为在tests/brokers/base/中只定义一次再由每个 Broker 继承。该目录下的文件basic.py、consume.py、publish.py、router.py、codec.py、middlewares.py、parser.py、requests.py、connection.py、fastapi.py、testclient.py、address.py、include_router.py、publish_command.py等分别描述一类行为任何 Broker 都必须通过继承满足同一套行为约定。每个 Broker 定义自己的 Config每个 Broker 在tests/brokers/broker/basic.py中定义自己的测试配置类。以 tests/brokers/kafka/basic.py 为例class KafkaTestcaseConfig(BaseTestcaseConfig[KafkaBroker]): def get_broker( self, apply_types: bool False, **kwargs: Any, ) - KafkaBroker: return KafkaBroker(apply_typesapply_types, **kwargs) def get_router(self, **kwargs: Any) - KafkaRouter: return KafkaRouter(**kwargs) def get_cancel_ack_subscriber_kwargs(self, queue: str) - dict[str, Any]: return { group_id: f{queue}-cancel-ack, auto_offset_reset: earliest, } class KafkaMemoryTestcaseConfig(KafkaTestcaseConfig): override def patch_broker( self, *brokers: KafkaBroker, **kwargs: Any, ) - Any: return TestKafkaBroker(*brokers, **kwargs)而基类 tests/brokers/base/basic.py 中的BaseTestcaseConfig提供了公共骨架timeout: float 3.0默认断言超时、抽象的get_broker/get_router工厂方法以及默认的patch_broker实现单 Broker 直接返回自身多 Broker 时用AsyncExitStack把它们作为异步上下文一并启动。测试类多重继承Config 行为套件具体测试类通过多重继承同时获得如何构造 BrokerConfig与要验证什么行为行为套件pytest.mark.kafka() pytest.mark.asyncio() class TestKafkaCodec(KafkaMemoryTestcaseConfig, CodecTestcase): pass这在 tests/brokers/kafka/test_codec.py 中真实存在同样的模式还出现在tests/brokers/confluent/test_codec.py、tests/brokers/nats/test_codec.py等文件中。真实 Broker 版本则叠加connected标记如 tests/brokers/kafka/test_consume.pypytest.mark.kafka() pytest.mark.connected() class TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ...两条规则新的跨 Broker 行为→ 放进tests/brokers/base/的某个基类让所有 Broker 自动继承该测试。Broker 专属行为→ 直接写在tests/brokers/broker/下对应的测试文件中。钩子面Hook surface基类测试用例中的成员其存在价值在于被某个 Broker 改写override。tests/brokers/base/address.py中的separator、declare_subscriber与publish就是这样的钩子separator默认是.MQTT 将其改写为/declare_subscriber被 Kafka 改写成pattern、RabbitMQ 改写成RabbitQueue键publish被 RabbitMQ 改写成 routing key 语义——每个钩子的 docstring 都写明是谁在改写它如What goes between two address segments; MQTT spells it/.、Kafka respells this withpattern, RabbitMQ with aRabbitQueuekey.。没有 Broker 改写的部分就直接内联在测试体内、放在它所服务的断言旁边并用一行注释携带它编译成的值# subscribe to queue.{level} subscriber self.declare_subscriber( broker, f{queue}{self.separator}{{level}}, queue, )这段代码取自 tests/brokers/base/address.py是钩子 内联注释模式的直接示范把抽象声明与具体编译结果同时呈现在测试里读者无需跳转即可理解。何时才添加新测试写新测试之前先问一个问题如果改动是错的现在有什么已经会变红可能是一个套件里已有的测试也可能是mypy对库代码的检查——例如一个 registrator 的签名已经通过faststream/broker/testing.py中的类型标注被静态检查覆盖。如果已有检查能捕获该缺陷就不要再加一个新测试。即便需求单上写着每个 case 都要有测试也不能覆盖这条规则——此时应指出是哪个既有检查覆盖了剩余部分而不是写一个只是复述它的测试。同时要遵守两条边界不要锁定 FastStream 不拥有的语言或标准库行为例如NamedTuple可解包、tuple 的语义——这些是 Python 自身的保证不是框架的契约。测试函数与类不写 docstring——函数名就是行为说明。唯一的例外是下面回归测试模式其 docstring 只放 issue URL。当断言需要解释时用紧贴在其上方的一行#注释而不是 docstring 里的长篇 prose。一个行为只写一个等式dirty-equals当测试需要从多个角度检查一个值——一个 tuple 的多个字段、一个 dict 的若干键、一个长度加一个元素——不要写成一串assert x[0] ...、assert x[1] ...或带found标志的循环。正确做法是用dirty-equals匹配器构建期望形状一次比较完成。这样失败时 pytest 会打印整个形状测试本身也读起来像对行为的一次性陈述# Claimed entries come first, with their previous deliveries and idle time assert received[:2] [ (pending_message, IsInt(ge1), IsInt(ge100)), (new_message, 0, 0), ] assert snapshot IsPartialDict({ delivery_counts: HasLen(size), idle_times: HasLen(size), })这正是 tests/brokers/redis/stream_claim.py 中真实使用的写法该文件从dirty_equals导入HasLen, IsInstance, IsInt, IsPartialDict。其中IsPartialDict接受 dict 字面量因此带点的键和枚举值读起来与它们镜像的配置完全一致一个已经会在缺失投递时失败的等式应该独立存在它前面再放一个assert event.is_set()并不会多传达任何信息——除非该事件本身是测试关注的独立事实。回归测试用完整 URL 命名用旧代码验证防御已修复 bug 的测试以完整 URL 命名 issue让被钉住的场景一键可达pytest.mark.xfail(reasonhttps://github.com/ag2ai/faststream/issues/2513) async def test_publisher_without_destination(self) - None: Fixes https://github.com/ag2ai/faststream/issues/2513.这段代码真实存在于 tests/brokers/kafka/test_test_client.py同类模式还出现在tests/brokers/confluent/test_test_client.py与tests/brokers/nats/test_test_client.py。要点URL 就是整个 docstring——不解释其底下的行为xfail/skip的 reason 也使用同一 URL测试体内的注释遵循 code-architecture 规则一行、直接位于所解释的断言之上。修复后的测试必须能在旧代码上变红一个在修复之后编写的测试其立足点是在旧代码上运行必须失败red。验证流程是回退修复 → 运行 → 恢复git show fix-commit -- faststream/ | git apply -R - uv run pytest tests/... -m not slow and not connected git checkout -- faststream/这一流程同样服务于套件瘦身两个因同一原因变红的测试应当合并为一个——保留声明承载更多信息的那一个例如转义花括号旁边还有 Path 参数优于只有转义花括号删除另一个在保留第三个之前先检查某 Broker 是否已经从test_router.py或test_path.py中获得了同样的覆盖。内存测试 vs 真实 Broker两种测试形态各有明确适用场景默认使用内存版TestBroker位于faststream/broker/testing.py如TestKafkaBroker、TestRabbitBroker、TestRedisBroker通过*MemoryTestcaseConfig接入。它快、随处可跑、无需connected标记是绝大多数行为测试的首选。当行为依赖真实 Broker 语义时acks、consumer groups、reconnects才使用真实 Broker普通*TestcaseConfigpytest.mark.connected()。连接设置来自tests/brokers/broker/conftest.py中的Settings数据类——例如 tests/brokers/redis/settings.py 定义了Settings(urlredis://localhost:6379, hostlocalhost, port6379)而 tests/brokers/redis/conftest.py 通过 session 级settingsfixture 暴露它。内存与真实 Broker 的切换完全由 Config 类控制KafkaMemoryTestcaseConfig.patch_broker返回TestKafkaBroker(*brokers, **kwargs)把真实 Broker 实例替换为内存替身而真实测试直接继承KafkaTestcaseConfig并在async with self.patch_broker(...)中原样启动 Broker。Fixtures 与工具全局 fixturestests/conftest.pytests/conftest.py 提供了一套可复用的全局 fixturesqueue唯一的 uuid 字符串str(uuid4())保证并行测试间的命名隔离event/event2/event3asyncio.Event用于异步断言消息已到达mock/mock2/async_mock函数作用域teardown 时通过reset_mock()自动重置contextContextRepo实例runnersession 作用域的CliRunnertyper testing用于 CLI 测试。条件跳过tests/marks.pytests/marks.py 定义了条件跳过标记skip_windows、skip_macos、pydantic_v1/pydantic_v2以及基于可选依赖探测的require_aiokafka、require_confluent、require_aiopika、require_redis含require_redis_v710、require_nats、require_mqtt。它们通过pytest.mark.skipif实现例如require_confluent pytest.mark.skipif(not HAS_CONFLUENT, reasonrequires confluent-kafka)。测试工具tests/tools.py 与 tests/mocks.pyspy_decoratortests/tools.py包装真实方法为一个间谍——通过.mock属性做调用断言同时保留原方法行为。异步方法先await mock(*args, **kwargs)再调用真实实现同步方法也先记录调用。它在 tests/brokers/kafka/test_consume.py 中大量用于验证 ack 行为patch.object(AIOKafkaConsumer, commit, spy_decorator(AIOKafkaConsumer.commit))随后断言m.mock.assert_called_once()手动 ack或assert not m.mock.calledmanual nack 不触发 commit。mock_pydantic_settings_envtests/mocks.pypatchpydantic_settings.sources.DotEnvSettingsSource._read_env_files用于环境驱动的 settings 测试。freezegun作为测试依赖可用见 pyproject.toml 的test-core依赖组。绝不从 conftest.py 导入禁止从conftest.py导入任何东西。pytest 会特殊加载 conftest 模块其 fixtures 被注入到被收集的文件中因此from .conftest import Settings或from tests.brokers.redis.conftest import ...会导致重复/不匹配的模块与令人困惑的收集错误。当 conftest 与测试文件需要同一个对象时把它声明在与它们相邻的普通 helper 模块中例如tests/brokers/redis/settings.py、tests/brokers/redis/basic.py然后两边都从那里导入。仓库正是这样做的tests/brokers/redis/conftest.py 通过from .settings import Settings导入而非在 conftest 内定义。关联技能本文所述的测试模式与仓库中另外三份技能文档配套使用dev-workflowDocker Broker 的管理与完整的 just 配方矩阵code-architecture被测代码位于何处、如何被组织成形documentation-writing文档代码片段docs snippets的测试位于tests/docs/下写文档改动时需要一并更新对应测试。【免费下载链接】faststreamAsynchronous Python framework for event-driven services. A thin client for Kafka, RabbitMQ, NATS, Redis and MQTT with full access to native broker features, plus AsyncAPI docs, in-memory tests and observability out of the box.项目地址: https://gitcode.com/GitHub_Trending/fa/faststream创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考