fix (fastapi): skip init=False fields when copying Dependant

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

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
     })
 
 

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

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")

tests/brokers/confluent/settings.py

@@ -0,0 +1,6 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Settings:
+    url: str = "localhost:9092"