modbus-server-nd/tests/test_register_store_context.py

import asyncio
import unittest

from pymodbus.constants import ExcCodes

from modbus_context import ReadonlyHoldingRegisterContext
from point_model import ModbusPoint
from register_store import RegisterStore, initialize_register_store


class RegisterStoreContextTest(unittest.TestCase):
    def test_store_read_write_and_invalid_address(self):
        store = RegisterStore()
        points = [ModbusPoint("a", "A", "float32", 1, 0)]
        initialize_register_store(points, store)

        self.assertEqual(store.read_holding_registers(1, 0, 2), [0, 0])
        store.write_internal(1, 0, [0x3F80, 0x0000])
        self.assertEqual(store.read_holding_registers(1, 0, 2), [0x3F80, 0x0000])
        self.assertEqual(store.read_holding_registers(1, 1, 2), ExcCodes.ILLEGAL_ADDRESS)

        with self.assertRaises(KeyError):
            store.write_internal(1, 2, [1])

    def test_context_is_readonly_holding_register_only(self):
        store = RegisterStore()
        initialize_register_store([ModbusPoint("a", "A", "int16", 1, 10)], store)
        context = ReadonlyHoldingRegisterContext(store)

        async def run():
            read_values = await context.async_getValues(1, 3, 10, 1)
            input_read = await context.async_getValues(1, 4, 10, 1)
            write_result = await context.async_setValues(1, 6, 10, [1])
            return read_values, input_read, write_result

        read_values, input_read, write_result = asyncio.run(run())

        self.assertEqual(read_values, [0])
        self.assertEqual(input_read, ExcCodes.ILLEGAL_FUNCTION)
        self.assertEqual(write_result, ExcCodes.ILLEGAL_FUNCTION)


if __name__ == "__main__":
    unittest.main()