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