robotmcp--ros-mcp-server/tests/integration/test_parameters.py

"""Integration tests for parameter tools.

These tests call the actual MCP tool functions (get_parameter, set_parameter,
has_parameter, delete_parameter, get_parameters, get_parameter_details)
against a live rosbridge container.

Parameter semantics differ across ROS versions:
- ROS 1 (melodic/noetic): global parameter server, slash-separated names
  (e.g. /turtlesim/background_r). The test seeds a known parameter via
  set_parameter before reading.
- ROS 2 (humble/jazzy): per-node parameters keyed "node:name" through
  rosbridge (e.g. /turtlesim:background_r). turtlesim populates background_r/g/b
  on launch.

get_parameters relies on rcl_interfaces/srv/ListParameters and is ROS 2-only;
on ROS 1 the tool returns a graceful "service does not exist" error, which the
test asserts directly.
"""

import time

import pytest

from ros_mcp.utils.rosapi_types import RosVersion, get_ros_version

pytestmark = [pytest.mark.integration]


_PARAMS = {
    RosVersion.ROS1: {
        "seed_name": "/ros_mcp_test_param",
        "seed_value": "42",
        # On ROS 1, the only param we know exists post-fixture is the one we seed.
        "existing_name": "/ros_mcp_test_param",
    },
    RosVersion.ROS2: {
        "seed_name": "/turtlesim:background_r",
        "seed_value": "100",
        # On ROS 2, turtlesim populates background_r on launch.
        "existing_name": "/turtlesim:background_r",
    },
}


def _params_for_current_version() -> dict:
    version = get_ros_version()
    return _PARAMS[version]


@pytest.fixture
def params():
    return _params_for_current_version()


@pytest.fixture
def ros1_seeded_param(tools, params):
    """On ROS 1, set a known parameter before the test and clean it up after."""
    if get_ros_version() != RosVersion.ROS1:
        yield params
        return
    tools["set_parameter"](name=params["seed_name"], value=params["seed_value"])
    yield params
    tools["delete_parameter"](name=params["seed_name"])


class TestGetParameter:
    def test_returns_value_for_known_parameter(self, tools, ros1_seeded_param):
        result = tools["get_parameter"](name=ros1_seeded_param["existing_name"])
        assert "value" in result
        assert result.get("successful") is True
        assert str(result["value"]).strip() != ""

    def test_empty_name_returns_error(self, tools):
        result = tools["get_parameter"](name="")
        assert "error" in result

    def test_whitespace_name_returns_error(self, tools):
        result = tools["get_parameter"](name="   ")
        assert "error" in result


class TestSetParameter:
    def test_set_succeeds(self, tools, params):
        version = get_ros_version()
        if version == RosVersion.ROS1:
            unique_name = f"/ros_mcp_test_set_{int(time.time_ns())}"
            result = tools["set_parameter"](name=unique_name, value="7")
            assert result.get("successful") is True
            tools["delete_parameter"](name=unique_name)
            return
        # ROS 2: mutating turtlesim's params, so capture and restore.
        target = params["seed_name"]
        original = tools["get_parameter"](name=target).get("value")
        try:
            result = tools["set_parameter"](name=target, value="7")
            assert result.get("successful") is True
        finally:
            if original is not None:
                tools["set_parameter"](name=target, value=str(original))

    def test_empty_name_returns_error(self, tools):
        result = tools["set_parameter"](name="", value="1")
        assert "error" in result


class TestHasParameter:
    def test_existing_parameter(self, tools, ros1_seeded_param):
        result = tools["has_parameter"](name=ros1_seeded_param["existing_name"])
        assert result["exists"] is True

    def test_nonexistent_parameter(self, tools):
        result = tools["has_parameter"](name="/ros_mcp_does_not_exist_xyz")
        assert result["exists"] is False

    def test_empty_name_returns_error(self, tools):
        result = tools["has_parameter"](name="")
        assert "error" in result


class TestDeleteParameter:
    def test_delete_round_trip(self, tools):
        if get_ros_version() == RosVersion.ROS2:
            pytest.skip(
                "ROS 2 delete-success is not exercisable against this setup: turtlesim "
                "silently drops undeclared params (set 'succeeds' but nothing persists "
                "to delete), and deleting a declared param crashes the rosapi node. The "
                "testable ROS 2 delete paths (nonexistent -> unsuccessful, empty name -> "
                "error) are covered by the other tests in this class."
            )
        name = f"/ros_mcp_test_delete_{int(time.time_ns())}"
        tools["set_parameter"](name=name, value="1")
        assert tools["has_parameter"](name=name)["exists"] is True
        result = tools["delete_parameter"](name=name)
        assert result.get("successful") is True
        assert tools["has_parameter"](name=name)["exists"] is False

    def test_nonexistent_returns_unsuccessful(self, tools):
        result = tools["delete_parameter"](name="/ros_mcp_does_not_exist_xyz")
        assert result.get("successful") is False

    def test_empty_name_returns_error(self, tools):
        result = tools["delete_parameter"](name="")
        assert "error" in result


class TestGetParameters:
    def test_returns_list_for_known_node(self, tools):
        result = tools["get_parameters"](node_name="turtlesim")
        if get_ros_version() == RosVersion.ROS1:
            # get_parameters relies on rcl_interfaces/srv/ListParameters which only
            # exists on ROS 2. The tool documents "Works only with ROS 2" and returns
            # a graceful error pointing at the missing per-node service.
            assert "error" in result
            assert "list_parameters" in result["error"]
            return
        assert "parameters" in result
        assert "parameter_count" in result
        assert isinstance(result["parameters"], list)

    def test_empty_node_returns_error(self, tools):
        result = tools["get_parameters"](node_name="")
        assert "error" in result


class TestGetParameterDetails:
    def test_returns_details_for_known_parameter(self, tools, ros1_seeded_param):
        # On ROS 1, the describe_parameters call fails internally but the tool
        # falls back to type inference from the value, so the test passes on both
        # versions. On ROS 2, describe_parameters supplies the type.
        result = tools["get_parameter_details"](name=ros1_seeded_param["existing_name"])
        assert result["exists"] is True
        assert "value" in result
        assert "type" in result

    def test_nonexistent_returns_error_response(self, tools):
        if get_ros_version() == RosVersion.ROS1:
            name = "/ros_mcp_definitely_does_not_exist"
        else:
            name = "/turtlesim:does_not_exist"
        result = tools["get_parameter_details"](name=name)
        assert result["exists"] is False

    def test_empty_name_returns_error(self, tools):
        result = tools["get_parameter_details"](name="")
        assert "error" in result