#!/usr/bin/env python3 """ TITAN MEMORY WHITE-BOX TEST SUITE ================================= Tests internal implementation paths and branches. Run: python -m pytest tests/test_titan_whitebox.py -v """ import os import sys import time import unittest from pathlib import Path from unittest.mock import patch, MagicMock, PropertyMock from datetime import datetime, timedelta # Add project root sys.path.insert(0, str(Path(__file__).parent.parent)) # Set API key for tests os.environ.setdefault('GEMINI_API_KEY', 'AIzaSyCT_rx0NusUJWoqtT7uxHAKEfHo129SJb8') class TestFileScannerWhiteBox(unittest.TestCase): """White-box tests for FileScanner internals.""" def test_scan_directory_handles_missing_dir(self): """T1: _scan_directory handles non-existent directory.""" from core.titan.file_scanner import FileScanner scanner = FileScanner(directories=['/nonexistent/path/xyz123']) files = scanner.scan() # Should return empty list, not crash self.assertEqual(files, []) def test_scan_directory_excludes_pycache(self): """T1b: _scan_directory excludes __pycache__ directories.""" from core.titan.file_scanner import FileScanner scanner = FileScanner() files = scanner.scan() for f in files: self.assertNotIn('__pycache__', str(f), f"Found pycache file: {f}") def test_scan_directory_excludes_patterns(self): """T1c: _scan_directory respects exclude patterns.""" from core.titan.file_scanner import FileScanner scanner = FileScanner(exclude_patterns=['*test*']) files = scanner.scan() for f in files: self.assertNotIn('test', f.name.lower(), f"Test file not excluded: {f}") def test_scan_handles_symlinks(self): """T1d: Scanner handles symbolic links gracefully.""" from core.titan.file_scanner import FileScanner # Create a symlink for testing test_dir = Path("/tmp/titan_symlink_test") test_dir.mkdir(exist_ok=True) test_file = test_dir / "real.py" test_file.write_text("# test") symlink = test_dir / "link.py" try: if not symlink.exists(): symlink.symlink_to(test_file) scanner = FileScanner(directories=[str(test_dir)]) scanner.follow_symlinks = False files = scanner.scan() # Should handle gracefully (either include or exclude, but not crash) self.assertIsInstance(files, list) finally: if symlink.exists(): symlink.unlink() if test_file.exists(): test_file.unlink() if test_dir.exists(): test_dir.rmdir() class TestCacheCreatorWhiteBox(unittest.TestCase): """White-box tests for CacheCreator internals.""" def test_batch_upload_respects_size(self): """T2: _batch_upload respects batch size limit.""" from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() # Create 100 dummy file paths files = [Path(f"/tmp/test_{i}.py") for i in range(100)] # Check that batching logic would split correctly batch_size = manager.BATCH_SIZE expected_batches = (len(files) + batch_size - 1) // batch_size batches = list(manager._create_batches(files)) self.assertEqual(len(batches), expected_batches) for batch in batches[:-1]: # All but last batch self.assertEqual(len(batch), batch_size) def test_wait_for_processing_timeout(self): """T3: _wait_for_processing has timeout.""" from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() # Mock a file that never finishes processing mock_file = MagicMock() mock_file.state.name = "PROCESSING" with patch.object(manager, '_get_file_state', return_value="PROCESSING"): start = time.time() result = manager._wait_for_file(mock_file, timeout=2) elapsed = time.time() - start # Should timeout within reasonable time self.assertLess(elapsed, 5, "Should timeout, not hang forever") def test_retry_logic_on_failure(self): """T3b: Retry logic activates on transient failures.""" from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() call_count = [0] def failing_upload(*args, **kwargs): call_count[0] += 1 if call_count[0] < 3: raise Exception("Transient error") return MagicMock(name="uploaded_file") with patch.object(manager, '_upload_single_file', side_effect=failing_upload): # Should retry and eventually succeed try: manager._upload_with_retry(Path("/tmp/test.py"), max_retries=3) except: pass self.assertGreaterEqual(call_count[0], 2, "Should have retried") class TestRefreshDaemonWhiteBox(unittest.TestCase): """White-box tests for RefreshDaemon internals.""" def test_calculate_refresh_time_correct(self): """T4: _calculate_refresh_time returns correct time.""" from core.titan.refresh_daemon import TitanRefreshDaemon from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() daemon = TitanRefreshDaemon(manager) # Test with 60 minute TTL, should refresh at 50 min ttl_minutes = 60 refresh_minutes = daemon._calculate_refresh_time(ttl_minutes) # Should be 10 minutes before expiry self.assertEqual(refresh_minutes, 50) def test_state_transitions_during_refresh(self): """T5: State transitions correctly during refresh cycle.""" from core.titan.refresh_daemon import TitanRefreshDaemon, DaemonState from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() daemon = TitanRefreshDaemon(manager) # Initial state self.assertEqual(daemon.state, DaemonState.IDLE) # Start daemon.start() time.sleep(0.1) self.assertEqual(daemon.state, DaemonState.RUNNING) # Stop daemon.stop() time.sleep(0.5) self.assertEqual(daemon.state, DaemonState.STOPPED) def test_refresh_failure_handling(self): """T5b: Daemon handles refresh failures correctly.""" from core.titan.refresh_daemon import TitanRefreshDaemon from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() daemon = TitanRefreshDaemon(manager, max_retries=3) # Track retry attempts daemon._failure_count = 0 # Simulate failed refresh daemon._on_refresh_failure("Test error") self.assertEqual(daemon._failure_count, 1) # After max failures, daemon should enter error state for _ in range(3): daemon._on_refresh_failure("Test error") self.assertGreaterEqual(daemon._failure_count, 3) class TestTitanBridgeWhiteBox(unittest.TestCase): """White-box tests for TitanBridge cache selection.""" def test_select_cache_picks_latest(self): """T6: _select_cache picks the most recent cache.""" from core.titan.cache_manager import TitanCacheManager, TitanMemory manager = TitanCacheManager() # Create mock caches with different times old_cache = TitanMemory( name="cachedContents/old", display_name="old_cache", model="gemini-2.0-flash-001", expire_time=(datetime.now() + timedelta(minutes=30)).isoformat(), token_count=1000, created_time=(datetime.now() - timedelta(hours=1)).isoformat() ) new_cache = TitanMemory( name="cachedContents/new", display_name="new_cache", model="gemini-2.0-flash-001", expire_time=(datetime.now() + timedelta(minutes=55)).isoformat(), token_count=1000, created_time=datetime.now().isoformat() ) # Mock list_caches to return both with patch.object(manager, 'list_caches', return_value=[old_cache, new_cache]): selected = manager._select_best_cache() # Should pick the newer one (more time until expiry) self.assertEqual(selected.name, "cachedContents/new") def test_fallback_creates_cache(self): """T6b: Fallback path creates new cache when none exists.""" from core.titan.cache_manager import TitanCacheManager manager = TitanCacheManager() with patch.object(manager, 'list_caches', return_value=[]): with patch.object(manager, 'create_cache') as mock_create: mock_create.return_value = MagicMock(name="new_cache") # This should trigger cache creation manager.ensure_cache_exists() mock_create.assert_called_once() class TestTokenCalculation(unittest.TestCase): """White-box tests for token calculation.""" def test_estimate_tokens_reasonable(self): """Token estimation is reasonable for Python files.""" from core.titan.file_scanner import FileScanner scanner = FileScanner() # Typical Python file ~100 lines, ~4 chars per token test_content = "def hello():\n print('world')\n" * 50 estimated = scanner._estimate_tokens(test_content) # Should be in reasonable range (not 0, not millions) self.assertGreater(estimated, 100) self.assertLess(estimated, 10000) if __name__ == '__main__': unittest.main(verbosity=2)