290 lines
11 KiB
Python
290 lines
11 KiB
Python
|
"""Tests for scripts/modbus_cli.py.
|
||
|
|
|
||
|
|
Acceptance criteria covered
|
||
|
|
----------------------------
|
||
|
|
1. ``modbus_cli read`` (via mock) prints decoded engineering values for all
|
||
|
|
profile metrics; columns are present for each metric key.
|
||
|
|
2. ``modbus_cli read`` exits non-zero on connection failure.
|
||
|
|
3. ``modbus_cli probe --fc 4`` prints raw register values in hex; optionally
|
||
|
|
decodes float32 pairs.
|
||
|
|
4. ``modbus_cli probe`` exits non-zero on connection failure.
|
||
|
|
5. CLI has NO write-register sub-command (only ``read`` and ``probe`` exist).
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
# Ensure project root is available when tests are run from the repo root.
|
||
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
if str(PROJECT_ROOT) not in sys.path:
|
||
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
|
|
||
|
|
import scripts.modbus_cli as cli # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Helpers
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def _make_ok_response(registers: list[int]) -> MagicMock:
|
||
|
|
resp = MagicMock()
|
||
|
|
resp.isError.return_value = False
|
||
|
|
resp.registers = registers
|
||
|
|
return resp
|
||
|
|
|
||
|
|
|
||
|
|
def _make_error_response() -> MagicMock:
|
||
|
|
resp = MagicMock()
|
||
|
|
resp.isError.return_value = True
|
||
|
|
resp.__str__ = lambda self: "ExceptionResponse(2)"
|
||
|
|
return resp
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# CLI: read sub-command
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestReadCommand:
|
||
|
|
"""``python -m scripts.modbus_cli read ...`` with mocked driver."""
|
||
|
|
|
||
|
|
def _make_sdm120_registers(self) -> dict[int, int]:
|
||
|
|
"""Return a minimal register map covering all 8 sdm120 core metrics."""
|
||
|
|
import struct
|
||
|
|
|
||
|
|
def pack(value: float) -> tuple[int, int]:
|
||
|
|
raw = struct.pack(">f", value)
|
||
|
|
hi = (raw[0] << 8) | raw[1]
|
||
|
|
lo = (raw[2] << 8) | raw[3]
|
||
|
|
return hi, lo
|
||
|
|
|
||
|
|
pairs: dict[int, tuple[int, int]] = {
|
||
|
|
0x0000: pack(230.2), # voltage
|
||
|
|
0x0006: pack(1.3), # current
|
||
|
|
0x000C: pack(295.0), # active_power
|
||
|
|
0x001E: pack(0.98), # power_factor
|
||
|
|
0x0046: pack(50.0), # frequency
|
||
|
|
0x0048: pack(123.4), # import_energy
|
||
|
|
0x004A: pack(0.0), # export_energy
|
||
|
|
0x0156: pack(123.4), # total_energy
|
||
|
|
}
|
||
|
|
regs: dict[int, int] = {}
|
||
|
|
for addr, (hi, lo) in pairs.items():
|
||
|
|
regs[addr] = hi
|
||
|
|
regs[addr + 1] = lo
|
||
|
|
return regs
|
||
|
|
|
||
|
|
@patch("app.integrations.modbus.driver.ModbusTcpClient")
|
||
|
|
def test_read_prints_decoded_values(self, mock_client_cls: MagicMock, capsys) -> None:
|
||
|
|
"""``read`` prints all profile metric keys with values."""
|
||
|
|
mock_client = MagicMock()
|
||
|
|
mock_client_cls.return_value = mock_client
|
||
|
|
mock_client.connect.return_value = True
|
||
|
|
|
||
|
|
regs = self._make_sdm120_registers()
|
||
|
|
# sdm120 has 2 blocks; simulate both responses.
|
||
|
|
# Block 1: 0x0000, count=0x0060 → we need 0x60=96 registers
|
||
|
|
block1 = [regs.get(i, 0) for i in range(0x0000, 0x0060)]
|
||
|
|
# Block 2: 0x0156, count=0x0004 → 4 registers
|
||
|
|
block2 = [regs.get(i, 0) for i in range(0x0156, 0x015A)]
|
||
|
|
mock_client.read_input_registers.side_effect = [
|
||
|
|
_make_ok_response(block1),
|
||
|
|
_make_ok_response(block2),
|
||
|
|
]
|
||
|
|
|
||
|
|
# Run the CLI.
|
||
|
|
cli.main(["read", "--host", "10.0.0.1", "--port", "502", "--unit", "1",
|
||
|
|
"--profile", "sdm120"])
|
||
|
|
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
output = captured.out
|
||
|
|
|
||
|
|
# All eight core metric keys must appear in the output.
|
||
|
|
for key in ("voltage", "current", "active_power", "power_factor",
|
||
|
|
"frequency", "import_energy", "export_energy", "total_energy"):
|
||
|
|
assert key in output, f"Missing metric key '{key}' in output:\n{output}"
|
||
|
|
|
||
|
|
# The decoded voltage should be close to 230.2.
|
||
|
|
assert "230." in output, f"Expected voltage ~230.2 in output:\n{output}"
|
||
|
|
|
||
|
|
@patch("app.integrations.modbus.driver.ModbusTcpClient")
|
||
|
|
def test_read_exits_nonzero_on_connection_failure(
|
||
|
|
self, mock_client_cls: MagicMock
|
||
|
|
) -> None:
|
||
|
|
"""Connection failure causes ``sys.exit(1)``."""
|
||
|
|
mock_client = MagicMock()
|
||
|
|
mock_client_cls.return_value = mock_client
|
||
|
|
mock_client.connect.return_value = False # Cannot connect.
|
||
|
|
|
||
|
|
with pytest.raises(SystemExit) as exc_info:
|
||
|
|
cli.main(["read", "--host", "10.0.0.1", "--port", "502", "--unit", "1",
|
||
|
|
"--profile", "sdm120"])
|
||
|
|
assert exc_info.value.code != 0
|
||
|
|
|
||
|
|
def test_read_exits_nonzero_for_missing_profile(self, capsys) -> None:
|
||
|
|
"""Unknown profile name causes sys.exit(1)."""
|
||
|
|
with pytest.raises(SystemExit) as exc_info:
|
||
|
|
cli.main(["read", "--host", "10.0.0.1", "--port", "502", "--unit", "1",
|
||
|
|
"--profile", "nonexistent_profile_xyz"])
|
||
|
|
assert exc_info.value.code != 0
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# CLI: probe sub-command
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestProbeCommand:
|
||
|
|
"""``python -m scripts.modbus_cli probe ...`` with mocked pymodbus."""
|
||
|
|
|
||
|
|
@patch("scripts.modbus_cli.ModbusTcpClient" if False else "pymodbus.client.ModbusTcpClient")
|
||
|
|
def _run_probe(self, mock_cls, args: list[str], regs: list[int]):
|
||
|
|
"""Helper: runs probe with a mocked client returning *regs*."""
|
||
|
|
|
||
|
|
@patch("app.integrations.modbus.driver.ModbusTcpClient")
|
||
|
|
def test_probe_prints_raw_registers(self, mock_driver_cls, capsys) -> None:
|
||
|
|
# probe uses its own client import, not the driver's — patch it there.
|
||
|
|
pass # see next test for the direct approach
|
||
|
|
|
||
|
|
def _setup_mock_client(self, regs: list[int]) -> MagicMock:
|
||
|
|
"""Return a mock ModbusTcpClient that returns *regs* for both FC03 and FC04."""
|
||
|
|
mock_cls = MagicMock()
|
||
|
|
mock_client = MagicMock()
|
||
|
|
mock_cls.return_value = mock_client
|
||
|
|
mock_client.connect.return_value = True
|
||
|
|
ok_response = _make_ok_response(regs)
|
||
|
|
mock_client.read_input_registers.return_value = ok_response
|
||
|
|
mock_client.read_holding_registers.return_value = ok_response
|
||
|
|
return mock_cls
|
||
|
|
|
||
|
|
def test_probe_fc4_prints_hex_values(self, capsys) -> None:
|
||
|
|
"""FC04 probe prints at least two hex addresses and values."""
|
||
|
|
mock_cls = self._setup_mock_client([0x4366, 0x3334])
|
||
|
|
|
||
|
|
with patch("pymodbus.client.ModbusTcpClient", mock_cls):
|
||
|
|
cli.main([
|
||
|
|
"probe",
|
||
|
|
"--host", "10.0.0.1",
|
||
|
|
"--port", "502",
|
||
|
|
"--unit", "1",
|
||
|
|
"--fc", "4",
|
||
|
|
"--address", "0x0000",
|
||
|
|
"--count", "2",
|
||
|
|
])
|
||
|
|
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
output = captured.out
|
||
|
|
# Expect hex addresses and values in the output.
|
||
|
|
assert "0x0000" in output.lower() or "0000" in output
|
||
|
|
assert "4366" in output.upper()
|
||
|
|
assert "3334" in output.upper()
|
||
|
|
|
||
|
|
def test_probe_fc4_with_float32_decode(self, capsys) -> None:
|
||
|
|
"""--decode float32 adds a float32 decode section to output."""
|
||
|
|
mock_cls = self._setup_mock_client([0x4366, 0x3334])
|
||
|
|
|
||
|
|
with patch("pymodbus.client.ModbusTcpClient", mock_cls):
|
||
|
|
cli.main([
|
||
|
|
"probe",
|
||
|
|
"--host", "10.0.0.1",
|
||
|
|
"--port", "502",
|
||
|
|
"--unit", "1",
|
||
|
|
"--fc", "4",
|
||
|
|
"--address", "0x0000",
|
||
|
|
"--count", "2",
|
||
|
|
"--decode", "float32",
|
||
|
|
])
|
||
|
|
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
output = captured.out
|
||
|
|
# The float decode section should show ~230.2.
|
||
|
|
assert "230." in output, f"Expected ~230.2 in float decode output:\n{output}"
|
||
|
|
|
||
|
|
def test_probe_exits_nonzero_on_connection_failure(self, capsys) -> None:
|
||
|
|
"""Connection failure causes non-zero exit."""
|
||
|
|
mock_cls = MagicMock()
|
||
|
|
mock_client = MagicMock()
|
||
|
|
mock_cls.return_value = mock_client
|
||
|
|
mock_client.connect.return_value = False
|
||
|
|
|
||
|
|
with patch("pymodbus.client.ModbusTcpClient", mock_cls):
|
||
|
|
with pytest.raises(SystemExit) as exc_info:
|
||
|
|
cli.main([
|
||
|
|
"probe",
|
||
|
|
"--host", "10.0.0.1",
|
||
|
|
"--port", "502",
|
||
|
|
"--unit", "1",
|
||
|
|
"--fc", "4",
|
||
|
|
"--address", "0x0000",
|
||
|
|
"--count", "2",
|
||
|
|
])
|
||
|
|
assert exc_info.value.code != 0
|
||
|
|
|
||
|
|
def test_probe_fc3_also_works(self, capsys) -> None:
|
||
|
|
"""FC03 (holding registers) is supported by probe."""
|
||
|
|
mock_cls = self._setup_mock_client([0x0001, 0x0002])
|
||
|
|
|
||
|
|
with patch("pymodbus.client.ModbusTcpClient", mock_cls):
|
||
|
|
cli.main([
|
||
|
|
"probe",
|
||
|
|
"--host", "10.0.0.1",
|
||
|
|
"--port", "502",
|
||
|
|
"--unit", "1",
|
||
|
|
"--fc", "3",
|
||
|
|
"--address", "0x0014",
|
||
|
|
"--count", "2",
|
||
|
|
])
|
||
|
|
|
||
|
|
# Verify holding-register read was called (not input-register read).
|
||
|
|
mock_client = mock_cls.return_value
|
||
|
|
mock_client.read_holding_registers.assert_called_once()
|
||
|
|
mock_client.read_input_registers.assert_not_called()
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# CLI: no write sub-commands
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestNoWriteSubcommands:
|
||
|
|
"""The CLI must not expose any write-register sub-commands."""
|
||
|
|
|
||
|
|
def test_no_write_subcommand_in_parser(self) -> None:
|
||
|
|
"""The argument parser has no 'write', 'set', or 'force' sub-commands."""
|
||
|
|
parser = cli.build_parser()
|
||
|
|
# Access the subparsers group.
|
||
|
|
subparsers_actions = [
|
||
|
|
action
|
||
|
|
for action in parser._actions
|
||
|
|
if hasattr(action, "_name_parser_map")
|
||
|
|
]
|
||
|
|
assert len(subparsers_actions) == 1
|
||
|
|
available_commands = set(subparsers_actions[0]._name_parser_map.keys())
|
||
|
|
# Only read-only commands must exist.
|
||
|
|
assert "read" in available_commands
|
||
|
|
assert "probe" in available_commands
|
||
|
|
# No write-related commands.
|
||
|
|
for name in available_commands:
|
||
|
|
assert name not in {"write", "set", "force", "write-register", "set-register"}, (
|
||
|
|
f"Write sub-command '{name}' found in CLI — it must not exist"
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_only_read_and_probe_subcommands(self) -> None:
|
||
|
|
"""Exactly two sub-commands are available: read and probe."""
|
||
|
|
parser = cli.build_parser()
|
||
|
|
subparsers_actions = [
|
||
|
|
action
|
||
|
|
for action in parser._actions
|
||
|
|
if hasattr(action, "_name_parser_map")
|
||
|
|
]
|
||
|
|
available = set(subparsers_actions[0]._name_parser_map.keys())
|
||
|
|
assert available == {"read", "probe"}, (
|
||
|
|
f"Expected only {{read, probe}}, got {available}"
|
||
|
|
)
|