home-automation/tests/test_modbus_driver.py

"""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_default_function_code_uses_fc04_input_registers(
        self, mock_client_cls: MagicMock
    ) -> None:
        """With no function_code given, read_blocks uses FC04 (read_input_registers)."""
        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])

        read_blocks("127.0.0.1", 502, 1, [{"start": 0x0000, "count": 2}])

        mock_client.read_input_registers.assert_called_once_with(0x0000, count=2, device_id=1)
        mock_client.read_holding_registers.assert_not_called()

    @patch("app.integrations.modbus.driver.ModbusTcpClient")
    def test_function_code_3_uses_fc03_holding_registers(
        self, mock_client_cls: MagicMock
    ) -> None:
        """function_code=3 dispatches FC03 (read_holding_registers), e.g. DDSU666."""
        mock_client = MagicMock()
        mock_client_cls.return_value = mock_client
        mock_client.connect.return_value = True
        mock_client.read_holding_registers.return_value = _make_ok_response(
            [0x4366, 0x3334, 0x3F80, 0x0000]
        )

        blocks = [{"start": 0x2000, "count": 4}]
        result = read_blocks("127.0.0.1", 502, 1, blocks, function_code=3)

        assert result == {0x2000: 0x4366, 0x2001: 0x3334, 0x2002: 0x3F80, 0x2003: 0x0000}
        mock_client.read_holding_registers.assert_called_once_with(
            0x2000, count=4, device_id=1
        )
        mock_client.read_input_registers.assert_not_called()

    @patch("app.integrations.modbus.driver.ModbusTcpClient")
    def test_invalid_function_code_raises_before_connecting(
        self, mock_client_cls: MagicMock
    ) -> None:
        """An unsupported function code is rejected without opening a connection."""
        mock_client = MagicMock()
        mock_client_cls.return_value = mock_client

        with pytest.raises(ModbusDriverError, match="Unsupported read function code"):
            read_blocks("127.0.0.1", 502, 1, [{"start": 0, "count": 2}], function_code=16)

        # No client should have been constructed or connected for a bad FC.
        mock_client_cls.assert_not_called()
        mock_client.connect.assert_not_called()

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