"""Tests for app/integrations/modbus/driver.py. All tests use mocks — no real hardware or network connection is required. Acceptance criteria covered ---------------------------- 1. ``registers_to_float(0x4366, 0x3334)`` decodes to ~230.2 (byte/word order correct). 2. ``read_blocks`` returns the expected register map on a successful response. 3. ``read_blocks`` raises ``ModbusConnectionError`` when the client cannot connect. 4. ``read_blocks`` raises ``ModbusConnectionError`` when ``ConnectionException`` is raised. 5. ``read_blocks`` raises ``ModbusResponseError`` when the response is a Modbus error frame. 6. ``read_blocks`` raises ``ModbusResponseError`` when register count mismatches. """ from __future__ import annotations import math from unittest.mock import MagicMock, patch import pytest from app.integrations.modbus.driver import ( ModbusConnectionError, ModbusDriverError, ModbusResponseError, read_blocks, registers_to_float, ) # --------------------------------------------------------------------------- # registers_to_float # --------------------------------------------------------------------------- class TestRegistersToFloat: """Big-endian float32 decoding.""" def test_voltage_reference_value(self) -> None: """PDF example: 0x4366,0x3334 → 230.2 V.""" result = registers_to_float(0x4366, 0x3334) # IEEE-754 nearest float32 for 230.2 is ~230.20001220703125 assert math.isclose(result, 230.2, rel_tol=1e-5), f"Got {result}" def test_demand_time_reference_value(self) -> None: """PDF example: 0x3F80,0x0000 → 1.0.""" result = registers_to_float(0x3F80, 0x0000) assert math.isclose(result, 1.0, rel_tol=1e-7) def test_network_node_reference_value(self) -> None: """PDF example: 0x4270,0x0000 → 60.0.""" result = registers_to_float(0x4270, 0x0000) assert math.isclose(result, 60.0, rel_tol=1e-7) def test_zero(self) -> None: result = registers_to_float(0x0000, 0x0000) assert result == 0.0 def test_word_order_matters(self) -> None: """Swapping hi/lo gives a different (wrong) value.""" correct = registers_to_float(0x4366, 0x3334) swapped = registers_to_float(0x3334, 0x4366) assert correct != swapped # --------------------------------------------------------------------------- # read_blocks — helpers for building mock pymodbus responses # --------------------------------------------------------------------------- def _make_ok_response(registers: list[int]) -> MagicMock: """Build a fake successful read_input_registers response.""" resp = MagicMock() resp.isError.return_value = False resp.registers = registers return resp def _make_error_response() -> MagicMock: """Build a fake Modbus exception response.""" resp = MagicMock() resp.isError.return_value = True resp.__str__ = lambda self: "ExceptionResponse(2)" return resp class TestReadBlocks: """Unit tests for read_blocks using a mocked ModbusTcpClient.""" @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_single_block_returns_register_map(self, mock_client_cls: MagicMock) -> None: """Successful single-block read returns correct {addr: value} map.""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True mock_client.read_input_registers.return_value = _make_ok_response( [0x4366, 0x3334, 0x3F80, 0x0000] ) blocks = [{"start": 0x0000, "count": 4}] result = read_blocks("127.0.0.1", 502, 1, blocks) assert result == {0: 0x4366, 1: 0x3334, 2: 0x3F80, 3: 0x0000} mock_client.read_input_registers.assert_called_once_with( 0x0000, count=4, device_id=1 ) mock_client.close.assert_called_once() @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_multiple_blocks_merged(self, mock_client_cls: MagicMock) -> None: """Multiple blocks are read and merged into a single dict.""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True # First block: 2 registers at 0x0000 resp1 = _make_ok_response([0x4366, 0x3334]) # Second block: 2 registers at 0x0156 resp2 = _make_ok_response([0x447A, 0x0000]) mock_client.read_input_registers.side_effect = [resp1, resp2] blocks = [{"start": 0x0000, "count": 2}, {"start": 0x0156, "count": 2}] result = read_blocks("127.0.0.1", 502, 1, blocks) assert result[0x0000] == 0x4366 assert result[0x0001] == 0x3334 assert result[0x0156] == 0x447A assert result[0x0157] == 0x0000 assert mock_client.read_input_registers.call_count == 2 @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_connect_failure_raises_connection_error( self, mock_client_cls: MagicMock ) -> None: """connect() returning False raises ModbusConnectionError.""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = False with pytest.raises(ModbusConnectionError, match="Could not connect"): read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) mock_client.close.assert_called_once() @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_connection_exception_raises_connection_error( self, mock_client_cls: MagicMock ) -> None: """pymodbus ConnectionException during connect() → ModbusConnectionError.""" from pymodbus.exceptions import ConnectionException mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.side_effect = ConnectionException("timeout") with pytest.raises(ModbusConnectionError, match="failed"): read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) mock_client.close.assert_called_once() @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_modbus_exception_response_raises_response_error( self, mock_client_cls: MagicMock ) -> None: """Modbus exception frame from device raises ModbusResponseError.""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True mock_client.read_input_registers.return_value = _make_error_response() with pytest.raises(ModbusResponseError, match="exception"): read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) mock_client.close.assert_called_once() @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_register_count_mismatch_raises_response_error( self, mock_client_cls: MagicMock ) -> None: """Response with wrong register count raises ModbusResponseError.""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True # Requested 4 but only got 2 mock_client.read_input_registers.return_value = _make_ok_response([0x4366, 0x3334]) with pytest.raises(ModbusResponseError, match="Expected 4 registers"): read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 4}]) @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_connection_exception_during_read_raises_connection_error( self, mock_client_cls: MagicMock ) -> None: """ConnectionException during register read → ModbusConnectionError.""" from pymodbus.exceptions import ConnectionException mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True mock_client.read_input_registers.side_effect = ConnectionException("dropped") with pytest.raises(ModbusConnectionError, match="Lost connection"): read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) mock_client.close.assert_called_once() @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_client_always_closed_on_error(self, mock_client_cls: MagicMock) -> None: """close() is always called even when an exception propagates.""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True mock_client.read_input_registers.return_value = _make_error_response() with pytest.raises(ModbusDriverError): read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) mock_client.close.assert_called_once() @patch("app.integrations.modbus.driver.ModbusTcpClient") def test_unit_id_passed_as_device_id(self, mock_client_cls: MagicMock) -> None: """unit_id is forwarded as device_id= keyword argument (pymodbus 3.13.x).""" mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.connect.return_value = True mock_client.read_input_registers.return_value = _make_ok_response([0x0001, 0x0002]) read_blocks("10.0.0.1", 502, unit_id=5, blocks=[{"start": 0, "count": 2}]) mock_client.read_input_registers.assert_called_once_with( 0, count=2, device_id=5 )