data-collector-mcp/tests/test_gateway_api.py

from __future__ import annotations

import unittest
from unittest.mock import patch

from data_collector_mcp import gateway_api


class GatewayApiTests(unittest.TestCase):
    def test_modbus_point_collect_test_posts_gateway_payload(self) -> None:
        response = {"code": 0, "msg": "success", "data": {"points": []}}
        with patch(
            "data_collector_mcp.gateway_api.find_project_config",
            return_value={"project_key": "dev-01", "base_url": "http://gateway.test"},
        ), patch(
            "data_collector_mcp.gateway_api.request_json",
            return_value=response,
        ) as request_json:
            result = gateway_api.modbus_point_collect_test(
                "dev-01",
                ip="192.168.1.10",
                port=502,
                slave_id=1,
                points=[{"function_code": 3, "address": 7, "type": "int16"}],
            )

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://gateway.test/api/dc-gateway/modbus/read_points",
            json_payload={
                "device_type": "ModbusTCP",
                "ip": "192.168.1.10",
                "port": 502,
                "word_byte_order": "ABCD",
                "address_base": 0,
                "slave_id": 1,
                "points": [{"function_code": 3, "address": 7, "type": "int16"}],
            },
        )

    def test_modbus_point_collect_test_keeps_custom_word_byte_order(self) -> None:
        with patch(
            "data_collector_mcp.gateway_api.find_project_config",
            return_value={"project_key": "dev-01", "base_url": "http://gateway.test"},
        ), patch(
            "data_collector_mcp.gateway_api.request_json",
            return_value={"code": 0},
        ) as request_json:
            gateway_api.modbus_point_collect_test(
                "dev-01",
                ip="192.168.1.10",
                port=502,
                slave_id=1,
                word_byte_order="DCBA",
                address_base=1,
                points=[{"function_code": 4, "address": 0, "type": "float32"}],
            )

        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["word_byte_order"], "DCBA")
        self.assertEqual(payload["address_base"], 1)


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