diff --git a/.agents/skills/testing-patterns/SKILL.md b/.agents/skills/testing-patterns/SKILL.md index 42cbec61a1d..2c87aaa2bab 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/faststream/_internal/fastapi/get_dependant.py b/faststream/_internal/fastapi/get_dependant.py index d84a6b8f79f..f91f05841ec 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 }) diff --git a/tests/brokers/base/publish_command.py b/tests/brokers/base/publish_command.py index 3fa1cad4a2a..7a4c09d6990 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/conftest.py b/tests/brokers/confluent/conftest.py index 44e82178c6d..86ae4b5592c 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 00000000000..3ceb1d6a125 --- /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_batch_body.py b/tests/brokers/confluent/test_batch_body.py index 33b6b169893..2d79e44c00c 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/confluent/test_connect.py b/tests/brokers/confluent/test_connect.py index 72e8e21038f..2a9ae010380 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 1eede17ed78..f856820f8f6 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 00000000000..3ceb1d6a125 --- /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_batch_body.py b/tests/brokers/kafka/test_batch_body.py index 3c80ccba775..aedfbe5a534 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/kafka/test_connect.py b/tests/brokers/kafka/test_connect.py index 2829d525b58..d72af8af69a 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 b0a3a8bb813..900a19e5e7c 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 00000000000..4b9eccd5643 --- /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 6a89b47cb48..7b6121cc627 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 09e34659ebf..d182a0ffec5 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 00000000000..034e35eec0c --- /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 c3ffec801b5..50212430fce 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/__init__.py b/tests/brokers/redis/cluster/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/brokers/redis/cluster/conftest.py b/tests/brokers/redis/cluster/conftest.py new file mode 100644 index 00000000000..ec8127c20d3 --- /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 00000000000..8100ef2b0be --- /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/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 1b8fc35829c..0b9f9a8800f 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 c356eb9f4f6..64cd881c8ed 100644 --- a/tests/brokers/redis/test_cluster_connect.py +++ b/tests/brokers/redis/cluster/test_cluster_connect.py @@ -14,7 +14,7 @@ from faststream.redis import RedisClusterBroker, StreamSub from tests.brokers.base.connection import BrokerConnectionTestcase -from .conftest import SettingsCluster +from .settings 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 102b92b053c..4965bbff436 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 42840d7082c..022c552a02b 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 42d115290d5..605be9a64cf 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 53bdca51e90..1747a2bb45c 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 cc84693959a..2be8e404ba6 100644 --- a/tests/brokers/redis/test_cluster_pubsub_more.py +++ b/tests/brokers/redis/cluster/test_cluster_pubsub_more.py @@ -4,9 +4,9 @@ import pytest from faststream.redis import RedisClusterBroker +from tests.brokers.redis.basic import RedisClusterTestcaseConfig -from .basic import RedisClusterTestcaseConfig -from .conftest import SettingsCluster +from .settings 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 927dc7c56ad..58187ba0ad0 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 579b834a413..01c187ccab4 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/conftest.py b/tests/brokers/redis/conftest.py index 345631ac41f..920f964e64d 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/sentinel/__init__.py b/tests/brokers/redis/sentinel/__init__.py new file mode 100644 index 00000000000..e69de29bb2d 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/tests/brokers/redis/settings.py b/tests/brokers/redis/settings.py new file mode 100644 index 00000000000..602a548bd78 --- /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 eaa98400e2a..1db88a6358c 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() diff --git a/tests/brokers/redis/test_test_client.py b/tests/brokers/redis/test_test_client.py index 19e4887ea58..1118d8ee09d 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, diff --git a/uv.lock b/uv.lock index 1477bd4b947..46016e05db2 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" },