modbus-server-nd/tests/test_value_refresh.py

import logging
import unittest

from point_model import ModbusPoint
from register_store import RegisterStore, initialize_register_store
from value_provider import ValueProvider
from value_refresh import ValueRefreshWorker


class FakeProvider(ValueProvider):
    def __init__(self, values):
        self.values = values

    def fetch_values(self, point_ids):
        return {point_id: self.values[point_id] for point_id in point_ids if point_id in self.values}


class FailingProvider(ValueProvider):
    def fetch_values(self, point_ids):
        raise RuntimeError("boom")


class FailingMiddleBatchProvider(ValueProvider):
    def __init__(self):
        self.called_batches = []

    def fetch_values(self, point_ids):
        self.called_batches.append(point_ids)
        if point_ids[0] == "p200":
            raise RuntimeError("boom")
        return {point_id: int(point_id[1:]) for point_id in point_ids}


class ValueRefreshTest(unittest.TestCase):
    def setUp(self):
        logging.disable(logging.CRITICAL)

    def tearDown(self):
        logging.disable(logging.NOTSET)

    def test_initial_missing_value_raises(self):
        points = [
            ModbusPoint("a", "A", "int16", 1, 0),
            ModbusPoint("b", "B", "int16", 1, 1),
        ]
        store = RegisterStore()
        initialize_register_store(points, store)
        worker = ValueRefreshWorker(points, FakeProvider({"a": 12}), store, 5, 200)

        with self.assertRaises(RuntimeError):
            worker.refresh_once(initial=True)

        self.assertEqual(store.read_holding_registers(1, 0, 2), [12, 0])

    def test_periodic_missing_value_keeps_old_value(self):
        points = [ModbusPoint("a", "A", "int16", 1, 0)]
        store = RegisterStore()
        initialize_register_store(points, store)
        ValueRefreshWorker(points, FakeProvider({"a": 5}), store, 5, 200).refresh_once(initial=True)
        ValueRefreshWorker(points, FakeProvider({}), store, 5, 200).refresh_once(initial=False)

        self.assertEqual(store.read_holding_registers(1, 0, 1), [5])

    def test_initial_provider_failure_raises(self):
        points = [ModbusPoint("a", "A", "int16", 1, 0)]
        store = RegisterStore()
        initialize_register_store(points, store)

        with self.assertRaises(RuntimeError):
            ValueRefreshWorker(points, FailingProvider(), store, 5, 200).refresh_once(initial=True)

        self.assertEqual(store.read_holding_registers(1, 0, 1), [0])

    def test_periodic_provider_failure_continues_next_batch(self):
        points = [ModbusPoint(f"p{i}", f"P{i}", "int16", 1, i) for i in range(401)]
        store = RegisterStore()
        initialize_register_store(points, store)
        provider = FailingMiddleBatchProvider()

        ValueRefreshWorker(points, provider, store, 5, 200).refresh_once(initial=False)

        self.assertEqual(len(provider.called_batches), 3)
        self.assertEqual(store.read_holding_registers(1, 0, 1), [0])
        self.assertEqual(store.read_holding_registers(1, 199, 3), [199, 0, 0])
        self.assertEqual(store.read_holding_registers(1, 400, 1), [400])


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