data-collector-gateway/intergration/modbus/test_modbus_read_api.py

import unittest

from common import (
    FUNC_CODE_COIL,
    FUNC_CODE_DISCRETE_INPUT,
    FUNC_CODE_HOLDING_REGISTER,
    FUNC_CODE_INPUT_REGISTER,
    all_type_point_specs,
    assert_frame_format,
    assert_response_contract,
    parse_frame,
    point_read_payload,
    post_json,
    raw_read_payload,
)


def expected_device(payload):
    return {
        "device_type": payload["device_type"],
        "ip": payload["ip"],
        "port": payload["port"],
        "word_byte_order": payload["word_byte_order"],
        "address_base": payload["address_base"],
        "slave_id": payload["slave_id"],
    }


class ModbusReadApiIntegrationTest(unittest.TestCase):
    def test_raw_read_returns_modbus_poll_style_communication(self):
        payload = raw_read_payload(FUNC_CODE_HOLDING_REGISTER, address=0, quantity=4)
        status, data = post_json(
            "/api/dc-gateway/modbus/read",
            payload,
        )

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        self.assertEqual("success", data["msg"])
        self.assertNotIn("values", data)
        self.assertNotIn("communication", data)
        self.assertNotIn("device", data)
        self.assertEqual(expected_device(payload), data["data"]["device"])
        communication = data["data"]["communication"]
        assert_frame_format(self, communication)

        tx_frames = [parse_frame(frame)[1] for frame in communication if frame.startswith("Tx:")]
        rx_frames = [parse_frame(frame)[1] for frame in communication if frame.startswith("Rx:")]
        self.assertTrue(tx_frames, data)
        self.assertTrue(rx_frames, data)

        tx = tx_frames[0]
        self.assertGreaterEqual(len(tx), 12)
        self.assertEqual(FUNC_CODE_HOLDING_REGISTER, tx[7])
        self.assertEqual([0x00, 0x00], tx[8:10])
        self.assertEqual([0x00, 0x04], tx[10:12])

        rx = rx_frames[0]
        self.assertGreaterEqual(len(rx), 9)
        self.assertEqual(FUNC_CODE_HOLDING_REGISTER, rx[7])
        self.assertEqual(8, rx[8])

    def test_read_points_returns_all_function_codes_and_types(self):
        payload = point_read_payload(all_type_point_specs())

        status, data = post_json("/api/dc-gateway/modbus/read_points", payload)

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        self.assertNotIn("communication", data)
        self.assertNotIn("device", data)
        self.assertEqual(expected_device(payload), data["data"]["device"])
        points = data["data"]["points"]
        self.assertEqual(len(payload["points"]), len(points))
        seen = {(point["function_code"], point["address"], point["type"], point.get("bit")) for point in points}
        expected = {(point["function_code"], point["address"], point["type"], point.get("bit")) for point in payload["points"]}
        self.assertEqual(expected, seen)
        self.assertTrue(any(point["function_code"] == FUNC_CODE_COIL for point in points))
        self.assertTrue(any(point["function_code"] == FUNC_CODE_DISCRETE_INPUT for point in points))
        self.assertTrue(any(point["function_code"] == FUNC_CODE_HOLDING_REGISTER for point in points))
        self.assertTrue(any(point["function_code"] == FUNC_CODE_INPUT_REGISTER for point in points))
        for point in points:
            self.assertIn("value", point)
            self.assertNotIn("address_base", point)
            self.assertNotIn("protocol_address", point)
            self.assertNotIn("raw_values", point)
            self.assertNotIn("communication", point)
            if point["type"] == "bool":
                self.assertIsInstance(point["value"], bool)
            elif point["type"].startswith("float"):
                self.assertIsInstance(point["value"], float)
            else:
                self.assertIsInstance(point["value"], int)

    def test_validation_error_returns_http_200_contract(self):
        status, data = post_json(
            "/api/dc-gateway/modbus/read",
            raw_read_payload(FUNC_CODE_HOLDING_REGISTER, address=0, quantity=126),
        )

        assert_response_contract(self, status, data)
        self.assertEqual(1, data["code"], data)
        self.assertIn("quantity", data["msg"])
        self.assertEqual([], data["data"]["communication"])


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