"""Tests for M5-T10: MqttManager + POST /api/config/mqtt/test. All paho interaction is mocked — no real broker is required. """ from __future__ import annotations import threading from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from app.integrations.mqtt import MqttManager, mqtt_source_client_id, mqtt_test_client_id # --------------------------------------------------------------------------- # Helpers / fixtures # --------------------------------------------------------------------------- def _make_settings( *, mqtt_enabled: bool = True, mqtt_broker_host: str = "broker.test", mqtt_broker_port: int = 1883, mqtt_username: str = "", mqtt_password: str = "", mqtt_tls_enabled: bool = False, mqtt_client_id: str = "home-automation", ha_discovery_prefix: str = "homeassistant", ): """Return a simple namespace that acts like a Settings object for MqttManager tests.""" s = MagicMock() s.mqtt_enabled = mqtt_enabled s.mqtt_broker_host = mqtt_broker_host s.mqtt_broker_port = mqtt_broker_port s.mqtt_username = mqtt_username s.mqtt_password = mqtt_password s.mqtt_tls_enabled = mqtt_tls_enabled s.mqtt_client_id = mqtt_client_id s.ha_discovery_prefix = ha_discovery_prefix return s def _login(client: TestClient) -> None: resp = client.post( "/api/auth/login", json={"username": "admin", "password": "test-password"}, ) assert resp.status_code == 200, f"Login failed: {resp.status_code}" # --------------------------------------------------------------------------- # MqttManager.is_configured # --------------------------------------------------------------------------- def test_is_configured_returns_false_when_disabled() -> None: manager = MqttManager() settings = _make_settings(mqtt_enabled=False, mqtt_broker_host="broker.test") assert manager.is_configured(settings) is False def test_is_configured_returns_false_when_host_empty() -> None: manager = MqttManager() settings = _make_settings(mqtt_enabled=True, mqtt_broker_host="") assert manager.is_configured(settings) is False def test_is_configured_returns_true_when_enabled_and_host_set() -> None: manager = MqttManager() settings = _make_settings(mqtt_enabled=True, mqtt_broker_host="broker.test") assert manager.is_configured(settings) is True # --------------------------------------------------------------------------- # MqttManager.connect — no-op when not configured # --------------------------------------------------------------------------- def test_connect_is_noop_when_disabled() -> None: manager = MqttManager() settings = _make_settings(mqtt_enabled=False) # Should not raise and should not create a client with patch("app.integrations.mqtt.mqtt.Client") as mock_client_cls: manager.connect(settings) mock_client_cls.assert_not_called() assert manager._client is None def test_connect_is_noop_when_host_empty() -> None: manager = MqttManager() settings = _make_settings(mqtt_broker_host="") with patch("app.integrations.mqtt.mqtt.Client") as mock_client_cls: manager.connect(settings) mock_client_cls.assert_not_called() assert manager._client is None # --------------------------------------------------------------------------- # MqttManager.connect — calls paho correctly when configured # --------------------------------------------------------------------------- def test_connect_creates_paho_client_with_version2() -> None: """connect() must pass CallbackAPIVersion.VERSION2 to paho.Client.""" manager = MqttManager() settings = _make_settings() mock_client = MagicMock() mock_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client) as mock_cls: with patch("app.integrations.mqtt.mqtt.CallbackAPIVersion"): manager.connect(settings) # Verify Client was instantiated with VERSION2 mock_cls.assert_called_once() call_kwargs = mock_cls.call_args # First positional arg or 'callback_api_version' kwarg assert call_kwargs is not None # loop_start must have been called mock_client.loop_start.assert_called_once() # connect must have been called with the right host/port mock_client.connect_async.assert_called_once_with( host="broker.test", port=1883, keepalive=60 ) def test_connect_uses_configured_client_id() -> None: manager = MqttManager() settings = _make_settings(mqtt_client_id="home-automation-dev") mock_client = MagicMock() with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client) as mock_cls: manager.connect(settings) assert mock_cls.call_args.kwargs["client_id"] == "home-automation-dev" def test_connect_same_settings_is_idempotent_even_when_disconnected() -> None: manager = MqttManager() settings = _make_settings() client = MagicMock() with patch("app.integrations.mqtt.mqtt.Client", return_value=client) as cls: manager.connect(settings) manager.connect(settings) cls.assert_called_once() client.loop_start.assert_called_once() client.connect_async.assert_called_once() def test_connect_replaces_client_when_paho_loop_thread_has_exited() -> None: manager = MqttManager() settings = _make_settings() old_client, recovered_client = MagicMock(), MagicMock() with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, recovered_client]): manager.connect(settings) old_client._thread = None manager._client_loop_started = True manager.connect(settings) old_client.disconnect.assert_called_once() old_client.loop_stop.assert_called_once() recovered_client.loop_start.assert_called_once() def test_loop_start_failure_cleans_installed_client_for_retry() -> None: manager = MqttManager() failed_client = MagicMock() failed_client.loop_start.return_value = -1 recovered_client = MagicMock() with patch("app.integrations.mqtt.mqtt.Client", side_effect=[failed_client, recovered_client]): manager.connect(_make_settings()) assert manager._client is None manager.connect(_make_settings()) failed_client.disconnect.assert_called_once() failed_client.loop_stop.assert_called_once() assert manager._client is recovered_client def test_concurrent_same_config_connect_creates_one_client() -> None: manager = MqttManager() settings = _make_settings() client = MagicMock() barrier = threading.Barrier(3) def connect() -> None: barrier.wait(timeout=1) manager.connect(settings) with patch("app.integrations.mqtt.mqtt.Client", return_value=client) as cls: workers = [threading.Thread(target=connect) for _ in range(2)] for worker in workers: worker.start() barrier.wait(timeout=1) for worker in workers: worker.join(timeout=2) assert all(not worker.is_alive() for worker in workers) cls.assert_called_once() client.loop_start.assert_called_once() @pytest.mark.parametrize("operation", ["disconnect", "reconnect"]) def test_teardown_can_join_loop_while_disconnect_callback_runs(operation: str) -> None: """A real callback thread must finish while teardown joins the fake loop.""" manager = MqttManager() callback_go = threading.Event() callback_entered = threading.Event() callback_done = threading.Event() clients: list[object] = [] class LoopClient: def __init__(self) -> None: self._thread: threading.Thread | None = None self.on_disconnect = None self.loop_joined = threading.Event() def connect_async(self, **_kwargs: object) -> None: pass def loop_start(self) -> int: def network_callback() -> None: callback_go.wait(timeout=2) callback_entered.set() try: self.on_disconnect(self, None, None, MagicMock(is_failure=True), None) finally: callback_done.set() self._thread = threading.Thread(target=network_callback, daemon=True) self._thread.start() return 0 def disconnect(self) -> None: callback_go.set() def loop_stop(self) -> None: assert callback_entered.wait(timeout=1), "callback thread did not start" assert self._thread is not None self._thread.join(timeout=1) assert not self._thread.is_alive(), "callback thread did not exit before loop_stop returned" self.loop_joined.set() def cleanup(self) -> None: callback_go.set() if self._thread is not None: self._thread.join(timeout=1) def make_client(**_kwargs: object) -> LoopClient: client = LoopClient() clients.append(client) return client try: with patch("app.integrations.mqtt.mqtt.Client", side_effect=make_client): manager.connect(_make_settings()) old_client = clients[0] if operation == "disconnect": manager.disconnect() assert manager._client is None else: manager.reconnect(_make_settings(mqtt_client_id="replacement")) assert manager._client is clients[1] assert callback_done.is_set() assert old_client.loop_joined.is_set() assert old_client._thread is not None and not old_client._thread.is_alive() finally: for client in clients: client.cleanup() @pytest.mark.parametrize("failure", ["constructor", "tls", "credentials", "connect_async", "loop_start"]) def test_initialization_failure_is_cleaned_and_next_connect_recovers(failure: str) -> None: manager = MqttManager() broken, recovered = MagicMock(), MagicMock() broken.connect_async.return_value = None if failure == "tls": broken.tls_set.side_effect = OSError("synthetic TLS failure") elif failure == "credentials": broken.username_pw_set.side_effect = OSError("synthetic credential failure") elif failure == "connect_async": broken.connect_async.side_effect = OSError("synthetic connect setup failure") elif failure == "loop_start": broken.loop_start.side_effect = OSError("synthetic loop failure") settings = _make_settings( mqtt_tls_enabled=failure == "tls", mqtt_username="user" if failure == "credentials" else "", mqtt_password="synthetic", ) first = ( [RuntimeError("synthetic constructor failure"), recovered] if failure == "constructor" else [broken, recovered] ) with patch("app.integrations.mqtt.mqtt.Client", side_effect=first): manager.connect(settings) assert manager._client is None manager.connect(settings) assert manager._client is recovered recovered.loop_start.assert_called_once() recovered.connect_async.assert_called_once() if failure in {"connect_async", "loop_start"}: broken.disconnect.assert_called_once() broken.loop_stop.assert_called_once() elif failure in {"tls", "credentials"}: broken.loop_start.assert_not_called() @pytest.mark.parametrize("old_is_failure", [False, True]) def test_old_connect_success_and_failure_callbacks_cannot_change_replacement( old_is_failure: bool, ) -> None: manager = MqttManager() old_client, new_client = MagicMock(), MagicMock() with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]): manager.connect(_make_settings()) manager.reconnect(_make_settings(mqtt_client_id="replacement")) manager._connected = True old_client.on_connect( old_client, None, None, MagicMock(is_failure=old_is_failure), None ) assert manager._client is new_client assert manager.is_connected new_client.subscribe.assert_not_called() def test_old_client_callbacks_cannot_change_replacement_state_or_dispatch() -> None: manager = MqttManager() settings = _make_settings() old_client, new_client = MagicMock(), MagicMock() received: list[bytes] = [] manager.subscribe("sensor/state", received.append) with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]): manager.connect(settings) manager.reconnect(_make_settings(mqtt_client_id="replacement")) manager._connected = True failed = MagicMock() failed.is_failure = True old_client.on_disconnect(old_client, None, None, failed, None) old_client.on_message(old_client, None, MagicMock(topic="sensor/state", payload=b"stale")) assert manager.is_connected assert received == [] new_client.on_message(new_client, None, MagicMock(topic="sensor/state", payload=b"fresh")) assert received == [b"fresh"] def test_deployment_client_id_variants_are_distinct() -> None: production_ids = { "home-automation", mqtt_source_client_id("home-automation", 7), mqtt_test_client_id("home-automation"), } development_ids = { "home-automation-dev", mqtt_source_client_id("home-automation-dev", 7), mqtt_test_client_id("home-automation-dev"), } assert len(production_ids) == 3 assert len(development_ids) == 3 assert production_ids.isdisjoint(development_ids) def test_connect_sets_credentials_when_username_provided() -> None: manager = MqttManager() settings = _make_settings(mqtt_username="user", mqtt_password="s3cr3t") mock_client = MagicMock() mock_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): manager.connect(settings) mock_client.username_pw_set.assert_called_once_with( username="user", password="s3cr3t" ) def test_connect_skips_credentials_when_username_empty() -> None: manager = MqttManager() settings = _make_settings(mqtt_username="", mqtt_password="") mock_client = MagicMock() mock_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): manager.connect(settings) mock_client.username_pw_set.assert_not_called() def test_connect_calls_tls_set_when_tls_enabled() -> None: manager = MqttManager() settings = _make_settings(mqtt_tls_enabled=True) mock_client = MagicMock() mock_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): manager.connect(settings) mock_client.tls_set.assert_called_once() # --------------------------------------------------------------------------- # MqttManager.connect — connection failure does not raise # --------------------------------------------------------------------------- def test_connect_does_not_raise_on_connection_failure() -> None: manager = MqttManager() settings = _make_settings() mock_client = MagicMock() mock_client.connect_async.side_effect = OSError("Connection refused") with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): # Must not raise — failures are caught and logged manager.connect(settings) # loop_stop should have been called to clean up the thread mock_client.loop_stop.assert_called() # Client reference should not be stored assert manager._client is None # --------------------------------------------------------------------------- # MqttManager.disconnect # --------------------------------------------------------------------------- def test_disconnect_stops_loop_and_clears_client() -> None: manager = MqttManager() settings = _make_settings() mock_client = MagicMock() mock_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): manager.connect(settings) manager._client = mock_client # ensure it's set manager.disconnect() mock_client.disconnect.assert_called() mock_client.loop_stop.assert_called() assert manager._client is None def test_disconnect_is_noop_when_not_connected() -> None: manager = MqttManager() # Should not raise when called before any connect manager.disconnect() # --------------------------------------------------------------------------- # MqttManager.publish # --------------------------------------------------------------------------- def test_publish_passes_topic_payload_retain_to_paho() -> None: manager = MqttManager() settings = _make_settings() mock_client = MagicMock() mock_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): manager.connect(settings) # Simulate connected state manager._connected = True manager._client = mock_client assert manager.publish("test/topic", '{"key": "value"}', retain=True, qos=1) is True mock_client.publish.assert_called_once_with( "test/topic", payload='{"key": "value"}', qos=1, retain=True ) def test_publish_is_noop_when_not_connected() -> None: manager = MqttManager() # No connect — should silently skip assert manager.publish("topic", "payload", retain=False) is False def test_publish_does_not_raise_on_paho_error() -> None: manager = MqttManager() mock_client = MagicMock() mock_client.publish.side_effect = RuntimeError("network error") manager._client = mock_client manager._connected = True # Must not raise assert manager.publish("topic", "payload") is False def test_publish_returns_false_when_paho_rejects_message() -> None: manager = MqttManager() mock_client = MagicMock() mock_client.publish.return_value.rc = 1 manager._client = mock_client manager._connected = True assert manager.publish("topic", "payload") is False # --------------------------------------------------------------------------- # MqttManager.reconnect # --------------------------------------------------------------------------- def test_reconnect_stops_old_client_and_starts_new_one() -> None: """reconnect() must tear down the existing client and establish a fresh one.""" manager = MqttManager() settings = _make_settings() # Simulate an already-connected client old_client = MagicMock() manager._client = old_client manager._connected = True new_client = MagicMock() new_client.connect_async.return_value = None with patch("app.integrations.mqtt.mqtt.Client", return_value=new_client): manager.reconnect(settings) # Old client must have been stopped old_client.disconnect.assert_called() old_client.loop_stop.assert_called() # New client must have been connected new_client.loop_start.assert_called_once() new_client.connect_async.assert_called_once_with( host="broker.test", port=1883, keepalive=60 ) @pytest.mark.parametrize( "settings", [ _make_settings(mqtt_enabled=False), _make_settings(mqtt_broker_host=""), ], ids=["disabled", "empty-host"], ) def test_unconfigured_reconnect_stops_main_client_but_keeps_dsmr_source(settings) -> None: """Changing main MQTT config must not interrupt independent DSMR clients.""" manager = MqttManager() main_client = MagicMock() source_client = MagicMock() received: list[bytes] = [] with patch( "app.integrations.mqtt.mqtt.Client", side_effect=[main_client, source_client], ): manager.connect(_make_settings()) assert manager.replace_source( 17, host="source.test", port=1883, username="", password="", tls_enabled=False, subscriptions={"meter/17": received.append}, ) manager.reconnect(settings) assert manager._client is None main_client.disconnect.assert_called_once() main_client.loop_stop.assert_called_once() assert manager._source_clients[17] is source_client assert manager.source_is_active(17) source_client.disconnect.assert_not_called() source_client.loop_stop.assert_not_called() message = MagicMock() message.topic = "meter/17" message.payload = b"reading" source_client.on_message(source_client, None, message) assert received == [b"reading"] manager.disconnect() # --------------------------------------------------------------------------- # POST /api/config/mqtt/test — session + CSRF guards # --------------------------------------------------------------------------- _MQTT_TEST_URL = "/api/config/mqtt/test" def test_mqtt_test_unauthenticated_returns_401(client: TestClient) -> None: response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) assert response.status_code == 401 def test_mqtt_test_authenticated_missing_csrf_returns_403(client: TestClient) -> None: _login(client) response = client.post(_MQTT_TEST_URL) assert response.status_code == 403 def test_mqtt_test_authenticated_empty_csrf_returns_403(client: TestClient) -> None: _login(client) response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": ""}) assert response.status_code == 403 # --------------------------------------------------------------------------- # POST /api/config/mqtt/test — three-state responses # --------------------------------------------------------------------------- _RUN_MQTT_TEST_PATH = "app.api.routes.api.config._run_mqtt_test" def test_mqtt_test_success_returns_200(client: TestClient) -> None: _login(client) with patch(_RUN_MQTT_TEST_PATH, return_value=None) as mock_run: response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) mock_run.assert_called_once() assert response.status_code == 200 body = response.json() assert body["result"] == "success" assert "message" in body def test_mqtt_test_config_error_returns_400(client: TestClient) -> None: _login(client) from app.api.routes.api.config import _MqttConfigurationError with patch(_RUN_MQTT_TEST_PATH, side_effect=_MqttConfigurationError("Host not configured")): response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) assert response.status_code == 400 body = response.json() assert body["result"] == "config-error" assert "Host not configured" in body["message"] def test_mqtt_test_connection_error_returns_502(client: TestClient) -> None: _login(client) from app.api.routes.api.config import _MqttConnectionError with patch(_RUN_MQTT_TEST_PATH, side_effect=_MqttConnectionError("connection refused")): response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) assert response.status_code == 502 body = response.json() assert body["result"] == "failed" assert "connection refused" in body["message"] def test_mqtt_test_response_does_not_echo_password(client: TestClient) -> None: """MQTT password must not appear in any test response body.""" _login(client) from app.api.routes.api.config import _MqttConnectionError with patch(_RUN_MQTT_TEST_PATH, side_effect=_MqttConnectionError("error: [redacted]")): response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "token"}) assert response.status_code == 502 # Actual password "s3cr3t" must not appear assert "s3cr3t" not in response.text def test_mqtt_test_empty_host_config_error(client: TestClient) -> None: """When broker host is empty, endpoint should return 400 config-error.""" _login(client) # The real _run_mqtt_test raises _MqttConfigurationError for empty host. # We don't mock it here — we let the real implementation run. # The default settings have mqtt_broker_host="" so this should 400. response = client.post(_MQTT_TEST_URL, headers={"X-CSRF-Token": "any-token"}) assert response.status_code == 400 body = response.json() assert body["result"] == "config-error" # --------------------------------------------------------------------------- # _run_mqtt_test unit tests (lower-level, mock paho entirely) # --------------------------------------------------------------------------- def test_run_mqtt_test_raises_config_error_when_host_empty() -> None: from app.api.routes.api.config import _run_mqtt_test, _MqttConfigurationError settings = _make_settings(mqtt_broker_host="") with pytest.raises(_MqttConfigurationError, match="host"): _run_mqtt_test(settings) def test_run_mqtt_test_raises_connection_error_on_os_error() -> None: from app.api.routes.api.config import _run_mqtt_test, _MqttConnectionError settings = _make_settings() mock_client = MagicMock() mock_client.connect.side_effect = OSError("Connection refused") # _run_mqtt_test imports paho.mqtt.client locally inside the function body, # so we patch via the real module path (paho.mqtt.client.Client). with patch("paho.mqtt.client.Client", return_value=mock_client): with pytest.raises(_MqttConnectionError, match="Cannot reach broker"): _run_mqtt_test(settings) # loop_stop must be called in the finally block mock_client.loop_stop.assert_called() def test_run_mqtt_test_raises_connection_error_on_timeout() -> None: """If connected_event never fires, should raise _MqttConnectionError.""" from app.api.routes.api.config import _run_mqtt_test, _MqttConnectionError settings = _make_settings() mock_client = MagicMock() mock_client.connect.return_value = None class _NeverSetEvent: def wait(self, timeout=None): return False # simulate timeout def set(self): pass def _make_event(): return _NeverSetEvent() with patch("paho.mqtt.client.Client", return_value=mock_client): with patch("threading.Event", side_effect=_make_event): with pytest.raises(_MqttConnectionError, match="timed out"): _run_mqtt_test(settings) def test_run_mqtt_test_uses_deployment_scoped_client_id() -> None: from app.api.routes.api.config import _run_mqtt_test, _MqttConnectionError settings = _make_settings(mqtt_client_id="home-automation-dev") mock_client = MagicMock() mock_client.connect.side_effect = OSError("Connection refused") with patch("paho.mqtt.client.Client", return_value=mock_client) as mock_cls: with pytest.raises(_MqttConnectionError): _run_mqtt_test(settings) assert mock_cls.call_args.kwargs["client_id"] == "home-automation-dev-test"