data-collector-mcp/tests/test_server_tools.py

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_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",
                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_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",
                ori_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_modbus_point_edit_uses_register_type_when_func_code_is_zero(self) -> None:
        with patch("data_collector_mcp.modbus_server.api_edit_modbus_point", return_value={"state": 0}) as api_edit:
            modbus_server.collector_modbus_point_edit(
                project_key="dev-01",
                ori_id=101,
                name="holding_register_uint16_edited",
                address=10,
                data_type="uint16",
                register_type="holding_register",
            )

        payload = api_edit.call_args.args[1]
        self.assertEqual(payload["register_type"], "holding_register")
        self.assertNotIn("func_code", payload)

    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_area": "DB"}]
        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",
                ori_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",
            {
                "ori_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_area_when_register_type_is_zero(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",
                ori_id=101,
                device_id=3,
                name="db_real",
                address="1.10",
                data_type="float32",
                register_area="DB",
                point_id="DB_REAL",
            )

        payload = api_edit.call_args.args[1]
        self.assertEqual(payload["register_area"], "DB")
        self.assertNotIn("register_type", 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",
                ori_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",
                ori_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["ori_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",
                ori_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()