Laravel MCP Companion

"""Tests for cache management functionality.""" import threading from pathlib import Path from unittest.mock import patch, mock_open from mcp_tools import get_file_content_cached, search_laravel_docs_impl, clear_caches class TestCacheManagement: """Test cases for cache size limits and LRU eviction.""" def setup_method(self): """Clear caches before each test.""" clear_caches() def teardown_method(self): """Clear caches after each test.""" clear_caches() def test_file_cache_size_limit(self): """Test that file cache respects 100-entry limit.""" # Mock file reading with patch('builtins.open', mock_open(read_data='content')): # Fill cache with 100 entries for i in range(100): content = get_file_content_cached(f'/path/file{i}.txt') assert content == 'content' # Access cache to verify it has 100 entries from mcp_tools import _file_content_cache assert len(_file_content_cache) == 100 # Add one more entry - should trigger cleanup get_file_content_cached('/path/file100.txt') # Cache should now have 81 entries (100 - 20 + 1) assert len(_file_content_cache) == 81 # First 20 files should be removed for i in range(20): assert f'/path/file{i}.txt' not in _file_content_cache # Files 20-99 should still be in cache for i in range(20, 100): assert f'/path/file{i}.txt' in _file_content_cache # The new file should be in cache assert '/path/file100.txt' in _file_content_cache def test_file_cache_lru_eviction_order(self): """Test that oldest entries are removed first (FIFO behavior).""" with patch('builtins.open', mock_open(read_data='content')): # Fill cache with entries in a specific order for i in range(100): get_file_content_cached(f'/path/ordered{i}.txt') # Trigger eviction get_file_content_cached('/path/trigger.txt') from mcp_tools import _file_content_cache # Files 0-19 should be evicted (oldest) for i in range(20): assert f'/path/ordered{i}.txt' not in _file_content_cache # Files 20-99 should remain for i in range(20, 100): assert f'/path/ordered{i}.txt' in _file_content_cache def test_search_cache_size_limit(self): """Test that search cache respects 100-entry limit.""" # Mock the actual search implementation with patch('mcp_tools.get_laravel_docs_metadata') as mock_metadata: mock_metadata.return_value = {'11.x': {'version': '11.x', 'files': []}} with patch('mcp_tools.get_file_content_cached') as mock_file_content: mock_file_content.return_value = "No match here" # Fill search cache with 100 entries for i in range(100): result = search_laravel_docs_impl(Path("/fake/docs"), f"query{i}", version="11.x") assert "No results found" in result # Check cache size from mcp_tools import _search_result_cache assert len(_search_result_cache) == 100 # Add one more - should trigger cleanup search_laravel_docs_impl(Path("/fake/docs"), "query100", version="11.x") # Should have 81 entries now assert len(_search_result_cache) == 81 # First 20 should be removed for i in range(20): cache_key = f"search:query{i}:11.x:True" assert cache_key not in _search_result_cache def test_concurrent_cache_access(self): """Test thread-safe cache operations.""" results = [] errors = [] def cache_writer(thread_id): """Write to cache from multiple threads.""" try: with patch('builtins.open', mock_open(read_data=f'content{thread_id}')): for i in range(10): content = get_file_content_cached(f'/path/thread{thread_id}_file{i}.txt') results.append((thread_id, i, content)) except Exception as e: errors.append((thread_id, str(e))) # Create multiple threads threads = [] for i in range(5): thread = threading.Thread(target=cache_writer, args=(i,)) threads.append(thread) thread.start() # Wait for all threads for thread in threads: thread.join() # Check no errors occurred assert len(errors) == 0 # Check all operations completed assert len(results) == 50 # 5 threads * 10 files each # Verify cache integrity from mcp_tools import _file_content_cache # Should have 50 entries (under the 100 limit) assert len(_file_content_cache) == 50 def test_cache_eviction_under_concurrent_load(self): """Test cache eviction works correctly with concurrent access.""" completed_operations = [] def fill_cache(start_idx): """Fill cache to trigger eviction.""" try: with patch('builtins.open', mock_open(read_data='content')): for i in range(start_idx, start_idx + 30): get_file_content_cached(f'/path/concurrent{i}.txt') completed_operations.append(i) except Exception: pass # Start 4 threads that will collectively add 120 entries threads = [] for i in range(4): thread = threading.Thread(target=fill_cache, args=(i * 30,)) threads.append(thread) thread.start() # Wait for completion for thread in threads: thread.join() # Verify operations completed assert len(completed_operations) == 120 # Check cache size is within limits from mcp_tools import _file_content_cache assert len(_file_content_cache) <= 100 # Should have evicted some entries assert len(_file_content_cache) >= 80 # At least 80 after eviction def test_cache_hit_behavior(self): """Test that cache hits don't affect eviction order.""" with patch('builtins.open', mock_open(read_data='content')) as mock_file: # Add first file content1 = get_file_content_cached('/path/file1.txt') assert mock_file.call_count == 1 # Access same file again - should be cache hit content2 = get_file_content_cached('/path/file1.txt') assert content1 == content2 assert mock_file.call_count == 1 # No new file read # Fill cache to near limit for i in range(2, 101): get_file_content_cached(f'/path/file{i}.txt') # Access first file again before eviction content3 = get_file_content_cached('/path/file1.txt') assert content3 == 'content' # Trigger eviction get_file_content_cached('/path/file101.txt') # The cache eviction removes the first 20 keys from the dict # Since Python 3.7+, dicts maintain insertion order from mcp_tools import _file_content_cache # Files should be evicted in insertion order: file1.txt through file20.txt for i in range(1, 21): assert f'/path/file{i}.txt' not in _file_content_cache, f"file{i}.txt should have been evicted" # Files 21-100 and 101 should still be in cache for i in range(21, 101): assert f'/path/file{i}.txt' in _file_content_cache, f"file{i}.txt should still be in cache" assert '/path/file101.txt' in _file_content_cache def test_cache_with_file_not_found(self): """Test cache behavior with file not found errors.""" # First call should attempt to read and get FileNotFoundError with patch('builtins.open', side_effect=FileNotFoundError): content1 = get_file_content_cached('/path/missing.txt') assert content1 == "File not found: /path/missing.txt" # Check that error response is NOT cached from mcp_tools import _file_content_cache assert '/path/missing.txt' not in _file_content_cache # Second call should also attempt to read with patch('builtins.open', mock_open(read_data='now exists')): content2 = get_file_content_cached('/path/missing.txt') assert content2 == 'now exists' # Now it should be cached assert '/path/missing.txt' in _file_content_cache def test_cache_with_encoding_error(self): """Test cache behavior with encoding errors.""" # Simulate encoding error with patch('builtins.open', side_effect=UnicodeDecodeError('utf-8', b'', 0, 1, 'invalid')): content = get_file_content_cached('/path/bad_encoding.txt') assert "Error reading file" in content # Should not be cached from mcp_tools import _file_content_cache assert '/path/bad_encoding.txt' not in _file_content_cache def test_search_cache_key_format(self): """Test search cache key generation.""" with patch('mcp_tools.get_laravel_docs_metadata') as mock_metadata: mock_metadata.return_value = {} # Test different parameter combinations search_laravel_docs_impl(Path("/fake/docs"), "test", version="11.x", include_external=True) search_laravel_docs_impl(Path("/fake/docs"), "test", version="11.x", include_external=False) search_laravel_docs_impl(Path("/fake/docs"), "test", version=None, include_external=True) from mcp_tools import _search_result_cache # Should have 3 different cache entries assert len(_search_result_cache) == 3 assert "search:test:11.x:True" in _search_result_cache assert "search:test:11.x:False" in _search_result_cache assert "search:test:None:True" in _search_result_cache def test_clear_caches_function(self): """Test that clear_caches properly empties both caches.""" with patch('builtins.open', mock_open(read_data='content')): # Add entries to both caches get_file_content_cached('/path/file.txt') with patch('mcp_tools.get_laravel_docs_metadata') as mock_metadata: mock_metadata.return_value = {} search_laravel_docs_impl(Path("/fake/docs"), "test") from mcp_tools import _file_content_cache, _search_result_cache assert len(_file_content_cache) > 0 assert len(_search_result_cache) > 0 # Clear caches clear_caches() # Both should be empty assert len(_file_content_cache) == 0 assert len(_search_result_cache) == 0