tests: fix tests

faststream/_internal/broker/broker.py

@@ -281,12 +281,12 @@ def setup_publisher(
     @property
     def _subscriber_setup_extra(self) -> "AnyDict":
         return {
-            "logger": self._state.logger_state.logger,
+            "logger": self._state.logger_state.logger.logger,
             "producer": self._producer,
             "graceful_timeout": self.graceful_timeout,
             "extra_context": {
                 "broker": self,
-                "logger": self._state.logger_state.logger,
+                "logger": self._state.logger_state.logger.logger,
             },
             # broker options
             "broker_parser": self._parser,

faststream/_internal/cli/main.py

@@ -250,6 +250,7 @@ def publish(
         if not app_obj.broker:
             raise ValueError("Broker instance not found in the app.")
 
+        app_obj._setup()
         result = anyio.run(publish_message, app_obj.broker, rpc, extra)
 
         if rpc:

faststream/_internal/cli/utils/logs.py

@@ -69,6 +69,9 @@ def set_log_level(level: int, app: "FastStream") -> None:
     if app.logger and getattr(app.logger, "setLevel", None):
         app.logger.setLevel(level)  # type: ignore[attr-defined]
 
-    broker_logger: Optional[LoggerProto] = getattr(app.broker, "logger", None)
-    if broker_logger is not None and getattr(broker_logger, "setLevel", None):
-        broker_logger.setLevel(level)  # type: ignore[attr-defined]
+    if app.broker:
+        broker_logger: Optional[LoggerProto] = (
+            app.broker._state.logger_state.logger.logger
+        )
+        if broker_logger is not None and getattr(broker_logger, "setLevel", None):
+            broker_logger.setLevel(level)  # type: ignore[attr-defined]

faststream/_internal/setup/fast_depends.py

@@ -5,7 +5,7 @@
     from faststream._internal.basic_types import Decorator
 
 
-@dataclass(slots=True)
+@dataclass
 class FastDependsData:
     apply_types: bool
     is_validate: bool

faststream/_internal/setup/logger.py

@@ -43,6 +43,8 @@ def make_logger_state(
 
 
 class _LoggerObject(Protocol):
+    logger: Optional["LoggerProto"]
+
     def log(
         self,
         message: str,
@@ -53,6 +55,9 @@ def log(
 
 
 class _NotSetLoggerObject(_LoggerObject):
+    def __init__(self) -> None:
+        self.logger = None
+
     def log(
         self,
         message: str,
@@ -64,6 +69,9 @@ def log(
 
 
 class _EmptyLoggerObject(_LoggerObject):
+    def __init__(self) -> None:
+        self.logger = None
+
     def log(
         self,
         message: str,
@@ -123,7 +131,7 @@ def __init__(self, log_fmt: Optional[str]) -> None:
         self._log_fmt = log_fmt
 
 
-@dataclass(slots=True)
+@dataclass
 class LoggerState(SetupAble):
     log_level: int
     params_storage: LoggerParamsStorage

tests/cli/rabbit/test_app.py

@@ -130,7 +130,9 @@ async def call2():
 
 
 @pytest.mark.asyncio
-async def test_startup_lifespan_before_broker_started(async_mock, app: FastStream):
+async def test_startup_lifespan_before_broker_started(
+    async_mock: AsyncMock, app: FastStream
+):
     @app.on_startup
     async def call():
         await async_mock.before()
@@ -171,7 +173,7 @@ async def call2():
 
 @pytest.mark.asyncio
 async def test_shutdown_lifespan_after_broker_stopped(
-    mock, async_mock, app: FastStream
+    mock, async_mock: AsyncMock, app: FastStream
 ):
     @app.after_shutdown
     async def call():
@@ -192,14 +194,15 @@ async def call_before():
 
 
 @pytest.mark.asyncio
-async def test_running(async_mock, app: FastStream):
+async def test_running(async_mock: AsyncMock, app: FastStream):
     app.exit()
 
     with patch.object(app.broker, "start", async_mock.broker_run), patch.object(
-        app.broker, "close", async_mock.broker_stopped
-    ):
+        app.broker, "connect", async_mock.broker_connect
+    ), patch.object(app.broker, "close", async_mock.broker_stopped):
         await app.run()
 
+    async_mock.broker_connect.assert_called_once()
     async_mock.broker_run.assert_called_once()
     async_mock.broker_stopped.assert_called_once()
 
@@ -217,7 +220,9 @@ async def f():
 
 
 @pytest.mark.asyncio
-async def test_running_lifespan_contextmanager(async_mock, mock: Mock, app: FastStream):
+async def test_running_lifespan_contextmanager(
+    async_mock: AsyncMock, mock: Mock, app: FastStream
+):
     @asynccontextmanager
     async def lifespan(env: str):
         mock.on(env)
@@ -228,10 +233,11 @@ async def lifespan(env: str):
     app.exit()
 
     with patch.object(app.broker, "start", async_mock.broker_run), patch.object(
-        app.broker, "close", async_mock.broker_stopped
-    ):
+        app.broker, "connect", async_mock.broker_connect
+    ), patch.object(app.broker, "close", async_mock.broker_stopped):
         await app.run(run_extra_options={"env": "test"})
 
+    async_mock.broker_connect.assert_called_once()
     async_mock.broker_run.assert_called_once()
     async_mock.broker_stopped.assert_called_once()
 
@@ -241,30 +247,32 @@ async def lifespan(env: str):
 
 @pytest.mark.asyncio
 @pytest.mark.skipif(IS_WINDOWS, reason="does not run on windows")
-async def test_stop_with_sigint(async_mock, app: FastStream):
-    with patch.object(app.broker, "start", async_mock.broker_run_sigint), patch.object(
-        app.broker, "close", async_mock.broker_stopped_sigint
-    ):
+async def test_stop_with_sigint(async_mock: AsyncMock, app: FastStream):
+    with patch.object(app.broker, "start", async_mock.broker_run), patch.object(
+        app.broker, "connect", async_mock.broker_connect
+    ), patch.object(app.broker, "close", async_mock.broker_stopped):
         async with anyio.create_task_group() as tg:
             tg.start_soon(app.run)
             tg.start_soon(_kill, signal.SIGINT)
 
-    async_mock.broker_run_sigint.assert_called_once()
-    async_mock.broker_stopped_sigint.assert_called_once()
+    async_mock.broker_connect.assert_called_once()
+    async_mock.broker_run.assert_called_once()
+    async_mock.broker_stopped.assert_called_once()
 
 
 @pytest.mark.asyncio
 @pytest.mark.skipif(IS_WINDOWS, reason="does not run on windows")
-async def test_stop_with_sigterm(async_mock, app: FastStream):
-    with patch.object(app.broker, "start", async_mock.broker_run_sigterm), patch.object(
-        app.broker, "close", async_mock.broker_stopped_sigterm
-    ):
+async def test_stop_with_sigterm(async_mock: AsyncMock, app: FastStream):
+    with patch.object(app.broker, "start", async_mock.broker_run), patch.object(
+        app.broker, "connect", async_mock.broker_connect
+    ), patch.object(app.broker, "close", async_mock.broker_stopped):
         async with anyio.create_task_group() as tg:
             tg.start_soon(app.run)
             tg.start_soon(_kill, signal.SIGTERM)
 
-    async_mock.broker_run_sigterm.assert_called_once()
-    async_mock.broker_stopped_sigterm.assert_called_once()
+    async_mock.broker_connect.assert_called_once()
+    async_mock.broker_run.assert_called_once()
+    async_mock.broker_stopped.assert_called_once()
 
 
 @pytest.mark.asyncio

tests/cli/rabbit/test_logs.py

@@ -20,12 +20,14 @@
 )
 def test_set_level(level, app: FastStream):
     level = get_log_level(level)
+    app._setup()
     set_log_level(level, app)
-    assert app.logger.level is app.broker.logger.level is level
+    broker_logger = app.broker._state.logger_state.logger.logger
+    assert app.logger.level is broker_logger.level is level
 
 
 @pytest.mark.parametrize(
-    ("level", "broker"),
+    ("level", "app"),
     (  # noqa: PT007
         pytest.param(
             logging.CRITICAL,
@@ -50,6 +52,7 @@ def test_set_level(level, app: FastStream):
     ),
 )
 def test_set_level_to_none(level, app: FastStream):
+    app._setup()
     set_log_level(get_log_level(level), app)
 
 

tests/cli/test_publish.py

@@ -1,3 +1,4 @@
+from typing import Tuple
 from unittest.mock import AsyncMock, patch
 
 from dirty_equals import IsPartialDict
@@ -14,23 +15,23 @@
 )
 
 
-def get_mock_app(broker_type, producer_type) -> FastStream:
+def get_mock_app(broker_type, producer_type) -> Tuple[FastStream, AsyncMock]:
     broker = broker_type()
     broker.connect = AsyncMock()
     mock_producer = AsyncMock(spec=producer_type)
     mock_producer.publish = AsyncMock()
     mock_producer._parser = AsyncMock()
     mock_producer._decoder = AsyncMock()
     broker._producer = mock_producer
-    return FastStream(broker)
+    return FastStream(broker), mock_producer
 
 
 @require_redis
 def test_publish_command_with_redis_options(runner):
     from faststream.redis import RedisBroker
     from faststream.redis.publisher.producer import RedisFastProducer
 
-    mock_app = get_mock_app(RedisBroker, RedisFastProducer)
+    mock_app, producer_mock = get_mock_app(RedisBroker, RedisFastProducer)
 
     with patch(
         "faststream._internal.cli.main.import_from_string",
@@ -57,8 +58,8 @@ def test_publish_command_with_redis_options(runner):
 
         assert result.exit_code == 0
 
-        assert mock_app.broker._producer.publish.call_args.args[0] == "hello world"
-        assert mock_app.broker._producer.publish.call_args.kwargs == IsPartialDict(
+        assert producer_mock.publish.call_args.args[0] == "hello world"
+        assert producer_mock.publish.call_args.kwargs == IsPartialDict(
             reply_to="tester",
             stream="streamname",
             list="listname",
@@ -72,7 +73,7 @@ def test_publish_command_with_confluent_options(runner):
     from faststream.confluent import KafkaBroker as ConfluentBroker
     from faststream.confluent.publisher.producer import AsyncConfluentFastProducer
 
-    mock_app = get_mock_app(ConfluentBroker, AsyncConfluentFastProducer)
+    mock_app, producer_mock = get_mock_app(ConfluentBroker, AsyncConfluentFastProducer)
 
     with patch(
         "faststream._internal.cli.main.import_from_string",
@@ -92,8 +93,9 @@ def test_publish_command_with_confluent_options(runner):
         )
 
         assert result.exit_code == 0
-        assert mock_app.broker._producer.publish.call_args.args[0] == "hello world"
-        assert mock_app.broker._producer.publish.call_args.kwargs == IsPartialDict(
+
+        assert producer_mock.publish.call_args.args[0] == "hello world"
+        assert producer_mock.publish.call_args.kwargs == IsPartialDict(
             topic="topicname",
             correlation_id="someId",
         )
@@ -104,7 +106,7 @@ def test_publish_command_with_kafka_options(runner):
     from faststream.kafka import KafkaBroker
     from faststream.kafka.publisher.producer import AioKafkaFastProducer
 
-    mock_app = get_mock_app(KafkaBroker, AioKafkaFastProducer)
+    mock_app, producer_mock = get_mock_app(KafkaBroker, AioKafkaFastProducer)
 
     with patch(
         "faststream._internal.cli.main.import_from_string",
@@ -124,8 +126,8 @@ def test_publish_command_with_kafka_options(runner):
         )
 
         assert result.exit_code == 0
-        assert mock_app.broker._producer.publish.call_args.args[0] == "hello world"
-        assert mock_app.broker._producer.publish.call_args.kwargs == IsPartialDict(
+        assert producer_mock.publish.call_args.args[0] == "hello world"
+        assert producer_mock.publish.call_args.kwargs == IsPartialDict(
             topic="topicname",
             correlation_id="someId",
         )
@@ -136,7 +138,7 @@ def test_publish_command_with_nats_options(runner):
     from faststream.nats import NatsBroker
     from faststream.nats.publisher.producer import NatsFastProducer
 
-    mock_app = get_mock_app(NatsBroker, NatsFastProducer)
+    mock_app, producer_mock = get_mock_app(NatsBroker, NatsFastProducer)
 
     with patch(
         "faststream._internal.cli.main.import_from_string",
@@ -159,8 +161,8 @@ def test_publish_command_with_nats_options(runner):
 
         assert result.exit_code == 0
 
-        assert mock_app.broker._producer.publish.call_args.args[0] == "hello world"
-        assert mock_app.broker._producer.publish.call_args.kwargs == IsPartialDict(
+        assert producer_mock.publish.call_args.args[0] == "hello world"
+        assert producer_mock.publish.call_args.kwargs == IsPartialDict(
             subject="subjectname",
             reply_to="tester",
             correlation_id="someId",
@@ -172,7 +174,7 @@ def test_publish_command_with_rabbit_options(runner):
     from faststream.rabbit import RabbitBroker
     from faststream.rabbit.publisher.producer import AioPikaFastProducer
 
-    mock_app = get_mock_app(RabbitBroker, AioPikaFastProducer)
+    mock_app, producer_mock = get_mock_app(RabbitBroker, AioPikaFastProducer)
 
     with patch(
         "faststream._internal.cli.main.import_from_string",
@@ -191,8 +193,8 @@ def test_publish_command_with_rabbit_options(runner):
 
         assert result.exit_code == 0
 
-        assert mock_app.broker._producer.publish.call_args.args[0] == "hello world"
-        assert mock_app.broker._producer.publish.call_args.kwargs == IsPartialDict(
+        assert producer_mock.publish.call_args.args[0] == "hello world"
+        assert producer_mock.publish.call_args.kwargs == IsPartialDict(
             {
                 "correlation_id": "someId",
             }
@@ -204,7 +206,7 @@ def test_publish_nats_request_command(runner: CliRunner):
     from faststream.nats import NatsBroker
     from faststream.nats.publisher.producer import NatsFastProducer
 
-    mock_app = get_mock_app(NatsBroker, NatsFastProducer)
+    mock_app, producer_mock = get_mock_app(NatsBroker, NatsFastProducer)
 
     with patch(
         "faststream._internal.cli.main.import_from_string",
@@ -224,8 +226,8 @@ def test_publish_nats_request_command(runner: CliRunner):
             ],
         )
 
-        assert mock_app.broker._producer.request.call_args.args[0] == "hello world"
-        assert mock_app.broker._producer.request.call_args.kwargs == IsPartialDict(
+        assert producer_mock.request.call_args.args[0] == "hello world"
+        assert producer_mock.request.call_args.kwargs == IsPartialDict(
             subject="subjectname",
             timeout=1.0,
         )