"""
Tests for Litmus MCP Server Tools
This test suite covers all MCP tools with both success and error scenarios.
Uses pytest with mocking to test stateless, header-based authentication.
"""
import pytest
from unittest.mock import Mock, MagicMock, patch
from starlette.requests import Request
# Import the tools from server
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
from server import (
get_litmusedge_driver_list,
get_devicehub_devices,
get_devicehub_device_tags,
get_current_value_of_devicehub_tag,
create_devicehub_device,
get_litmusedge_friendly_name,
set_litmusedge_friendly_name,
get_cloud_activation_status,
get_all_containers_on_litmusedge,
run_docker_container_on_litmusedge,
get_current_value_on_topic,
get_multiple_values_from_topic,
)
# ==================== Fixtures ====================
@pytest.fixture
def mock_request():
"""Create a mock MCP request with valid headers"""
request = Mock(spec=Request)
request.headers = {
"EDGE_URL": "https://test-edge.local:8443",
"EDGE_API_CLIENT_ID": "test-client-id",
"EDGE_API_CLIENT_SECRET": "test-secret",
"VALIDATE_CERTIFICATE": "false",
}
return request
@pytest.fixture
def mock_request_missing_headers():
"""Create a mock request with missing required headers"""
request = Mock(spec=Request)
request.headers = {
"EDGE_URL": "https://test-edge.local:8443",
# Missing CLIENT_ID and SECRET
}
return request
@pytest.fixture
def mock_connection():
"""Create a mock Litmus Edge connection"""
return MagicMock()
# ==================== Test: get_litmusedge_driver_list ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.list_all_drivers")
def test_get_litmusedge_driver_list_success(
mock_list_drivers, mock_get_creds, mock_request
):
"""Test successfully retrieving driver list"""
# Setup mocks
mock_get_creds.return_value = MagicMock()
# Create mock driver objects
mock_driver1 = MagicMock()
mock_driver1.name = "ModbusTCP"
mock_driver2 = MagicMock()
mock_driver2.name = "OPCUA"
mock_list_drivers.return_value = [mock_driver1, mock_driver2]
# Execute
result = get_litmusedge_driver_list(mock_request)
# Verify
assert isinstance(result, list)
assert len(result) >= 2
mock_get_creds.assert_called_once_with(mock_request)
def test_get_litmusedge_driver_list_missing_headers(mock_request_missing_headers):
"""Test driver list with missing authentication headers"""
with pytest.raises(Exception): # Should raise McpError
get_litmusedge_driver_list(mock_request_missing_headers)
# ==================== Test: get_devicehub_devices ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.devices.list_devices")
def test_get_devicehub_devices_success(mock_list_devices, mock_get_creds, mock_request):
"""Test successfully retrieving devices"""
# Setup mocks
mock_get_creds.return_value = MagicMock()
mock_device = MagicMock()
mock_device.name = "TestDevice"
mock_device.__dict__ = {
"name": "TestDevice",
"driver": "ModbusTCP",
"enabled": True,
}
mock_list_devices.return_value = [mock_device]
# Execute
result = get_devicehub_devices(mock_request)
# Verify
assert isinstance(result, dict)
assert "TestDevice" in result
assert result["TestDevice"]["name"] == "TestDevice"
mock_get_creds.assert_called_once_with(mock_request)
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.devices.list_devices")
def test_get_devicehub_devices_empty(mock_list_devices, mock_get_creds, mock_request):
"""Test retrieving devices when none exist"""
mock_get_creds.return_value = MagicMock()
mock_list_devices.return_value = []
result = get_devicehub_devices(mock_request)
assert isinstance(result, dict)
assert len(result) == 0
# ==================== Test: get_devicehub_device_tags ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.devices.list_devices")
@patch("server.tags.list_registers_from_single_device")
def test_get_devicehub_device_tags_success(
mock_list_tags, mock_list_devices, mock_get_creds, mock_request
):
"""Test successfully retrieving device tags"""
# Setup mocks
mock_get_creds.return_value = MagicMock()
mock_device = MagicMock()
mock_device.name = "TestDevice"
mock_list_devices.return_value = [mock_device]
mock_tag = MagicMock()
mock_tag.tag_name = "Temperature"
mock_tag.__dict__ = {
"tag_name": "Temperature",
"address": "40001",
"data_type": "FLOAT",
}
mock_list_tags.return_value = [mock_tag]
# Execute
result = get_devicehub_device_tags(mock_request, "TestDevice")
# Verify
assert isinstance(result, dict)
assert "Temperature" in result
assert result["Temperature"]["tag_name"] == "Temperature"
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.devices.list_devices")
def test_get_devicehub_device_tags_device_not_found(
mock_list_devices, mock_get_creds, mock_request
):
"""Test error when device doesn't exist"""
mock_get_creds.return_value = MagicMock()
mock_list_devices.return_value = []
with pytest.raises(Exception) as exc_info:
get_devicehub_device_tags(mock_request, "NonExistentDevice")
assert "not found" in str(exc_info.value).lower()
# ==================== Test: get_current_value_of_devicehub_tag ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.devices.list_devices")
@patch("server.tags.list_registers_from_single_device")
@patch("server.asyncio.run")
def test_get_current_value_of_devicehub_tag_success(
mock_asyncio_run, mock_list_tags, mock_list_devices, mock_get_creds, mock_request
):
"""Test successfully reading tag value"""
# Setup mocks
mock_get_creds.return_value = MagicMock()
mock_device = MagicMock()
mock_device.name = "TestDevice"
mock_list_devices.return_value = [mock_device]
mock_topic = MagicMock()
mock_topic.direction = "Output"
mock_topic.topic = "test/topic/output"
mock_tag = MagicMock()
mock_tag.tag_name = "Temperature"
mock_tag.topics = [mock_topic]
mock_list_tags.return_value = [mock_tag]
mock_asyncio_run.return_value = {
"value": 25.5,
"timestamp": 1234567890,
"quality": "good",
}
# Execute
result = get_current_value_of_devicehub_tag(
mock_request, "TestDevice", tag_name="Temperature"
)
# Verify
assert isinstance(result, dict)
assert result["value"] == 25.5
assert "timestamp" in result
@patch("server._get_litmus_creds_from_mcp_client_headers")
def test_get_current_value_of_devicehub_tag_missing_params(
mock_get_creds, mock_request
):
"""Test error when neither tag_name nor tag_id provided"""
mock_get_creds.return_value = MagicMock()
with pytest.raises(Exception) as exc_info:
get_current_value_of_devicehub_tag(mock_request, "TestDevice")
assert "tag_name or tag_id is required" in str(exc_info.value).lower()
# ==================== Test: create_devicehub_device ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.list_all_drivers")
@patch("server.devices.create_device")
def test_create_devicehub_device_success(
mock_create_device, mock_list_drivers, mock_get_creds, mock_request
):
"""Test successfully creating a device"""
# Setup mocks
mock_get_creds.return_value = MagicMock()
mock_driver = MagicMock()
mock_driver.name = "ModbusTCP"
mock_driver.id = "driver-123"
mock_driver.get_default_properties.return_value = {"ip": "192.168.1.1", "port": 502}
mock_list_drivers.return_value = [mock_driver]
mock_created = MagicMock()
mock_created.__dict__ = {"id": "device-456", "name": "NewDevice"}
mock_create_device.return_value = mock_created
# Execute
result = create_devicehub_device(mock_request, "NewDevice", "ModbusTCP")
# Verify
assert isinstance(result, dict)
assert result["name"] == "NewDevice"
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.list_all_drivers")
def test_create_devicehub_device_invalid_driver(
mock_list_drivers, mock_get_creds, mock_request
):
"""Test error with invalid driver name"""
mock_get_creds.return_value = MagicMock()
mock_driver = MagicMock()
mock_driver.name = "ModbusTCP"
mock_list_drivers.return_value = [mock_driver]
with pytest.raises(Exception) as exc_info:
create_devicehub_device(mock_request, "NewDevice", "InvalidDriver")
assert "not found" in str(exc_info.value).lower()
# ==================== Test: get_litmusedge_friendly_name ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.network.get_friendly_name")
def test_get_litmusedge_friendly_name_success(
mock_get_name, mock_get_creds, mock_request
):
"""Test successfully getting device friendly name"""
mock_get_creds.return_value = MagicMock()
mock_get_name.return_value = "Factory_Gateway_01"
result = get_litmusedge_friendly_name(mock_request)
assert result == "Factory_Gateway_01"
assert isinstance(result, str)
# ==================== Test: set_litmusedge_friendly_name ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.network.set_friendly_name")
def test_set_litmusedge_friendly_name_success(
mock_set_name, mock_get_creds, mock_request
):
"""Test successfully setting device friendly name"""
mock_get_creds.return_value = MagicMock()
mock_set_name.return_value = None
result = set_litmusedge_friendly_name(mock_request, "New_Gateway_Name")
assert "updated" in result.lower()
assert "New_Gateway_Name" in result
mock_set_name.assert_called_once()
# ==================== Test: get_cloud_activation_status ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.device_management.show_cloud_registration_status")
def test_get_cloud_activation_status_success(
mock_get_status, mock_get_creds, mock_request
):
"""Test successfully getting cloud activation status"""
mock_get_creds.return_value = MagicMock()
mock_get_status.return_value = {
"status": "activated",
"connected": True,
"last_sync": "2025-01-15T10:30:00Z",
}
result = get_cloud_activation_status(mock_request)
assert isinstance(result, dict)
assert result["status"] == "activated"
assert result["connected"] is True
# ==================== Test: get_all_containers_on_litmusedge ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.list_all_containers")
def test_get_all_containers_success(mock_list_containers, mock_get_creds, mock_request):
"""Test successfully listing containers"""
mock_get_creds.return_value = MagicMock()
mock_list_containers.return_value = [
{"name": "node-red", "image": "nodered/node-red:latest", "status": "running"},
{"name": "influxdb", "image": "influxdb:2.0", "status": "running"},
]
result = get_all_containers_on_litmusedge(mock_request)
assert isinstance(result, list)
assert len(result) == 2
assert result[0]["name"] == "node-red"
# ==================== Test: run_docker_container_on_litmusedge ====================
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.run_container")
def test_run_docker_container_success(mock_run_container, mock_get_creds, mock_request):
"""Test successfully running a docker container"""
mock_get_creds.return_value = MagicMock()
mock_run_container.return_value = {"id": "container-abc123"}
result = run_docker_container_on_litmusedge(
mock_request, "docker run -d --name test-app nginx:latest"
)
assert isinstance(result, str)
assert "container-abc123" in result
@patch("server._get_litmus_creds_from_mcp_client_headers")
@patch("server.run_container")
def test_run_docker_container_no_id(mock_run_container, mock_get_creds, mock_request):
"""Test running container when no ID returned"""
mock_get_creds.return_value = MagicMock()
mock_run_container.return_value = {}
result = run_docker_container_on_litmusedge(mock_request, "docker run nginx")
assert "Unknown" in result or "container" in result.lower()
# ==================== Test: get_current_value_on_topic (async) ====================
@pytest.mark.asyncio
@patch("server.nc_single_topic")
async def test_get_current_value_on_topic_success(mock_nc_single):
"""Test successfully getting value from NATS topic"""
mock_nc_single.return_value = {
"value": 42.5,
"timestamp": 1234567890,
"quality": "good",
}
result = await get_current_value_on_topic("test/topic")
assert isinstance(result, dict)
assert result["value"] == 42.5
assert "timestamp" in result
@pytest.mark.asyncio
async def test_get_current_value_on_topic_with_custom_source():
"""Test getting value with custom NATS source"""
with patch("server.nc_single_topic") as mock_nc_single:
mock_nc_single.return_value = {"value": 100}
result = await get_current_value_on_topic(
"custom/topic", nats_source="192.168.1.100", nats_port="4223"
)
assert result["value"] == 100
# ==================== Test: get_multiple_values_from_topic (async) ====================
@pytest.mark.asyncio
@patch("server.collect_multiple_values_from_topic")
async def test_get_multiple_values_from_topic_success(mock_collect):
"""Test successfully collecting multiple values"""
mock_collect.return_value = {
"values": [10.0, 20.0, 30.0, 40.0, 50.0],
"humanTimestamps": [
"2025-01-15 10:00:00",
"2025-01-15 10:00:01",
"2025-01-15 10:00:02",
"2025-01-15 10:00:03",
"2025-01-15 10:00:04",
],
}
result = await get_multiple_values_from_topic("test/topic", num_samples=5)
assert isinstance(result, dict)
assert "values" in result
assert "humanTimestamps" in result
assert len(result["values"]) == 5
assert len(result["humanTimestamps"]) == 5
@pytest.mark.asyncio
async def test_get_multiple_values_from_topic_default_samples():
"""Test collecting values with default sample count"""
with patch("server.collect_multiple_values_from_topic") as mock_collect:
mock_collect.return_value = {"values": [0] * 10, "humanTimestamps": [""] * 10}
result = await get_multiple_values_from_topic("test/topic")
# Default should be 10 samples
assert len(result["values"]) == 10
# ==================== Integration-Style Tests ====================
def test_full_authentication_flow_with_valid_headers(mock_request):
"""Test that valid headers allow tool execution"""
with patch("server._get_litmus_creds_from_mcp_client_headers") as mock_get_creds:
with patch("server.list_all_drivers") as mock_list_drivers:
mock_get_creds.return_value = MagicMock()
mock_driver = MagicMock()
mock_driver.name = "Test"
mock_list_drivers.return_value = [mock_driver]
# Should not raise any exception
result = get_litmusedge_driver_list(mock_request)
assert result is not None
def test_full_authentication_flow_with_missing_headers(mock_request_missing_headers):
"""Test that missing headers prevent tool execution"""
with pytest.raises(Exception):
get_litmusedge_driver_list(mock_request_missing_headers)
# ==================== Run Tests ====================
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])