69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
"""Offline regression coverage for scheduler-driven MQTT recovery."""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.integrations.mqtt import MqttManager
|
|
from app.main import _run_scheduled_ha_state_publish
|
|
|
|
|
|
def test_scheduler_reuses_manager_across_disconnects_then_subscribes_and_publishes() -> None:
|
|
session = MagicMock()
|
|
session_local = MagicMock(return_value=session)
|
|
settings = MagicMock(
|
|
mqtt_enabled=True,
|
|
mqtt_broker_host="synthetic-broker",
|
|
mqtt_broker_port=1883,
|
|
mqtt_username="",
|
|
mqtt_password="",
|
|
mqtt_tls_enabled=False,
|
|
mqtt_client_id="offline-test",
|
|
)
|
|
manager = MqttManager()
|
|
clients: list[MagicMock] = []
|
|
|
|
def make_client(*_args: object, **_kwargs: object) -> MagicMock:
|
|
client = MagicMock()
|
|
clients.append(client)
|
|
return client
|
|
|
|
manager.subscribe("homeassistant/status", lambda _payload: None)
|
|
|
|
def publish_states(_session: object) -> None:
|
|
manager.publish("homeassistant/state", b"state", retain=True)
|
|
|
|
with (
|
|
patch("app.main.get_session_local", return_value=session_local),
|
|
patch("app.main.build_runtime_settings", return_value=settings),
|
|
patch("app.main.get_settings", return_value=MagicMock()),
|
|
patch("app.main.mqtt_manager", manager),
|
|
patch("app.main.publish_discovery") as publish_discovery,
|
|
patch("app.main.publish_states", side_effect=publish_states),
|
|
patch("app.integrations.mqtt.mqtt.Client", side_effect=make_client),
|
|
):
|
|
for _ in range(3):
|
|
_run_scheduled_ha_state_publish()
|
|
assert len(clients) == 1
|
|
assert clients[0].loop_start.call_count == 1
|
|
clients[0].on_disconnect(
|
|
clients[0], None, None, MagicMock(is_failure=True), None
|
|
)
|
|
assert not manager.is_connected
|
|
|
|
# A subsequent scheduler turn sees Paho's existing loop retrying; it
|
|
# must not replace that client. Simulate the successful CONNACK.
|
|
_run_scheduled_ha_state_publish()
|
|
assert len(clients) == 1
|
|
clients[0].on_connect(clients[0], None, None, MagicMock(is_failure=False), None)
|
|
|
|
assert manager.is_connected
|
|
clients[0].subscribe.assert_called_with("homeassistant/status")
|
|
_run_scheduled_ha_state_publish()
|
|
assert len(clients) == 1
|
|
clients[0].publish.assert_called_once_with(
|
|
"homeassistant/state", payload=b"state", qos=0, retain=True
|
|
)
|
|
assert publish_discovery.call_count == 5
|
|
assert session.close.call_count == 5
|
|
manager.disconnect()
|