update MQTT tests

Author: avaldebe

src/pms/extra/cli.py

@@ -1,7 +1,7 @@
 import json
 import sys
 from dataclasses import fields
-from typing import Annotated
+from typing import Annotated, Optional
 
 if sys.version_info >= (3, 10):
     from typing import TypeAlias
@@ -58,11 +58,11 @@ def influxdb(
 MQTT_HOST: TypeAlias = Annotated[str, typer.Option("--mqtt-host", help="mqtt server")]
 MQTT_PORT: TypeAlias = Annotated[int, typer.Option("--mqtt-port", help="server port")]
 MQTT_USER: TypeAlias = Annotated[
-    str,
+    Optional[str],
     typer.Option("--mqtt-user", envvar="MQTT_USER", help="server username", show_default=False),
 ]
 MQTT_PASS: TypeAlias = Annotated[
-    str,
+    Optional[str],
     typer.Option("--mqtt-pass", envvar="MQTT_PASS", help="server password", show_default=False),
 ]
 
@@ -72,8 +72,8 @@ def mqtt(
     topic: Annotated[str, typer.Option("--topic", "-t", help="mqtt root/topic")] = "homie/test",
     host: MQTT_HOST = "test.mosquitto.org",
     port: MQTT_PORT = 1883,
-    user: MQTT_USER = "",
-    word: MQTT_PASS = "",
+    user: MQTT_USER = None,
+    word: MQTT_PASS = None,
 ):
     """Read sensor and push PM measurements to a MQTT server"""
     try:
@@ -109,8 +109,8 @@ def bridge(
     mqtt_topic: MQTT_TOPIC = "homie/+/+/+",
     mqtt_host: MQTT_HOST = "test.mosquitto.org",
     mqtt_port: MQTT_PORT = 1883,
-    mqtt_user: MQTT_USER = "",
-    mqtt_pass: MQTT_PASS = "",
+    mqtt_user: MQTT_USER = None,
+    mqtt_pass: MQTT_PASS = None,
     db_host: DB_HOST = "influxdb",
     db_port: DB_PORT = 8086,
     db_user: DB_USER = "root",

src/pms/extra/mqtt.py

@@ -8,10 +8,12 @@
 
 
 class Publisher(Protocol):
-    def __call__(self, data: dict[str, int | str]) -> None: ...
+    def __call__(self, data: dict[str, int | float | str]) -> None: ...
 
 
-def publisher(*, topic: str, host: str, port: int, username: str, password: str) -> Publisher:
+def publisher(
+    *, topic: str, host: str, port: int, username: str | None, password: str | None
+) -> Publisher:
     """returns function to publish to `topic` at `host`"""
     c = Client(client_id=topic)
     c.enable_logger(logger)  # type:ignore[arg-type]
@@ -25,7 +27,7 @@ def publisher(*, topic: str, host: str, port: int, username: str, password: str)
     c.connect(host, port)
     c.loop_start()
 
-    def pub(data: dict[str, int | str]) -> None:
+    def pub(data: dict[str, int | float | str]) -> None:
         for k, v in data.items():
             c.publish(f"{topic}/{k}", v, 1, True)
 
@@ -38,6 +40,13 @@ class Data(NamedTuple):
     measurement: str
     value: float
 
+    def __str__(self):
+        return f"{self.date:%F %T},{self.location},{self.measurement},{self.value}"
+
+    @property
+    def date(self) -> datetime:
+        return datetime.fromtimestamp(self.time)
+
     @staticmethod
     def now() -> int:
         """current time as seconds since epoch"""
@@ -63,8 +72,8 @@ def decode(cls, topic: str, payload: str, *, time: int | None = None) -> Data:
 
         try:
             value = float(payload)
-        except ValueError:
-            raise UserWarning(f"non numeric payload: {payload}")
+        except ValueError as e:
+            raise UserWarning(f"non numeric payload: {payload}") from e
         else:
             return cls(time, location, measurement, value)
 
@@ -73,8 +82,8 @@ def subscribe(
     topic: str,
     host: str,
     port: int,
-    username: str,
-    password: str,
+    username: str | None,
+    password: str | None,
     *,
     on_sensordata: Callable[[Data], None],
 ) -> None:

tests/conftest.py

@@ -64,12 +64,12 @@ class CapturedData(Enum):
     """Captured data from tests/captured_data"""
 
     _ignore_ = "name capt CapturedData reader"
+    value: tuple[RawData]
 
+    CapturedData = vars()
     with captured_data_reader(data=CAPTURED_DATA) as reader:
-        CapturedData = vars()
         for name in (s.name for s in Sensor):
-            capt = tuple(reader(name))
-            if capt:
+            if capt := tuple(reader(name)):
                 CapturedData[name] = capt
 
     def __str__(self) -> str:
@@ -106,6 +106,7 @@ def msg_hex(self) -> Iterator[str]:
                 yield message.hex
 
     def samples(self, command: str) -> int:
+        logger.debug(f"{self.name} {len(self.value)} obs")
         return len(self.value) - (command == "mqtt")
 
     def options(self, command: str, *, debug: bool = False) -> list[str]:
@@ -151,11 +152,10 @@ def captured_data(request: pytest.FixtureRequest) -> CapturedData:
 
 @pytest.fixture
 def replay_time(monkeypatch: pytest.MonkeyPatch, captured_data: CapturedData) -> None:
-    """mock datetime at `pms.core.sensor`, `pms.sensors.base` and `pms.sensors.mqtt`"""
-    captured_data = captured_data
+    """mock datetime at `pms.core.sensor` and `pms.sensors.base`"""
 
     class mock_datetime(datetime):
-        message_timestamp = captured_data.message_timestamp
+        _timestamp = captured_data.message_timestamp
 
         @classmethod
         def fromtimestamp(cls, t, tz=captured_data.tzinfo):
@@ -164,11 +164,10 @@ def fromtimestamp(cls, t, tz=captured_data.tzinfo):
 
         @classmethod
         def now(cls, tz=captured_data.tzinfo):
-            return cls.fromtimestamp(next(cls.message_timestamp), tz)
+            return cls.fromtimestamp(next(cls._timestamp), tz)
 
     monkeypatch.setattr("pms.core.sensor.datetime", mock_datetime)
     monkeypatch.setattr("pms.sensors.base.datetime", mock_datetime)
-    monkeypatch.setattr("pms.extra.mqtt.datetime", mock_datetime)
 
 
 @pytest.fixture

tests/extra/test_cli.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+from collections.abc import Iterator
 from dataclasses import fields
 from typing import Callable
 
 import pytest
+from loguru import logger
 from typer.testing import CliRunner
 
 from pms.main import main
@@ -12,21 +14,64 @@
 
 
 @pytest.fixture()
-def mock_mqtt_publisher(monkeypatch: pytest.MonkeyPatch):
-    """mock pms.extra.mqtt.publisher"""
-    from pms.extra.mqtt import Publisher
+def mock_mqtt_client(captured_data, monkeypatch: pytest.MonkeyPatch):
+    from pms.extra.mqtt import Data
+
+    def mqtt_messages() -> Iterator[Data]:
+        for obs in captured_data.obs:
+            for field in fields(obs):
+                if not field.metadata:
+                    continue
+                yield Data(obs.time, "test", field.name, getattr(obs, field.name))
+
+    class MockClient:
+        _message = mqtt_messages()
+
+        def __init__(self, *, client_id: str):
+            assert client_id in {"homie/test", "homie/+/+/+"}
 
-    def publisher(*, topic: str, host: str, port: int, username: str, password: str) -> Publisher:
-        def pub(data: dict[str, int | str]) -> None:
+        def enable_logger(self, logger):  # noqa: F811
             pass
 
-        return pub
+        def username_pw_set(self, username: str | None, password: str | None = None):
+            assert username is None
+            assert password is None
+
+        def will_set(self, topic, payload=None, qos=0, retain=False, properties=None):
+            assert topic.endswith("$online")
+            assert payload == "false"
+            assert qos == 1
+            assert retain is True
+
+        def connect(self, host, port=1883):
+            assert host == "test.mosquitto.org"
+            assert port == 1883  # MQTT, unencrypted, unauthenticated
+
+        def publish(self, topic, payload=None, qos=0, retain=False, properties=None):
+            logger.debug(f"{topic} = {payload}")
+            assert topic.startswith("homie/test/")
+            assert isinstance(payload, (str, float, int))
+            assert qos == 1
+            assert retain is True
+            if isinstance(payload, (float, int)):
+                logger.debug(msg := next(self._message))
+                assert Data.decode(topic, payload, time=msg.time) == msg
+
+        def _handle_on_message(self, message):
+            logger.debug("_handle_on_message")
+            assert self.on_message is not None
+
+        def loop_start(self):
+            logger.debug("loop_start")
+
+        def loop_forever(self, timeout=1, retry_first_connection=False):
+            logger.debug("loop_forever")
 
-    monkeypatch.setattr("pms.extra.mqtt.publisher", publisher)
+    monkeypatch.setattr("pms.extra.mqtt.Client", MockClient)
 
 
 @pytest.fixture()
-def mock_mqtt_subscribe(captured_data, monkeypatch: pytest.MonkeyPatch, replay_time):
+def mock_mqtt_subscribe(captured_data, monkeypatch: pytest.MonkeyPatch):
     """mock ms.extra.mqtt.subscribe"""
     from pms.extra.mqtt import Data
 
@@ -60,16 +105,16 @@ def publisher(*, host: str, port: int, username: str, password: str, db_name: st
         def pub(*, time: int, tags: dict[str, str], data: dict[str, float]) -> None:
             tag = ",".join(f"{k},{v}" for k, v in tags.items())
             for key, val in data.items():
-                print(f"{time},{tag},{key},{val}")
+                logger.debug(f"{time},{tag},{key},{val}")
 
         return pub
 
     monkeypatch.setattr("pms.extra.influxdb.publisher", publisher)
 
 
-@pytest.mark.usefixtures("mock_mqtt_publisher")
+@pytest.mark.usefixtures("mock_mqtt_client")
 def test_mqtt(capture):
-    result = runner.invoke(main, capture.options("mqtt"))
+    result = runner.invoke(main, capture.options("mqtt", debug=True))
     assert result.exit_code == 0
 
 

tests/extra/test_mqtt.py

@@ -20,30 +20,50 @@ def test_decode(location, measurement, value, secs=1_567_201_793):
 @pytest.mark.parametrize(
     "topic,payload,error",
     [
-        pytest.param("short/topic", "27.00", "topic total length: 2", id="short topic"),
-        pytest.param("too/long/topic/+/+/+", "27.00", "topic total length: 6", id="long topic"),
+        pytest.param(
+            "short/topic",
+            "27.00",
+            r"topic total length: 2",
+            id="short topic",
+        ),
+        pytest.param(
+            "too/long/topic/+/+/+",
+            "27.00",
+            "topic total length: 6",
+            id="long topic",
+        ),
         pytest.param(
             "sneaky/system/topic/$+",
             "27.00",
-            "system topic: sneaky/system/topic/$+",
+            r"system topic: sneaky/system/topic/\$\+",
             id="system topic",
         ),
         pytest.param(
             "other/$system/topic/+",
             "27.00",
-            "system topic: other/$system/topic/+",
+            r"system topic: other/\$system/topic/\+",
             id="system topic",
         ),
-        pytest.param("non/numeric/payload/+", "OK", "non numeric payload: OK", id="NaN payload"),
         pytest.param(
-            "non/numeric/payload/+", "value27", "non numeric payload: value27", id="NaN payload"
+            "non/numeric/payload/+",
+            "OK",
+            r"non numeric payload: OK",
+            id="NaN payload",
+        ),
+        pytest.param(
+            "non/numeric/payload/+",
+            "value27",
+            r"non numeric payload: value27",
+            id="NaN payload",
         ),
         pytest.param(
-            "non/numeric/payload/+", "27,00", "non numeric payload: 27,00", id="NaN payload"
+            "non/numeric/payload/+",
+            "27,00",
+            r"non numeric payload: 27,00",
+            id="NaN payload",
         ),
     ],
 )
 def test_decode_error(topic, payload, error, secs=1_567_201_793):
-    with pytest.raises(Exception) as e:
+    with pytest.raises(UserWarning, match=error):
         mqtt.Data.decode(topic, payload, time=secs)
-    assert str(e.value) == error