""" Tests unitaires pour StorageManager """ import pytest import tempfile import shutil from pathlib import Path import numpy as np from datetime import datetime from core.persistence import StorageManager # Mock classes pour les tests (en attendant l'implémentation complète) class MockRawSession: """Mock de RawSession pour les tests""" def __init__(self, session_id, started_at, events=None, screenshots=None): self.session_id = session_id self.started_at = started_at self.start_time = started_at # Alias pour compatibilité self.events = events or [] self.screenshots = screenshots or [] self.agent_version = "test_v1.0.0" self.environment = {"os": "test"} self.user = {"name": "test_user"} self.context = {"test": "context"} def to_json(self): return { "schema_version": "rawsession_v1", "session_id": self.session_id, "agent_version": self.agent_version, "environment": self.environment, "user": self.user, "context": self.context, "started_at": self.started_at, "start_time": self.started_at, "ended_at": None, "events": self.events, "screenshots": self.screenshots } @classmethod def from_json(cls, data): session = cls( session_id=data["session_id"], started_at=data.get("started_at", data.get("start_time")), events=data.get("events", []), screenshots=data.get("screenshots", []) ) session.agent_version = data.get("agent_version", "test_v1.0.0") session.environment = data.get("environment", {"os": "test"}) session.user = data.get("user", {"name": "test_user"}) session.context = data.get("context", {"test": "context"}) return session class MockObject: """Mock object pour accès par attributs""" def __init__(self, **kwargs): for key, value in kwargs.items(): if isinstance(value, dict): setattr(self, key, MockObject(**value)) else: setattr(self, key, value) class MockScreenState: """Mock de ScreenState pour les tests""" def __init__(self, state_id, timestamp, raw=None, perception=None, context=None, window=None, session_id="test_session"): self.state_id = state_id self.screen_state_id = state_id # Alias pour compatibilité avec le vrai modèle self.timestamp = timestamp self.session_id = session_id # Convertir dicts en objets pour accès par attributs self.raw = MockObject(**(raw or {})) self.perception = MockObject(**(perception or {})) self.context = MockObject(**(context or {})) self.window = MockObject(**(window or {"app_name": "test.exe", "window_title": "Test", "screen_resolution": [1920, 1080]})) def to_json(self): # Convertir les MockObjects en dicts def to_dict(obj): if isinstance(obj, MockObject): return {k: to_dict(v) for k, v in obj.__dict__.items()} return obj return { "schema_version": "screenstate_v1", "screen_state_id": self.state_id, "state_id": self.state_id, "session_id": self.session_id, "timestamp": self.timestamp, "window": to_dict(self.window), "raw": to_dict(self.raw), "perception": to_dict(self.perception), "context": to_dict(self.context) } @classmethod def from_json(cls, data): return cls( state_id=data.get("screen_state_id", data.get("state_id")), timestamp=data["timestamp"], session_id=data.get("session_id", "test_session"), window=data.get("window", {"app_name": "test.exe", "window_title": "Test", "screen_resolution": [1920, 1080]}), raw=data.get("raw", {}), perception=data.get("perception", {}), context=data.get("context", {}) ) @pytest.fixture def temp_storage(): """Crée un répertoire temporaire pour les tests.""" temp_dir = tempfile.mkdtemp() storage = StorageManager(base_path=temp_dir) yield storage # Cleanup shutil.rmtree(temp_dir) @pytest.fixture def sample_raw_session(): """Crée une RawSession de test.""" # Créer une session minimale avec to_json/from_json session = MockRawSession( session_id="test_session_001", started_at=datetime.now().isoformat(), events=[], screenshots=[] ) return session @pytest.fixture def sample_screen_state(): """Crée un ScreenState de test.""" # Créer un state minimal avec to_json/from_json state = MockScreenState( state_id="test_state_001", timestamp=datetime.now().isoformat(), window={ "app_name": "test.exe", "window_title": "Test", "screen_resolution": [1920, 1080], "workspace": "main" }, raw={ "screenshot_path": "test.png", "capture_method": "mss", "file_size_bytes": 1024 }, perception={ "embedding": { "provider": "openclip_ViT-B-32", "vector_id": "test_vector_001", "dimensions": 512 }, "detected_text": ["Hello", "World"], "text_detection_method": "qwen_vl", "confidence_avg": 0.95 }, context={ "current_workflow_candidate": None, "workflow_step": None, "user_id": "test_user", "tags": [], "business_variables": {} } ) return state class TestStorageManagerBasics: """Tests de base du StorageManager.""" def test_initialization(self, temp_storage): """Test que le StorageManager initialise correctement les répertoires.""" assert temp_storage.base_path.exists() assert (temp_storage.base_path / "sessions").exists() assert (temp_storage.base_path / "screen_states").exists() assert (temp_storage.base_path / "embeddings").exists() assert (temp_storage.base_path / "faiss_index").exists() assert (temp_storage.base_path / "workflows").exists() def test_get_date_path(self, temp_storage): """Test que les chemins de date sont créés correctement.""" date_path = temp_storage._get_date_path("sessions") today = datetime.now().strftime("%Y-%m-%d") assert today in str(date_path) assert date_path.exists() class TestRawSessionPersistence: """Tests de persistence pour RawSession.""" def test_save_raw_session(self, temp_storage, sample_raw_session): """Test sauvegarde d'une RawSession.""" filepath = temp_storage.save_raw_session(sample_raw_session) assert filepath.exists() assert filepath.suffix == ".json" assert "session_" in filepath.name def test_load_raw_session(self, temp_storage, sample_raw_session): """Test chargement d'une RawSession.""" # Sauvegarder filepath = temp_storage.save_raw_session(sample_raw_session) # Charger loaded_session = temp_storage.load_raw_session(filepath) assert loaded_session.session_id == sample_raw_session.session_id assert len(loaded_session.events) == len(sample_raw_session.events) assert len(loaded_session.screenshots) == len(sample_raw_session.screenshots) def test_raw_session_round_trip(self, temp_storage, sample_raw_session): """Test round-trip: save puis load doit retourner les mêmes données.""" filepath = temp_storage.save_raw_session(sample_raw_session) loaded_session = temp_storage.load_raw_session(filepath) # Vérifier que les données sont identiques assert loaded_session.session_id == sample_raw_session.session_id # Le vrai modèle convertit started_at en datetime, donc on compare les ISO strings assert loaded_session.started_at.isoformat() == sample_raw_session.started_at assert len(loaded_session.events) == len(sample_raw_session.events) def test_list_sessions(self, temp_storage, sample_raw_session): """Test listage des sessions.""" # Sauvegarder quelques sessions temp_storage.save_raw_session(sample_raw_session, "session_001") temp_storage.save_raw_session(sample_raw_session, "session_002") # Lister sessions = temp_storage.list_sessions() assert len(sessions) == 2 assert all("session_id" in s for s in sessions) class TestScreenStatePersistence: """Tests de persistence pour ScreenState.""" def test_save_screen_state(self, temp_storage, sample_screen_state): """Test sauvegarde d'un ScreenState.""" filepath = temp_storage.save_screen_state(sample_screen_state) assert filepath.exists() assert filepath.suffix == ".json" assert "state_" in filepath.name def test_load_screen_state(self, temp_storage, sample_screen_state): """Test chargement d'un ScreenState.""" # Sauvegarder filepath = temp_storage.save_screen_state(sample_screen_state) # Charger loaded_state = temp_storage.load_screen_state(filepath) assert loaded_state.screen_state_id == sample_screen_state.state_id assert loaded_state.raw.screenshot_path == sample_screen_state.raw.screenshot_path def test_screen_state_round_trip(self, temp_storage, sample_screen_state): """Test round-trip pour ScreenState.""" filepath = temp_storage.save_screen_state(sample_screen_state) loaded_state = temp_storage.load_screen_state(filepath) assert loaded_state.screen_state_id == sample_screen_state.state_id # Le vrai modèle a window comme objet séparé, pas dans context assert loaded_state.window.window_title == sample_screen_state.window.window_title class TestEmbeddingPersistence: """Tests de persistence pour embeddings.""" def test_save_embedding(self, temp_storage): """Test sauvegarde d'un embedding.""" vector = np.random.rand(512).astype(np.float32) filepath = temp_storage.save_embedding( vector, embedding_id="test_001", embedding_type="state" ) assert filepath.exists() assert filepath.suffix == ".npy" def test_load_embedding(self, temp_storage): """Test chargement d'un embedding.""" original_vector = np.random.rand(512).astype(np.float32) # Sauvegarder temp_storage.save_embedding( original_vector, embedding_id="test_001", embedding_type="state" ) # Charger loaded_vector, metadata = temp_storage.load_embedding( embedding_id="test_001", embedding_type="state" ) assert np.allclose(loaded_vector, original_vector) def test_embedding_with_metadata(self, temp_storage): """Test sauvegarde d'embedding avec métadonnées.""" vector = np.random.rand(512).astype(np.float32) metadata = { "source": "test", "model": "openclip" } filepath = temp_storage.save_embedding( vector, embedding_id="test_001", embedding_type="state", metadata=metadata ) # Vérifier que le fichier de métadonnées existe metadata_file = filepath.with_suffix('.json') assert metadata_file.exists() # Charger et vérifier loaded_vector, loaded_metadata = temp_storage.load_embedding( embedding_id="test_001", embedding_type="state" ) assert loaded_metadata["source"] == "test" assert loaded_metadata["model"] == "openclip" def test_save_embeddings_batch(self, temp_storage): """Test sauvegarde en batch.""" embeddings = { "emb_001": np.random.rand(512).astype(np.float32), "emb_002": np.random.rand(512).astype(np.float32), "emb_003": np.random.rand(512).astype(np.float32) } paths = temp_storage.save_embeddings_batch(embeddings, embedding_type="state") assert len(paths) == 3 assert all(p.exists() for p in paths) def test_list_embeddings(self, temp_storage): """Test listage des embeddings.""" # Sauvegarder quelques embeddings for i in range(3): vector = np.random.rand(512).astype(np.float32) temp_storage.save_embedding( vector, embedding_id=f"test_{i:03d}", embedding_type="state" ) # Lister embeddings = temp_storage.list_embeddings(embedding_type="state") assert len(embeddings) == 3 assert all("embedding_id" in e for e in embeddings) class TestStorageStats: """Tests des statistiques de stockage.""" def test_get_storage_stats(self, temp_storage, sample_raw_session): """Test récupération des statistiques.""" # Sauvegarder quelques fichiers temp_storage.save_raw_session(sample_raw_session) temp_storage.save_embedding( np.random.rand(512).astype(np.float32), embedding_id="test_001", embedding_type="state" ) # Récupérer les stats stats = temp_storage.get_storage_stats() assert "sessions" in stats assert "embeddings" in stats assert "total_size_mb" in stats assert stats["sessions"] >= 1 assert stats["embeddings"] >= 1 class TestCleanup: """Tests du nettoyage des fichiers.""" def test_cleanup_old_files(self, temp_storage): """Test nettoyage des vieux fichiers.""" # Pour ce test, on ne peut pas facilement créer de vieux fichiers # On teste juste que la méthode s'exécute sans erreur deleted = temp_storage.cleanup_old_files(days_to_keep=30) assert isinstance(deleted, dict) assert "sessions" in deleted assert "screen_states" in deleted assert "embeddings" in deleted if __name__ == "__main__": pytest.main([__file__, "-v"])