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_devices_batches_then_matches_largest_id(self) -> None:
        self._patch_project()
        devices = [
            {
                "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,
                "group_id": 10,
            },
            {
                "name": "modbus_tcp_2",
                "device_type": 1,
                "ip": "127.0.0.2",
                "port": 5503,
                "slave_id": 2,
                "byte_order": 1,
                "word_order": 1,
                "address_base": 0,
                "group_id": 10,
            },
        ]
        device_list = {
            "state": 0,
            "devices": [
                {
                    "id": 10,
                    "name": "group",
                    "type": "devicegroup",
                    "groups": [
                        {
                            "id": 3,
                            "name": "modbus_tcp_1",
                            "type": "modbus",
                            "device_type": 1,
                            "ip": "127.0.0.1",
                            "port": 5502,
                            "slave_id": 1,
                            "group_id": 10,
                            "address_offset": 0,
                        },
                        {
                            "id": 9,
                            "name": "modbus_tcp_1",
                            "type": "modbus",
                            "device_type": 1,
                            "ip": "127.0.0.1",
                            "port": 5502,
                            "slave_id": 1,
                            "group_id": 10,
                            "address_offset": 0,
                        },
                        {
                            "id": 11,
                            "name": "modbus_tcp_2",
                            "type": "modbus",
                            "device_type": 1,
                            "ip": "127.0.0.2",
                            "port": 5503,
                            "slave_id": 2,
                            "group_id": 10,
                            "address_offset": 0,
                        },
                    ],
                }
            ],
        }
        with patch(
            "data_collector_mcp.collector_api.request_json",
            side_effect=[{"state": 0}, {"state": 0}, device_list],
        ) as request_json:
            result = collector_api.create_modbus_devices("dev-01", devices)

        self.assertEqual(result["state"], 0)
        self.assertEqual(result["summary"], {"total": 2, "created": 2, "matched": 2, "failed": 0})
        self.assertEqual(result["results"][0]["device_id"], 9)
        self.assertEqual(result["results"][1]["device_id"], 11)
        self.assertEqual(request_json.call_args_list[0].args[1], "http://collector.test/api/collector/device")
        self.assertEqual(request_json.call_args_list[1].args[1], "http://collector.test/api/collector/device")
        self.assertEqual(request_json.call_args_list[2].args[1], "http://collector.test/api/collector/device?num_points=false")

    def test_create_modbus_points_batches_requests(self) -> None:
        self._patch_project()
        points = [
            {"device_id": 9, "name": "temperature", "func_code": 3, "address": 10, "type": "uint16"},
            {"device_id": 9, "name": "pressure", "func_code": 3, "address": 11, "type": "uint16"},
        ]
        with patch(
            "data_collector_mcp.collector_api.request_json",
            side_effect=[{"state": 0}, {"state": 0}],
        ) as request_json:
            result = collector_api.create_modbus_points("dev-01", points)

        self.assertEqual(result["state"], 0)
        self.assertEqual(result["summary"], {"total": 2, "success": 2, "failed": 0})
        self.assertEqual(request_json.call_count, 2)
        self.assertEqual(request_json.call_args_list[0].args[1], "http://collector.test/api/collector/modbus/point/add_collect_point")

    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_create_s7_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_s7_device(
                "dev-01",
                {
                    "name": "s7_1200_1",
                    "ip": "127.0.0.1",
                    "rock": 0,
                    "slot": 1,
                    "group_id": 10,
                },
            )

        self.assertEqual(result, response)
        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"], "s7")
        self.assertEqual(payload["device_type"], 1)
        self.assertEqual(payload["port"], 102)
        self.assertEqual(payload["tsap_conn_type"], "PG")
        self.assertEqual(payload["device_group_id"], 10)
        self.assertEqual(payload["rock"], 0)
        self.assertEqual(payload["slot"], 1)
        self.assertNotIn("group_id", payload)

    def test_create_s7_devices_batches_then_matches_largest_id(self) -> None:
        self._patch_project()
        devices = [
            {
                "name": "s7_1200_1",
                "ip": "127.0.0.1",
                "rock": 0,
                "slot": 1,
                "device_group_id": 10,
            }
        ]
        device_list = {
            "state": 0,
            "devices": [
                {
                    "id": 5,
                    "name": "s7_1200_1",
                    "type": "s7",
                    "ip": "127.0.0.1",
                    "port": 102,
                    "rock": 0,
                    "slot": 1,
                    "device_type": 1,
                    "tsap_conn_type": "PG",
                    "group_id": 10,
                },
                {
                    "id": 8,
                    "name": "s7_1200_1",
                    "type": "s7",
                    "ip": "127.0.0.1",
                    "port": 102,
                    "rock": 0,
                    "slot": 1,
                    "device_type": 1,
                    "tsap_conn_type": "PG",
                    "group_id": 10,
                },
            ],
        }
        with patch(
            "data_collector_mcp.collector_api.request_json",
            side_effect=[{"state": 0}, device_list],
        ) as request_json:
            result = collector_api.create_s7_devices("dev-01", devices)

        self.assertEqual(result["state"], 0)
        self.assertEqual(result["summary"], {"total": 1, "created": 1, "matched": 1, "failed": 0})
        self.assertEqual(result["results"][0]["device_id"], 8)
        self.assertEqual(request_json.call_args_list[0].args[1], "http://collector.test/api/collector/device")
        self.assertEqual(request_json.call_args_list[1].args[1], "http://collector.test/api/collector/device?num_points=false")

    def test_create_s7_points_batches_requests(self) -> None:
        self._patch_project()
        points = [
            {"device_id": 8, "name": "db_real", "register_area": "DB", "address": "1.10", "type": "REAL"},
            {"device_id": 8, "name": "m_bool", "register_area": "M", "address": "10.2", "type": "BOOL"},
        ]
        with patch(
            "data_collector_mcp.collector_api.request_json",
            side_effect=[{"state": 0}, {"state": 0}],
        ) as request_json:
            result = collector_api.create_s7_points("dev-01", points)

        self.assertEqual(result["state"], 0)
        self.assertEqual(result["summary"], {"total": 2, "success": 2, "failed": 0})
        self.assertEqual(request_json.call_count, 2)
        self.assertEqual(request_json.call_args_list[0].args[1], "http://collector.test/api/collector/s7/point/add")

    def test_edit_s7_device_maps_ori_id_and_posts_to_update_endpoint(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.edit_s7_device(
                "dev-01",
                {
                    "ori_id": 3,
                    "name": "s7_edited",
                    "ip": "127.0.0.1",
                    "rock": 0,
                    "slot": 2,
                    "tsap_conn_type": "op",
                },
            )

        self.assertEqual(request_json.call_args.args[1], "http://collector.test/api/collector/s7/device/update")
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["id"], 3)
        self.assertEqual(payload["tsap_conn_type"], "OP")
        self.assertNotIn("ori_id", payload)

    def test_create_s7_point_normalizes_aliases_and_posts_to_collector(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.create_s7_point(
                "dev-01",
                {
                    "device_id": 3,
                    "name": "db_real",
                    "point_id": "DB_REAL",
                    "register_area": "DB",
                    "address": "1.10",
                    "type": "REAL",
                    "group_id": 5,
                },
            )

        self.assertEqual(request_json.call_args.args[1], "http://collector.test/api/collector/s7/point/add")
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["register_type"], 4)
        self.assertEqual(payload["data_type"], "float32")
        self.assertEqual(payload["group_Id"], 5)
        self.assertEqual(payload["scale_ratio"], 1)
        self.assertNotIn("type", payload)
        self.assertNotIn("register_area", payload)
        self.assertNotIn("group_id", payload)

    def test_edit_s7_point_maps_ori_id_and_requires_device_id(self) -> None:
        self._patch_project()
        with patch(
            "data_collector_mcp.collector_api.request_json",
            return_value={"state": 0},
        ) as request_json:
            collector_api.edit_s7_point(
                "dev-01",
                {
                    "ori_id": 101,
                    "device_id": 3,
                    "name": "m_bool",
                    "register_type": 3,
                    "address": "10.2",
                    "data_type": "BOOL",
                },
            )

        self.assertEqual(request_json.call_args.args[1], "http://collector.test/api/collector/s7/point/update")
        payload = request_json.call_args.kwargs["json_payload"]
        self.assertEqual(payload["id"], 101)
        self.assertEqual(payload["data_type"], "bool")
        self.assertEqual(payload["register_type"], 3)
        self.assertEqual(payload["group_Id"], 0)
        self.assertNotIn("ori_id", payload)

        with self.assertRaisesRegex(ValueError, "payload.device_id is required"):
            collector_api.edit_s7_point(
                "dev-01",
                {
                    "ori_id": 101,
                    "name": "m_bool",
                    "register_type": 3,
                    "address": "10.2",
                    "data_type": "bool",
                },
            )

    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()