tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

M5-T03: add Modbus TCP driver, YAML profile framework, sdm120, modbus_cli

Browse Source
This commit is contained in:
tliu93
2026-06-22 13:03:50 +02:00
parent d9c82e39fb
commit c89cea4953
11 changed files with 1631 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_modbus_profiles.py
+259
View File
@@ -0,0 +1,259 @@
"""Tests for app/integrations/modbus/profiles.py.
Acceptance criteria covered
----------------------------
1. ``load_profile("sdm120")`` succeeds and returns a valid ``ModbusProfile``.
2. ``decode(profile, registers)`` maps known register values to correct keys/values.
3. ``load_profile`` raises a recognisable error for a missing profile.
4. ``load_profile`` raises a recognisable error for a profile that fails validation.
5. ``list_profiles()`` returns at least one entry containing ``"sdm120"``.
6. ``decode`` silently skips metrics whose registers are absent from the map.
"""
from __future__ import annotations
import math
import textwrap
from pathlib import Path
from unittest.mock import patch
import pytest
import yaml
from app.integrations.modbus.profiles import (
ModbusProfile,
ProfileNotFoundError,
ProfileValidationError,
decode,
list_profiles,
load_profile,
)
# ---------------------------------------------------------------------------
# load_profile — happy path
# ---------------------------------------------------------------------------
class TestLoadProfileSdm120:
"""Validate that the shipped sdm120.yaml loads and validates correctly."""
def test_load_returns_modbus_profile(self) -> None:
profile = load_profile("sdm120")
assert isinstance(profile, ModbusProfile)
def test_name_and_description(self) -> None:
profile = load_profile("sdm120")
assert profile.name == "sdm120"
assert "SDM120" in profile.description or "sdm120" in profile.description.lower()
def test_function_code_is_4(self) -> None:
profile = load_profile("sdm120")
assert profile.function_code == 4
def test_word_and_byte_order_big(self) -> None:
profile = load_profile("sdm120")
assert profile.word_order == "big"
assert profile.byte_order == "big"
def test_has_at_least_two_blocks(self) -> None:
profile = load_profile("sdm120")
assert len(profile.blocks) >= 2
def test_has_core_metrics(self) -> None:
"""All eight core metrics are present."""
profile = load_profile("sdm120")
keys = {m.key for m in profile.metrics}
expected = {
"voltage",
"current",
"active_power",
"power_factor",
"frequency",
"import_energy",
"export_energy",
"total_energy",
}
assert expected.issubset(keys), f"Missing keys: {expected - keys}"
def test_voltage_address_is_0x0000(self) -> None:
profile = load_profile("sdm120")
voltage = next(m for m in profile.metrics if m.key == "voltage")
assert voltage.address == 0x0000
def test_frequency_address_is_0x0046(self) -> None:
profile = load_profile("sdm120")
freq = next(m for m in profile.metrics if m.key == "frequency")
assert freq.address == 0x0046
def test_total_energy_address_is_0x0156(self) -> None:
profile = load_profile("sdm120")
te = next(m for m in profile.metrics if m.key == "total_energy")
assert te.address == 0x0156
def test_energy_metrics_have_state_class(self) -> None:
profile = load_profile("sdm120")
energy_keys = {"import_energy", "export_energy", "total_energy"}
for m in profile.metrics:
if m.key in energy_keys:
assert m.state_class == "total_increasing", (
f"{m.key} should have state_class='total_increasing'"
)
def test_no_deployment_fields_in_profile(self) -> None:
"""Profile YAML must not contain unit_id or friendly_name."""
profile_path = (
Path(__file__).parent.parent
/ "app/integrations/modbus/profiles/sdm120.yaml"
)
with profile_path.open() as fh:
raw = yaml.safe_load(fh)
assert "unit_id" not in raw, "unit_id is a deployment field, not a protocol field"
assert "friendly_name" not in raw, "friendly_name is a deployment field"
# ---------------------------------------------------------------------------
# load_profile — error cases
# ---------------------------------------------------------------------------
class TestLoadProfileErrors:
def test_missing_profile_raises_not_found(self) -> None:
with pytest.raises(ProfileNotFoundError, match="nonexistent_profile"):
load_profile("nonexistent_profile")
def test_invalid_yaml_content_raises_validation_error(self, tmp_path: Path) -> None:
"""A profile missing required fields raises ProfileValidationError."""
bad_yaml = textwrap.dedent(
"""\
name: bad_profile
description: Missing required fields
# function_code, word_order, byte_order, blocks, metrics are all absent
"""
)
bad_path = tmp_path / "bad_profile.yaml"
bad_path.write_text(bad_yaml)
# Temporarily patch the profiles directory to include our bad file.
fake_dir = tmp_path
with patch("app.integrations.modbus.profiles._PROFILES_DIR", fake_dir):
with pytest.raises(ProfileValidationError, match="bad_profile"):
load_profile("bad_profile")
def test_wrong_type_in_yaml_raises_validation_error(self, tmp_path: Path) -> None:
"""A profile with wrong field types raises ProfileValidationError."""
bad_yaml = textwrap.dedent(
"""\
name: typed_wrong
description: Wrong type for function_code
function_code: "four" # should be int
word_order: big
byte_order: big
blocks: []
metrics: []
"""
)
bad_path = tmp_path / "typed_wrong.yaml"
bad_path.write_text(bad_yaml)
with patch("app.integrations.modbus.profiles._PROFILES_DIR", tmp_path):
with pytest.raises(ProfileValidationError, match="typed_wrong"):
load_profile("typed_wrong")
# ---------------------------------------------------------------------------
# decode — register-to-value mapping
# ---------------------------------------------------------------------------
class TestDecode:
"""Verify decode() maps known register pairs to correct float values."""
def _make_registers(self, items: dict[int, tuple[int, int]]) -> dict[int, int]:
"""Build a register map from ``{address: (hi, lo)}`` pairs."""
regs: dict[int, int] = {}
for addr, (hi, lo) in items.items():
regs[addr] = hi
regs[addr + 1] = lo
return regs
def test_voltage_decoded_correctly(self) -> None:
"""0x4366,0x3334 at address 0x0000 → ~230.2V."""
profile = load_profile("sdm120")
registers = self._make_registers({0x0000: (0x4366, 0x3334)})
# Only voltage registers provided; others will be skipped.
result = decode(profile, registers)
assert "voltage" in result
assert math.isclose(result["voltage"], 230.2, rel_tol=1e-5)
def test_total_energy_decoded_correctly(self) -> None:
"""Float32 at address 0x0156 decoded as total_energy."""
profile = load_profile("sdm120")
# 0x447A0000 → 1000.0
registers = self._make_registers({0x0156: (0x447A, 0x0000)})
result = decode(profile, registers)
assert "total_energy" in result
assert math.isclose(result["total_energy"], 1000.0, rel_tol=1e-5)
def test_missing_registers_skipped(self) -> None:
"""Metrics whose register addresses are absent are not in the output."""
profile = load_profile("sdm120")
# Only supply voltage registers; all others missing.
registers = {0x0000: 0x4366, 0x0001: 0x3334}
result = decode(profile, registers)
assert "voltage" in result
assert "current" not in result
assert "active_power" not in result
def test_full_sdm120_payload_keys(self) -> None:
"""All 8 core keys appear when all registers are supplied."""
profile = load_profile("sdm120")
# Supply a minimal register map covering all 8 core metrics.
regs = self._make_registers(
{
0x0000: (0x4366, 0x3334), # voltage ~230.2V
0x0006: (0x3F80, 0x0000), # current ~1.0A
0x000C: (0x4393, 0x8000), # active_power ~295.0W
0x001E: (0x3F7A, 0xE148), # power_factor ~0.98
0x0046: (0x4248, 0x0000), # frequency ~50.0Hz
0x0048: (0x42F6, 0x6666), # import_energy ~123.2kWh
0x004A: (0x0000, 0x0000), # export_energy ~0.0kWh
0x0156: (0x42F6, 0x6666), # total_energy ~123.2kWh
}
)
result = decode(profile, regs)
expected_keys = {
"voltage", "current", "active_power", "power_factor",
"frequency", "import_energy", "export_energy", "total_energy",
}
assert expected_keys == set(result.keys())
def test_decode_does_not_modify_registers(self) -> None:
"""decode() must not mutate the input register dict."""
profile = load_profile("sdm120")
original = {0x0000: 0x4366, 0x0001: 0x3334}
registers = dict(original)
decode(profile, registers)
assert registers == original
# ---------------------------------------------------------------------------
# list_profiles
# ---------------------------------------------------------------------------
class TestListProfiles:
def test_contains_sdm120(self) -> None:
profiles = list_profiles()
names = [name for name, _ in profiles]
assert "sdm120" in names
def test_returns_list_of_tuples(self) -> None:
profiles = list_profiles()
assert isinstance(profiles, list)
for item in profiles:
assert isinstance(item, tuple)
assert len(item) == 2
name, description = item
assert isinstance(name, str)
assert isinstance(description, str)