wehub-resource-sync
db1d565b64
Integration Tests / melodic (push) Has been cancelled
Integration Tests / noetic (push) Has been cancelled
Integration Tests / humble (push) Has been cancelled
Integration Tests / jazzy (push) Has been cancelled
Ruff Lint & Format / ruff (push) Has been cancelled
Sync main to develop / Check if sync is needed (push) Has been cancelled
Sync main to develop / Sync main to develop (push) Has been cancelled
98 lines
4.1 KiB
Python
98 lines
4.1 KiB
Python
"""Integration tests for node tools (step 2).
|
|
|
|
These tests call the actual MCP tool functions (get_nodes, get_node_details)
|
|
against a live rosbridge container.
|
|
|
|
Note: calling get_node_details for nonexistent nodes crashes rosapi_node
|
|
on ROS 2 (#273). Tests only query nodes confirmed to exist.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
pytestmark = [pytest.mark.integration]
|
|
|
|
|
|
class TestGetNodes:
|
|
"""Verify get_nodes MCP tool returns the running node list."""
|
|
|
|
def test_returns_nodes(self, tools):
|
|
"""get_nodes should return nodes and node_count."""
|
|
result = tools["get_nodes"]()
|
|
assert "nodes" in result
|
|
assert "node_count" in result
|
|
assert result["node_count"] > 0
|
|
assert result["node_count"] == len(result["nodes"])
|
|
|
|
def test_includes_turtlesim(self, tools):
|
|
"""turtlesim node should be present (launched by Docker container)."""
|
|
result = tools["get_nodes"]()
|
|
nodes = result["nodes"]
|
|
assert any("/turtlesim" in n for n in nodes), f"turtlesim not in {nodes}"
|
|
|
|
def test_includes_rosbridge(self, tools):
|
|
"""rosbridge node should be present."""
|
|
result = tools["get_nodes"]()
|
|
nodes = result["nodes"]
|
|
assert any("rosbridge" in n for n in nodes), f"rosbridge not in {nodes}"
|
|
|
|
def test_node_count_at_least_three(self, tools):
|
|
"""Should have at least 3 nodes: turtlesim, rosbridge, rosapi."""
|
|
result = tools["get_nodes"]()
|
|
assert result["node_count"] >= 3, (
|
|
f"Expected >= 3, got {result['node_count']}: {result['nodes']}"
|
|
)
|
|
|
|
def test_all_nodes_are_ros_names(self, tools):
|
|
"""Every node name should be a string starting with /."""
|
|
result = tools["get_nodes"]()
|
|
for node in result["nodes"]:
|
|
assert isinstance(node, str), f"Expected string, got {type(node)}: {node}"
|
|
assert node.startswith("/"), f"Node name should start with /: {node}"
|
|
|
|
|
|
class TestGetNodeDetails:
|
|
"""Verify get_node_details MCP tool returns publishers/subscribers/services.
|
|
|
|
Only queries nodes confirmed to exist — calling get_node_details for
|
|
nonexistent nodes crashes rosapi_node on ROS 2 (#273).
|
|
"""
|
|
|
|
def test_turtlesim_details(self, tools):
|
|
"""get_node_details for /turtlesim should return publishers and subscribers."""
|
|
result = tools["get_node_details"](node="/turtlesim")
|
|
assert result["node"] == "/turtlesim"
|
|
assert result["publisher_count"] > 0, "turtlesim should have publishers"
|
|
assert result["subscriber_count"] > 0, "turtlesim should have subscribers"
|
|
assert any("pose" in p for p in result["publishers"]), f"No pose in {result['publishers']}"
|
|
assert any("cmd_vel" in s for s in result["subscribers"]), (
|
|
f"No cmd_vel in {result['subscribers']}"
|
|
)
|
|
|
|
def test_rosbridge_has_services(self, tools):
|
|
"""get_node_details for rosbridge should return services."""
|
|
# Find rosbridge node name dynamically (varies by distro)
|
|
nodes_result = tools["get_nodes"]()
|
|
rosbridge = next((n for n in nodes_result["nodes"] if "rosbridge" in n), None)
|
|
assert rosbridge is not None
|
|
|
|
result = tools["get_node_details"](node=rosbridge)
|
|
assert result["node"] == rosbridge
|
|
assert result["service_count"] > 0, "rosbridge should have services"
|
|
|
|
def test_detail_counts_match_lists(self, tools):
|
|
"""publisher_count/subscriber_count/service_count should match list lengths."""
|
|
result = tools["get_node_details"](node="/turtlesim")
|
|
assert result["publisher_count"] == len(result["publishers"])
|
|
assert result["subscriber_count"] == len(result["subscribers"])
|
|
assert result["service_count"] == len(result["services"])
|
|
|
|
def test_empty_node_name_returns_error(self, tools):
|
|
"""get_node_details with empty string should return error dict."""
|
|
result = tools["get_node_details"](node="")
|
|
assert "error" in result
|
|
|
|
def test_whitespace_node_name_returns_error(self, tools):
|
|
"""get_node_details with whitespace should return error dict."""
|
|
result = tools["get_node_details"](node=" ")
|
|
assert "error" in result
|