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_edit_modbus_device_maps_aliases_and_posts_to_legacy_endpoint(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.edit_modbus_device(
                "dev-01",
                {
                    "ori_id": 1,
                    "name": "modbus_tcp_edited",
                    "device_type": 1,
                    "ip": "127.0.0.1",
                    "port": 5502,
                    "slave_id": 1,
                    "byte_order": 2,
                    "word_order": 2,
                    "address_base": 1,
                    "group_id": 10,
                },
            )

        self.assertEqual(result, response)
        self.assertEqual(
            request_json.call_args.args[:3],
            (
                "POST",
                "http://collector.test/api/collector/modbus/device/edit",
                "token",
            ),
        )
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["ori_id"], 1)
        self.assertEqual(payload["type"], 1)
        self.assertEqual(payload["address_offset"], 1)
        self.assertEqual(payload["device_group_id"], 10)
        self.assertEqual(payload["timeout"], 3)
        self.assertEqual(payload["alarm_interval"], 90)
        self.assertEqual(payload["collect_interval"], 5)
        self.assertEqual(payload["retry_times"], 0)
        self.assertEqual(payload["mode"], 0)
        self.assertNotIn("device_type", payload)
        self.assertNotIn("address_base", payload)
        self.assertNotIn("group_id", payload)

    def test_edit_modbus_device_supports_rtu_serial_port(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.edit_modbus_device(
                "dev-01",
                {
                    "ori_id": 1,
                    "name": "modbus_rtu_edited",
                    "type": 2,
                    "serial_port": "COM3",
                    "slave_id": 1,
                    "byte_order": 1,
                    "word_order": 1,
                },
            )

        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["type"], 2)
        self.assertEqual(payload["serial_port"], "COM3")

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

        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.edit_modbus_device("dev-01", payload)

    def test_edit_modbus_device_requires_serial_port_for_rtu(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.serial_port is required"):
            collector_api.edit_modbus_device(
                "dev-01",
                {
                    "ori_id": 1,
                    "name": "modbus_rtu_edited",
                    "type": 2,
                    "slave_id": 1,
                    "byte_order": 1,
                    "word_order": 1,
                },
            )

    def test_edit_modbus_device_rejects_invalid_connection_type(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.type must be one of 1, 2, 3, 4, 5"):
            collector_api.edit_modbus_device(
                "dev-01",
                {
                    "ori_id": 1,
                    "name": "modbus_tcp_edited",
                    "type": 0,
                    "ip": "127.0.0.1",
                    "port": 5502,
                    "slave_id": 1,
                    "byte_order": 1,
                    "word_order": 1,
                },
            )

    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["point_id"], "HR_UINT16")
        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_defaults_point_id_to_empty_string(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": "holding_register_uint16",
                    "func_code": 3,
                    "address": 10,
                    "type": "uint16",
                },
            )

        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["point_id"], "")

    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_edit_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.edit_modbus_point(
                "dev-01",
                {
                    "ori_id": 101,
                    "name": "holding_register_uint16_edited",
                    "point_id": "HR_UINT16_EDITED",
                    "register_type": "holding_register",
                    "address": 10,
                    "type": "WORD",
                },
            )

        self.assertEqual(result, response)
        self.assertEqual(
            request_json.call_args.args[:3],
            (
                "POST",
                "http://collector.test/api/collector/modbus/point/edit_collect_point",
                "token",
            ),
        )
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["ori_id"], 101)
        self.assertEqual(payload["point_id"], "HR_UINT16_EDITED")
        self.assertEqual(payload["func_code"], 3)
        self.assertEqual(payload["type"], "uint16")
        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.assertNotIn("register_type", payload)

    def test_edit_modbus_point_defaults_point_id_to_empty_string(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.edit_modbus_point(
                "dev-01",
                {
                    "ori_id": 101,
                    "name": "holding_register_uint16_edited",
                    "func_code": 3,
                    "address": 10,
                    "type": "uint16",
                },
            )

        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["point_id"], "")

    def test_edit_modbus_point_requires_ori_id(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "payload.ori_id is required"):
            collector_api.edit_modbus_point(
                "dev-01",
                {
                    "name": "holding_register_uint16_edited",
                    "func_code": 3,
                    "address": 10,
                    "type": "uint16",
                },
            )

    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",
        )

    def test_connect_device_posts_connected_status(self) -> None:
        self._patch_project()
        response = {"state": 0, "data": {"status": 2, "running_status": 0}}
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value=response,
        ) as request_json:
            result = collector_api.connect_device("dev-01", device_id=1, device_type="modbus")

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://collector.test/api/collector/common/device/set_connect_status",
            "token",
            json_payload={"id": 1, "type": "modbus", "status": 2},
        )

    def test_disconnect_device_posts_disconnected_status(self) -> None:
        self._patch_project()
        response = {"state": 0, "data": {"status": 1, "running_status": 0}}
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value=response,
        ) as request_json:
            result = collector_api.disconnect_device("dev-01", device_id=1, device_type="modbus")

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://collector.test/api/collector/common/device/set_connect_status",
            "token",
            json_payload={"id": 1, "type": "modbus", "status": 1},
        )

    def test_list_device_points_posts_device_and_group(self) -> None:
        self._patch_project()
        response = {"state": 0, "data": {"point": [], "total": 0}}
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value=response,
        ) as request_json:
            result = collector_api.list_device_points(
                "dev-01",
                device_id=1,
                device_type="modbus",
                group_id=100,
            )

        self.assertEqual(result, response)
        request_json.assert_called_once_with(
            "POST",
            "http://collector.test/api/collector/common/device/get_collect_point",
            "token",
            json_payload={"id": 1, "type": "modbus", "group_id": 100},
        )

    def test_device_common_tools_validate_ids(self) -> None:
        self._patch_project()
        with self.assertRaisesRegex(ValueError, "device_id must be a positive integer"):
            collector_api.connect_device("dev-01", device_id=0)
        with self.assertRaisesRegex(ValueError, "group_id must be a non-negative integer"):
            collector_api.list_device_points("dev-01", device_id=1, group_id=-1)
        with self.assertRaisesRegex(ValueError, "device_type is required"):
            collector_api.disconnect_device("dev-01", device_id=1, device_type="")


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