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)

    def test_s7_raw_read_posts_gateway_payload(self) -> None:
        response = {"code": 0, "msg": "success", "data": {"communication": []}}
        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.s7_raw_read(
                "dev-01",
                ip="192.168.1.10",
                rock=0,
                slot=1,
                read={"area": "DB", "db": 1, "start": 0, "size": 4},
            )

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://gateway.test/api/dc-gateway/s7/read",
            json_payload={
                "device_type": "S7TCP",
                "ip": "192.168.1.10",
                "port": 102,
                "rock": 0,
                "slot": 1,
                "tsap_conn_type": "PG",
                "read": {"area": "DB", "db": 1, "start": 0, "size": 4},
            },
        )

    def test_s7_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.s7_point_collect_test(
                "dev-01",
                ip="192.168.1.10",
                port=1102,
                rock=0,
                slot=1,
                tsap_conn_type="OP",
                points=[{"area": "DB", "db": 1, "start": 0, "type": "float32"}],
            )

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://gateway.test/api/dc-gateway/s7/read_points",
            json_payload={
                "device_type": "S7TCP",
                "ip": "192.168.1.10",
                "port": 1102,
                "rock": 0,
                "slot": 1,
                "tsap_conn_type": "OP",
                "points": [{"area": "DB", "db": 1, "start": 0, "type": "float32"}],
            },
        )

    def test_s7_connect_scan_posts_ip_only(self) -> None:
        response = {"code": 0, "msg": "success", "data": {"available": []}}
        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.s7_connect_scan("dev-01", ip="192.168.1.10")

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://gateway.test/api/dc-gateway/s7/connect_scan",
            json_payload={"ip": "192.168.1.10"},
        )


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