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_device_defaults_smart200_tsap_to_pg(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_device( "dev-01", { "name": "smart200_1", "ip": "127.0.0.1", "rock": 0, "slot": 1, "device_type": 3, }, ) payload = request_json.call_args.kwargs["json_payload"] self.assertEqual(payload["device_type"], 3) self.assertEqual(payload["tsap_conn_type"], "PG") 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_uses_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", { "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_uses_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", { "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", { "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="") def test_create_bacnet_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_bacnet_device( "dev-01", { "name": "bacnet_1", "ip": "192.168.1.20", "bacnet_device_id": 12345, }, ) 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"], "bacnet") self.assertEqual(payload["device_type"], 1) self.assertEqual(payload["port"], 47808) self.assertEqual(payload["bacnet_net"], 0) self.assertEqual(payload["asp_ip"], "") self.assertEqual(payload["timeout"], 3) self.assertFalse(payload["is_persistent"]) self.assertEqual(payload["group_id"], 0) self.assertEqual(payload["alarm_interval"], 90) self.assertEqual(payload["collect_interval"], 5) def test_create_bacnet_devices_batches_then_matches_largest_id(self) -> None: self._patch_project() devices = [{"name": "bacnet_1", "ip": "192.168.1.20", "bacnet_device_id": 12345, "group_id": 10}] device_list = { "state": 0, "devices": [ { "id": 1, "name": "group", "type": "devicegroup", "groups": [ { "id": 3, "name": "bacnet_1", "type": "bacnet", "device_type": 1, "ip": "192.168.1.20", "port": 47808, "bacnet_device_id": 12345, "bacnet_net": 0, "group_id": 10, }, { "id": 9, "name": "bacnet_1", "type": "bacnet", "device_type": 1, "ip": "192.168.1.20", "port": 47808, "bacnet_device_id": 12345, "bacnet_net": 0, "group_id": 10, }, ], } ], } with patch( "data_collector_mcp.collector_api.request_json", side_effect=[{"state": 0}, device_list], ): result = collector_api.create_bacnet_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"], 9) def test_edit_bacnet_device_maps_to_legacy_endpoint(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.edit_bacnet_device( "dev-01", { "ori_id": 9, "name": "bacnet_edited", "ip": "192.168.1.21", "bacnet_device_id": 54321, "bacnet_net": 2, "device_group_id": 10, }, ) self.assertEqual(result, response) self.assertEqual( request_json.call_args.args[:3], ("POST", "http://collector.test/api/collector/bacnet/device/edit", "token"), ) payload = request_json.call_args.kwargs["json_payload"] self.assertEqual(payload["ori_id"], 9) self.assertEqual(payload["type"], 1) self.assertEqual(payload["device_id"], "54321") self.assertEqual(payload["net"], 2) self.assertEqual(payload["device_group_id"], 10) self.assertNotIn("bacnet_device_id", payload) self.assertNotIn("bacnet_net", payload) def test_create_bacnet_point_wraps_single_point_batch_payload(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_bacnet_point( "dev-01", { "device_id": 9, "name": "zone_temperature", "object_type": "AnalogInput", "object_id": 1, "point_id": "AI_TEMP", "value_type": 4, "invalid_values": "-9999", "valid_range_start": 0, "valid_range_end": 100, }, ) self.assertEqual(result, response) self.assertEqual( request_json.call_args.args[:3], ("POST", "http://collector.test/api/collector/bacnet/point/add_collect_point", "token"), ) payload = request_json.call_args.kwargs["json_payload"] self.assertEqual(payload["device_id"], 9) self.assertEqual(len(payload["points"]), 1) point = payload["points"][0] self.assertEqual(point["object_name"], "zone_temperature") self.assertEqual(point["name"], "zone_temperature") self.assertEqual(point["object_type"], "AnalogInput") self.assertEqual(point["object_id"], 1) self.assertEqual(point["point_id"], "AI_TEMP") self.assertEqual(point["value_type"], 4) self.assertEqual(point["scale_ratio"], 1) self.assertEqual(point["invalid_values"], "-9999") self.assertEqual(point["valid_range_start"], 0) self.assertEqual(point["valid_range_end"], 100) def test_bacnet_point_payloads_normalize_object_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_bacnet_point( "dev-01", { "device_id": 9, "name": "zone_temperature", "object_type": "analog-input", "object_id": 1, }, ) payload = request_json.call_args.kwargs["json_payload"] self.assertEqual(payload["points"][0]["object_type"], "AnalogInput") with patch( "data_collector_mcp.collector_api.request_json", return_value={"state": 0}, ) as request_json: collector_api.edit_bacnet_point( "dev-01", { "id": 101, "name": "zone_temperature_edited", "object_type": "analogInput", "object_id": 1, }, ) payload = request_json.call_args.kwargs["json_payload"] self.assertEqual(payload["object_type"], "AnalogInput") def test_edit_bacnet_point_uses_id(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.edit_bacnet_point( "dev-01", { "id": 101, "name": "zone_temperature_edited", "object_type": "AnalogInput", "object_id": 1, "point_id": "AI_TEMP_EDITED", "invalid_values": "-9999", }, ) self.assertEqual(result, response) self.assertEqual( request_json.call_args.args[:3], ("POST", "http://collector.test/api/collector/bacnet/point/edit", "token"), ) payload = request_json.call_args.kwargs["json_payload"] self.assertEqual(payload["id"], 101) self.assertEqual(payload["object_name"], "zone_temperature_edited") self.assertEqual(payload["point_id"], "AI_TEMP_EDITED") self.assertEqual(payload["invalid_values"], "-9999") self.assertIsNone(payload["valid_range_start"]) self.assertNotIn("ori_id", payload) def test_bacnet_payloads_validate_required_fields(self) -> None: self._patch_project() with self.assertRaisesRegex(ValueError, "payload.bacnet_device_id is required"): collector_api.create_bacnet_device("dev-01", {"name": "bacnet_1", "ip": "192.168.1.20"}) with self.assertRaisesRegex(ValueError, "payload.object_name is required"): collector_api.create_bacnet_point( "dev-01", {"device_id": 1, "object_type": "AnalogInput", "object_id": 1}, ) with self.assertRaisesRegex(ValueError, "payload.object_id must be between"): collector_api.edit_bacnet_point( "dev-01", {"id": 1, "name": "bad", "object_type": "AnalogInput", "object_id": 4_194_304}, ) if __name__ == "__main__": unittest.main()