From d9e82038dc8cffc41c4067e1f6624ffdc8a9232e Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 24 Aug 2026 01:02:23 +0200 Subject: [PATCH] M8-R02: report DSMR MQTT source health --- app/integrations/mqtt.py | 36 +++- app/services/dsmr_ingest.py | 92 ++++++++- tests/test_dsmr_ingest.py | 259 ++++++++++++++++++++++++++ tests/test_dsmr_subscription_apply.py | 15 ++ tests/test_meter_source_api.py | 8 +- tests/test_mqtt_subscribe.py | 82 ++++++++ 6 files changed, 483 insertions(+), 9 deletions(-) diff --git a/app/integrations/mqtt.py b/app/integrations/mqtt.py index 0fb575d..5bf111f 100644 --- a/app/integrations/mqtt.py +++ b/app/integrations/mqtt.py @@ -248,6 +248,7 @@ class MqttManager: tls_enabled: bool, subscriptions: dict[str, Callable[[bytes], None]], base_client_id: str = "home-automation", + state_handler: Callable[[str], None] | None = None, ) -> bool: """Replace one source-owned client and its handlers. @@ -259,6 +260,7 @@ class MqttManager: with self._source_lifecycle_lock: self._stop_source_client(source_id) if not host: + self._report_source_state(state_handler, "error", source_id) return False self._next_source_generation += 1 generation = self._next_source_generation @@ -279,14 +281,18 @@ class MqttManager: if not self._is_current_source_client(source_id, generation, connected_client): return if reason_code.is_failure: + self._source_connected.discard(source_id) logger.warning("DSMR MQTT connection refused for source_id=%s", source_id) - return - self._source_connected.add(source_id) - for topic in captured_subscriptions: - try: - connected_client.subscribe(topic) - except Exception: - logger.exception("DSMR MQTT re-subscribe failed for source_id=%s", source_id) + state = "error" + else: + self._source_connected.add(source_id) + state = "online" + for topic in captured_subscriptions: + try: + connected_client.subscribe(topic) + except Exception: + logger.exception("DSMR MQTT re-subscribe failed for source_id=%s", source_id) + self._report_source_state(state_handler, state, source_id) def _on_disconnect( disconnected_client: mqtt.Client, @@ -299,6 +305,7 @@ class MqttManager: if not self._is_current_source_client(source_id, generation, disconnected_client): return self._source_connected.discard(source_id) + self._report_source_state(state_handler, "error", source_id) def _on_message( message_client: mqtt.Client, @@ -334,6 +341,7 @@ class MqttManager: client.tls_set() except Exception: logger.exception("DSMR MQTT TLS setup failed for source_id=%s", source_id) + self._report_source_state(state_handler, "error", source_id) return False if username: client.username_pw_set(username=username, password=password or None) @@ -345,11 +353,13 @@ class MqttManager: self._source_subscriptions[source_id] = captured_subscriptions self._source_generations[source_id] = generation self._source_states[source_id] = _SourceClientState(client, generation) + self._report_source_state(state_handler, "connecting", source_id) client.loop_start() try: client.connect(host=host, port=port, keepalive=60) except Exception: logger.exception("DSMR MQTT connect failed (source_id=%s, host=%s)", source_id, host) + self._report_source_state(state_handler, "error", source_id) self._stop_source_client(source_id) return False return True @@ -541,6 +551,18 @@ class MqttManager: and self._source_clients.get(source_id) is client ) + @staticmethod + def _report_source_state( + state_handler: Callable[[str], None] | None, state: str, source_id: int + ) -> None: + """Invoke an optional health callback without exposing connection credentials.""" + if state_handler is None: + return + try: + state_handler(state) + except Exception: + logger.exception("DSMR MQTT source state update failed for source_id=%s", source_id) + # --------------------------------------------------------------------------- # Module-level singleton — shared across lifespan and route handlers diff --git a/app/services/dsmr_ingest.py b/app/services/dsmr_ingest.py index 6d71b4d..1716cb0 100644 --- a/app/services/dsmr_ingest.py +++ b/app/services/dsmr_ingest.py @@ -185,9 +185,11 @@ def apply_dsmr_subscription(settings: "Settings | None" = None) -> None: # Every source owns a distinct MQTT client, so equal topics from # different sources/brokers are dispatchable. A telegram and tariff # topic on the *same* client would overwrite one handler, however. + rejected_source_ids: set[int] = set() for source_id, snapshot in list(desired.items()): if snapshot.tariff_topic and snapshot.topic == snapshot.tariff_topic: logger.error("DSMR source_id=%s rejected: telegram/tariff topic collision", source_id) + rejected_source_ids.add(source_id) desired.pop(source_id) with _subscription_lock: @@ -203,6 +205,19 @@ def apply_dsmr_subscription(settings: "Settings | None" = None) -> None: _subscription_tokens.pop(source_id, None) set_current_tariff(source_id, None) + # A disabled source has no installed MQTT owner. Persist that fact + # after invalidating its callback token, so a retained callback cannot + # revive an earlier online state while teardown is in progress. + for source_id in stale_source_ids: + _mark_disabled_source_inactive(source_id) + + # A topic collision is a configuration error for an enabled source, + # not a disabled-state transition. It must therefore replace any + # earlier online state even when this process started without a + # matching runtime subscription to tear down. + for source_id in rejected_source_ids: + _mark_rejected_source_error(source_id) + for source_id in stale_source_ids: mqtt_manager.remove_source(source_id) @@ -250,6 +265,9 @@ def apply_dsmr_subscription(settings: "Settings | None" = None) -> None: tls_enabled=snapshot.tls_enabled, subscriptions=handlers, base_client_id=base_client_id, + state_handler=lambda state, captured=snapshot, captured_token=token: ( + handle_captured_source_state(captured, captured_token, state) + ), ) if not applied: with _subscription_lock: @@ -293,6 +311,72 @@ def handle_captured_tariff_message( handle_tariff_message(payload_bytes, snapshot.source_id) +def handle_captured_source_state(snapshot: DsmrSourceSnapshot, token: object, state: str) -> None: + """Persist one active generation's connection health in a short DB session.""" + with _subscription_lock: + if _subscription_tokens.get(snapshot.source_id) is not token: + return + session_local = get_session_local() + session = session_local() + try: + source = session.get(MeterSource, snapshot.source_id) + if source is None or not source.enabled or source.kind != "dsmr_mqtt": + return + source.status = state + source.last_error = "MQTT connection failed." if state == "error" else None + source.updated_at = datetime.now(timezone.utc) + session.commit() + except Exception: + session.rollback() + logger.exception("DSMR source health update failed for source_id=%s", snapshot.source_id) + finally: + session.close() + + +def _mark_disabled_source_inactive(source_id: int) -> None: + """Clear an obsolete online health state for a disabled DSMR source. + + The caller has already invalidated the source's generation token. This + helper deliberately opens its own short session so reconcile never shares + a callback-thread transaction. Deleted sources simply have no row left + to update. + """ + session_local = get_session_local() + session = session_local() + try: + source = session.get(MeterSource, source_id) + if source is None or source.enabled or source.kind != "dsmr_mqtt": + return + source.status = "unknown" + source.last_error = None + source.updated_at = datetime.now(timezone.utc) + session.commit() + except Exception: + session.rollback() + logger.exception("DSMR disabled source health update failed for source_id=%s", source_id) + finally: + session.close() + + +def _mark_rejected_source_error(source_id: int) -> None: + """Persist a non-sensitive error for an enabled source rejected by reconcile.""" + session_local = get_session_local() + session = session_local() + try: + source = session.get(MeterSource, source_id) + if source is None or not source.enabled or source.kind != "dsmr_mqtt": + return + source.status = "error" + source.last_error = "DSMR source configuration invalid." + source.updated_at = datetime.now(timezone.utc) + session.commit() + except Exception: + session.rollback() + logger.exception("DSMR rejected source health update failed for source_id=%s", source_id) + finally: + session.close() + + def _handle_message_inner(payload_bytes: bytes, snapshot: DsmrSourceSnapshot) -> None: try: data = json.loads(payload_bytes) @@ -331,7 +415,13 @@ def _handle_message_inner(payload_bytes: bytes, snapshot: DsmrSourceSnapshot) -> payload=data, ) ) - session.commit() + source = session.get(MeterSource, snapshot.source_id) + if source is not None and source.enabled and source.kind == "dsmr_mqtt": + source.status = "online" + source.last_seen_at = datetime.now(timezone.utc) + source.last_error = None + source.updated_at = datetime.now(timezone.utc) + session.commit() except sqlalchemy.exc.IntegrityError: session.rollback() except Exception: diff --git a/tests/test_dsmr_ingest.py b/tests/test_dsmr_ingest.py index 11fceed..66ca69e 100644 --- a/tests/test_dsmr_ingest.py +++ b/tests/test_dsmr_ingest.py @@ -31,6 +31,7 @@ from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from app.models.energy import DsmrReading +from app.models.meter_source import MeterSource # --------------------------------------------------------------------------- @@ -174,6 +175,264 @@ def test_second_00_persists_full_frame(dsmr_db): assert payload["phase_voltage_l2"] is None +def test_accepted_message_marks_source_online_and_updates_last_seen(dsmr_db): + """A valid, accepted telegram clears stale diagnostics without changing its payload.""" + engine, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + with Session(engine) as session: + source = session.get(MeterSource, 1) + assert source is not None + source.enabled = True + source.status = "error" + source.last_error = "old error" + session.commit() + + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) + + with Session(engine) as session: + source = session.get(MeterSource, 1) + assert source is not None + assert source.status == "online" + assert source.last_error is None + assert source.last_seen_at is not None + + +def test_source_state_callback_persists_only_current_generation(dsmr_db, monkeypatch): + """Connection callbacks use a short session and stale generations leave health untouched.""" + engine, SessionLocal = dsmr_db + from app.services import dsmr_ingest + + snapshot = _make_settings() + current = object() + monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {snapshot.source_id: current}) + with Session(engine) as session: + source = session.get(MeterSource, snapshot.source_id) + assert source is not None + source.enabled = True + session.commit() + with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal): + dsmr_ingest.handle_captured_source_state(snapshot, current, "connecting") + dsmr_ingest.handle_captured_source_state(snapshot, current, "error") + dsmr_ingest.handle_captured_source_state(snapshot, object(), "online") + + with Session(engine) as session: + source = session.get(MeterSource, snapshot.source_id) + assert source is not None + assert source.status == "error" + assert source.last_error == "MQTT connection failed." + + +def test_source_health_is_isolated_between_sources(dsmr_db, monkeypatch): + """Connection and message health changes only touch their own source row.""" + engine, SessionLocal = dsmr_db + from app.services import dsmr_ingest + from app.services.dsmr_ingest import DsmrSourceSnapshot + + first = _make_settings() + second = DsmrSourceSnapshot(2, "second/topic", "", 10) + first_token = object() + second_token = object() + monkeypatch.setattr( + dsmr_ingest, "_subscription_tokens", {first.source_id: first_token, second.source_id: second_token} + ) + with Session(engine) as session: + first_source = session.get(MeterSource, first.source_id) + assert first_source is not None + first_source.enabled = True + session.add(MeterSource( + id=2, name="Second DSMR", kind="dsmr_mqtt", enabled=True, config={}, + status="online", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), + )) + session.commit() + + with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal): + dsmr_ingest.handle_captured_source_state(first, first_token, "connecting") + dsmr_ingest.handle_captured_source_state(second, second_token, "error") + dsmr_ingest.handle_message(_payload(_SAMPLE_TELEGRAM), first) + + with Session(engine) as session: + first_source = session.get(MeterSource, first.source_id) + second_source = session.get(MeterSource, second.source_id) + assert first_source is not None and second_source is not None + assert first_source.status == "online" + assert first_source.last_error is None + assert first_source.last_seen_at is not None + assert second_source.status == "error" + assert second_source.last_error == "MQTT connection failed." + assert second_source.last_seen_at is None + + +def test_disable_reconcile_persists_unknown_and_rejects_retained_state_callback(dsmr_db, monkeypatch): + """Disable invalidates the generation before clearing a stale online health state.""" + engine, SessionLocal = dsmr_db + from app.services import dsmr_ingest + + class FakeMqtt: + def __init__(self) -> None: + self.removed: list[int] = [] + self.state_handler = None + self.active: set[int] = set() + + def replace_source(self, source_id: int, **kwargs: object) -> bool: + self.active.add(source_id) + self.state_handler = kwargs["state_handler"] + return True + + def remove_source(self, source_id: int) -> None: + self.removed.append(source_id) + self.active.discard(source_id) + + def source_is_active(self, source_id: int) -> bool: + return source_id in self.active + + fake = FakeMqtt() + snapshot = _make_settings() + monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {}) + monkeypatch.setattr(dsmr_ingest, "_tariffs", {}) + monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal) + monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake) + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + with Session(engine) as session: + source = session.get(MeterSource, snapshot.source_id) + assert source is not None + source.enabled = True + source.status = "online" + session.commit() + dsmr_ingest.apply_dsmr_subscription() + assert fake.state_handler is not None + + with Session(engine) as session: + source = session.get(MeterSource, snapshot.source_id) + assert source is not None + source.enabled = False + session.commit() + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: []) + dsmr_ingest.apply_dsmr_subscription() + fake.state_handler("online") + + with Session(engine) as session: + source = session.get(MeterSource, snapshot.source_id) + assert source is not None + assert source.status == "unknown" + assert source.last_error is None + assert fake.removed == [snapshot.source_id] + + +def test_topic_collision_persists_error_when_reconcile_removes_active_client(dsmr_db, monkeypatch): + """A rejected enabled source cannot retain health from its removed client.""" + engine, SessionLocal = dsmr_db + from app.services import dsmr_ingest + + class FakeMqtt: + def __init__(self) -> None: + self.removed: list[int] = [] + self.active: set[int] = set() + + def replace_source(self, source_id: int, **kwargs: object) -> bool: + self.active.add(source_id) + return True + + def remove_source(self, source_id: int) -> None: + self.removed.append(source_id) + self.active.discard(source_id) + + def source_is_active(self, source_id: int) -> bool: + return source_id in self.active + + fake = FakeMqtt() + valid = _make_settings(dsmr_mqtt_topic="dsmr/telegram") + collision = _make_settings(dsmr_mqtt_topic="dsmr/telegram") + collision = dsmr_ingest.DsmrSourceSnapshot( + collision.source_id, + collision.topic, + collision.topic, + collision.sample_interval_s, + ) + monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {}) + monkeypatch.setattr(dsmr_ingest, "_tariffs", {}) + monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal) + monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake) + with Session(engine) as session: + source = session.get(MeterSource, valid.source_id) + assert source is not None + source.enabled = True + source.status = "online" + session.commit() + + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [valid]) + dsmr_ingest.apply_dsmr_subscription() + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision]) + dsmr_ingest.apply_dsmr_subscription() + + with Session(engine) as session: + source = session.get(MeterSource, valid.source_id) + assert source is not None + assert source.enabled is True + assert source.status == "error" + assert source.last_error == "DSMR source configuration invalid." + assert fake.removed == [valid.source_id] + assert fake.active == set() + + +def test_topic_collision_persists_error_on_startup_without_runtime_client(dsmr_db, monkeypatch): + """A collision corrects stale persisted online health without an installed client.""" + engine, SessionLocal = dsmr_db + from app.services import dsmr_ingest + + collision = dsmr_ingest.DsmrSourceSnapshot(1, "dsmr/telegram", "dsmr/telegram", 10) + monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {}) + monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal) + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision]) + with Session(engine) as session: + source = session.get(MeterSource, collision.source_id) + assert source is not None + source.enabled = True + source.status = "online" + session.commit() + + dsmr_ingest.apply_dsmr_subscription() + + with Session(engine) as session: + source = session.get(MeterSource, collision.source_id) + assert source is not None + assert source.status == "error" + assert source.last_error == "DSMR source configuration invalid." + + +def test_topic_collision_does_not_change_a_disabled_source(dsmr_db, monkeypatch): + """A concurrent disable is not overwritten by collision error handling.""" + engine, SessionLocal = dsmr_db + from app.services import dsmr_ingest + + collision = dsmr_ingest.DsmrSourceSnapshot(1, "dsmr/telegram", "dsmr/telegram", 10) + monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {}) + monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal) + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision]) + with Session(engine) as session: + source = session.get(MeterSource, collision.source_id) + assert source is not None + source.enabled = False + source.status = "unknown" + source.last_error = None + session.commit() + + dsmr_ingest.apply_dsmr_subscription() + + with Session(engine) as session: + source = session.get(MeterSource, collision.source_id) + assert source is not None + assert source.status == "unknown" + assert source.last_error is None + + def test_second_10_persists(dsmr_db): """A telegram with second=10 (another 10s boundary) must also be persisted.""" _, SessionLocal = dsmr_db diff --git a/tests/test_dsmr_subscription_apply.py b/tests/test_dsmr_subscription_apply.py index 9625a96..f2224ce 100644 --- a/tests/test_dsmr_subscription_apply.py +++ b/tests/test_dsmr_subscription_apply.py @@ -46,6 +46,7 @@ def fake_mqtt(monkeypatch): monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake) monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {}) monkeypatch.setattr(dsmr_ingest, "_tariffs", {}) return fake @@ -186,3 +187,17 @@ def test_same_snapshot_aba_rejects_retained_handler(fake_mqtt, monkeypatch): old_handler(b"stale") # type: ignore[operator] fake_mqtt.handlers[1]["same"](b"fresh") # type: ignore[operator] assert received == [(b"fresh", snapshot)] + + +def test_source_state_handler_is_generation_scoped(fake_mqtt, monkeypatch): + snapshot = _source(1, "same", broker_host="one.test") + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + dsmr_ingest.apply_dsmr_subscription() + stale = fake_mqtt.replace_calls[-1][1]["state_handler"] + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: []) + dsmr_ingest.apply_dsmr_subscription() + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + dsmr_ingest.apply_dsmr_subscription() + + monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: (_ for _ in ()).throw(AssertionError())) + stale("error") # type: ignore[operator] diff --git a/tests/test_meter_source_api.py b/tests/test_meter_source_api.py index 6b0a541..5d5a9f0 100644 --- a/tests/test_meter_source_api.py +++ b/tests/test_meter_source_api.py @@ -302,6 +302,12 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth _login(client) source = _create_source(client, config={"username": "private-user", "password": "private-secret"}) channel_uuid = _add_channel(engine, source["uuid"]) + with Session(engine) as session: + source_model = session.query(MeterSource).filter_by(uuid=source["uuid"]).one() + source_model.status = "online" + source_model.last_seen_at = datetime.now(UTC) + source_model.last_error = None + session.commit() listed = client.get("/api/energy/sources") assert listed.status_code == 200 assert listed.json()["total"] >= 1 @@ -332,7 +338,7 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth "uuid", "label", "suggested_commodity", "unit", "device_type", "latest_value", "latest_at", "latest_quality", "binding_count", "bound_meter_ids", "binding_summary", } - assert channels.json()["source_status"] == "unknown" + assert channels.json()["source_status"] == "online" meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ "label": "Contract meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial", diff --git a/tests/test_mqtt_subscribe.py b/tests/test_mqtt_subscribe.py index bf89091..3ef998a 100644 --- a/tests/test_mqtt_subscribe.py +++ b/tests/test_mqtt_subscribe.py @@ -437,6 +437,88 @@ def test_source_sync_connack_before_connect_returns_subscribes_all_topics() -> N assert client.subscribed == ["telegram/topic", "tariff/topic"] +def test_source_health_tracks_connack_disconnect_and_ignores_stale_callbacks() -> None: + """A source is connecting until CONNACK, and old generations cannot rewrite health.""" + manager = MqttManager() + old_client = MagicMock() + new_client = MagicMock() + old_states: list[str] = [] + new_states: list[str] = [] + kwargs = { + "host": "broker.test", + "port": 1883, + "username": "", + "password": "", + "tls_enabled": False, + "subscriptions": {"topic": lambda _payload: None}, + } + with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]): + assert manager.replace_source(1, **kwargs, state_handler=old_states.append) + assert old_states == ["connecting"] + assert manager.replace_source(1, **kwargs, state_handler=new_states.append) + + accepted = MagicMock() + accepted.is_failure = False + old_client.on_connect(old_client, None, MagicMock(), accepted, None) + old_client.on_disconnect(old_client, None, MagicMock(), MagicMock(), None) + assert old_states == ["connecting"] + + new_client.on_connect(new_client, None, MagicMock(), accepted, None) + new_client.on_disconnect(new_client, None, MagicMock(), MagicMock(), None) + assert new_states == ["connecting", "online", "error"] + + +def test_source_health_rejected_connack_reports_error_without_disconnect() -> None: + """A failed CONNACK is distinct from a later disconnect callback.""" + manager = MqttManager() + client = MagicMock() + states: list[str] = [] + with patch("app.integrations.mqtt.mqtt.Client", return_value=client): + assert manager.replace_source( + 1, host="broker.test", port=1883, username="", password="", tls_enabled=False, + subscriptions={"topic": lambda _payload: None}, state_handler=states.append, + ) + + accepted = MagicMock() + accepted.is_failure = False + refused = MagicMock() + refused.is_failure = True + client.on_connect(client, None, MagicMock(), accepted, None) + client.on_connect(client, None, MagicMock(), refused, None) + + assert states == ["connecting", "online", "error"] + assert 1 not in manager._source_connected + + +def test_source_health_disconnect_then_reconnect_connack_is_ordered_and_isolated() -> None: + """One source's reconnect sequence cannot change another source's health.""" + manager = MqttManager() + first_client = MagicMock() + second_client = MagicMock() + first_states: list[str] = [] + second_states: list[str] = [] + kwargs = { + "host": "broker.test", + "port": 1883, + "username": "", + "password": "", + "tls_enabled": False, + "subscriptions": {"topic": lambda _payload: None}, + } + with patch("app.integrations.mqtt.mqtt.Client", side_effect=[first_client, second_client]): + assert manager.replace_source(1, **kwargs, state_handler=first_states.append) + assert manager.replace_source(2, **kwargs, state_handler=second_states.append) + + accepted = MagicMock() + accepted.is_failure = False + first_client.on_connect(first_client, None, MagicMock(), accepted, None) + first_client.on_disconnect(first_client, None, MagicMock(), MagicMock(), None) + first_client.on_connect(first_client, None, MagicMock(), accepted, None) + + assert first_states == ["connecting", "online", "error", "online"] + assert second_states == ["connecting"] + + def test_source_teardown_with_joining_loop_stop_waits_for_callback_before_aba() -> None: """loop_stop may join a callback that needs the manager lock to finish.""" manager = MqttManager()