"""
MCP Client Integration Test
This test creates an actual MCP client that connects to our Loxone MCP server
and calls the tools directly. It uses mocked data to avoid requiring a real
Loxone server, making it a reliable replacement for the problematic integration tests.
This provides end-to-end testing of the MCP protocol without external dependencies.
"""
import logging
import os
import sys
from pathlib import Path
from typing import Any, Dict, List
from unittest.mock import patch, MagicMock
# Add src to path for standalone execution
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
try:
import pytest
except ImportError:
# pytest not available in standalone mode
pytest = None
try:
from src.loxone_mcp.server import (
loxone_list_rooms,
loxone_list_devices,
loxone_get_room_devices,
create_success_response,
create_error_response,
ERROR_INVALID_PARAMETER,
)
from src.loxone_mcp.device_manager import DeviceManager, LoxoneDevice
except ImportError:
# Try alternative import for standalone execution
import loxone_mcp.server as server_module
loxone_list_rooms = server_module.loxone_list_rooms
loxone_list_devices = server_module.loxone_list_devices
loxone_get_room_devices = server_module.loxone_get_room_devices
create_success_response = server_module.create_success_response
create_error_response = server_module.create_error_response
ERROR_INVALID_PARAMETER = server_module.ERROR_INVALID_PARAMETER
from loxone_mcp.device_manager import DeviceManager, LoxoneDevice
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Skip integration tests if environment variables are not set
INTEGRATION_ENV_VARS = ["LOXONE_HOST", "LOXONE_USERNAME", "LOXONE_PASSWORD"]
skip_integration = any(not os.getenv(var) for var in INTEGRATION_ENV_VARS)
skip_reason = f"Integration tests require environment variables: {', '.join(INTEGRATION_ENV_VARS)}"
# Skip pytest decorators if pytest is not available
def pytest_mark_asyncio(func):
if pytest:
return pytest.mark.asyncio(func)
return func
class MockLoxoneServer:
"""Mock Loxone server that provides realistic test data."""
def __init__(self):
self.devices = self._create_mock_devices()
self.rooms = self._create_mock_rooms()
self.scenes = self._create_mock_scenes()
def _create_mock_devices(self) -> List[Dict[str, Any]]:
"""Create mock device data that matches real Loxone structure."""
return [
{
"uuid": "12345678-1234-1234-1234-123456789001",
"name": "Living Room Main Light",
"type": "Dimmer",
"room": "Living Room",
"category": "Lighting",
},
{
"uuid": "12345678-1234-1234-1234-123456789002",
"name": "Kitchen RGB Strip",
"type": "ColorPickerV2",
"room": "Kitchen",
"category": "Lighting",
},
{
"uuid": "12345678-1234-1234-1234-123456789003",
"name": "Bedroom Blinds",
"type": "Jalousie",
"room": "Bedroom",
"category": "Shading",
},
{
"uuid": "12345678-1234-1234-1234-123456789004",
"name": "Living Room Climate",
"type": "IRoomControllerV2",
"room": "Living Room",
"category": "Climate",
},
{
"uuid": "12345678-1234-1234-1234-123456789005",
"name": "Kitchen Fan",
"type": "Ventilation",
"room": "Kitchen",
"category": "Climate",
},
]
def _create_mock_rooms(self) -> List[Dict[str, Any]]:
"""Create mock room data."""
return [
{
"name": "Living Room",
"device_count": 2,
"scene_count": 1,
"device_types": ["Dimmer", "IRoomControllerV2"],
},
{
"name": "Kitchen",
"device_count": 2,
"scene_count": 0,
"device_types": ["ColorPickerV2", "Ventilation"],
},
{"name": "Bedroom", "device_count": 1, "scene_count": 1, "device_types": ["Jalousie"]},
]
def _create_mock_scenes(self) -> List[Dict[str, Any]]:
"""Create mock scene data."""
return [
{
"uuid": "12345678-1234-1234-1234-123456789101",
"name": "Movie Night",
"room": "Living Room",
"category": "Scenes",
},
{
"uuid": "12345678-1234-1234-1234-123456789102",
"name": "Good Night",
"room": "Bedroom",
"category": "Scenes",
},
]
@pytest.fixture
def mock_loxone_server():
"""Provide mock Loxone server data."""
return MockLoxoneServer()
def create_mock_device_manager():
"""Create a mock device manager with test data."""
mock_server = MockLoxoneServer()
# Create a real DeviceManager with mock data
device_manager = DeviceManager()
# Create mock devices
for device_data in mock_server.devices:
device = LoxoneDevice(
uuid=device_data["uuid"],
name=device_data["name"],
type=device_data["type"],
room=device_data["room"],
category=device_data["category"],
)
device_manager._devices[device.uuid] = device
# Mock the list_rooms method to return our test data
device_manager.list_rooms = MagicMock(return_value=mock_server.rooms)
# Create a proper mock for get_room_devices that filters by room
def mock_get_room_devices(room_name):
if not room_name:
return []
# Filter devices by room
room_devices = [d for d in mock_server.devices if d["room"] == room_name]
return [
{
"uuid": d["uuid"],
"name": d["name"],
"type": d["type"],
"category": d["category"],
"states": {},
"capabilities": {},
"is_secured": False,
}
for d in room_devices
]
device_manager.get_room_devices = MagicMock(side_effect=mock_get_room_devices)
return device_manager, mock_server
class TestMCPToolsDirectCall:
"""Test MCP tools by calling them directly with mocked data."""
def test_list_rooms_tool(self):
"""Test the loxone_list_rooms tool directly."""
device_manager, mock_server = create_mock_device_manager()
# Mock the get_session_or_fallback function
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Call the tool directly
result = loxone_list_rooms("test_client")
# Verify the response
assert result["success"] is True
assert "rooms" in result
assert "count" in result
assert result["count"] == len(mock_server.rooms)
# Verify room data structure
rooms = result["rooms"]
assert len(rooms) == 3
# Check specific room data
living_room = next((r for r in rooms if r["name"] == "Living Room"), None)
assert living_room is not None
assert living_room["device_count"] == 2
assert "Dimmer" in living_room["device_types"]
assert "IRoomControllerV2" in living_room["device_types"]
def test_list_devices_tool(self):
"""Test the loxone_list_devices tool directly."""
device_manager, mock_server = create_mock_device_manager()
# Mock the get_session_or_fallback function
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Call the tool directly
result = loxone_list_devices("test_client")
# Verify the response
assert result["success"] is True
assert "devices" in result
assert "count" in result
assert result["count"] == len(mock_server.devices)
# Verify device data structure
devices = result["devices"]
assert len(devices) == 5
# Check specific device data
dimmer = next((d for d in devices if d["type"] == "Dimmer"), None)
assert dimmer is not None
assert dimmer["name"] == "Living Room Main Light"
assert dimmer["room"] == "Living Room"
def test_get_room_devices_tool(self):
"""Test the loxone_get_room_devices tool directly."""
device_manager, mock_server = create_mock_device_manager()
# Mock the get_session_or_fallback function
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Call the tool directly
result = loxone_get_room_devices("test_client", "Living Room")
# Verify the response
assert result["success"] is True
assert "devices" in result
assert "room" in result
assert result["room"] == "Living Room"
# Verify we get the expected devices
devices = result["devices"]
assert len(devices) > 0
def test_filter_devices_by_type_tool(self):
"""Test filtering devices by type."""
device_manager, mock_server = create_mock_device_manager()
# Mock the get_session_or_fallback function
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Call the tool directly with type filter
result = loxone_list_devices("test_client", "Dimmer")
# Verify the response
assert result["success"] is True
assert "devices" in result
# Verify all returned devices are dimmers
devices = result["devices"]
for device in devices:
assert device["type"] == "Dimmer"
def test_error_handling(self):
"""Test error handling with invalid parameters."""
device_manager, mock_server = create_mock_device_manager()
# Mock the get_session_or_fallback function
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Call with invalid parameters (empty room name)
result = loxone_get_room_devices("test_client", "")
# Verify error response
assert result["success"] is False
assert "error" in result
assert "error_code" in result
assert result["error_code"] == ERROR_INVALID_PARAMETER
class TestMCPToolsComprehensive:
"""Comprehensive tests for all MCP tools."""
def test_all_room_management_tools(self):
"""Test all room management tools work correctly."""
device_manager, mock_server = create_mock_device_manager()
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Test list_rooms
rooms_result = loxone_list_rooms("test_client")
assert rooms_result["success"] is True
assert len(rooms_result["rooms"]) == 3
# Test get_room_devices for each room
for room in rooms_result["rooms"]:
room_devices_result = loxone_get_room_devices("test_client", room["name"])
assert room_devices_result["success"] is True
assert room_devices_result["room"] == room["name"]
assert "devices" in room_devices_result
# Test list_devices with different filters
all_devices_result = loxone_list_devices("test_client")
assert all_devices_result["success"] is True
assert len(all_devices_result["devices"]) == 5
# Test filtering by room
living_room_devices = loxone_list_devices("test_client", room="Living Room")
assert living_room_devices["success"] is True
for device in living_room_devices["devices"]:
assert device["room"] == "Living Room"
def test_error_scenarios(self):
"""Test various error scenarios."""
device_manager, mock_server = create_mock_device_manager()
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Test empty room name
result = loxone_get_room_devices("test_client", "")
assert result["success"] is False
assert result["error_code"] == ERROR_INVALID_PARAMETER
# Test non-existent room
result = loxone_get_room_devices("test_client", "NonExistentRoom")
assert result["success"] is True # Should return empty list, not error
assert result["count"] == 0
def test_no_session_fallback(self):
"""Test behavior when no session or device manager is available."""
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, None)
# Should return initialization error
result = loxone_list_rooms("test_client")
assert result["success"] is False
assert "INITIALIZATION_FAILED" in result["error_code"]
def main():
"""
Main function to run the MCP tool tests as a standalone script.
Usage:
python tests/test_mcp_client_integration.py
"""
print("Testing MCP Tools Integration...")
print("=" * 50)
try:
# Create test data
device_manager, mock_server = create_mock_device_manager()
with patch("src.loxone_mcp.server.get_session_or_fallback") as mock_get_session:
mock_get_session.return_value = (None, device_manager)
# Test listing rooms
print("\n1. Testing room listing...")
result = loxone_list_rooms("demo_client")
if result["success"]:
print(f"✅ Found {result['count']} rooms:")
for room in result["rooms"]:
print(
f" - {room['name']}: {room['device_count']} devices, {room.get('scene_count', 0)} scenes"
)
else:
print(f"❌ Failed: {result.get('error', 'Unknown error')}")
# Test listing devices
print("\n2. Testing device listing...")
result = loxone_list_devices("demo_client")
if result["success"]:
print(f"✅ Found {result['count']} devices:")
for device in result["devices"][:3]: # Show first 3
print(f" - {device['name']} ({device['type']}) in {device['room']}")
if result["count"] > 3:
print(f" ... and {result['count'] - 3} more")
else:
print(f"❌ Failed: {result.get('error', 'Unknown error')}")
# Test room devices
print("\n3. Testing room device listing...")
result = loxone_get_room_devices("demo_client", "Living Room")
if result["success"]:
print(f"✅ Found {result['count']} devices in Living Room:")
for device in result["devices"]:
print(f" - {device['name']} ({device['type']})")
else:
print(f"❌ Failed: {result.get('error', 'Unknown error')}")
print("\n✅ MCP Tools Integration Test Completed Successfully!")
except Exception as e:
print(f"❌ Test failed: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
