data-collector-mcp/tests/test_collector_api.py

from __future__ import annotations

import unittest
from unittest.mock import patch

from data_collector_mcp import collector_api


class CollectorApiTests(unittest.TestCase):
    def _patch_project(self):
        patches = [
            patch(
                "data_collector_mcp.collector_api.find_project_config",
                return_value={
                    "project_key": "dev-01",
                    "data_collector_base_url": "http://collector.test",
                },
            ),
            patch(
                "data_collector_mcp.collector_api.resolve_project_token",
                return_value="token",
            ),
        ]
        for item in patches:
            item.start()
            self.addCleanup(item.stop)

    def test_create_modbus_device_merges_defaults_and_posts_to_collector(self) -> None:
        self._patch_project()
        response = {"state": 0, "state_info": "成功"}
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value=response,
        ) as request_json:
            result = collector_api.create_modbus_device(
                "dev-01",
                {
                    "name": "modbus_tcp_1",
                    "device_type": 1,
                    "ip": "127.0.0.1",
                    "port": 5502,
                    "slave_id": 1,
                    "byte_order": 2,
                    "word_order": 2,
                    "address_base": 1,
                },
            )

        self.assertEqual(result, response)
        request_json.assert_called_once()
        self.assertEqual(request_json.call_args.args[:3], ("POST", "http://collector.test/api/collector/device", "token"))
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["type"], "modbus")
        self.assertEqual(payload["device_type"], 1)
        self.assertEqual(payload["timeout"], 3)
        self.assertEqual(payload["alarm_interval"], 90)
        self.assertEqual(payload["collect_interval"], 5)
        self.assertEqual(payload["byte_order"], 2)
        self.assertEqual(payload["word_order"], 2)
        self.assertEqual(payload["address_offset"], 1)
        self.assertNotIn("address_base", payload)
        self.assertEqual(payload["name"], "modbus_tcp_1")

    def test_create_modbus_device_requires_required_fields(self) -> None:
        self._patch_project()
        required_fields = [
            "name",
            "device_type",
            "ip",
            "port",
            "slave_id",
            "word_order",
            "byte_order",
            "address_base",
        ]
        base_payload = {
            "name": "modbus_tcp_1",
            "device_type": 1,
            "ip": "127.0.0.1",
            "port": 5502,
            "slave_id": 1,
            "byte_order": 1,
            "word_order": 1,
            "address_base": 0,
        }

        for field_name in required_fields:
            with self.subTest(field_name=field_name):
                payload = dict(base_payload)
                payload.pop(field_name)
                with self.assertRaisesRegex(ValueError, f"payload.{field_name} is required"):
                    collector_api.create_modbus_device("dev-01", payload)

    def test_create_modbus_point_merges_defaults_and_posts_to_collector(self) -> None:
        self._patch_project()
        response = {"state": 0, "state_info": "成功", "data": None}
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value=response,
        ) as request_json:
            result = collector_api.create_modbus_point(
                "dev-01",
                {
                    "device_id": 1,
                    "name": "holding_register_uint16",
                    "point_id": "HR_UINT16",
                    "func_code": 3,
                    "address": 10,
                    "type": "uint16",
                },
            )

        self.assertEqual(result, response)
        self.assertEqual(
            request_json.call_args.args[:3],
            (
                "POST",
                "http://collector.test/api/collector/modbus/point/add_collect_point",
                "token",
            ),
        )
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["scale_ratio"], 1)
        self.assertEqual(payload["value_offset"], 0)
        self.assertEqual(payload["group_id"], 0)
        self.assertEqual(payload["invalid_values"], "")
        self.assertIsNone(payload["valid_range_start"])
        self.assertIsNone(payload["valid_range_end"])
        self.assertEqual(payload["bit"], 0)
        self.assertEqual(payload["func_code"], 3)

    def test_create_modbus_point_normalizes_type_alias_and_register_type(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.create_modbus_point(
                "dev-01",
                {
                    "device_id": 1,
                    "name": "temperature",
                    "point_id": "TEMP",
                    "register_type": "holding_register",
                    "address": 10,
                    "type": "SHORT",
                },
            )

        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["func_code"], 3)
        self.assertEqual(payload["type"], "int16")
        self.assertNotIn("register_type", payload)

    def test_create_modbus_point_requires_name(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.name is required"):
            collector_api.create_modbus_point(
                "dev-01",
                {
                    "device_id": 1,
                    "func_code": 3,
                    "address": 10,
                    "type": "int16",
                },
            )

    def test_create_modbus_point_requires_register_type_or_func_code(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.register_type is required"):
            collector_api.create_modbus_point(
                "dev-01",
                {
                    "device_id": 1,
                    "name": "temperature",
                    "address": 10,
                    "type": "int16",
                },
            )

    def test_create_modbus_point_requires_address(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.address is required"):
            collector_api.create_modbus_point(
                "dev-01",
                {
                    "device_id": 1,
                    "name": "temperature",
                    "func_code": 3,
                    "type": "int16",
                },
            )

    def test_create_modbus_point_rejects_unknown_type(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.type is invalid"):
            collector_api.create_modbus_point(
                "dev-01",
                {
                    "device_id": 1,
                    "name": "temperature",
                    "func_code": 3,
                    "address": 10,
                    "type": "SHORT_REAL",
                },
            )

    def test_list_devices_defaults_num_points_false(self) -> None:
        self._patch_project()
        response = {"state": 0, "devices": []}
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value=response,
        ) as request_json:
            result = collector_api.list_devices("dev-01")

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "GET",
            "http://collector.test/api/collector/device?num_points=false",
            "token",
            json_payload=None,
        )

    def test_list_devices_can_enable_num_points(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.list_devices("dev-01", num_points=True)

        self.assertEqual(
            request_json.call_args.args[1],
            "http://collector.test/api/collector/device?num_points=true",
        )


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