1176 lines
44 KiB
Python
1176 lines
44 KiB
Python
"""Tests for M5-T11: HA Discovery publish + state publish.
|
|
|
|
Coverage:
|
|
1. build_discovery_payload: correct HA topic format, device block, unique_id
|
|
derived from uuid, device_class, unit_of_measurement, state_class.
|
|
2. publish_discovery: enabled entities get retained config; disabled entities
|
|
get empty (clear) payload; uses retained=True.
|
|
3. publish_states: calls value_getter; pushes correct state topic.
|
|
4. publish_device_state: after successful poll pushes state; after failed poll
|
|
pushes online=False (availability "offline").
|
|
5. MQTT not enabled/configured → entire chain is no-op; does not crash or raise.
|
|
6. Discovery not enabled → no-op.
|
|
7. Availability topic included in discovery config.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_app_alembic_config(database_url: str) -> Config:
|
|
cfg = Config("alembic_app.ini")
|
|
cfg.set_main_option("sqlalchemy.url", database_url)
|
|
return cfg
|
|
|
|
|
|
def _make_device(
|
|
session: Session,
|
|
*,
|
|
uuid: str,
|
|
friendly_name: str = "Test Meter",
|
|
host: str = "10.0.0.1",
|
|
port: int = 502,
|
|
unit_id: int = 1,
|
|
profile: str = "sdm120",
|
|
poll_interval_s: int = 5,
|
|
enabled: bool = True,
|
|
last_poll_ok: bool | None = None,
|
|
) -> Any:
|
|
from app.models.modbus import ModbusDevice
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
device = ModbusDevice(
|
|
uuid=uuid,
|
|
friendly_name=friendly_name,
|
|
host=host,
|
|
port=port,
|
|
unit_id=unit_id,
|
|
profile=profile,
|
|
poll_interval_s=poll_interval_s,
|
|
enabled=enabled,
|
|
last_poll_ok=last_poll_ok,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.flush()
|
|
return device
|
|
|
|
|
|
def _enable_entity(session: Session, key: str) -> None:
|
|
"""Insert an enabled ExposedEntityToggle row for *key*."""
|
|
from app.models.expose import ExposedEntityToggle
|
|
|
|
toggle = ExposedEntityToggle(
|
|
key=key,
|
|
enabled=True,
|
|
updated_at=datetime.now(tz=timezone.utc),
|
|
)
|
|
session.add(toggle)
|
|
session.flush()
|
|
|
|
|
|
def _make_settings(
|
|
*,
|
|
mqtt_enabled: bool = True,
|
|
ha_discovery_enabled: bool = True,
|
|
ha_discovery_prefix: str = "homeassistant",
|
|
) -> MagicMock:
|
|
s = MagicMock()
|
|
s.mqtt_enabled = mqtt_enabled
|
|
s.ha_discovery_enabled = ha_discovery_enabled
|
|
s.ha_discovery_prefix = ha_discovery_prefix
|
|
return s
|
|
|
|
|
|
def _make_mock_manager(*, is_connected: bool = True) -> MagicMock:
|
|
"""Return a MagicMock that mimics MqttManager with is_connected as an attribute."""
|
|
mgr = MagicMock()
|
|
mgr.is_connected = is_connected
|
|
return mgr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def disco_db(tmp_path: Path):
|
|
"""Temporary SQLite DB at Alembic head for discovery tests."""
|
|
db_path = tmp_path / "disco_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
command.upgrade(_make_app_alembic_config(db_url), "head")
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
yield engine
|
|
engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. build_discovery_payload
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_discovery_payload_topic_format() -> None:
|
|
"""Topic must follow <prefix>/<component>/<node_id>/<object_id>/config."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "aabbccdd-1234-5678-0000-111122223333"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="My Meter")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.voltage",
|
|
component="sensor",
|
|
device=device,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="My Meter Voltage",
|
|
)
|
|
|
|
topic, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
# node_id replaces hyphens with underscores
|
|
node_id = uuid_val.replace("-", "_")
|
|
# object_id: entity.key dots+hyphens → underscores
|
|
object_id = f"modbus_{uuid_val.replace('-', '_')}_voltage"
|
|
expected_topic = f"homeassistant/sensor/{node_id}/{object_id}/config"
|
|
assert topic == expected_topic, f"Got: {topic!r}\nExpected: {expected_topic!r}"
|
|
|
|
|
|
def test_build_discovery_payload_unique_id_from_uuid() -> None:
|
|
"""unique_id must be derived from device uuid and metric key (not friendly_name)."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "11111111-2222-3333-4444-555555555555"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Meter A")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.current",
|
|
component="sensor",
|
|
device=device,
|
|
device_class="current",
|
|
unit="A",
|
|
name="Meter A Current",
|
|
)
|
|
|
|
_, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
# unique_id must contain the device uuid (normalised)
|
|
uid = config["unique_id"].replace("-", "").replace("_", "")
|
|
uuid_normalised = uuid_val.replace("-", "")
|
|
assert uuid_normalised in uid, (
|
|
f"unique_id {config['unique_id']!r} must contain device uuid {uuid_val!r}"
|
|
)
|
|
# And the metric key
|
|
assert "current" in config["unique_id"]
|
|
# Must not contain friendly_name directly
|
|
assert "Meter" not in config["unique_id"]
|
|
|
|
|
|
def test_build_discovery_payload_unique_id_is_stable_across_renames() -> None:
|
|
"""Changing friendly_name must not change unique_id."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "aaaabbbb-cccc-dddd-eeee-ffffaaaabbbb"
|
|
entity_key = f"modbus.{uuid_val}.voltage"
|
|
|
|
device_a = DeviceInfo(identifiers=("modbus", uuid_val), name="Old Name")
|
|
entity_a = ExposableEntity(
|
|
key=entity_key,
|
|
component="sensor",
|
|
device=device_a,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Old Name Voltage",
|
|
)
|
|
|
|
device_b = DeviceInfo(identifiers=("modbus", uuid_val), name="New Name")
|
|
entity_b = ExposableEntity(
|
|
key=entity_key,
|
|
component="sensor",
|
|
device=device_b,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="New Name Voltage",
|
|
)
|
|
|
|
_, config_a = build_discovery_payload(entity_a, prefix="homeassistant")
|
|
_, config_b = build_discovery_payload(entity_b, prefix="homeassistant")
|
|
|
|
assert config_a["unique_id"] == config_b["unique_id"], (
|
|
"unique_id must not change when friendly_name changes"
|
|
)
|
|
|
|
|
|
def test_build_discovery_payload_device_block() -> None:
|
|
"""Config must include a device block with identifiers and name."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "cccccccc-0000-0000-0000-000000000001"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Friendly Name")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.voltage",
|
|
component="sensor",
|
|
device=device,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Friendly Name Voltage",
|
|
)
|
|
|
|
_, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
assert "device" in config
|
|
assert "identifiers" in config["device"]
|
|
assert uuid_val in config["device"]["identifiers"]
|
|
assert config["device"]["name"] == "Friendly Name"
|
|
|
|
|
|
def test_build_discovery_payload_sensor_fields() -> None:
|
|
"""Sensor payload must include device_class, unit_of_measurement, state_class."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "dddddddd-0000-0000-0000-000000000001"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Energy Meter")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.import_energy",
|
|
component="sensor",
|
|
device=device,
|
|
device_class="energy",
|
|
unit="kWh",
|
|
name="Energy Meter Import Energy",
|
|
state_class="total_increasing",
|
|
)
|
|
|
|
_, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
assert config["device_class"] == "energy"
|
|
assert config["unit_of_measurement"] == "kWh"
|
|
assert config["state_class"] == "total_increasing"
|
|
assert "state_topic" in config
|
|
assert "availability" in config
|
|
|
|
|
|
def test_build_discovery_payload_binary_sensor() -> None:
|
|
"""Binary sensor (online) payload must include payload_on/off but not unit."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "eeeeeeee-0000-0000-0000-000000000001"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Meter Online")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.online",
|
|
component="binary_sensor",
|
|
device=device,
|
|
device_class="connectivity",
|
|
unit="",
|
|
name="Meter Online Online",
|
|
)
|
|
|
|
_, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
assert config.get("payload_on") == "ON"
|
|
assert config.get("payload_off") == "OFF"
|
|
assert "unit_of_measurement" not in config
|
|
|
|
|
|
def test_build_discovery_payload_state_topic_matches_node_and_object() -> None:
|
|
"""state_topic must follow <prefix>/<component>/<node_id>/<object_id>/state."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload
|
|
|
|
uuid_val = "12345678-abcd-efab-0000-abcdef012345"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Meter")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.voltage",
|
|
component="sensor",
|
|
device=device,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Meter Voltage",
|
|
)
|
|
|
|
_, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
state_topic = config["state_topic"]
|
|
assert "homeassistant/sensor/" in state_topic
|
|
assert state_topic.endswith("/state")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. publish_discovery: retained, enabled vs disabled
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_publish_discovery_sends_retained_for_enabled_entity(disco_db) -> None:
|
|
"""Enabled entities must be published with retain=True and non-empty payload."""
|
|
uuid_val = "11110000-0000-0000-0000-000000000001"
|
|
|
|
with Session(disco_db) as session:
|
|
_make_device(session, uuid=uuid_val, friendly_name="Retained Meter")
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
_enable_entity(session, voltage_key)
|
|
session.commit()
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
publish_calls: list[tuple[str, Any, bool]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
publish_calls.append((topic, payload, retain))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_discovery
|
|
|
|
with Session(disco_db) as session:
|
|
publish_discovery(session)
|
|
|
|
# Find the voltage entity's config publish (non-empty retained)
|
|
node_id = uuid_val.replace("-", "_")
|
|
voltage_obj_id = f"modbus_{uuid_val.replace('-', '_')}_voltage"
|
|
expected_voltage_topic = f"homeassistant/sensor/{node_id}/{voltage_obj_id}/config"
|
|
|
|
retained_non_empty = [
|
|
(t, p, r) for t, p, r in publish_calls
|
|
if r and p not in (b"", "", None) and t == expected_voltage_topic
|
|
]
|
|
assert len(retained_non_empty) >= 1, (
|
|
f"Expected retained non-empty publish for voltage config topic. "
|
|
f"All calls: {[(t, type(p).__name__, r) for t, p, r in publish_calls]}"
|
|
)
|
|
|
|
|
|
def test_publish_discovery_sends_empty_payload_for_disabled_entity(disco_db) -> None:
|
|
"""Disabled entities must be published with an empty payload (clearing the config)."""
|
|
uuid_val = "22220000-0000-0000-0000-000000000001"
|
|
|
|
with Session(disco_db) as session:
|
|
_make_device(session, uuid=uuid_val, friendly_name="Disabled Entity Meter")
|
|
# Do NOT enable the voltage entity — it stays disabled by default.
|
|
session.commit()
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
publish_calls: list[tuple[str, Any, bool]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
publish_calls.append((topic, payload, retain))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_discovery
|
|
|
|
with Session(disco_db) as session:
|
|
publish_discovery(session)
|
|
|
|
# All config topic publishes for disabled entities should have empty payload + retained
|
|
config_calls = [(t, p, r) for t, p, r in publish_calls if "config" in t]
|
|
assert len(config_calls) > 0, "Expected at least one config topic publish"
|
|
|
|
empty_retained = [
|
|
(t, p, r) for t, p, r in config_calls if p in (b"", "", None) and r
|
|
]
|
|
assert len(empty_retained) > 0, (
|
|
f"Expected at least one empty-payload retained publish for disabled entities. "
|
|
f"Config calls: {[(t, type(p).__name__, r) for t, p, r in config_calls]}"
|
|
)
|
|
|
|
|
|
def test_publish_discovery_enabled_vs_disabled_payload(disco_db) -> None:
|
|
"""Enabled entities get non-empty JSON payload; disabled entities get empty payload."""
|
|
uuid_val = "33330000-0000-0000-0000-000000000001"
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
|
|
with Session(disco_db) as session:
|
|
_make_device(session, uuid=uuid_val, friendly_name="Mixed Meter")
|
|
_enable_entity(session, voltage_key) # enable only voltage
|
|
session.commit()
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
publish_calls: list[tuple[str, Any, bool]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
publish_calls.append((topic, payload, retain))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_discovery
|
|
|
|
with Session(disco_db) as session:
|
|
publish_discovery(session)
|
|
|
|
config_calls = [(t, p, r) for t, p, r in publish_calls if "config" in t]
|
|
assert len(config_calls) > 0, "Expected discovery config publishes"
|
|
|
|
node_id = uuid_val.replace("-", "_")
|
|
voltage_obj_id = f"modbus_{uuid_val.replace('-', '_')}_voltage"
|
|
voltage_config_topic = f"homeassistant/sensor/{node_id}/{voltage_obj_id}/config"
|
|
|
|
# The voltage entity's config topic should have non-empty JSON payload
|
|
voltage_call = next(
|
|
((t, p, r) for t, p, r in config_calls if t == voltage_config_topic), None
|
|
)
|
|
assert voltage_call is not None, (
|
|
f"Expected config publish for voltage topic {voltage_config_topic!r}. "
|
|
f"Got topics: {[t for t, _, _ in config_calls]}"
|
|
)
|
|
_t, payload, retain = voltage_call
|
|
assert payload not in (b"", "", None), "Enabled entity should get non-empty config payload"
|
|
assert retain is True, "Discovery config must be retained"
|
|
|
|
# All other config publishes should be empty (disabled entities)
|
|
other_calls = [(t, p, r) for t, p, r in config_calls if t != voltage_config_topic]
|
|
for _t, p, _r in other_calls:
|
|
assert p in (b"", "", None), (
|
|
f"Disabled entity config at {_t!r} must have empty payload, got {p!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. publish_states
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_publish_states_calls_value_getter_for_enabled_sensor() -> None:
|
|
"""publish_states must call value_getter and publish result to state topic."""
|
|
uuid_val = "44440000-0000-0000-0000-000000000001"
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
getter_calls = []
|
|
|
|
def _fake_getter(_sess):
|
|
getter_calls.append(1)
|
|
return 230.5
|
|
|
|
published: list[tuple[str, Any]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
published.append((topic, payload))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
from app.integrations.expose import CatalogEntry, DeviceInfo, ExposableEntity
|
|
|
|
device_info = DeviceInfo(identifiers=("modbus", uuid_val), name="State Meter")
|
|
mock_entity = ExposableEntity(
|
|
key=voltage_key,
|
|
component="sensor",
|
|
device=device_info,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="State Meter Voltage",
|
|
value_getter=_fake_getter,
|
|
)
|
|
mock_catalog = [CatalogEntry(entity=mock_entity, enabled=True)]
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
patch("app.services.ha_discovery.build_catalog", return_value=mock_catalog),
|
|
):
|
|
from app.services.ha_discovery import publish_states
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_states(session)
|
|
|
|
assert len(getter_calls) == 1, "value_getter should have been called once"
|
|
# Check that a state topic was published with the value
|
|
state_publishes = [(t, p) for t, p in published if "state" in t]
|
|
assert len(state_publishes) == 1
|
|
assert "230.5" in state_publishes[0][1]
|
|
|
|
|
|
def test_publish_states_skips_disabled_entities() -> None:
|
|
"""publish_states must not publish state for disabled entities."""
|
|
uuid_val = "55550000-0000-0000-0000-000000000001"
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
getter_calls = []
|
|
|
|
def _fake_getter(_sess):
|
|
getter_calls.append(1)
|
|
return 230.5
|
|
|
|
from app.integrations.expose import CatalogEntry, DeviceInfo, ExposableEntity
|
|
|
|
device_info = DeviceInfo(identifiers=("modbus", uuid_val), name="Disabled Meter")
|
|
mock_entity = ExposableEntity(
|
|
key=voltage_key,
|
|
component="sensor",
|
|
device=device_info,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Disabled Meter Voltage",
|
|
value_getter=_fake_getter,
|
|
)
|
|
# The entity is disabled
|
|
mock_catalog = [CatalogEntry(entity=mock_entity, enabled=False)]
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
patch("app.services.ha_discovery.build_catalog", return_value=mock_catalog),
|
|
):
|
|
from app.services.ha_discovery import publish_states
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_states(session)
|
|
|
|
assert len(getter_calls) == 0, "value_getter must not be called for disabled entity"
|
|
mock_mgr.publish.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. publish_device_state (poll hook)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_publish_device_state_pushes_availability_and_state() -> None:
|
|
"""After a successful poll, publish_device_state must push 'online' and state values."""
|
|
uuid_val = "66660000-0000-0000-0000-000000000001"
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
published: list[tuple[str, Any]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
published.append((topic, payload))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
from app.integrations.expose import CatalogEntry, DeviceInfo, ExposableEntity
|
|
|
|
device_info = DeviceInfo(identifiers=("modbus", uuid_val), name="Poll Meter")
|
|
sensor_entity = ExposableEntity(
|
|
key=voltage_key,
|
|
component="sensor",
|
|
device=device_info,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Poll Meter Voltage",
|
|
value_getter=lambda _sess: 231.0,
|
|
)
|
|
online_entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.online",
|
|
component="binary_sensor",
|
|
device=device_info,
|
|
device_class="connectivity",
|
|
unit="",
|
|
name="Poll Meter Online",
|
|
)
|
|
mock_catalog = [
|
|
CatalogEntry(entity=sensor_entity, enabled=True),
|
|
CatalogEntry(entity=online_entity, enabled=True),
|
|
]
|
|
|
|
mock_device = MagicMock()
|
|
mock_device.uuid = uuid_val
|
|
mock_device.last_poll_ok = True
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
patch("app.services.ha_discovery.build_catalog", return_value=mock_catalog),
|
|
):
|
|
from app.services.ha_discovery import publish_device_state
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_device_state(session, mock_device)
|
|
|
|
topics_published = [t for t, _ in published]
|
|
|
|
# Availability topic must be published
|
|
avail_calls = [t for t in topics_published if "availability" in t]
|
|
assert len(avail_calls) >= 1, "Expected availability topic publish"
|
|
avail_payload = next(p for t, p in published if "availability" in t)
|
|
assert avail_payload == "online", f"Expected 'online' payload, got {avail_payload!r}"
|
|
|
|
# State topic for voltage must be published
|
|
state_calls = [t for t in topics_published if "state" in t and "sensor" in t]
|
|
assert len(state_calls) >= 1, "Expected state topic publish for sensor entity"
|
|
state_payload = next(p for t, p in published if "state" in t and "sensor" in t)
|
|
assert "231.0" in state_payload
|
|
|
|
|
|
def test_publish_device_state_publishes_offline_on_failed_poll() -> None:
|
|
"""When last_poll_ok=False, publish_device_state must publish 'offline' availability."""
|
|
uuid_val = "77770000-0000-0000-0000-000000000001"
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
published: list[tuple[str, Any]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
published.append((topic, payload))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
mock_device = MagicMock()
|
|
mock_device.uuid = uuid_val
|
|
mock_device.last_poll_ok = False
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
patch("app.services.ha_discovery.build_catalog", return_value=[]),
|
|
):
|
|
from app.services.ha_discovery import publish_device_state
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_device_state(session, mock_device)
|
|
|
|
# Must have published availability = "offline"
|
|
avail_calls = [(t, p) for t, p in published if "availability" in t]
|
|
assert len(avail_calls) >= 1, "Expected at least one availability topic publish"
|
|
assert avail_calls[0][1] == "offline", (
|
|
f"Expected 'offline' availability, got {avail_calls[0][1]!r}"
|
|
)
|
|
|
|
# No sensor state publishes expected (device is offline)
|
|
state_calls = [(t, p) for t, p in published if "state" in t]
|
|
assert len(state_calls) == 0, "Must not publish state values when device is offline"
|
|
|
|
|
|
def test_publish_device_offline_publishes_offline_topic() -> None:
|
|
"""publish_device_offline must publish 'offline' to the availability topic."""
|
|
uuid_val = "88880000-0000-0000-0000-000000000001"
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
published: list[tuple[str, Any]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
published.append((topic, payload))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_device_offline
|
|
from unittest.mock import MagicMock as _MagicMock
|
|
|
|
publish_device_offline(_MagicMock(), uuid_val)
|
|
|
|
assert len(published) == 1
|
|
topic, payload = published[0]
|
|
assert "availability" in topic
|
|
assert payload == "offline"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. MQTT not enabled/not connected → entire chain no-op
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_should_publish_true_when_db_enabled_bootstrap_disabled(disco_db) -> None:
|
|
"""_should_publish must use DB-merged settings: if DB has mqtt_enabled=True and
|
|
ha_discovery_enabled=True (bootstrap has both False), publish should proceed.
|
|
|
|
This is the core regression test for Issue 2: bare get_settings() returns bootstrap
|
|
(False); build_runtime_settings returns DB-merged (True).
|
|
"""
|
|
from app.services.config_page import build_runtime_settings
|
|
from app.services.ha_discovery import _should_publish
|
|
from app.models.config import AppConfigEntry
|
|
from datetime import UTC, datetime
|
|
|
|
now = datetime.now(UTC)
|
|
|
|
# Write MQTT + discovery enabled into app_config (DB level)
|
|
with Session(disco_db) as session:
|
|
session.add(AppConfigEntry(key="MQTT_ENABLED", value="true", updated_at=now))
|
|
session.add(AppConfigEntry(key="MQTT_BROKER_HOST", value="broker.test", updated_at=now))
|
|
session.add(AppConfigEntry(key="HA_DISCOVERY_ENABLED", value="true", updated_at=now))
|
|
session.commit()
|
|
|
|
# Bootstrap has mqtt_enabled=False and ha_discovery_enabled=False (env defaults)
|
|
from app.config import Settings
|
|
bootstrap = Settings(_env_file=None, mqtt_enabled=False, ha_discovery_enabled=False,
|
|
app_database_url=str(disco_db.url))
|
|
|
|
with Session(disco_db) as session:
|
|
runtime = build_runtime_settings(session, bootstrap)
|
|
|
|
# Simulate a connected broker for the _should_publish check
|
|
mock_mgr_connected = MagicMock()
|
|
mock_mgr_connected.is_connected = True
|
|
|
|
with patch("app.services.ha_discovery.mqtt_manager", mock_mgr_connected):
|
|
result = _should_publish(runtime)
|
|
|
|
assert result is True, (
|
|
"_should_publish must return True when DB has mqtt_enabled=True + ha_discovery_enabled=True, "
|
|
"even if bootstrap Settings has both False"
|
|
)
|
|
|
|
|
|
def test_publish_discovery_noop_when_mqtt_disabled() -> None:
|
|
"""publish_discovery must be a no-op when mqtt_enabled=False."""
|
|
settings = _make_settings(mqtt_enabled=False, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=False)
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_discovery
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_discovery(session) # must not raise
|
|
|
|
mock_mgr.publish.assert_not_called()
|
|
|
|
|
|
def test_publish_states_noop_when_not_connected() -> None:
|
|
"""publish_states must be a no-op when MQTT is not connected."""
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
# is_connected=False simulates no broker connection
|
|
mock_mgr = _make_mock_manager(is_connected=False)
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_states
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_states(session) # must not raise
|
|
|
|
mock_mgr.publish.assert_not_called()
|
|
|
|
|
|
def test_publish_discovery_noop_when_ha_discovery_disabled() -> None:
|
|
"""publish_discovery must be a no-op when ha_discovery_enabled=False."""
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=False)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_discovery
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_discovery(session) # must not raise
|
|
|
|
mock_mgr.publish.assert_not_called()
|
|
|
|
|
|
def test_publish_device_state_noop_when_mqtt_disabled() -> None:
|
|
"""publish_device_state must be a no-op when MQTT is not configured."""
|
|
settings = _make_settings(mqtt_enabled=False, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=False)
|
|
|
|
mock_device = MagicMock()
|
|
mock_device.uuid = "99990000-0000-0000-0000-000000000001"
|
|
mock_device.last_poll_ok = True
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_device_state
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_device_state(session, mock_device) # must not raise
|
|
|
|
mock_mgr.publish.assert_not_called()
|
|
|
|
|
|
def test_publish_device_offline_noop_when_mqtt_disabled() -> None:
|
|
"""publish_device_offline must be a no-op when MQTT is not enabled."""
|
|
settings = _make_settings(mqtt_enabled=False, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=False)
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
):
|
|
from app.services.ha_discovery import publish_device_offline
|
|
from unittest.mock import MagicMock as _MagicMock
|
|
|
|
publish_device_offline(_MagicMock(), "some-uuid") # must not raise
|
|
|
|
mock_mgr.publish.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. Poll loop integration: MQTT errors must not crash poll
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_poll_device_mqtt_error_does_not_crash_poll(tmp_path: Path) -> None:
|
|
"""MQTT publish failure in poll_device must not propagate; poll must still succeed."""
|
|
db_path = tmp_path / "poll_mqtt_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
command.upgrade(_make_app_alembic_config(db_url), "head")
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
|
|
from app.models.modbus import ModbusDevice, ModbusReading
|
|
from app.services.modbus_poll import poll_device
|
|
from sqlalchemy import select
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(engine) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="MQTT Crash Test",
|
|
host="10.0.0.1",
|
|
port=502,
|
|
unit_id=1,
|
|
profile="sdm120",
|
|
poll_interval_s=5,
|
|
enabled=True,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.commit()
|
|
device_id = device.id
|
|
|
|
KNOWN_PAYLOAD = {"voltage": 230.0, "current": 1.0}
|
|
|
|
with (
|
|
patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
|
|
patch("app.services.modbus_poll.driver.read_blocks") as mock_read,
|
|
patch("app.services.modbus_poll.profiles.decode") as mock_decode,
|
|
# Make publish_device_state raise to simulate MQTT failure
|
|
patch(
|
|
"app.services.modbus_poll.publish_device_state",
|
|
side_effect=RuntimeError("Simulated MQTT failure"),
|
|
) as mock_ha,
|
|
):
|
|
mock_profile = MagicMock()
|
|
mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
|
|
mock_load.return_value = mock_profile
|
|
mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334}
|
|
mock_decode.return_value = KNOWN_PAYLOAD
|
|
|
|
# Must not raise despite MQTT failure
|
|
result = poll_device(session, device)
|
|
|
|
assert result is not None, "poll_device must succeed even if MQTT publish raises"
|
|
mock_ha.assert_called_once()
|
|
|
|
# Reading must have been persisted
|
|
readings = session.execute(select(ModbusReading)).scalars().all()
|
|
assert len(readings) == 1
|
|
assert readings[0].device_id == device_id
|
|
assert readings[0].payload == KNOWN_PAYLOAD
|
|
|
|
engine.dispose()
|
|
|
|
|
|
def test_poll_device_failed_push_offline_error_does_not_crash_poll(tmp_path: Path) -> None:
|
|
"""When driver fails and publish_device_offline also raises, poll must not propagate."""
|
|
db_path = tmp_path / "poll_offline_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
command.upgrade(_make_app_alembic_config(db_url), "head")
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
|
|
from app.models.modbus import ModbusDevice
|
|
from app.services.modbus_poll import poll_device
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(engine) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="Offline Crash Test",
|
|
host="10.0.0.2",
|
|
port=502,
|
|
unit_id=1,
|
|
profile="sdm120",
|
|
poll_interval_s=5,
|
|
enabled=True,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.commit()
|
|
|
|
with (
|
|
patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
|
|
patch(
|
|
"app.services.modbus_poll.driver.read_blocks",
|
|
side_effect=OSError("Device unreachable"),
|
|
),
|
|
patch(
|
|
"app.services.modbus_poll.publish_device_offline",
|
|
side_effect=RuntimeError("Simulated MQTT offline failure"),
|
|
) as mock_offline,
|
|
):
|
|
mock_profile = MagicMock()
|
|
mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
|
|
mock_load.return_value = mock_profile
|
|
|
|
# Must not raise despite driver AND offline publish both failing
|
|
result = poll_device(session, device)
|
|
|
|
assert result is None, "poll_device must return None on driver failure"
|
|
mock_offline.assert_called_once()
|
|
|
|
engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 7. Availability topic included in discovery config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_discovery_payload_includes_availability_topic() -> None:
|
|
"""Discovery config must include an 'availability' section with the device's avail topic."""
|
|
from app.integrations.expose import DeviceInfo, ExposableEntity
|
|
from app.services.ha_discovery import build_discovery_payload, _availability_topic
|
|
|
|
uuid_val = "abcdef00-0000-0000-0000-000000000001"
|
|
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Avail Meter")
|
|
entity = ExposableEntity(
|
|
key=f"modbus.{uuid_val}.voltage",
|
|
component="sensor",
|
|
device=device,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Avail Meter Voltage",
|
|
)
|
|
|
|
_, config = build_discovery_payload(entity, prefix="homeassistant")
|
|
|
|
expected_avail_topic = _availability_topic(uuid_val, "homeassistant")
|
|
assert "availability" in config
|
|
avail_topics = [a["topic"] for a in config["availability"]]
|
|
assert expected_avail_topic in avail_topics, (
|
|
f"Expected availability topic {expected_avail_topic!r} in config"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 8. Discovery retained flag
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_publish_discovery_uses_retain_true_for_enabled_entity() -> None:
|
|
"""publish_discovery must call publish() with retain=True for enabled entities."""
|
|
uuid_val = "ffffffff-0000-0000-0000-000000000001"
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
from app.integrations.expose import CatalogEntry, DeviceInfo, ExposableEntity
|
|
|
|
device_info = DeviceInfo(identifiers=("modbus", uuid_val), name="Retain Test Meter")
|
|
entity = ExposableEntity(
|
|
key=voltage_key,
|
|
component="sensor",
|
|
device=device_info,
|
|
device_class="voltage",
|
|
unit="V",
|
|
name="Retain Test Meter Voltage",
|
|
)
|
|
catalog = [CatalogEntry(entity=entity, enabled=True)]
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
patch("app.services.ha_discovery.build_catalog", return_value=catalog),
|
|
):
|
|
from app.services.ha_discovery import publish_discovery
|
|
from sqlalchemy import create_engine as _ce
|
|
|
|
eng = _ce("sqlite:///:memory:")
|
|
with Session(eng) as session:
|
|
publish_discovery(session)
|
|
|
|
# Verify that publish was called with retain=True
|
|
calls = mock_mgr.publish.call_args_list
|
|
assert len(calls) == 1, f"Expected exactly 1 publish call, got {len(calls)}"
|
|
# The call signature is publish(topic, payload, retain=True)
|
|
call_kwargs = calls[0].kwargs
|
|
assert call_kwargs.get("retain") is True, (
|
|
f"Expected retain=True in publish call, got kwargs={call_kwargs}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 9. Real provider value_getter integration test (R1 coverage)
|
|
# Uses the real modbus provider — does NOT mock build_catalog — to verify
|
|
# that the true value retrieval path works end-to-end. If value_getter
|
|
# were reverted to a stub (return None), this test would fail.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_real_provider_value_getter_returns_latest_reading_value(disco_db) -> None:
|
|
"""publish_device_state must publish the real decoded metric value from the DB.
|
|
|
|
This test exercises the full value_getter chain without mocking build_catalog:
|
|
1. Creates a real ModbusDevice and enables its voltage entity.
|
|
2. Inserts a ModbusReading with a known payload.
|
|
3. Calls publish_device_state using the real modbus provider catalog.
|
|
4. Asserts that the voltage state topic carries the expected numeric value.
|
|
5. Asserts that last_poll_ok=True → binary_sensor online entity → "ON".
|
|
|
|
Regression guard: if value_getter reverts to a stub (returns None), the sensor
|
|
state publish is skipped and the assertions on step 4 fail.
|
|
"""
|
|
uuid_val = "eeee0000-cafe-babe-0000-000000000001"
|
|
voltage_key = f"modbus.{uuid_val}.voltage"
|
|
online_key = f"modbus.{uuid_val}.online"
|
|
|
|
# --- DB setup: device + enabled toggle + a reading ---
|
|
from datetime import datetime, timezone
|
|
|
|
from app.models.expose import ExposedEntityToggle
|
|
from app.models.modbus import ModbusDevice, ModbusReading
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
|
|
with Session(disco_db) as session:
|
|
device = ModbusDevice(
|
|
uuid=uuid_val,
|
|
friendly_name="Real Provider Meter",
|
|
host="10.0.0.99",
|
|
port=502,
|
|
unit_id=1,
|
|
profile="sdm120",
|
|
poll_interval_s=5,
|
|
enabled=True,
|
|
last_poll_ok=True,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.flush()
|
|
|
|
# Enable the voltage entity and online binary_sensor.
|
|
session.add(ExposedEntityToggle(key=voltage_key, enabled=True, updated_at=now))
|
|
session.add(ExposedEntityToggle(key=online_key, enabled=True, updated_at=now))
|
|
|
|
# Insert a reading with a known voltage value.
|
|
reading = ModbusReading(
|
|
device_id=device.id,
|
|
recorded_at=now,
|
|
payload={"voltage": 231.5, "current": 1.2, "active_power": 275.0},
|
|
)
|
|
session.add(reading)
|
|
session.commit()
|
|
|
|
# --- MQTT setup ---
|
|
settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True)
|
|
mock_mgr = _make_mock_manager(is_connected=True)
|
|
|
|
published: list[tuple[str, Any]] = []
|
|
|
|
def _capture(topic, payload, *, retain=False, qos=0):
|
|
published.append((topic, payload))
|
|
|
|
mock_mgr.publish.side_effect = _capture
|
|
|
|
# Build a mock device object (only uuid + last_poll_ok needed by publish_device_state).
|
|
mock_device = MagicMock()
|
|
mock_device.uuid = uuid_val
|
|
mock_device.last_poll_ok = True
|
|
|
|
with (
|
|
patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
|
|
patch("app.services.ha_discovery.mqtt_manager", mock_mgr),
|
|
# NOTE: build_catalog is NOT mocked — we use the real modbus provider.
|
|
):
|
|
from app.services.ha_discovery import publish_device_state
|
|
|
|
with Session(disco_db) as session:
|
|
publish_device_state(session, mock_device)
|
|
|
|
# --- Assertions ---
|
|
topics_to_payloads: dict[str, Any] = {t: p for t, p in published}
|
|
|
|
# Availability topic must be "online".
|
|
avail_topic = f"homeassistant/modbus/{uuid_val.replace('-', '_')}/availability"
|
|
assert avail_topic in topics_to_payloads, (
|
|
f"Availability topic {avail_topic!r} not published. All topics: {list(topics_to_payloads)}"
|
|
)
|
|
assert topics_to_payloads[avail_topic] == "online", (
|
|
f"Expected 'online', got {topics_to_payloads[avail_topic]!r}"
|
|
)
|
|
|
|
# Voltage state topic must carry the real value from the reading (231.5).
|
|
node_id = uuid_val.replace("-", "_")
|
|
voltage_obj_id = f"modbus_{uuid_val.replace('-', '_')}_voltage"
|
|
voltage_state_topic = f"homeassistant/sensor/{node_id}/{voltage_obj_id}/state"
|
|
assert voltage_state_topic in topics_to_payloads, (
|
|
f"Voltage state topic {voltage_state_topic!r} not published. "
|
|
f"Sensor state topics: {[t for t in topics_to_payloads if 'sensor' in t and 'state' in t]}"
|
|
)
|
|
assert "231.5" in str(topics_to_payloads[voltage_state_topic]), (
|
|
f"Expected '231.5' in voltage state payload, "
|
|
f"got {topics_to_payloads[voltage_state_topic]!r}"
|
|
)
|
|
|
|
# Online binary_sensor state topic must be "ON" (last_poll_ok=True).
|
|
online_obj_id = f"modbus_{uuid_val.replace('-', '_')}_online"
|
|
online_state_topic = f"homeassistant/binary_sensor/{node_id}/{online_obj_id}/state"
|
|
assert online_state_topic in topics_to_payloads, (
|
|
f"Online state topic {online_state_topic!r} not published. "
|
|
f"Topics: {list(topics_to_payloads)}"
|
|
)
|
|
assert topics_to_payloads[online_state_topic] == "ON", (
|
|
f"Expected 'ON' for online sensor, got {topics_to_payloads[online_state_topic]!r}"
|
|
)
|