from __future__ import annotations import asyncio import inspect import unittest from unittest.mock import patch from data_collector_mcp import app, bacnet_server, common_server, modbus_server, s7_server class ServerToolTests(unittest.TestCase): def test_app_imports_tool_modules_for_registration(self) -> None: self.assertIs(app.common_server, common_server) self.assertIs(app.modbus_server, modbus_server) self.assertIs(app.s7_server, s7_server) self.assertIs(app.bacnet_server, bacnet_server) def test_app_registers_expected_mcp_tools(self) -> None: tools = asyncio.run(app.mcp.list_tools()) tool_names = {tool.name for tool in tools} self.assertEqual( tool_names, { "project.list", "collector.device_list", "collector.device_connect", "collector.device_disconnect", "collector.device_points", "modbus.raw_read", "modbus.point_collect_test", "bacnet.point_collect_test", "bacnet.point_search", "bacnet.bbmd_whois", "s7.raw_read", "s7.point_collect_test", "s7.connect_scan", "collector.modbus_device_create", "collector.modbus_device_edit", "collector.modbus_point_create", "collector.modbus_point_edit", "collector.s7_device_create", "collector.s7_device_edit", "collector.s7_point_create", "collector.s7_point_edit", "collector.bacnet_device_create", "collector.bacnet_device_edit", "collector.bacnet_point_create", "collector.bacnet_point_edit", }, ) def test_common_tools_include_field_descriptions_and_output_schema(self) -> None: tools = {tool.name: tool for tool in asyncio.run(app.mcp.list_tools())} project_list = tools["project.list"] project_props = project_list.output_schema["properties"] self.assertIn("projects", project_props) project_item_props = project_props["projects"]["items"]["properties"] self.assertEqual(project_item_props["project_key"]["description"], "项目标识，用于其他采集工具的 project_key 参数。") device_list = tools["collector.device_list"] self.assertEqual( device_list.parameters["properties"]["project_key"]["description"], "项目标识，来自 project.list 返回的 project_key。", ) self.assertEqual(device_list.parameters["properties"]["num_points"]["default"], False) self.assertIn("devices", device_list.output_schema["properties"]) connect = tools["collector.device_connect"] self.assertEqual( connect.parameters["properties"]["device_type"]["description"], "设备协议类型，可传 modbus、s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。", ) connect_data_schema = connect.output_schema["properties"]["data"]["anyOf"][0] self.assertIn("running_status", connect_data_schema["properties"]) disconnect = tools["collector.device_disconnect"] self.assertEqual(disconnect.parameters["properties"]["device_type"]["default"], "modbus") self.assertIn("data", disconnect.output_schema["properties"]) device_points = tools["collector.device_points"] self.assertEqual(device_points.parameters["properties"]["group_id"]["default"], 0) points_data_schema = device_points.output_schema["properties"]["data"]["anyOf"][0] point_props = points_data_schema["properties"]["point"]["items"]["properties"] self.assertEqual(point_props["present_value"]["description"], "当前内存中的点位最新值。") def test_protocol_tools_include_field_descriptions_and_output_schema(self) -> None: tools = {tool.name: tool for tool in asyncio.run(app.mcp.list_tools())} point_search = tools["bacnet.point_search"] self.assertEqual( point_search.parameters["properties"]["bacnet_device_id"]["description"], "BACnet 设备对象实例号，范围 0..4194303。", ) point_output_props = point_search.output_schema["properties"] self.assertIn("code", point_output_props) point_data_schema = point_output_props["data"]["anyOf"][0] point_props = point_data_schema["properties"]["points"]["items"]["properties"] self.assertEqual(point_props["present_value"]["description"], "当前值；读取 BACnet present-value 得到。") device_create = tools["collector.bacnet_device_create"] device_item_props = device_create.parameters["properties"]["devices"]["items"]["properties"] self.assertEqual(device_item_props["ip"]["description"], "BACnet/IP 设备地址。") create_output_props = device_create.output_schema["properties"] self.assertIn("summary", create_output_props) self.assertIn("results", create_output_props) device_edit_props = tools["collector.bacnet_device_edit"].parameters["properties"] self.assertIn("id", device_edit_props) self.assertNotIn("ori_id", device_edit_props) point_edit_props = tools["collector.bacnet_point_edit"].parameters["properties"] self.assertIn("id", point_edit_props) self.assertNotIn("ori_id", point_edit_props) self.assertNotIn("data", tools["collector.bacnet_point_edit"].output_schema["properties"]) modbus_collect = tools["modbus.point_collect_test"] modbus_point_props = modbus_collect.parameters["properties"]["points"]["items"]["properties"] self.assertEqual(modbus_point_props["address"]["description"], "寄存器地址。") self.assertIn("data", modbus_collect.output_schema["properties"]) modbus_output_point_props = ( modbus_collect.output_schema["properties"]["data"]["anyOf"][0]["properties"]["points"]["items"]["properties"] ) self.assertIn("value", modbus_output_point_props) self.assertNotIn("present_value", modbus_output_point_props) for tool_name in ("collector.modbus_device_edit", "collector.modbus_point_edit"): props = tools[tool_name].parameters["properties"] self.assertIn("id", props) self.assertNotIn("ori_id", props) self.assertNotIn("register_type", tools["collector.modbus_point_edit"].parameters["properties"]) self.assertIn("1=Read Coils", tools["collector.modbus_point_edit"].parameters["properties"]["func_code"]["description"]) self.assertNotIn("data", tools["collector.modbus_point_edit"].output_schema["properties"]) s7_collect = tools["s7.point_collect_test"] s7_point_props = s7_collect.parameters["properties"]["points"]["items"]["properties"] self.assertEqual(s7_point_props["area"]["description"], "读取区域，可用 DB、M、I、Q、V。") self.assertIn("data", s7_collect.output_schema["properties"]) s7_output_point_props = s7_collect.output_schema["properties"]["data"]["anyOf"][0]["properties"]["points"]["items"]["properties"] self.assertIn("value", s7_output_point_props) self.assertNotIn("present_value", s7_output_point_props) for tool_name in ("collector.s7_device_edit", "collector.s7_point_edit"): props = tools[tool_name].parameters["properties"] self.assertIn("id", props) self.assertNotIn("ori_id", props) self.assertNotIn("data", tools["collector.s7_point_edit"].output_schema["properties"]) s7_create_point_props = tools["collector.s7_point_create"].parameters["properties"]["points"]["items"]["properties"] self.assertIn("register_type", s7_create_point_props) self.assertNotIn("register_area", s7_create_point_props) self.assertNotIn("register_area", tools["collector.s7_point_edit"].parameters["properties"]) def test_project_list_filters_enabled_projects_and_sorts(self) -> None: with patch( "data_collector_mcp.common_server.load_projects_config", return_value=[ { "project_key": "z-prod", "project_name": "Prod", "base_url": "http://gateway.prod", "data_collector_base_url": "http://collector.prod", "enabled": True, }, { "project_key": "disabled", "project_name": "Disabled", "base_url": "http://gateway.disabled", "data_collector_base_url": "http://collector.disabled", "enabled": False, }, { "project_key": "a-dev", "project_name": "Dev", "base_url": "http://gateway.dev", "data_collector_base_url": "http://collector.dev", "enabled": True, }, ], ): result = common_server.project_list() self.assertEqual( result, { "projects": [ { "project_key": "a-dev", "project_name": "Dev", }, { "project_key": "z-prod", "project_name": "Prod", }, ], "total": 2, }, ) for project in result["projects"]: self.assertNotIn("base_url", project) self.assertNotIn("data_collector_base_url", project) def test_common_device_tools_forward_to_api(self) -> None: with patch("data_collector_mcp.common_server.api_list_devices", return_value={"state": 0}) as list_devices: self.assertEqual(common_server.collector_device_list("dev-01", num_points=True), {"state": 0}) list_devices.assert_called_once_with("dev-01", num_points=True) with patch("data_collector_mcp.common_server.api_connect_device", return_value={"state": 0}) as connect: self.assertEqual( common_server.collector_device_connect("dev-01", device_id=1, device_type="s7"), {"state": 0}, ) connect.assert_called_once_with("dev-01", device_id=1, device_type="s7") with patch("data_collector_mcp.common_server.api_disconnect_device", return_value={"state": 0}) as disconnect: self.assertEqual( common_server.collector_device_disconnect("dev-01", device_id=1, device_type="modbus"), {"state": 0}, ) disconnect.assert_called_once_with("dev-01", device_id=1, device_type="modbus") with patch("data_collector_mcp.common_server.api_list_device_points", return_value={"state": 0}) as points: self.assertEqual( common_server.collector_device_points("dev-01", device_id=1, device_type="s7", group_id=2), {"state": 0}, ) points.assert_called_once_with("dev-01", device_id=1, device_type="s7", group_id=2) def test_modbus_device_create_accepts_batch_devices(self) -> None: signature = inspect.signature(modbus_server.collector_modbus_device_create) self.assertIn("devices", signature.parameters) 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, } ] with patch("data_collector_mcp.modbus_server.api_create_modbus_devices", return_value={"state": 0}) as api_create: result = modbus_server.collector_modbus_device_create( project_key="dev-01", devices=devices, ) self.assertEqual(result, {"state": 0}) api_create.assert_called_once_with("dev-01", devices) def test_modbus_raw_read_forwards_to_api(self) -> None: read = {"function_code": 3, "address": 0, "quantity": 4} with patch("data_collector_mcp.modbus_server.api_modbus_raw_read", return_value={"code": 0}) as raw_read: result = modbus_server.modbus_raw_read( project_key="dev-01", ip="192.168.1.10", port=502, slave_id=1, read=read, word_byte_order="DCBA", address_base=1, ) self.assertEqual(result, {"code": 0}) raw_read.assert_called_once_with( "dev-01", ip="192.168.1.10", port=502, slave_id=1, read=read, device_type="ModbusTCP", word_byte_order="DCBA", address_base=1, ) def test_modbus_point_create_accepts_batch_points(self) -> None: points = [ { "device_id": 1, "name": "temperature", "address": 10, "type": "uint16", "func_code": 3, "scale_ratio": 2, "value_offset": 1, "group_id": 5, "invalid_values": "-9999", "valid_range_start": 0, "valid_range_end": 100, } ] with patch("data_collector_mcp.modbus_server.api_create_modbus_points", return_value={"state": 0}) as api_create: result = modbus_server.collector_modbus_point_create(project_key="dev-01", points=points) self.assertEqual(result, {"state": 0}) api_create.assert_called_once_with("dev-01", points) def test_modbus_device_edit_uses_explicit_parameters(self) -> None: signature = inspect.signature(modbus_server.collector_modbus_device_edit) self.assertNotIn("payload", signature.parameters) with patch("data_collector_mcp.modbus_server.api_edit_modbus_device", return_value={"state": 0}) as api_edit: result = modbus_server.collector_modbus_device_edit( project_key="dev-01", 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_offset=1, device_group_id=10, ) self.assertEqual(result, {"state": 0}) api_edit.assert_called_once_with( "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, "serial_port": "", "timeout": 3, "is_persistent": False, "baud_rate": 0, "data_bit": 0, "parity": 0, "stop_bit": 0, "mode": 0, "address_offset": 1, "retry_times": 0, "device_group_id": 10, "alarm_interval": 90, "collect_interval": 5, }, ) def test_modbus_point_edit_uses_explicit_parameters_with_func_code(self) -> None: signature = inspect.signature(modbus_server.collector_modbus_point_edit) self.assertNotIn("payload", signature.parameters) with patch("data_collector_mcp.modbus_server.api_edit_modbus_point", return_value={"state": 0}) as api_edit: result = modbus_server.collector_modbus_point_edit( project_key="dev-01", id=101, name="holding_register_uint16_edited", address=10, data_type="uint16", func_code=3, point_id="HR_UINT16_EDITED", scale_ratio=2, value_offset=1, group_id=5, invalid_values="-9999", valid_range_start=0, valid_range_end=100, ) self.assertEqual(result, {"state": 0}) api_edit.assert_called_once_with( "dev-01", { "ori_id": 101, "name": "holding_register_uint16_edited", "address": 10, "type": "uint16", "point_id": "HR_UINT16_EDITED", "scale_ratio": 2, "value_offset": 1, "group_id": 5, "invalid_values": "-9999", "valid_range_start": 0, "valid_range_end": 100, "bit": 0, "describe": "", "func_code": 3, }, ) def test_s7_device_create_accepts_batch_devices(self) -> None: signature = inspect.signature(s7_server.collector_s7_device_create) self.assertIn("devices", signature.parameters) devices = [{"name": "s7_1200_1", "ip": "127.0.0.1", "rock": 0, "slot": 1, "tsap_conn_type": "OP"}] with patch("data_collector_mcp.s7_server.api_create_s7_devices", return_value={"state": 0}) as api_create: result = s7_server.collector_s7_device_create( project_key="dev-01", devices=devices, ) self.assertEqual(result, {"state": 0}) api_create.assert_called_once_with( "dev-01", [{"name": "s7_1200_1", "ip": "127.0.0.1", "rock": 0, "slot": 1, "tsap_conn_type": "OP"}], ) def test_s7_point_create_accepts_batch_points(self) -> None: points = [{"device_id": 3, "name": "db_real", "address": "1.10", "type": "REAL", "register_type": 4}] with patch("data_collector_mcp.s7_server.api_create_s7_points", return_value={"state": 0}) as api_create: result = s7_server.collector_s7_point_create(project_key="dev-01", points=points) self.assertEqual(result, {"state": 0}) api_create.assert_called_once_with("dev-01", points) def test_s7_device_edit_uses_explicit_parameters(self) -> None: signature = inspect.signature(s7_server.collector_s7_device_edit) self.assertNotIn("payload", signature.parameters) with patch("data_collector_mcp.s7_server.api_edit_s7_device", return_value={"state": 0}) as api_edit: result = s7_server.collector_s7_device_edit( project_key="dev-01", id=3, name="s7_edited", ip="127.0.0.1", rock=0, slot=1, tsap_conn_type="OP", ) self.assertEqual(result, {"state": 0}) api_edit.assert_called_once_with( "dev-01", { "id": 3, "name": "s7_edited", "ip": "127.0.0.1", "rock": 0, "slot": 1, "port": 102, "device_type": 1, "tsap_conn_type": "OP", "is_persistent": False, "device_group_id": 0, "timeout": 3, "alarm_interval": 90, "collect_interval": 5, }, ) def test_s7_point_edit_uses_register_type(self) -> None: signature = inspect.signature(s7_server.collector_s7_point_edit) self.assertNotIn("payload", signature.parameters) with patch("data_collector_mcp.s7_server.api_edit_s7_point", return_value={"state": 0}) as api_edit: s7_server.collector_s7_point_edit( project_key="dev-01", id=101, device_id=3, name="db_real", address="1.10", data_type="float32", register_type=4, point_id="DB_REAL", ) payload = api_edit.call_args.args[1] self.assertEqual(payload["register_type"], 4) self.assertNotIn("register_area", payload) def test_s7_gateway_tools_forward_to_api(self) -> None: with patch("data_collector_mcp.s7_server.api_s7_raw_read", return_value={"code": 0}) as raw_read: self.assertEqual( s7_server.s7_raw_read( project_key="dev-01", ip="192.168.1.10", rock=0, slot=1, read={"area": "DB", "db": 1, "start": 0, "size": 4}, ), {"code": 0}, ) raw_read.assert_called_once_with( "dev-01", ip="192.168.1.10", rock=0, slot=1, read={"area": "DB", "db": 1, "start": 0, "size": 4}, device_type="S7-1200", port=102, tsap_conn_type=None, ) with patch("data_collector_mcp.s7_server.api_s7_point_collect_test", return_value={"code": 0}) as point_test: self.assertEqual( s7_server.s7_point_collect_test( project_key="dev-01", ip="192.168.1.10", port=1102, rock=0, slot=1, tsap_conn_type="BASIC", points=[{"area": "M", "start": 0, "type": "bool"}], ), {"code": 0}, ) point_test.assert_called_once_with( "dev-01", ip="192.168.1.10", rock=0, slot=1, points=[{"area": "M", "start": 0, "type": "bool"}], device_type="S7-1200", port=1102, tsap_conn_type="BASIC", ) with patch("data_collector_mcp.s7_server.api_s7_connect_scan", return_value={"code": 0}) as connect_scan: self.assertEqual(s7_server.s7_connect_scan("dev-01", ip="192.168.1.10"), {"code": 0}) connect_scan.assert_called_once_with("dev-01", ip="192.168.1.10") def test_bacnet_gateway_tools_forward_to_api(self) -> None: with patch( "data_collector_mcp.bacnet_server.api_bacnet_point_collect_test", return_value={"code": 0}, ) as point_test: self.assertEqual( bacnet_server.bacnet_point_collect_test( project_key="dev-01", ip="192.168.1.20", bacnet_device_id=12345, port=47809, points=[{"object_type": "AnalogInput", "object_id": 1}], ), {"code": 0}, ) point_test.assert_called_once_with( "dev-01", ip="192.168.1.20", bacnet_device_id=12345, port=47809, points=[{"object_type": "AnalogInput", "object_id": 1}], ) with patch( "data_collector_mcp.bacnet_server.api_bacnet_point_search", return_value={"code": 0}, ) as point_search: self.assertEqual( bacnet_server.bacnet_point_search( project_key="dev-01", ip="192.168.1.20", bacnet_device_id=12345, ), {"code": 0}, ) point_search.assert_called_once_with( "dev-01", ip="192.168.1.20", bacnet_device_id=12345, port=47808, ) with patch( "data_collector_mcp.bacnet_server.api_bacnet_bbmd_whois", return_value={"code": 0}, ) as bbmd_whois: self.assertEqual(bacnet_server.bacnet_bbmd_whois("dev-01"), {"code": 0}) bbmd_whois.assert_called_once_with("dev-01") def test_bacnet_collector_tools_forward_to_api(self) -> None: devices = [{"name": "bacnet_1", "ip": "192.168.1.20", "bacnet_device_id": 12345}] with patch( "data_collector_mcp.bacnet_server.api_create_bacnet_devices", return_value={"state": 0}, ) as api_create: result = bacnet_server.collector_bacnet_device_create(project_key="dev-01", devices=devices) self.assertEqual(result, {"state": 0}) api_create.assert_called_once_with("dev-01", devices) with patch( "data_collector_mcp.bacnet_server.api_edit_bacnet_device", return_value={"state": 0}, ) as api_edit: result = bacnet_server.collector_bacnet_device_edit( project_key="dev-01", id=9, name="bacnet_edited", ip="192.168.1.21", bacnet_device_id=54321, device_group_id=10, ) self.assertEqual(result, {"state": 0}) api_edit.assert_called_once_with( "dev-01", { "ori_id": 9, "name": "bacnet_edited", "ip": "192.168.1.21", "bacnet_device_id": 54321, "port": 47808, "bacnet_net": 0, "asp_ip": "", "is_persistent": False, "device_group_id": 10, "timeout": 3, "alarm_interval": 90, "collect_interval": 5, }, ) points = [ { "device_id": 9, "name": "zone_temperature", "object_type": "AnalogInput", "object_id": 1, "scale_ratio": 2, "value_offset": 1, "group_id": 5, "invalid_values": "-9999", "valid_range_start": 0, "valid_range_end": 100, } ] with patch( "data_collector_mcp.bacnet_server.api_create_bacnet_points", return_value={"state": 0}, ) as api_point_create: result = bacnet_server.collector_bacnet_point_create(project_key="dev-01", points=points) self.assertEqual(result, {"state": 0}) api_point_create.assert_called_once_with("dev-01", points) with patch( "data_collector_mcp.bacnet_server.api_edit_bacnet_point", return_value={"state": 0}, ) as api_point_edit: result = bacnet_server.collector_bacnet_point_edit( project_key="dev-01", id=101, name="zone_temperature_edited", object_type="AnalogInput", object_id=1, point_id="AI_TEMP_EDITED", ) self.assertEqual(result, {"state": 0}) payload = api_point_edit.call_args.args[1] self.assertEqual(api_point_edit.call_args.args[0], "dev-01") self.assertEqual(payload["id"], 101) self.assertEqual(payload["name"], "zone_temperature_edited") self.assertEqual(payload["object_type"], "AnalogInput") self.assertEqual(payload["object_id"], 1) self.assertEqual(payload["point_id"], "AI_TEMP_EDITED") with patch( "data_collector_mcp.bacnet_server.api_edit_bacnet_point", return_value={"state": 0}, ) as api_point_edit: bacnet_server.collector_bacnet_point_edit( project_key="dev-01", id=101, name="zone_temperature_edited", object_type="AnalogInput", object_id=1, scale_ratio=2, value_offset=1, group_id=5, invalid_values="-9999", valid_range_start=0, valid_range_end=100, ) payload = api_point_edit.call_args.args[1] self.assertEqual(payload["scale_ratio"], 2) self.assertEqual(payload["value_offset"], 1) self.assertEqual(payload["group_id"], 5) self.assertEqual(payload["invalid_values"], "-9999") self.assertEqual(payload["valid_range_start"], 0) self.assertEqual(payload["valid_range_end"], 100) if __name__ == "__main__": unittest.main()