From 9d0797e3de001ac8e0fc58d6e6731df5ca3d89c8 Mon Sep 17 00:00:00 2001 From: Lancetnik Date: Wed, 29 Jul 2026 20:28:40 +0300 Subject: [PATCH 1/4] fix (fastapi): skip init=False fields when copying Dependant Dataclass fields declared with `field(init=False)` cannot be passed to the constructor, so building `_FastStreamDependant` from a parent `Dependant` that has any such field raised a TypeError. Co-Authored-By: Claude Opus 5 (1M context) --- faststream/_internal/fastapi/get_dependant.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/faststream/_internal/fastapi/get_dependant.py b/faststream/_internal/fastapi/get_dependant.py index d84a6b8f79..f91f05841e 100644 --- a/faststream/_internal/fastapi/get_dependant.py +++ b/faststream/_internal/fastapi/get_dependant.py @@ -30,7 +30,9 @@ class _FastStreamDependant(Dependant): def _extend_fastapi_dependant(dependant: Dependant) -> _FastStreamDependant: """Copy a native FastAPI dependant into an extensible subclass.""" return _FastStreamDependant(**{ - field.name: getattr(dependant, field.name) for field in fields(dependant) + field.name: getattr(dependant, field.name) + for field in fields(dependant) + if field.init }) From 10bc91e46627167233ce77ed91231b5bdcaf6b7a Mon Sep 17 00:00:00 2001 From: Lancetnik Date: Wed, 29 Jul 2026 20:36:23 +0300 Subject: [PATCH 2/4] chore: group redis cluster and sentinel tests into subpackages Move tests/brokers/redis/test_cluster*.py into tests/brokers/redis/cluster/ and test_sentinel.py into tests/brokers/redis/sentinel/. Also bump the project version to 0.7.3 in uv.lock. Co-Authored-By: Claude Opus 5 (1M context) --- tests/brokers/redis/cluster/__init__.py | 0 tests/brokers/redis/{ => cluster}/test_cluster.py | 0 .../brokers/redis/{ => cluster}/test_cluster_codec.py | 3 +-- .../redis/{ => cluster}/test_cluster_connect.py | 3 +-- .../redis/{ => cluster}/test_cluster_consume.py | 3 +-- .../redis/{ => cluster}/test_cluster_include_router.py | 3 +-- .../redis/{ => cluster}/test_cluster_middlewares.py | 6 ++++-- .../redis/{ => cluster}/test_cluster_more_unit.py | 0 .../redis/{ => cluster}/test_cluster_publish.py | 3 +-- .../redis/{ => cluster}/test_cluster_pubsub_more.py | 5 ++--- .../redis/{ => cluster}/test_cluster_requests.py | 3 +-- .../redis/{ => cluster}/test_cluster_test_client.py | 10 ++++++++-- tests/brokers/redis/sentinel/__init__.py | 0 tests/brokers/redis/{ => sentinel}/test_sentinel.py | 0 uv.lock | 2 +- 15 files changed, 21 insertions(+), 20 deletions(-) create mode 100644 tests/brokers/redis/cluster/__init__.py rename tests/brokers/redis/{ => cluster}/test_cluster.py (100%) rename tests/brokers/redis/{ => cluster}/test_cluster_codec.py (73%) rename tests/brokers/redis/{ => cluster}/test_cluster_connect.py (99%) rename tests/brokers/redis/{ => cluster}/test_cluster_consume.py (82%) rename tests/brokers/redis/{ => cluster}/test_cluster_include_router.py (85%) rename tests/brokers/redis/{ => cluster}/test_cluster_middlewares.py (85%) rename tests/brokers/redis/{ => cluster}/test_cluster_more_unit.py (100%) rename tests/brokers/redis/{ => cluster}/test_cluster_publish.py (99%) rename tests/brokers/redis/{ => cluster}/test_cluster_pubsub_more.py (98%) rename tests/brokers/redis/{ => cluster}/test_cluster_requests.py (92%) rename tests/brokers/redis/{ => cluster}/test_cluster_test_client.py (93%) create mode 100644 tests/brokers/redis/sentinel/__init__.py rename tests/brokers/redis/{ => sentinel}/test_sentinel.py (100%) diff --git a/tests/brokers/redis/cluster/__init__.py b/tests/brokers/redis/cluster/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/brokers/redis/test_cluster.py b/tests/brokers/redis/cluster/test_cluster.py similarity index 100% rename from tests/brokers/redis/test_cluster.py rename to tests/brokers/redis/cluster/test_cluster.py diff --git a/tests/brokers/redis/test_cluster_codec.py b/tests/brokers/redis/cluster/test_cluster_codec.py similarity index 73% rename from tests/brokers/redis/test_cluster_codec.py rename to tests/brokers/redis/cluster/test_cluster_codec.py index 1b8fc35829..0b9f9a8800 100644 --- a/tests/brokers/redis/test_cluster_codec.py +++ b/tests/brokers/redis/cluster/test_cluster_codec.py @@ -1,8 +1,7 @@ import pytest from tests.brokers.base.codec import CodecTestcase - -from .basic import RedisClusterMemoryTestcaseConfig +from tests.brokers.redis.basic import RedisClusterMemoryTestcaseConfig @pytest.mark.redis_cluster() diff --git a/tests/brokers/redis/test_cluster_connect.py b/tests/brokers/redis/cluster/test_cluster_connect.py similarity index 99% rename from tests/brokers/redis/test_cluster_connect.py rename to tests/brokers/redis/cluster/test_cluster_connect.py index c356eb9f4f..7adbd9f485 100644 --- a/tests/brokers/redis/test_cluster_connect.py +++ b/tests/brokers/redis/cluster/test_cluster_connect.py @@ -13,8 +13,7 @@ from faststream.redis import RedisClusterBroker, StreamSub from tests.brokers.base.connection import BrokerConnectionTestcase - -from .conftest import SettingsCluster +from tests.brokers.redis.conftest import SettingsCluster @pytest.mark.connected() diff --git a/tests/brokers/redis/test_cluster_consume.py b/tests/brokers/redis/cluster/test_cluster_consume.py similarity index 82% rename from tests/brokers/redis/test_cluster_consume.py rename to tests/brokers/redis/cluster/test_cluster_consume.py index 102b92b053..4965bbff43 100644 --- a/tests/brokers/redis/test_cluster_consume.py +++ b/tests/brokers/redis/cluster/test_cluster_consume.py @@ -1,8 +1,7 @@ import pytest from tests.brokers.base.consume import BrokerRealConsumeTestcase - -from .basic import RedisClusterTestcaseConfig +from tests.brokers.redis.basic import RedisClusterTestcaseConfig @pytest.mark.connected() diff --git a/tests/brokers/redis/test_cluster_include_router.py b/tests/brokers/redis/cluster/test_cluster_include_router.py similarity index 85% rename from tests/brokers/redis/test_cluster_include_router.py rename to tests/brokers/redis/cluster/test_cluster_include_router.py index 42840d7082..022c552a02 100644 --- a/tests/brokers/redis/test_cluster_include_router.py +++ b/tests/brokers/redis/cluster/test_cluster_include_router.py @@ -4,8 +4,7 @@ IncludePublisherTestcase, IncludeSubscriberTestcase, ) - -from .basic import RedisClusterTestcaseConfig +from tests.brokers.redis.basic import RedisClusterTestcaseConfig @pytest.mark.redis_cluster() diff --git a/tests/brokers/redis/test_cluster_middlewares.py b/tests/brokers/redis/cluster/test_cluster_middlewares.py similarity index 85% rename from tests/brokers/redis/test_cluster_middlewares.py rename to tests/brokers/redis/cluster/test_cluster_middlewares.py index 42d115290d..605be9a64c 100644 --- a/tests/brokers/redis/test_cluster_middlewares.py +++ b/tests/brokers/redis/cluster/test_cluster_middlewares.py @@ -5,8 +5,10 @@ MiddlewareTestcase, MiddlewaresOrderTestcase, ) - -from .basic import RedisClusterMemoryTestcaseConfig, RedisClusterTestcaseConfig +from tests.brokers.redis.basic import ( + RedisClusterMemoryTestcaseConfig, + RedisClusterTestcaseConfig, +) @pytest.mark.redis_cluster() diff --git a/tests/brokers/redis/test_cluster_more_unit.py b/tests/brokers/redis/cluster/test_cluster_more_unit.py similarity index 100% rename from tests/brokers/redis/test_cluster_more_unit.py rename to tests/brokers/redis/cluster/test_cluster_more_unit.py diff --git a/tests/brokers/redis/test_cluster_publish.py b/tests/brokers/redis/cluster/test_cluster_publish.py similarity index 99% rename from tests/brokers/redis/test_cluster_publish.py rename to tests/brokers/redis/cluster/test_cluster_publish.py index 53bdca51e9..1747a2bb45 100644 --- a/tests/brokers/redis/test_cluster_publish.py +++ b/tests/brokers/redis/cluster/test_cluster_publish.py @@ -7,8 +7,7 @@ from faststream import Context from faststream.redis import RedisResponse from tests.brokers.base.publish import BrokerPublishTestcase - -from .basic import RedisClusterTestcaseConfig +from tests.brokers.redis.basic import RedisClusterTestcaseConfig @pytest.mark.connected() diff --git a/tests/brokers/redis/test_cluster_pubsub_more.py b/tests/brokers/redis/cluster/test_cluster_pubsub_more.py similarity index 98% rename from tests/brokers/redis/test_cluster_pubsub_more.py rename to tests/brokers/redis/cluster/test_cluster_pubsub_more.py index cc84693959..0e1b38863d 100644 --- a/tests/brokers/redis/test_cluster_pubsub_more.py +++ b/tests/brokers/redis/cluster/test_cluster_pubsub_more.py @@ -4,9 +4,8 @@ import pytest from faststream.redis import RedisClusterBroker - -from .basic import RedisClusterTestcaseConfig -from .conftest import SettingsCluster +from tests.brokers.redis.basic import RedisClusterTestcaseConfig +from tests.brokers.redis.conftest import SettingsCluster @pytest.mark.slow() diff --git a/tests/brokers/redis/test_cluster_requests.py b/tests/brokers/redis/cluster/test_cluster_requests.py similarity index 92% rename from tests/brokers/redis/test_cluster_requests.py rename to tests/brokers/redis/cluster/test_cluster_requests.py index 927dc7c56a..58187ba0ad 100644 --- a/tests/brokers/redis/test_cluster_requests.py +++ b/tests/brokers/redis/cluster/test_cluster_requests.py @@ -3,8 +3,7 @@ from faststream import BaseMiddleware from faststream.redis.parser import BinaryMessageFormatV1 from tests.brokers.base.requests import RequestsTestcase - -from .basic import RedisClusterMemoryTestcaseConfig +from tests.brokers.redis.basic import RedisClusterMemoryTestcaseConfig class Mid(BaseMiddleware): diff --git a/tests/brokers/redis/test_cluster_test_client.py b/tests/brokers/redis/cluster/test_cluster_test_client.py similarity index 93% rename from tests/brokers/redis/test_cluster_test_client.py rename to tests/brokers/redis/cluster/test_cluster_test_client.py index 579b834a41..01c187ccab 100644 --- a/tests/brokers/redis/test_cluster_test_client.py +++ b/tests/brokers/redis/cluster/test_cluster_test_client.py @@ -3,8 +3,7 @@ from faststream.redis import ListSub, StreamSub from faststream.redis.testing import FakeProducer from tests.brokers.base.testclient import BrokerTestclientTestcase - -from .basic import RedisClusterMemoryTestcaseConfig +from tests.brokers.redis.basic import RedisClusterMemoryTestcaseConfig @pytest.mark.redis_cluster() @@ -12,6 +11,13 @@ class TestClusterTestClient(RedisClusterMemoryTestcaseConfig, BrokerTestclientTestcase): """TestClient tests for RedisClusterBroker (memory-based).""" + @pytest.mark.connected() + async def test_broker_with_real_patches_publishers_and_subscribers( + self, + queue: str, + ) -> None: + await super().test_broker_with_real_patches_publishers_and_subscribers(queue) + async def test_broker_gets_patched_attrs_within_cm(self) -> None: await super().test_broker_gets_patched_attrs_within_cm(FakeProducer) diff --git a/tests/brokers/redis/sentinel/__init__.py b/tests/brokers/redis/sentinel/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/brokers/redis/test_sentinel.py b/tests/brokers/redis/sentinel/test_sentinel.py similarity index 100% rename from tests/brokers/redis/test_sentinel.py rename to tests/brokers/redis/sentinel/test_sentinel.py diff --git a/uv.lock b/uv.lock index 1477bd4b94..46016e05db 100644 --- a/uv.lock +++ b/uv.lock @@ -851,7 +851,7 @@ wheels = [ [[package]] name = "faststream" -version = "0.7.2" +version = "0.7.3" source = { editable = "." } dependencies = [ { name = "anyio" }, From f4c3aff93dc3cd6608521e0b9506ec9cdc8f3765 Mon Sep 17 00:00:00 2001 From: Lancetnik Date: Wed, 29 Jul 2026 20:52:37 +0300 Subject: [PATCH 3/4] chore(tests): move shared settings out of conftest modules Importing from a conftest.py is unsafe: pytest loads conftest modules specially, so a direct import can yield a duplicated/mismatched module. Declare the Settings dataclasses in plain settings.py helpers next to each broker's tests and import them from both the conftest and the test files. Also move SettingsCluster and the settings_cluster fixture down into tests/brokers/redis/cluster/, where they are the only users. Co-Authored-By: Claude Opus 5 (1M context) --- .agents/skills/testing-patterns/SKILL.md | 2 ++ tests/brokers/confluent/conftest.py | 7 +----- tests/brokers/confluent/settings.py | 6 +++++ tests/brokers/confluent/test_connect.py | 2 +- tests/brokers/kafka/conftest.py | 7 +----- tests/brokers/kafka/settings.py | 6 +++++ tests/brokers/kafka/test_connect.py | 2 +- tests/brokers/mqtt/conftest.py | 8 +------ tests/brokers/mqtt/settings.py | 7 ++++++ tests/brokers/mqtt/test_connect.py | 2 +- tests/brokers/nats/conftest.py | 7 +----- tests/brokers/nats/settings.py | 6 +++++ tests/brokers/nats/test_connect.py | 2 +- tests/brokers/redis/cluster/conftest.py | 8 +++++++ tests/brokers/redis/cluster/settings.py | 11 +++++++++ .../redis/cluster/test_cluster_connect.py | 3 ++- .../redis/cluster/test_cluster_pubsub_more.py | 3 ++- tests/brokers/redis/conftest.py | 24 +------------------ tests/brokers/redis/settings.py | 8 +++++++ tests/brokers/redis/test_connect.py | 2 +- 20 files changed, 68 insertions(+), 55 deletions(-) create mode 100644 tests/brokers/confluent/settings.py create mode 100644 tests/brokers/kafka/settings.py create mode 100644 tests/brokers/mqtt/settings.py create mode 100644 tests/brokers/nats/settings.py create mode 100644 tests/brokers/redis/cluster/conftest.py create mode 100644 tests/brokers/redis/cluster/settings.py create mode 100644 tests/brokers/redis/settings.py diff --git a/.agents/skills/testing-patterns/SKILL.md b/.agents/skills/testing-patterns/SKILL.md index 42cbec61a1..2c87aaa2ba 100644 --- a/.agents/skills/testing-patterns/SKILL.md +++ b/.agents/skills/testing-patterns/SKILL.md @@ -76,6 +76,8 @@ class TestConsume(KafkaTestcaseConfig, BrokerRealConsumeTestcase): ... - `tests/mocks.py`: `mock_pydantic_settings_env` for env-driven settings tests. - `dirty-equals` and `freezegun` are available as test deps. +**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. diff --git a/tests/brokers/confluent/conftest.py b/tests/brokers/confluent/conftest.py index 44e82178c6..86ae4b5592 100644 --- a/tests/brokers/confluent/conftest.py +++ b/tests/brokers/confluent/conftest.py @@ -1,13 +1,8 @@ -from dataclasses import dataclass - import pytest from faststream.confluent import KafkaRouter - -@dataclass -class Settings: - url: str = "localhost:9092" +from .settings import Settings @pytest.fixture(scope="session") diff --git a/tests/brokers/confluent/settings.py b/tests/brokers/confluent/settings.py new file mode 100644 index 0000000000..3ceb1d6a12 --- /dev/null +++ b/tests/brokers/confluent/settings.py @@ -0,0 +1,6 @@ +from dataclasses import dataclass + + +@dataclass +class Settings: + url: str = "localhost:9092" diff --git a/tests/brokers/confluent/test_connect.py b/tests/brokers/confluent/test_connect.py index 72e8e21038..2a9ae01038 100644 --- a/tests/brokers/confluent/test_connect.py +++ b/tests/brokers/confluent/test_connect.py @@ -5,7 +5,7 @@ from faststream.confluent.helpers import config from tests.brokers.base.connection import BrokerConnectionTestcase -from .conftest import Settings +from .settings import Settings @pytest.mark.connected() diff --git a/tests/brokers/kafka/conftest.py b/tests/brokers/kafka/conftest.py index 1eede17ed7..f856820f8f 100644 --- a/tests/brokers/kafka/conftest.py +++ b/tests/brokers/kafka/conftest.py @@ -1,13 +1,8 @@ -from dataclasses import dataclass - import pytest from faststream.kafka import KafkaRouter - -@dataclass -class Settings: - url: str = "localhost:9092" +from .settings import Settings @pytest.fixture(scope="session") diff --git a/tests/brokers/kafka/settings.py b/tests/brokers/kafka/settings.py new file mode 100644 index 0000000000..3ceb1d6a12 --- /dev/null +++ b/tests/brokers/kafka/settings.py @@ -0,0 +1,6 @@ +from dataclasses import dataclass + + +@dataclass +class Settings: + url: str = "localhost:9092" diff --git a/tests/brokers/kafka/test_connect.py b/tests/brokers/kafka/test_connect.py index 2829d525b5..d72af8af69 100644 --- a/tests/brokers/kafka/test_connect.py +++ b/tests/brokers/kafka/test_connect.py @@ -5,7 +5,7 @@ from faststream.kafka import KafkaBroker from tests.brokers.base.connection import BrokerConnectionTestcase -from .conftest import Settings +from .settings import Settings @pytest.mark.kafka() diff --git a/tests/brokers/mqtt/conftest.py b/tests/brokers/mqtt/conftest.py index b0a3a8bb81..900a19e5e7 100644 --- a/tests/brokers/mqtt/conftest.py +++ b/tests/brokers/mqtt/conftest.py @@ -1,14 +1,8 @@ -from dataclasses import dataclass - import pytest from faststream.mqtt.broker.router import MQTTRouter - -@dataclass -class Settings: - host: str = "localhost" - port: int = 1883 +from .settings import Settings @pytest.fixture(scope="session") diff --git a/tests/brokers/mqtt/settings.py b/tests/brokers/mqtt/settings.py new file mode 100644 index 0000000000..4b9eccd564 --- /dev/null +++ b/tests/brokers/mqtt/settings.py @@ -0,0 +1,7 @@ +from dataclasses import dataclass + + +@dataclass +class Settings: + host: str = "localhost" + port: int = 1883 diff --git a/tests/brokers/mqtt/test_connect.py b/tests/brokers/mqtt/test_connect.py index 6a89b47cb4..7b6121cc62 100644 --- a/tests/brokers/mqtt/test_connect.py +++ b/tests/brokers/mqtt/test_connect.py @@ -5,7 +5,7 @@ from faststream.mqtt.broker.broker import MQTTBroker from tests.brokers.base.connection import BrokerConnectionTestcase -from .conftest import Settings +from .settings import Settings @pytest.mark.connected() diff --git a/tests/brokers/nats/conftest.py b/tests/brokers/nats/conftest.py index 09e34659eb..d182a0ffec 100644 --- a/tests/brokers/nats/conftest.py +++ b/tests/brokers/nats/conftest.py @@ -1,13 +1,8 @@ -from dataclasses import dataclass - import pytest from faststream.nats import JStream, NatsRouter - -@dataclass -class Settings: - url: str = "nats://localhost:4222" +from .settings import Settings @pytest.fixture(scope="session") diff --git a/tests/brokers/nats/settings.py b/tests/brokers/nats/settings.py new file mode 100644 index 0000000000..034e35eec0 --- /dev/null +++ b/tests/brokers/nats/settings.py @@ -0,0 +1,6 @@ +from dataclasses import dataclass + + +@dataclass +class Settings: + url: str = "nats://localhost:4222" diff --git a/tests/brokers/nats/test_connect.py b/tests/brokers/nats/test_connect.py index c3ffec801b..50212430fc 100644 --- a/tests/brokers/nats/test_connect.py +++ b/tests/brokers/nats/test_connect.py @@ -6,7 +6,7 @@ from faststream.nats import NatsBroker from tests.brokers.base.connection import BrokerConnectionTestcase -from .conftest import Settings +from .settings import Settings @pytest.mark.connected() diff --git a/tests/brokers/redis/cluster/conftest.py b/tests/brokers/redis/cluster/conftest.py new file mode 100644 index 0000000000..ec8127c20d --- /dev/null +++ b/tests/brokers/redis/cluster/conftest.py @@ -0,0 +1,8 @@ +import pytest + +from .settings import SettingsCluster + + +@pytest.fixture(scope="session") +def settings_cluster(): + return SettingsCluster() diff --git a/tests/brokers/redis/cluster/settings.py b/tests/brokers/redis/cluster/settings.py new file mode 100644 index 0000000000..8100ef2b0b --- /dev/null +++ b/tests/brokers/redis/cluster/settings.py @@ -0,0 +1,11 @@ +from dataclasses import dataclass, field + + +@dataclass +class SettingsCluster: + url: str = "redis://127.0.0.1:7001" + host: str = "127.0.0.1" + port: int = 7001 + startup_nodes: list[tuple[str, int]] = field( + default_factory=lambda: [("127.0.0.1", 7002), ("127.0.0.1", 7003)], + ) diff --git a/tests/brokers/redis/cluster/test_cluster_connect.py b/tests/brokers/redis/cluster/test_cluster_connect.py index 7adbd9f485..64cd881c8e 100644 --- a/tests/brokers/redis/cluster/test_cluster_connect.py +++ b/tests/brokers/redis/cluster/test_cluster_connect.py @@ -13,7 +13,8 @@ from faststream.redis import RedisClusterBroker, StreamSub from tests.brokers.base.connection import BrokerConnectionTestcase -from tests.brokers.redis.conftest import SettingsCluster + +from .settings import SettingsCluster @pytest.mark.connected() diff --git a/tests/brokers/redis/cluster/test_cluster_pubsub_more.py b/tests/brokers/redis/cluster/test_cluster_pubsub_more.py index 0e1b38863d..2be8e404ba 100644 --- a/tests/brokers/redis/cluster/test_cluster_pubsub_more.py +++ b/tests/brokers/redis/cluster/test_cluster_pubsub_more.py @@ -5,7 +5,8 @@ from faststream.redis import RedisClusterBroker from tests.brokers.redis.basic import RedisClusterTestcaseConfig -from tests.brokers.redis.conftest import SettingsCluster + +from .settings import SettingsCluster @pytest.mark.slow() diff --git a/tests/brokers/redis/conftest.py b/tests/brokers/redis/conftest.py index 345631ac41..920f964e64 100644 --- a/tests/brokers/redis/conftest.py +++ b/tests/brokers/redis/conftest.py @@ -1,25 +1,8 @@ -from dataclasses import dataclass, field - import pytest from faststream.redis import RedisRouter - -@dataclass -class Settings: - url: str = "redis://localhost:6379" - host: str = "localhost" - port: int = 6379 - - -@dataclass -class SettingsCluster: - url: str = "redis://127.0.0.1:7001" - host: str = "127.0.0.1" - port: int = 7001 - startup_nodes: list[tuple[str, int]] = field( - default_factory=lambda: [("127.0.0.1", 7002), ("127.0.0.1", 7003)], - ) +from .settings import Settings @pytest.fixture(scope="session") @@ -27,11 +10,6 @@ def settings(): return Settings() -@pytest.fixture(scope="session") -def settings_cluster(): - return SettingsCluster() - - @pytest.fixture() def router(): return RedisRouter() diff --git a/tests/brokers/redis/settings.py b/tests/brokers/redis/settings.py new file mode 100644 index 0000000000..602a548bd7 --- /dev/null +++ b/tests/brokers/redis/settings.py @@ -0,0 +1,8 @@ +from dataclasses import dataclass + + +@dataclass +class Settings: + url: str = "redis://localhost:6379" + host: str = "localhost" + port: int = 6379 diff --git a/tests/brokers/redis/test_connect.py b/tests/brokers/redis/test_connect.py index eaa98400e2..1db88a6358 100644 --- a/tests/brokers/redis/test_connect.py +++ b/tests/brokers/redis/test_connect.py @@ -5,7 +5,7 @@ from faststream.redis import RedisBroker from tests.brokers.base.connection import BrokerConnectionTestcase -from .conftest import Settings +from .settings import Settings @pytest.mark.connected() From 3e9a2cefcfc3d551bc222c7d08a10f2454cc26cf Mon Sep 17 00:00:00 2001 From: Lancetnik Date: Wed, 29 Jul 2026 21:08:03 +0300 Subject: [PATCH 4/4] chore(tests): deduplicate batch key alignment tests The kafka and confluent test_batch_body.py modules were identical modulo naming: 500 lines whose only real differences were the import aliases and `m.key` vs `m.key()`. Worse, the confluent module imported `faststream.kafka.response` for its non-async tests, so confluent's own `batch_bodies` setter had no unit coverage at all. Move the shared cases into a `BatchKeysTestcase` in tests/brokers/base, driven by a single `KEY_ALIGNMENT_CASES` table, and leave each broker a thin subclass supplying its own classes and key accessor. Keep only the e2e test: `key_for()` is reached through the TestBroker batch publish path, so the middleware test covers the setter, _align_keys and key_for as a superset of what the unit test checked. Also drop `test_random_bodies` -- same inputs as `test_keys_order` with a membership assertion too weak to catch a mis-association -- folding its unique cases in as exact-equality params, and drop the redis `test_stream_same_group_executes_message_exactly_once` duplicate after strengthening the remaining test to assert the payload. Co-Authored-By: Claude Opus 5 (1M context) --- tests/brokers/base/publish_command.py | 151 +++++++++++- tests/brokers/confluent/test_batch_body.py | 256 +-------------------- tests/brokers/kafka/test_batch_body.py | 246 +------------------- tests/brokers/redis/test_test_client.py | 28 +-- 4 files changed, 170 insertions(+), 511 deletions(-) diff --git a/tests/brokers/base/publish_command.py b/tests/brokers/base/publish_command.py index 3fa1cad4a2..7a4c09d699 100644 --- a/tests/brokers/base/publish_command.py +++ b/tests/brokers/base/publish_command.py @@ -1,8 +1,9 @@ +from collections.abc import Sequence from typing import Any import pytest -from faststream import Response +from faststream import BaseMiddleware, Context, Response from faststream.response import ensure_response from faststream.response.response import ( BatchPublishCommand, @@ -75,3 +76,151 @@ def test_batch_bodies_empty_setter(self) -> None: assert cmd.batch_bodies == () assert cmd.body is None assert cmd.extra_bodies == () + + +BODY_A, BODY_B, BODY_C = "body-A", "body-B", "body-C" + + +def reverse(bodies: Sequence[Any]) -> tuple[Any, ...]: + return tuple(reversed(bodies)) + + +def drop_first(bodies: Sequence[Any]) -> tuple[Any, ...]: + return tuple(bodies[1:]) + + +def keep(bodies: Sequence[Any]) -> tuple[Any, ...]: + return tuple(bodies) + + +def dedup(bodies: Sequence[Any]) -> tuple[Any, ...]: + return tuple(dict.fromkeys(bodies)) + + +SHUFFLED = ( + (BODY_A, b"1"), + (BODY_C, b"2"), + (BODY_B, b"3"), + (BODY_A, b"4"), + (BODY_B, b"5"), + (BODY_A, b"6"), + (BODY_A, b"7"), +) + +# (original (body, key) pairs, batch_bodies mutation, expected (body, key) pairs) +KEY_ALIGNMENT_CASES = ( + pytest.param( + ((BODY_A, b"key-A"), (BODY_B, b"key-B")), + reverse, + ((BODY_B, b"key-B"), (BODY_A, b"key-A")), + id="reversed", + ), + pytest.param( + ((BODY_A, b"key-A"), (BODY_B, b"key-B"), (BODY_C, b"key-C")), + drop_first, + ((BODY_B, b"key-B"), (BODY_C, b"key-C")), + id="first-dropped", + ), + pytest.param( + ((BODY_A, b"key-A"), (BODY_B, b"key-B"), (BODY_B, b"key-B")), + keep, + ((BODY_A, b"key-A"), (BODY_B, b"key-B"), (BODY_B, b"key-B")), + id="untouched-with-duplicates", + ), + pytest.param( + ((BODY_A, b"key-A"), (BODY_B, b"key-B"), (BODY_B, b"key-B")), + reverse, + ((BODY_B, b"key-B"), (BODY_B, b"key-B"), (BODY_A, b"key-A")), + id="reversed-with-duplicates", + ), + pytest.param( + ((BODY_A, b"key-1"), (BODY_A, b"key-2"), (BODY_B, b"key-3")), + dedup, + ((BODY_A, b"key-1"), (BODY_B, b"key-3")), + id="deduplicated", + ), + pytest.param(SHUFFLED, keep, SHUFFLED, id="untouched-shuffled"), + # Every pair below is one of the original ones, but identical bodies keep their + # keys in the original relative order instead of being reversed along with them -- + # equal bodies are indistinguishable, so any of their keys is a valid match. + pytest.param( + SHUFFLED, + reverse, + ( + (BODY_A, b"1"), + (BODY_A, b"4"), + (BODY_B, b"3"), + (BODY_A, b"6"), + (BODY_B, b"5"), + (BODY_C, b"2"), + (BODY_A, b"7"), + ), + id="reversed-shuffled", + ), + pytest.param( + SHUFFLED, + drop_first, + SHUFFLED[1:], + id="first-dropped-shuffled", + marks=pytest.mark.xfail( + reason="Identical bodies can't be traced back to their keys once a body is removed, since keys are not unique.", + ), + ), +) + + +class BatchKeysTestcase: + """Per-message keys must keep following their bodies when `batch_bodies` is replaced. + + Applies to Kafka-like brokers, where every batch element carries its own key. + """ + + publish_command_cls: type[BatchPublishCommand] + publish_message_cls: type[Any] + + @staticmethod + def get_message_key(raw_message: Any) -> bytes: + return raw_message.key + + @pytest.mark.asyncio() + @pytest.mark.parametrize( + ("pairs", "mutate", "expected"), + KEY_ALIGNMENT_CASES, + ) + async def test_publish_middleware_keeps_keys_aligned( + self, + queue: str, + pairs, + mutate, + expected, + ) -> None: + publish_command_cls = self.publish_command_cls + + class MutatingMiddleware(BaseMiddleware): + async def publish_scope(self, call_next, cmd): + if isinstance(cmd, publish_command_cls): + cmd.batch_bodies = mutate(cmd.batch_bodies) + return await call_next(cmd) + + broker = self.get_broker(apply_types=True, middlewares=(MutatingMiddleware,)) + + received = [] + + @broker.subscriber(queue, batch=True) + async def handler(msgs, raw=Context("message")) -> None: + received.extend( + zip( + msgs, + (self.get_message_key(m) for m in raw.raw_message), + strict=True, + ), + ) + + async with self.patch_broker(broker) as br: + await br.start() + await br.publish_batch( + *(self.publish_message_cls(body, key=key) for body, key in pairs), + topic=queue, + ) + + assert tuple(received) == expected diff --git a/tests/brokers/confluent/test_batch_body.py b/tests/brokers/confluent/test_batch_body.py index 33b6b16989..2d79e44c00 100644 --- a/tests/brokers/confluent/test_batch_body.py +++ b/tests/brokers/confluent/test_batch_body.py @@ -1,252 +1,18 @@ -import pytest - -from faststream import BaseMiddleware, Context -from faststream.confluent import KafkaBroker, TestKafkaBroker -from faststream.confluent.response import ( - KafkaPublishCommand as ConfluentPublishCommand, - KafkaPublishMessage as ConfluentPublishMessage, -) -from faststream.kafka.response import KafkaPublishCommand, KafkaPublishMessage -from faststream.response.publish_type import PublishType - -body_a = "body-A" -body_b = "body-B" -body_c = "body-C" - - -def delivered(cmd): - return [(body, cmd.key_for(i)) for i, body in enumerate(cmd.batch_bodies)] - - -def reverse_bodies(batch_bodies): - return tuple(reversed(batch_bodies)) - - -def remove_body(batch_bodies): - return tuple(batch_bodies[1:]) - - -def do_nothing(batch_bodies): - return batch_bodies - - -def dedup_bodies(batch_bodies): - seen = set() - deduped = [] - for body in batch_bodies: - if body not in seen: - seen.add(body) - deduped.append(body) - return tuple(deduped) - - -@pytest.mark.kafka() -@pytest.mark.parametrize( - ("kafka_messages", "changing_pattern", "expected"), - ( - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-C", key=b"key-C"), - ], - remove_body, - [("body-B", b"key-B"), ("body-C", b"key-C")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - do_nothing, - [("body-A", b"key-A"), ("body-B", b"key-B"), ("body-B", b"key-B")], - ), - ( - [ - KafkaPublishMessage(body_a, key=b"key-1"), - KafkaPublishMessage(body_c, key=b"key-2"), - KafkaPublishMessage(body_b, key=b"key-3"), - KafkaPublishMessage(body_a, key=b"key-4"), - KafkaPublishMessage(body_b, key=b"key-5"), - KafkaPublishMessage(body_a, key=b"key-6"), - KafkaPublishMessage(body_a, key=b"key-7"), - KafkaPublishMessage(body_b, key=b"key-8"), - ], - do_nothing, - [ - ("body-A", b"key-1"), - ("body-C", b"key-2"), - ("body-B", b"key-3"), - ("body-A", b"key-4"), - ("body-B", b"key-5"), - ("body-A", b"key-6"), - ("body-A", b"key-7"), - ("body-B", b"key-8"), - ], - ), - ), -) -def test_keys_order(kafka_messages, changing_pattern, expected) -> None: - - cmd = KafkaPublishCommand( - *kafka_messages, - topic="topic", - _publish_type=PublishType.PUBLISH, - ) - cmd.batch_bodies = changing_pattern(cmd.batch_bodies) - delivered_cmd = delivered(cmd) - assert delivered_cmd == expected +from typing import Any +import pytest -@pytest.mark.kafka() -@pytest.mark.parametrize( - ("kafka_messages", "changing_pattern", "expected"), - ( - ( - [ - KafkaPublishMessage(body_a, key=b"key-1"), - KafkaPublishMessage(body_c, key=b"key-2"), - KafkaPublishMessage(body_b, key=b"key-3"), - KafkaPublishMessage(body_a, key=b"key-4"), - KafkaPublishMessage(body_b, key=b"key-5"), - KafkaPublishMessage(body_a, key=b"key-6"), - KafkaPublishMessage(body_a, key=b"key-7"), - KafkaPublishMessage(body_b, key=b"key-8"), - ], - reverse_bodies, - [ - ("body-A", b"key-1"), - ("body-C", b"key-2"), - ("body-B", b"key-3"), - ("body-A", b"key-4"), - ("body-B", b"key-5"), - ("body-A", b"key-6"), - ("body-A", b"key-7"), - ("body-B", b"key-8"), - ], - ), - pytest.param( - [ - KafkaPublishMessage(body_a, key=b"key-1"), - KafkaPublishMessage(body_c, key=b"key-2"), - KafkaPublishMessage(body_b, key=b"key-3"), - KafkaPublishMessage(body_a, key=b"key-4"), - KafkaPublishMessage(body_b, key=b"key-5"), - KafkaPublishMessage(body_a, key=b"key-6"), - KafkaPublishMessage(body_a, key=b"key-7"), - KafkaPublishMessage(body_b, key=b"key-8"), - ], - remove_body, - [ - ("body-C", b"key-2"), - ("body-B", b"key-3"), - ("body-A", b"key-4"), - ("body-B", b"key-5"), - ("body-A", b"key-6"), - ("body-A", b"key-7"), - ("body-B", b"key-8"), - ], - marks=pytest.mark.xfail( - reason="It is not possible to track the relationship between identical bodies and keys after removing bodies, as the keys are not unique." - ), - ), - ), -) -def test_random_bodies(kafka_messages, changing_pattern, expected) -> None: +from faststream.confluent.response import KafkaPublishCommand, KafkaPublishMessage +from tests.brokers.base.publish_command import BatchKeysTestcase - cmd = KafkaPublishCommand( - *kafka_messages, - topic="topic", - _publish_type=PublishType.PUBLISH, - ) - cmd.batch_bodies = changing_pattern(cmd.batch_bodies) - delivered_cmd = delivered(cmd) - for body in expected: - assert body in delivered_cmd +from .basic import ConfluentMemoryTestcaseConfig @pytest.mark.confluent() -@pytest.mark.asyncio() -@pytest.mark.parametrize( - ("kafka_messages", "changing_pattern", "expected"), - ( - ( - [ - ConfluentPublishMessage("body-A", key=b"key-A"), - ConfluentPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - ConfluentPublishMessage("body-A", key=b"key-A"), - ConfluentPublishMessage("body-B", key=b"key-B"), - ConfluentPublishMessage("body-C", key=b"key-C"), - ], - remove_body, - [("body-B", b"key-B"), ("body-C", b"key-C")], - ), - ( - [ - ConfluentPublishMessage("body-A", key=b"key-A"), - ConfluentPublishMessage("body-B", key=b"key-B"), - ConfluentPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - ConfluentPublishMessage("body-A", key=b"key-1"), - ConfluentPublishMessage("body-A", key=b"key-2"), - ConfluentPublishMessage("body-B", key=b"key-3"), - ], - dedup_bodies, - [("body-A", b"key-1"), ("body-B", b"key-3")], - ), - ), -) -async def test_publish_middleware_keeps_keys_aligned( - queue: str, - kafka_messages, - changing_pattern, - expected, -) -> None: - received = [] - - class MutatingMiddleware(BaseMiddleware): - async def publish_scope(self, call_next, cmd): - if isinstance(cmd, ConfluentPublishCommand): - cmd.batch_bodies = changing_pattern(cmd.batch_bodies) - return await call_next(cmd) - - broker = KafkaBroker(apply_types=True, middlewares=(MutatingMiddleware,)) - - @broker.subscriber(queue, batch=True) - async def handler(msgs, raw=Context("message")) -> None: - received.extend(zip(msgs, (m.key() for m in raw.raw_message), strict=True)) - - async with TestKafkaBroker(broker) as br: - await br.start() - await br.publish_batch(*kafka_messages, topic=queue) +class TestBatchKeys(ConfluentMemoryTestcaseConfig, BatchKeysTestcase): + publish_command_cls = KafkaPublishCommand + publish_message_cls = KafkaPublishMessage - assert received == expected + @staticmethod + def get_message_key(raw_message: Any) -> bytes: + return raw_message.key() diff --git a/tests/brokers/kafka/test_batch_body.py b/tests/brokers/kafka/test_batch_body.py index 3c80ccba77..aedfbe5a53 100644 --- a/tests/brokers/kafka/test_batch_body.py +++ b/tests/brokers/kafka/test_batch_body.py @@ -1,248 +1,12 @@ import pytest -from faststream import BaseMiddleware, Context -from faststream.kafka import KafkaBroker, TestKafkaBroker from faststream.kafka.response import KafkaPublishCommand, KafkaPublishMessage -from faststream.response.publish_type import PublishType +from tests.brokers.base.publish_command import BatchKeysTestcase -body_a = "body-A" -body_b = "body-B" -body_c = "body-C" - - -def delivered(cmd): - return [(body, cmd.key_for(i)) for i, body in enumerate(cmd.batch_bodies)] - - -def reverse_bodies(batch_bodies): - return tuple(reversed(batch_bodies)) - - -def remove_body(batch_bodies): - return tuple(batch_bodies[1:]) - - -def do_nothing(batch_bodies): - return batch_bodies - - -def dedup_bodies(batch_bodies): - seen = set() - deduped = [] - for body in batch_bodies: - if body not in seen: - seen.add(body) - deduped.append(body) - return tuple(deduped) +from .basic import KafkaMemoryTestcaseConfig @pytest.mark.kafka() -@pytest.mark.parametrize( - ("kafka_messages", "changing_pattern", "expected"), - ( - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-C", key=b"key-C"), - ], - remove_body, - [("body-B", b"key-B"), ("body-C", b"key-C")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - do_nothing, - [("body-A", b"key-A"), ("body-B", b"key-B"), ("body-B", b"key-B")], - ), - ( - [ - KafkaPublishMessage(body_a, key=b"key-1"), - KafkaPublishMessage(body_c, key=b"key-2"), - KafkaPublishMessage(body_b, key=b"key-3"), - KafkaPublishMessage(body_a, key=b"key-4"), - KafkaPublishMessage(body_b, key=b"key-5"), - KafkaPublishMessage(body_a, key=b"key-6"), - KafkaPublishMessage(body_a, key=b"key-7"), - KafkaPublishMessage(body_b, key=b"key-8"), - ], - do_nothing, - [ - ("body-A", b"key-1"), - ("body-C", b"key-2"), - ("body-B", b"key-3"), - ("body-A", b"key-4"), - ("body-B", b"key-5"), - ("body-A", b"key-6"), - ("body-A", b"key-7"), - ("body-B", b"key-8"), - ], - ), - ), -) -def test_keys_order(kafka_messages, changing_pattern, expected) -> None: - - cmd = KafkaPublishCommand( - *kafka_messages, - topic="topic", - _publish_type=PublishType.PUBLISH, - ) - cmd.batch_bodies = changing_pattern(cmd.batch_bodies) - delivered_cmd = delivered(cmd) - assert delivered_cmd == expected - - -@pytest.mark.kafka() -@pytest.mark.parametrize( - ("kafka_messages", "changing_pattern", "expected"), - ( - ( - [ - KafkaPublishMessage(body_a, key=b"key-1"), - KafkaPublishMessage(body_c, key=b"key-2"), - KafkaPublishMessage(body_b, key=b"key-3"), - KafkaPublishMessage(body_a, key=b"key-4"), - KafkaPublishMessage(body_b, key=b"key-5"), - KafkaPublishMessage(body_a, key=b"key-6"), - KafkaPublishMessage(body_a, key=b"key-7"), - KafkaPublishMessage(body_b, key=b"key-8"), - ], - reverse_bodies, - [ - ("body-A", b"key-1"), - ("body-C", b"key-2"), - ("body-B", b"key-3"), - ("body-A", b"key-4"), - ("body-B", b"key-5"), - ("body-A", b"key-6"), - ("body-A", b"key-7"), - ("body-B", b"key-8"), - ], - ), - pytest.param( - [ - KafkaPublishMessage(body_a, key=b"key-1"), - KafkaPublishMessage(body_c, key=b"key-2"), - KafkaPublishMessage(body_b, key=b"key-3"), - KafkaPublishMessage(body_a, key=b"key-4"), - KafkaPublishMessage(body_b, key=b"key-5"), - KafkaPublishMessage(body_a, key=b"key-6"), - KafkaPublishMessage(body_a, key=b"key-7"), - KafkaPublishMessage(body_b, key=b"key-8"), - ], - remove_body, - [ - ("body-C", b"key-2"), - ("body-B", b"key-3"), - ("body-A", b"key-4"), - ("body-B", b"key-5"), - ("body-A", b"key-6"), - ("body-A", b"key-7"), - ("body-B", b"key-8"), - ], - marks=pytest.mark.xfail( - reason="It is not possible to track the relationship between identical bodies and keys after removing bodies, as the keys are not unique." - ), - ), - ), -) -def test_random_bodies(kafka_messages, changing_pattern, expected) -> None: - - cmd = KafkaPublishCommand( - *kafka_messages, - topic="topic", - _publish_type=PublishType.PUBLISH, - ) - cmd.batch_bodies = changing_pattern(cmd.batch_bodies) - delivered_cmd = delivered(cmd) - for body in expected: - assert body in delivered_cmd - - -@pytest.mark.kafka() -@pytest.mark.asyncio() -@pytest.mark.parametrize( - ("kafka_messages", "changing_pattern", "expected"), - ( - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-C", key=b"key-C"), - ], - remove_body, - [("body-B", b"key-B"), ("body-C", b"key-C")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-A"), - KafkaPublishMessage("body-B", key=b"key-B"), - KafkaPublishMessage("body-B", key=b"key-B"), - ], - reverse_bodies, - [("body-B", b"key-B"), ("body-B", b"key-B"), ("body-A", b"key-A")], - ), - ( - [ - KafkaPublishMessage("body-A", key=b"key-1"), - KafkaPublishMessage("body-A", key=b"key-2"), - KafkaPublishMessage("body-B", key=b"key-3"), - ], - dedup_bodies, - [("body-A", b"key-1"), ("body-B", b"key-3")], - ), - ), -) -async def test_publish_middleware_keeps_keys_aligned( - queue: str, - kafka_messages, - changing_pattern, - expected, -) -> None: - received = [] - - class MutatingMiddleware(BaseMiddleware): - async def publish_scope(self, call_next, cmd): - if isinstance(cmd, KafkaPublishCommand): - cmd.batch_bodies = changing_pattern(cmd.batch_bodies) - return await call_next(cmd) - - broker = KafkaBroker(apply_types=True, middlewares=(MutatingMiddleware,)) - - @broker.subscriber(queue, batch=True) - async def handler(msgs, raw=Context("message")) -> None: - received.extend(zip(msgs, (m.key for m in raw.raw_message), strict=True)) - - async with TestKafkaBroker(broker) as br: - await br.start() - await br.publish_batch(*kafka_messages, topic=queue) - - assert received == expected +class TestBatchKeys(KafkaMemoryTestcaseConfig, BatchKeysTestcase): + publish_command_cls = KafkaPublishCommand + publish_message_cls = KafkaPublishMessage diff --git a/tests/brokers/redis/test_test_client.py b/tests/brokers/redis/test_test_client.py index 19e4887ea5..1118d8ee09 100644 --- a/tests/brokers/redis/test_test_client.py +++ b/tests/brokers/redis/test_test_client.py @@ -222,30 +222,10 @@ async def subscriber2(msg) -> None: ... async with self.patch_broker(broker) as br: await br.publish("hello", stream=queue) - # only one consumer in the group should receive the message - assert {subscriber1.mock.call_count, subscriber2.mock.call_count} == { - 0, - 1, - } - - async def test_stream_same_group_executes_message_exactly_once( - self, - queue: str, - ) -> None: - broker = self.get_broker() - - @broker.subscriber( - stream=StreamSub(queue, group="workers", consumer="claimer"), - ) - @broker.subscriber( - stream=StreamSub(queue, group="workers", consumer="w1"), - ) - async def worker(msg: str) -> None: ... - - async with self.patch_broker(broker) as br: - await br.publish("hello", stream=queue) - - worker.mock.assert_called_once_with("hello") + # exactly one consumer of the group handles the message + called = [m for m in (subscriber1.mock, subscriber2.mock) if m.call_count] + assert len(called) == 1 + called[0].assert_called_once_with("hello") async def test_stream_different_groups_all_receive( self,