"""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)