tliu93
ef48032adb
- mqtt.py: subscription registry, on_message dispatch (swallows handler errors), re-subscribe on (re)connect; publish/connect/disconnect/reconnect unchanged. - app/services/dsmr_ingest.py: handle_message stores the full telegram frame (no allowlist; gas/phases/null preserved), 10s downsample by timestamp.second, source_id idempotency (select + IntegrityError rollback). Net-thread safe. - main.py registers the DSMR subscription only when dsmr_ingest_enabled.
287 lines
9.3 KiB
Python
287 lines
9.3 KiB
Python
"""Tests for M6-T06: MqttManager subscribe + on_message dispatch.
|
|
|
|
All paho interaction is mocked — no real broker is required.
|
|
|
|
Covers:
|
|
- subscribe() registers topic → handler in the internal registry.
|
|
- on_message dispatches the payload to the correct handler.
|
|
- A handler that raises does NOT propagate the exception out of on_message.
|
|
- Existing publish() behaviour is not regressed.
|
|
- When already connected, subscribe() calls client.subscribe() immediately.
|
|
- on_connect re-subscribes to all registered topics after (re-)connect.
|
|
- subscribe() before connect: topic is subscribed when connect fires.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.integrations.mqtt import MqttManager
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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,
|
|
):
|
|
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
|
|
return s
|
|
|
|
|
|
def _make_mqtt_message(topic: str, payload: bytes) -> MagicMock:
|
|
"""Construct a minimal mock of a paho MQTTMessage."""
|
|
msg = MagicMock()
|
|
msg.topic = topic
|
|
msg.payload = payload
|
|
return msg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# subscribe() — registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_subscribe_registers_handler() -> None:
|
|
"""subscribe() must store the handler in the internal _subscriptions dict."""
|
|
manager = MqttManager()
|
|
handler = MagicMock()
|
|
|
|
manager.subscribe("test/topic", handler)
|
|
|
|
assert "test/topic" in manager._subscriptions
|
|
assert manager._subscriptions["test/topic"] is handler
|
|
|
|
|
|
def test_subscribe_overwrites_previous_handler() -> None:
|
|
"""Subscribing to the same topic twice replaces the handler."""
|
|
manager = MqttManager()
|
|
h1 = MagicMock()
|
|
h2 = MagicMock()
|
|
|
|
manager.subscribe("test/topic", h1)
|
|
manager.subscribe("test/topic", h2)
|
|
|
|
assert manager._subscriptions["test/topic"] is h2
|
|
|
|
|
|
def test_subscribe_when_not_connected_does_not_call_paho_subscribe() -> None:
|
|
"""subscribe() before connect must not attempt to call client.subscribe()."""
|
|
real_manager = MqttManager()
|
|
handler = MagicMock()
|
|
|
|
mock_client = MagicMock()
|
|
real_manager._client = None
|
|
real_manager._connected = False
|
|
|
|
real_manager.subscribe("dsmr/json", handler)
|
|
|
|
# No client → subscribe should not have been called on paho
|
|
mock_client.subscribe.assert_not_called()
|
|
|
|
|
|
def test_subscribe_when_connected_calls_paho_subscribe_immediately() -> None:
|
|
"""subscribe() while connected must call client.subscribe(topic) right away."""
|
|
manager = MqttManager()
|
|
mock_client = MagicMock()
|
|
manager._client = mock_client
|
|
manager._connected = True
|
|
|
|
handler = MagicMock()
|
|
manager.subscribe("dsmr/json", handler)
|
|
|
|
mock_client.subscribe.assert_called_once_with("dsmr/json")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# on_message dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _extract_on_message(manager: MqttManager, settings) -> object:
|
|
"""Connect with a mocked paho client and return the on_message callback that
|
|
was registered on the client mock."""
|
|
mock_client = MagicMock()
|
|
mock_client.connect.return_value = None
|
|
|
|
with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
|
|
manager.connect(settings)
|
|
|
|
# The callback was assigned as: client.on_message = _on_message
|
|
return mock_client.on_message
|
|
|
|
|
|
def test_on_message_dispatches_to_registered_handler() -> None:
|
|
"""on_message must invoke the handler registered for the message's topic."""
|
|
manager = MqttManager()
|
|
settings = _make_settings()
|
|
|
|
received_payloads: list[bytes] = []
|
|
|
|
def handler(payload: bytes) -> None:
|
|
received_payloads.append(payload)
|
|
|
|
manager.subscribe("dsmr/json", handler)
|
|
|
|
on_message = _extract_on_message(manager, settings)
|
|
|
|
msg = _make_mqtt_message("dsmr/json", b'{"id": 1}')
|
|
on_message(MagicMock(), None, msg) # simulate paho calling the callback
|
|
|
|
assert received_payloads == [b'{"id": 1}']
|
|
|
|
|
|
def test_on_message_ignores_unknown_topic() -> None:
|
|
"""on_message for an unregistered topic must not raise and must not call any handler."""
|
|
manager = MqttManager()
|
|
settings = _make_settings()
|
|
handler = MagicMock()
|
|
manager.subscribe("other/topic", handler)
|
|
|
|
on_message = _extract_on_message(manager, settings)
|
|
|
|
msg = _make_mqtt_message("unknown/topic", b"data")
|
|
# Must not raise
|
|
on_message(MagicMock(), None, msg)
|
|
|
|
handler.assert_not_called()
|
|
|
|
|
|
def test_on_message_swallows_handler_exception() -> None:
|
|
"""If the handler raises, on_message must NOT propagate the exception."""
|
|
manager = MqttManager()
|
|
settings = _make_settings()
|
|
|
|
def bad_handler(payload: bytes) -> None:
|
|
raise RuntimeError("handler exploded")
|
|
|
|
manager.subscribe("dsmr/json", bad_handler)
|
|
on_message = _extract_on_message(manager, settings)
|
|
|
|
msg = _make_mqtt_message("dsmr/json", b"{}")
|
|
# This must not raise
|
|
on_message(MagicMock(), None, msg)
|
|
|
|
|
|
def test_on_message_does_not_crash_on_handler_exception_multiple_calls() -> None:
|
|
"""After a handler exception the manager remains functional for subsequent messages."""
|
|
manager = MqttManager()
|
|
settings = _make_settings()
|
|
|
|
call_count = [0]
|
|
|
|
def flaky_handler(payload: bytes) -> None:
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise ValueError("first call fails")
|
|
|
|
manager.subscribe("dsmr/json", flaky_handler)
|
|
on_message = _extract_on_message(manager, settings)
|
|
|
|
msg1 = _make_mqtt_message("dsmr/json", b"first")
|
|
msg2 = _make_mqtt_message("dsmr/json", b"second")
|
|
|
|
on_message(MagicMock(), None, msg1) # should not raise despite exception in handler
|
|
on_message(MagicMock(), None, msg2) # handler called again
|
|
|
|
assert call_count[0] == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# on_connect re-subscribes registered topics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_on_connect_subscribes_all_registered_topics() -> None:
|
|
"""When connect fires successfully, _on_connect must subscribe every registered topic."""
|
|
manager = MqttManager()
|
|
settings = _make_settings()
|
|
|
|
manager.subscribe("topic/a", MagicMock())
|
|
manager.subscribe("topic/b", MagicMock())
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.connect.return_value = None
|
|
|
|
# Capture the on_connect callback that was registered on the mock client
|
|
with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
|
|
manager.connect(settings)
|
|
|
|
on_connect_cb = mock_client.on_connect
|
|
|
|
# Simulate broker accepting the connection (reason_code.is_failure == False)
|
|
reason_code = MagicMock()
|
|
reason_code.is_failure = False
|
|
on_connect_cb(mock_client, None, MagicMock(), reason_code, None)
|
|
|
|
# client.subscribe should have been called once per topic during on_connect
|
|
subscribe_calls = [c[0][0] for c in mock_client.subscribe.call_args_list]
|
|
assert "topic/a" in subscribe_calls
|
|
assert "topic/b" in subscribe_calls
|
|
|
|
|
|
def test_on_connect_does_not_subscribe_on_failure() -> None:
|
|
"""When the broker rejects the connection, _on_connect must NOT subscribe."""
|
|
manager = MqttManager()
|
|
settings = _make_settings()
|
|
manager.subscribe("topic/a", MagicMock())
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.connect.return_value = None
|
|
|
|
with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
|
|
manager.connect(settings)
|
|
|
|
on_connect_cb = mock_client.on_connect
|
|
reason_code = MagicMock()
|
|
reason_code.is_failure = True
|
|
|
|
mock_client.subscribe.reset_mock()
|
|
on_connect_cb(mock_client, None, MagicMock(), reason_code, None)
|
|
|
|
mock_client.subscribe.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Existing publish() is not regressed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_publish_still_works_after_subscribe_registered() -> None:
|
|
"""Adding a subscription must not break publish() behaviour."""
|
|
manager = MqttManager()
|
|
mock_client = MagicMock()
|
|
mock_client.connect.return_value = None
|
|
manager._client = mock_client
|
|
manager._connected = True
|
|
|
|
manager.subscribe("dsmr/json", MagicMock())
|
|
manager.publish("home/sensor", b"data", retain=True, qos=1)
|
|
|
|
mock_client.publish.assert_called_once_with(
|
|
"home/sensor", payload=b"data", qos=1, retain=True
|
|
)
|
|
|
|
|
|
def test_publish_skipped_when_not_connected_regression() -> None:
|
|
"""publish() must remain a no-op when not connected, even with subscriptions."""
|
|
manager = MqttManager()
|
|
manager.subscribe("dsmr/json", MagicMock())
|
|
|
|
# No client set up — publish should be silent
|
|
manager.publish("topic", b"payload") # must not raise
|