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