"""
Unit tests for SmartThingsMCP client operations.
Tests the SmartThingsMCPClient and client-side caching functionality.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
from datetime import datetime, timedelta
class TestClientCaching:
"""Test client-side caching functionality."""
def test_cache_hit(self):
"""Test cache hit for repeated calls."""
cache = {}
cache_key = "list_devices_loc-1"
cached_value = {"items": [{"id": "d1", "name": "Device 1"}]}
# First call - cache miss
cache[cache_key] = {
"data": cached_value,
"timestamp": datetime.now()
}
# Check cache
assert cache_key in cache
assert cache[cache_key]["data"] == cached_value
def test_cache_miss(self):
"""Test cache miss for uncached calls."""
cache = {}
cache_key = "list_devices_loc-1"
assert cache_key not in cache
def test_cache_ttl_expiration(self):
"""Test that cached entries expire based on TTL."""
cache = {}
cache_key = "list_devices_loc-1"
ttl = 300 # 5 minutes
# Add entry with old timestamp
old_time = datetime.now() - timedelta(seconds=ttl + 1)
cache[cache_key] = {
"data": {"items": []},
"timestamp": old_time
}
# Check if expired
entry = cache.get(cache_key)
is_expired = (datetime.now() - entry["timestamp"]).total_seconds() > ttl
assert is_expired
def test_cache_invalidation_on_write(self):
"""Test that cache is cleared on write operations."""
cache = {
"list_devices_loc-1": {"data": {"items": []}},
"list_rooms_loc-1": {"data": {"items": []}}
}
# Simulate write operation - invalidate cache
cache.clear()
assert len(cache) == 0
def test_cache_statistics_tracking(self):
"""Test cache hit/miss statistics."""
stats = {
"hits": 0,
"misses": 0
}
# Simulate hits
stats["hits"] += 1
stats["hits"] += 1
stats["misses"] += 1
assert stats["hits"] == 2
assert stats["misses"] == 1
assert stats["hits"] / (stats["hits"] + stats["misses"]) == 2/3
class TestClientAuthentication:
"""Test client authentication."""
def test_bearer_token_header(self):
"""Test that Bearer token is properly set."""
token = "test-token-12345"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
assert headers["Authorization"] == f"Bearer {token}"
assert headers["Authorization"].startswith("Bearer ")
def test_invalid_token_handling(self):
"""Test handling of invalid tokens."""
invalid_token = None
assert invalid_token is None
def test_token_refresh(self):
"""Test token refresh mechanism."""
old_token = "old-token"
new_token = "new-token"
# Simulate token refresh
current_token = old_token
current_token = new_token
assert current_token == new_token
class TestClientDeviceOperations:
"""Test device-related client operations."""
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_list_devices_caching(self, mock_request):
"""Test that list_devices result is cached."""
mock_request.return_value = {
"items": [{"id": "d1", "name": "Device 1"}]
}
# First call should hit API
result1 = mock_request("token", "GET", "https://api.smartthings.com/v1/devices")
# Second call should use cache (in real implementation)
result2 = result1 # Simulating cached result
assert result1 == result2
assert result1["items"][0]["id"] == "d1"
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_get_device_status(self, mock_request):
"""Test getting device status."""
mock_request.return_value = {
"id": "d1",
"status": "online",
"components": []
}
result = mock_request("token", "GET",
"https://api.smartthings.com/v1/devices/d1/status")
assert result["id"] == "d1"
assert result["status"] == "online"
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_execute_command_cache_invalidation(self, mock_request):
"""Test that execute_command invalidates relevant caches."""
mock_request.return_value = {"status": "ACCEPTED"}
# Simulate cache before command
cache = {
"list_devices_loc-1": {"data": {}},
"device_d1_status": {"data": {}}
}
# Execute command - should invalidate cache
result = mock_request("token", "POST",
"https://api.smartthings.com/v1/devices/d1/commands",
json={})
# In real implementation, cache would be cleared
cache.clear()
assert result["status"] == "ACCEPTED"
assert len(cache) == 0
class TestClientRoomOperations:
"""Test room-related client operations."""
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_list_rooms_caching(self, mock_request):
"""Test that list_rooms result is cached."""
mock_request.return_value = {
"items": [
{"id": "room-1", "name": "Living Room"},
{"id": "room-2", "name": "Bedroom"}
]
}
result = mock_request("token", "GET",
"https://api.smartthings.com/v1/rooms?locationId=loc-1")
assert len(result["items"]) == 2
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_get_devices_in_room(self, mock_request):
"""Test getting devices in a specific room."""
mock_request.return_value = {
"items": [
{"id": "d1", "name": "Light", "roomId": "room-1"},
{"id": "d2", "name": "Switch", "roomId": "room-1"}
]
}
result = mock_request("token", "GET",
"https://api.smartthings.com/v1/devices?roomId=room-1")
assert all(d["roomId"] == "room-1" for d in result["items"])
class TestClientRuleOperations:
"""Test rule-related client operations."""
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_list_rules_caching(self, mock_request):
"""Test that list_rules result is cached."""
mock_request.return_value = {
"items": [
{"id": "rule-1", "name": "Evening Lights"},
{"id": "rule-2", "name": "Morning Coffee"}
]
}
result = mock_request("token", "GET",
"https://api.smartthings.com/v1/rules?locationId=loc-1")
assert len(result["items"]) == 2
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_create_rule_cache_invalidation(self, mock_request):
"""Test that creating a rule invalidates list_rules cache."""
mock_request.return_value = {
"id": "rule-new",
"name": "New Rule"
}
result = mock_request("token", "POST",
"https://api.smartthings.com/v1/rules",
json={"name": "New Rule"})
assert result["id"] == "rule-new"
class TestClientLocationOperations:
"""Test location-related client operations."""
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_list_locations_caching(self, mock_request):
"""Test that list_locations result is cached."""
mock_request.return_value = {
"items": [
{"id": "loc-1", "name": "Home"},
{"id": "loc-2", "name": "Cabin"}
]
}
result = mock_request("token", "GET",
"https://api.smartthings.com/v1/locations")
assert len(result["items"]) == 2
@patch('SmartThingsMCP.modules.server.common.make_request')
def test_get_location_details(self, mock_request):
"""Test getting location details."""
mock_request.return_value = {
"id": "loc-1",
"name": "Home",
"countryCode": "US",
"timeZoneId": "America/New_York"
}
result = mock_request("token", "GET",
"https://api.smartthings.com/v1/locations/loc-1")
assert result["id"] == "loc-1"
class TestClientErrorHandling:
"""Test client error handling."""
def test_handle_401_unauthorized(self):
"""Test handling of 401 Unauthorized response."""
response = {
"status_code": 401,
"error": "Unauthorized"
}
assert response["status_code"] == 401
def test_handle_404_not_found(self):
"""Test handling of 404 Not Found response."""
response = {
"status_code": 404,
"error": "Not Found"
}
assert response["status_code"] == 404
def test_handle_500_server_error(self):
"""Test handling of 500 Server Error response."""
response = {
"status_code": 500,
"error": "Internal Server Error"
}
assert response["status_code"] == 500
def test_handle_network_error(self):
"""Test handling of network errors."""
error = None
try:
# Simulate connection error
raise ConnectionError("Failed to connect to server")
except ConnectionError as e:
error = e
assert error is not None
assert "Failed to connect" in str(error)
class TestClientPerformance:
"""Test client performance aspects."""
def test_cache_reduces_api_calls(self):
"""Test that caching reduces number of API calls."""
api_call_count = 0
def mock_api_call():
nonlocal api_call_count
api_call_count += 1
return {"data": "result"}
cache = {}
# First call - not cached
result1 = mock_api_call()
cache["key"] = result1
# Second call - use cache instead of API
result2 = cache.get("key")
# Should only have 1 API call
assert api_call_count == 1
assert result1 == result2
def test_lru_cache_eviction(self):
"""Test LRU (Least Recently Used) cache eviction."""
max_size = 3
cache = {}
# Add items to cache
for i in range(max_size + 1):
if len(cache) >= max_size:
# Remove least recently used (first item)
oldest_key = list(cache.keys())[0]
del cache[oldest_key]
cache[f"key-{i}"] = f"value-{i}"
# Should only contain max_size items
assert len(cache) == max_size
