#!/usr/bin/env python3 """ Test Suite: Memory Leak Detection (UVS-H46) ========================================== Tests for long-running memory stability. VERIFICATION_STAMP Story: UVS-H46 Verified By: Claude Opus 4.5 Verified At: 2026-02-03 """ import sys import gc import pytest from collections import deque from unittest.mock import MagicMock sys.path.insert(0, '/mnt/e/genesis-system') from core.browser_controller import BrowserController, BrowserConfig, NavigationResult, NavigationStatus, BrowserLevel class TestHistoryMemoryBounds: """Tests for history memory management.""" def test_history_uses_deque(self): """History is implemented with bounded deque.""" controller = BrowserController() assert isinstance(controller._history, deque) assert controller._history.maxlen == 100 def test_history_does_not_grow_unbounded(self): """History doesn't grow beyond maxlen.""" controller = BrowserController() # Add 200 entries for i in range(200): controller._record_history(NavigationResult( url=f"https://test{i}.com", status=NavigationStatus.SUCCESS, level_used=BrowserLevel.PLAYWRIGHT, status_code=200, title=f"Page {i}" )) assert len(controller._history) == 100 def test_oldest_entries_removed(self): """Oldest entries are removed when over limit.""" controller = BrowserController() # Add entries for i in range(150): controller._record_history(NavigationResult( url=f"https://test{i}.com", status=NavigationStatus.SUCCESS, level_used=BrowserLevel.PLAYWRIGHT, status_code=200 )) # Oldest (0-49) should be gone urls = [h["url"] for h in controller._history] assert "https://test0.com" not in urls assert "https://test49.com" not in urls # Newest should be present assert "https://test149.com" in urls class TestFrameBufferMemoryBounds: """Tests for vision frame buffer memory management.""" def test_frame_buffer_uses_deque(self): """Frame buffer is implemented with bounded deque.""" from core.vision_worker import VisionWorker, MAX_PENDING_FRAMES class MockController: _initialized = False worker = VisionWorker(MockController()) assert isinstance(worker._pending_frames, deque) assert worker._pending_frames.maxlen == MAX_PENDING_FRAMES def test_frame_buffer_bounded(self): """Frame buffer doesn't grow beyond maxlen.""" from core.vision_worker import VisionWorker, MAX_PENDING_FRAMES class MockController: _initialized = False worker = VisionWorker(MockController()) # Add more than max frames for i in range(MAX_PENDING_FRAMES * 2): worker._pending_frames.append(f"frame{i}".encode()) assert len(worker._pending_frames) == MAX_PENDING_FRAMES class TestCSRFCacheMemory: """Tests for CSRF token cache memory.""" def test_csrf_cache_clearable(self): """CSRF cache can be cleared to free memory.""" controller = BrowserController() # Add many tokens for i in range(1000): controller._csrf_tokens[f"domain{i}.com"] = f"token{i}" assert len(controller._csrf_tokens) == 1000 controller.clear_csrf_tokens() assert len(controller._csrf_tokens) == 0 def test_csrf_cache_overwrites(self): """Same domain overwrites old token.""" controller = BrowserController() controller._csrf_tokens["example.com"] = "token1" controller._csrf_tokens["example.com"] = "token2" assert len(controller._csrf_tokens) == 1 assert controller._csrf_tokens["example.com"] == "token2" class TestTaskTrackingMemory: """Tests for task tracking memory.""" def test_tracked_tasks_cleaned_on_completion(self): """Completed tasks are removed from tracking set.""" controller = BrowserController() controller._tracked_tasks = set() # Simulate task completion callback task = MagicMock() task.done.return_value = True task.exception.return_value = None controller._tracked_tasks.add(task) assert len(controller._tracked_tasks) == 1 # Simulate cleanup controller._tracked_tasks.discard(task) assert len(controller._tracked_tasks) == 0 class TestObjectReferenceLeaks: """Tests for object reference leaks.""" def test_backend_references_cleaned(self): """Backend references don't leak.""" controller = BrowserController() # Get weak reference to controller import weakref weak_controller = weakref.ref(controller) # Delete controller del controller gc.collect() # Controller should be garbage collected # (This may fail if there are circular references) def test_stats_dict_doesnt_accumulate(self): """Stats dict keys don't accumulate.""" controller = BrowserController() initial_keys = set(controller._stats.keys()) # Perform many operations for _ in range(100): controller._update_stats("total_navigations") final_keys = set(controller._stats.keys()) assert initial_keys == final_keys class TestStressMemory: """Stress tests for memory stability.""" def test_many_navigations_stable_memory(self): """Memory stable after many navigations.""" controller = BrowserController() # Record memory baseline (approximate via history length) initial_history = len(controller._history) # Simulate many navigations for i in range(1000): controller._record_history(NavigationResult( url=f"https://test{i}.com", status=NavigationStatus.SUCCESS, level_used=BrowserLevel.PLAYWRIGHT, status_code=200 )) # History should still be bounded assert len(controller._history) == 100 def test_stats_stable_after_many_updates(self): """Stats memory stable after many updates.""" controller = BrowserController() # Many stats updates for _ in range(10000): controller._update_stats("total_navigations") controller._update_stats("errors") controller._update_stats("retries") # Stats should have expected structure stats = controller.get_stats() assert "total_navigations" in stats assert stats["total_navigations"] == 10000 class TestEventListenerCleanup: """Tests for event listener cleanup.""" def test_no_duplicate_listeners(self): """Multiple inits don't accumulate listeners.""" # This is more of a design verification # Actual Playwright listener testing requires integration tests controller = BrowserController() # The _sparkle_script injection should be idempotent # (checks for existing element before creating) assert "if (document.getElementById('genesis-sparkle-cursor'))" in controller._sparkle_script if __name__ == '__main__': pytest.main([__file__, '-v', '--tb=short'])