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

    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.point_collect_test",
                "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",
            },
        )

    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",
                        "base_url": "http://gateway.dev",
                        "data_collector_base_url": "http://collector.dev",
                    },
                    {
                        "project_key": "z-prod",
                        "project_name": "Prod",
                        "base_url": "http://gateway.prod",
                        "data_collector_base_url": "http://collector.prod",
                    },
                ],
                "total": 2,
            },
        )

    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_uses_explicit_parameters(self) -> None:
        signature = inspect.signature(modbus_server.collector_modbus_device_create)
        self.assertNotIn("payload", signature.parameters)

        with patch("data_collector_mcp.modbus_server.api_create_modbus_device", return_value={"state": 0}) as api_create:
            result = modbus_server.collector_modbus_device_create(
                project_key="dev-01",
                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,
            )

        self.assertEqual(result, {"state": 0})
        api_create.assert_called_once_with(
            "dev-01",
            {
                "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,
                "serial_port": "",
                "timeout": 3,
                "is_persistent": True,
                "baud_rate": 0,
                "data_bit": 0,
                "parity": 0,
                "stop_bit": 0,
                "mode": 0,
                "retry_times": 0,
                "group_id": 10,
                "alarm_interval": 90,
                "collect_interval": 5,
            },
        )

    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": True,
                "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",
            )

        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": 1,
                "value_offset": 0,
                "group_id": 0,
                "invalid_values": "",
                "valid_range_start": None,
                "valid_range_end": None,
                "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_uses_explicit_parameters(self) -> None:
        signature = inspect.signature(s7_server.collector_s7_device_create)
        self.assertNotIn("payload", signature.parameters)

        with patch("data_collector_mcp.s7_server.api_create_s7_device", return_value={"state": 0}) as api_create:
            result = s7_server.collector_s7_device_create(
                project_key="dev-01",
                name="s7_1200_1",
                ip="127.0.0.1",
                rock=0,
                slot=1,
                tsap_conn_type="OP",
            )

        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,
                "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_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")


if __name__ == "__main__":
    unittest.main()