data-collector-mcp/tests/test_server_tools.py

from __future__ import annotations

import inspect
import unittest
from unittest.mock import patch

from data_collector_mcp import server


class ServerToolTests(unittest.TestCase):
    def test_modbus_device_edit_uses_explicit_parameters(self) -> None:
        signature = inspect.signature(server.collector_modbus_device_edit)
        self.assertNotIn("payload", signature.parameters)

        with patch("data_collector_mcp.server.api_edit_modbus_device", return_value={"state": 0}) as api_edit:
            result = 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(server.collector_modbus_point_edit)
        self.assertNotIn("payload", signature.parameters)

        with patch("data_collector_mcp.server.api_edit_modbus_point", return_value={"state": 0}) as api_edit:
            result = 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.server.api_edit_modbus_point", return_value={"state": 0}) as api_edit:
            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)


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