260 lines
9.9 KiB
Python
260 lines
9.9 KiB
Python
"""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)
|