""" Tests unitaires pour VersionedStore - Fiche #22 Auto-Heal Hybride Tests pour le système de versioning de l'apprentissage réversible. Tests avec fonctionnalité réelle sans simulation. Auteur: Dom, Alice Kiro - 23 décembre 2024 """ import json import pytest import shutil import sqlite3 import tempfile import time from datetime import datetime from pathlib import Path from core.learning.versioned_store import VersionedStore, VersionInfo class TestVersionInfo: """Tests pour la classe VersionInfo""" def test_creation(self): """Test de création de VersionInfo""" now = datetime.now() version = VersionInfo( version_id="v001", created_at=now, workflow_id="test_workflow", success_rate_before=0.85, success_rate_after=None, components_versioned=["prototypes", "faiss"] ) assert version.version_id == "v001" assert version.created_at == now assert version.workflow_id == "test_workflow" assert version.success_rate_before == 0.85 assert version.success_rate_after is None assert version.components_versioned == ["prototypes", "faiss"] def test_serialization(self): """Test de sérialisation/désérialisation""" now = datetime.now() original = VersionInfo( version_id="v002", created_at=now, workflow_id="test_workflow", success_rate_before=0.90, success_rate_after=0.75, components_versioned=["prototypes", "faiss", "memory"] ) # Sérialisation data = original.to_dict() assert data['version_id'] == "v002" assert data['success_rate_before'] == 0.90 assert data['success_rate_after'] == 0.75 assert data['components_versioned'] == ["prototypes", "faiss", "memory"] # Désérialisation restored = VersionInfo.from_dict(data) assert restored.version_id == original.version_id assert restored.workflow_id == original.workflow_id assert restored.success_rate_before == original.success_rate_before assert restored.success_rate_after == original.success_rate_after assert restored.components_versioned == original.components_versioned assert abs((restored.created_at - original.created_at).total_seconds()) < 1 class TestVersionedStore: """Tests pour la classe VersionedStore""" def setup_method(self): """Setup pour chaque test""" # Créer un répertoire temporaire pour les tests self.temp_dir = Path(tempfile.mkdtemp()) self.versioned_store = VersionedStore(base_path=self.temp_dir) # Créer des données de test réalistes self._create_realistic_test_data() def teardown_method(self): """Cleanup après chaque test""" if self.temp_dir.exists(): shutil.rmtree(self.temp_dir) def _create_realistic_test_data(self): """Créer des données de test réalistes qui reflètent l'usage réel du système""" workflow_id = "test_workflow" # Créer des prototypes réalistes (format JSON utilisé par le système) prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id prototypes_dir.mkdir(parents=True, exist_ok=True) # Prototype 1: Bouton de connexion prototype1 = { "id": "login_button_prototype", "ui_type": "button", "semantic_role": "submit", "text_content": "Se connecter", "visual_features": { "color": "#007bff", "size": {"width": 120, "height": 40} }, "embedding": [0.1, 0.2, 0.3, 0.4, 0.5] * 100, # 500 dimensions simulées "confidence": 0.95, "usage_count": 15 } (prototypes_dir / "login_button.json").write_text(json.dumps(prototype1, indent=2)) # Prototype 2: Champ de saisie prototype2 = { "id": "input_field_prototype", "ui_type": "input", "semantic_role": "textbox", "placeholder": "Nom d'utilisateur", "visual_features": { "border_color": "#ccc", "size": {"width": 200, "height": 30} }, "embedding": [0.2, 0.3, 0.4, 0.5, 0.6] * 100, "confidence": 0.88, "usage_count": 12 } (prototypes_dir / "input_field.json").write_text(json.dumps(prototype2, indent=2)) # Créer un index FAISS réaliste avec métadonnées faiss_dir = self.temp_dir / "faiss_index" / f"workflow_{workflow_id}" faiss_dir.mkdir(parents=True, exist_ok=True) # Simuler un fichier FAISS (contenu binaire réaliste) faiss_content = b'\x00\x01\x02\x03' * 1000 # Contenu binaire simulé (faiss_dir / "index.faiss").write_bytes(faiss_content) # Métadonnées FAISS réalistes faiss_metadata = { "dimension": 512, "index_type": "IVF", "metric": "cosine", "total_vectors": 150, "nlist": 50, "nprobe": 8, "created_at": datetime.now().isoformat(), "last_updated": datetime.now().isoformat() } (faiss_dir / "metadata.json").write_text(json.dumps(faiss_metadata, indent=2)) # Créer une base de données SQLite réaliste avec schéma du système db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: # Schéma réaliste basé sur le système target memory conn.execute(""" CREATE TABLE target_elements ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, element_id TEXT NOT NULL, ui_type TEXT, semantic_role TEXT, text_content TEXT, bbox_x INTEGER, bbox_y INTEGER, bbox_width INTEGER, bbox_height INTEGER, confidence REAL, last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, success_count INTEGER DEFAULT 0, failure_count INTEGER DEFAULT 0 ) """) # Insérer des données réalistes test_elements = [ (workflow_id, "btn_login", "button", "submit", "Se connecter", 100, 200, 120, 40, 0.95, 15, 1), (workflow_id, "input_username", "input", "textbox", "", 100, 150, 200, 30, 0.88, 12, 0), (workflow_id, "input_password", "input", "password", "", 100, 180, 200, 30, 0.90, 12, 0) ] conn.executemany(""" INSERT INTO target_elements (workflow_id, element_id, ui_type, semantic_role, text_content, bbox_x, bbox_y, bbox_width, bbox_height, confidence, success_count, failure_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, test_elements) conn.commit() def test_initialization(self): """Test d'initialisation du VersionedStore""" assert self.versioned_store.base_path == self.temp_dir # Vérifier que les répertoires ont été créés assert self.versioned_store.prototypes_path.exists() assert self.versioned_store.faiss_path.exists() assert self.versioned_store.memory_snapshots_path.exists() assert self.versioned_store.versions_metadata_path.exists() def test_generate_version_id(self): """Test de génération d'ID de version avec timestamps réels""" workflow_id = "test_workflow" # Générer deux IDs avec un petit délai version_id1 = self.versioned_store._generate_version_id(workflow_id) time.sleep(0.01) # Petit délai pour assurer des timestamps différents version_id2 = self.versioned_store._generate_version_id(workflow_id) # Vérifier le format assert version_id1.startswith("v") assert workflow_id in version_id1 assert version_id1 != version_id2 # Doivent être différents # Vérifier que les IDs contiennent des timestamps valides timestamp_part1 = version_id1.split("_")[0][1:] # Enlever le 'v' timestamp_part2 = version_id2.split("_")[0][1:] # Vérifier le format de timestamp (YYYYMMDD_HHMMSS) assert len(timestamp_part1) == 8 # YYYYMMDD assert timestamp_part1.isdigit() assert len(timestamp_part2) == 8 assert timestamp_part2.isdigit() def test_snapshot_version_success(self): """Test de création de snapshot réussie avec données réalistes""" workflow_id = "test_workflow" success_rate = 0.85 # Créer un snapshot version_id = self.versioned_store.snapshot_version(workflow_id, success_rate) # Vérifier que l'ID de version a été généré correctement assert version_id.startswith("v") assert workflow_id in version_id # Vérifier que les métadonnées ont été créées metadata_path = self.versioned_store._get_version_metadata_path(workflow_id, version_id) assert metadata_path.exists() # Vérifier le contenu des métadonnées with open(metadata_path, 'r') as f: metadata = json.load(f) assert metadata['version_id'] == version_id assert metadata['workflow_id'] == workflow_id assert metadata['success_rate_before'] == success_rate assert metadata['success_rate_after'] is None assert "prototypes" in metadata['components_versioned'] assert "faiss" in metadata['components_versioned'] assert "memory" in metadata['components_versioned'] # Vérifier que les prototypes ont été copiés avec le bon contenu prototypes_version_path = self.versioned_store.prototypes_path / version_id assert prototypes_version_path.exists() login_button_file = prototypes_version_path / "login_button.json" assert login_button_file.exists() # Vérifier le contenu du prototype copié with open(login_button_file, 'r') as f: prototype_data = json.load(f) assert prototype_data['id'] == "login_button_prototype" assert prototype_data['ui_type'] == "button" assert prototype_data['confidence'] == 0.95 # Vérifier que l'index FAISS a été copié faiss_version_path = self.versioned_store.faiss_path / f"workflow_{workflow_id}" / version_id assert faiss_version_path.exists() assert (faiss_version_path / "index.faiss").exists() # Vérifier le contenu des métadonnées FAISS faiss_metadata_file = faiss_version_path / "metadata.json" assert faiss_metadata_file.exists() with open(faiss_metadata_file, 'r') as f: faiss_metadata = json.load(f) assert faiss_metadata['dimension'] == 512 assert faiss_metadata['index_type'] == "IVF" # Vérifier que la base de données a été sauvegardée memory_snapshot_path = self.versioned_store.memory_snapshots_path / f"{workflow_id}_{version_id}.db" assert memory_snapshot_path.exists() # Vérifier le contenu de la base de données sauvegardée with sqlite3.connect(str(memory_snapshot_path)) as conn: cursor = conn.execute("SELECT COUNT(*) FROM target_elements WHERE workflow_id = ?", (workflow_id,)) count = cursor.fetchone()[0] assert count == 3 # 3 éléments insérés dans _create_realistic_test_data def test_snapshot_version_no_data(self): """Test de création de snapshot sans données existantes""" workflow_id = "nonexistent_workflow" version_id = self.versioned_store.snapshot_version(workflow_id, 0.0) # Le snapshot devrait être créé même sans données assert version_id.startswith("v") # Vérifier les métadonnées metadata_path = self.versioned_store._get_version_metadata_path(workflow_id, version_id) assert metadata_path.exists() with open(metadata_path, 'r') as f: metadata = json.load(f) # Aucun composant ne devrait être versionné assert len(metadata['components_versioned']) == 1 # Seulement memory (base vide) def test_list_versions(self): """Test de listage des versions""" workflow_id = "test_workflow" # Créer plusieurs versions version1 = self.versioned_store.snapshot_version(workflow_id, 0.80) version2 = self.versioned_store.snapshot_version(workflow_id, 0.85) version3 = self.versioned_store.snapshot_version(workflow_id, 0.90) # Lister les versions versions = self.versioned_store.list_versions(workflow_id) # Vérifier le nombre et l'ordre (plus récente en premier) assert len(versions) == 3 assert versions[0].version_id == version3 # Plus récente assert versions[1].version_id == version2 assert versions[2].version_id == version1 # Plus ancienne # Vérifier les taux de succès assert versions[0].success_rate_before == 0.90 assert versions[1].success_rate_before == 0.85 assert versions[2].success_rate_before == 0.80 def test_rollback_to_previous_success(self): """Test de rollback réussi avec vérification du contenu réel""" workflow_id = "test_workflow" # Créer une version initiale version_id = self.versioned_store.snapshot_version(workflow_id, 0.85) # Modifier les données originales de façon réaliste prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id # Modifier un prototype existant modified_prototype = { "id": "login_button_prototype", "ui_type": "button", "semantic_role": "submit", "text_content": "Connexion", # Texte modifié "visual_features": { "color": "#ff0000", # Couleur modifiée "size": {"width": 150, "height": 45} # Taille modifiée }, "embedding": [0.9, 0.8, 0.7, 0.6, 0.5] * 100, # Embedding modifié "confidence": 0.75, # Confiance réduite "usage_count": 20 # Usage augmenté } (prototypes_dir / "login_button.json").write_text(json.dumps(modified_prototype, indent=2)) # Ajouter un nouveau prototype new_prototype = { "id": "new_button_prototype", "ui_type": "button", "semantic_role": "cancel", "text_content": "Annuler", "confidence": 0.60 } (prototypes_dir / "new_button.json").write_text(json.dumps(new_prototype, indent=2)) # Modifier la base de données db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: # Modifier un élément existant conn.execute(""" UPDATE target_elements SET confidence = 0.70, success_count = 20 WHERE element_id = 'btn_login' """) # Ajouter un nouvel élément conn.execute(""" INSERT INTO target_elements (workflow_id, element_id, ui_type, semantic_role, text_content, confidence) VALUES (?, 'btn_cancel', 'button', 'cancel', 'Annuler', 0.60) """, (workflow_id,)) conn.commit() # Effectuer le rollback success = self.versioned_store.rollback_to_previous(workflow_id, version_id) assert success is True # Vérifier que les prototypes ont été restaurés with open(prototypes_dir / "login_button.json", 'r') as f: restored_prototype = json.load(f) # Vérifier que les valeurs originales ont été restaurées assert restored_prototype['text_content'] == "Se connecter" # Valeur originale assert restored_prototype['visual_features']['color'] == "#007bff" # Couleur originale assert restored_prototype['confidence'] == 0.95 # Confiance originale assert restored_prototype['usage_count'] == 15 # Usage original # Vérifier que le nouveau prototype a été supprimé assert not (prototypes_dir / "new_button.json").exists() # Vérifier que la base de données a été restaurée with sqlite3.connect(str(db_path)) as conn: # Vérifier l'élément modifié cursor = conn.execute(""" SELECT confidence, success_count FROM target_elements WHERE element_id = 'btn_login' """) result = cursor.fetchone() assert result[0] == 0.95 # Confiance originale assert result[1] == 15 # Success count original # Vérifier que le nouvel élément a été supprimé cursor = conn.execute(""" SELECT COUNT(*) FROM target_elements WHERE element_id = 'btn_cancel' """) count = cursor.fetchone()[0] assert count == 0 def test_rollback_to_latest(self): """Test de rollback vers la version la plus récente""" workflow_id = "test_workflow" # Créer plusieurs versions version1 = self.versioned_store.snapshot_version(workflow_id, 0.80) version2 = self.versioned_store.snapshot_version(workflow_id, 0.85) # Rollback sans spécifier de version (devrait prendre la plus récente) success = self.versioned_store.rollback_to_previous(workflow_id) assert success is True def test_rollback_nonexistent_version(self): """Test de rollback vers une version inexistante""" workflow_id = "test_workflow" success = self.versioned_store.rollback_to_previous(workflow_id, "nonexistent_version") assert success is False def test_rollback_no_versions(self): """Test de rollback sans versions disponibles""" workflow_id = "nonexistent_workflow" success = self.versioned_store.rollback_to_previous(workflow_id) assert success is False def test_cleanup_old_versions(self): """Test de nettoyage des anciennes versions""" workflow_id = "test_workflow" # Créer 7 versions versions = [] for i in range(7): version_id = self.versioned_store.snapshot_version(workflow_id, 0.8 + i * 0.01) versions.append(version_id) # Vérifier qu'on a bien 7 versions all_versions = self.versioned_store.list_versions(workflow_id) assert len(all_versions) == 7 # Nettoyer en gardant seulement 3 versions self.versioned_store.cleanup_old_versions(workflow_id, keep_count=3) # Vérifier qu'il ne reste que 3 versions remaining_versions = self.versioned_store.list_versions(workflow_id) assert len(remaining_versions) == 3 # Vérifier que ce sont les 3 plus récentes remaining_ids = [v.version_id for v in remaining_versions] assert versions[-1] in remaining_ids # Plus récente assert versions[-2] in remaining_ids assert versions[-3] in remaining_ids assert versions[0] not in remaining_ids # Plus ancienne supprimée def test_update_version_success_rate(self): """Test de mise à jour du taux de succès""" workflow_id = "test_workflow" # Créer une version version_id = self.versioned_store.snapshot_version(workflow_id, 0.85) # Mettre à jour le taux de succès success = self.versioned_store.update_version_success_rate(workflow_id, version_id, 0.92) assert success is True # Vérifier que la mise à jour a été sauvegardée version_info = self.versioned_store._load_version_info(workflow_id, version_id) assert version_info is not None assert version_info.success_rate_after == 0.92 def test_update_version_success_rate_nonexistent(self): """Test de mise à jour du taux de succès pour une version inexistante""" workflow_id = "test_workflow" success = self.versioned_store.update_version_success_rate(workflow_id, "nonexistent", 0.92) assert success is False def test_get_version_stats(self): """Test de récupération des statistiques de versions""" workflow_id = "test_workflow" # Créer plusieurs versions avec différents composants version1 = self.versioned_store.snapshot_version(workflow_id, 0.80) version2 = self.versioned_store.snapshot_version(workflow_id, 0.85) # Mettre à jour les taux de succès après self.versioned_store.update_version_success_rate(workflow_id, version1, 0.82) self.versioned_store.update_version_success_rate(workflow_id, version2, 0.88) # Obtenir les statistiques stats = self.versioned_store.get_version_stats(workflow_id) assert stats['total_versions'] == 2 assert stats['latest_version']['version_id'] == version2 assert stats['average_success_rate_before'] == 0.825 # (0.80 + 0.85) / 2 assert stats['average_success_rate_after'] == 0.85 # (0.82 + 0.88) / 2 assert stats['versions_with_after_rate'] == 2 # Vérifier la distribution des composants components_dist = stats['components_distribution'] assert components_dist['prototypes'] == 2 assert components_dist['faiss'] == 2 assert components_dist['memory'] == 2 def test_get_version_stats_no_versions(self): """Test de récupération des statistiques sans versions""" workflow_id = "nonexistent_workflow" stats = self.versioned_store.get_version_stats(workflow_id) assert stats['total_versions'] == 0 assert stats['latest_version'] is None assert stats['average_success_rate_before'] == 0.0 assert stats['average_success_rate_after'] == 0.0 assert stats['components_distribution'] == {} def test_version_prototypes_only(self): """Test de versioning des prototypes seulement avec données réalistes""" workflow_id = "prototypes_only_workflow" # Créer seulement des prototypes réalistes prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id prototypes_dir.mkdir(parents=True, exist_ok=True) # Créer un prototype réaliste prototype_data = { "id": "simple_button_prototype", "ui_type": "button", "semantic_role": "action", "text_content": "Valider", "visual_features": { "color": "#28a745", "size": {"width": 100, "height": 35} }, "embedding": [0.3, 0.4, 0.5, 0.6, 0.7] * 100, "confidence": 0.92, "usage_count": 8, "created_at": datetime.now().isoformat() } (prototypes_dir / "simple_button.json").write_text(json.dumps(prototype_data, indent=2)) # Créer un snapshot version_id = self.versioned_store.snapshot_version(workflow_id, 0.75) # Vérifier les métadonnées version_info = self.versioned_store._load_version_info(workflow_id, version_id) assert "prototypes" in version_info.components_versioned # Vérifier que le prototype a été correctement versionné versioned_prototype_path = self.versioned_store.prototypes_path / version_id / "simple_button.json" assert versioned_prototype_path.exists() with open(versioned_prototype_path, 'r') as f: versioned_data = json.load(f) assert versioned_data['id'] == "simple_button_prototype" assert versioned_data['confidence'] == 0.92 assert versioned_data['usage_count'] == 8 def test_error_handling_during_snapshot(self): """Test de gestion d'erreur pendant la création de snapshot avec conditions réelles""" workflow_id = "test_workflow" # Créer une condition d'erreur réelle : remplir le disque ou créer un conflit de fichier import os import stat # Créer d'abord un répertoire source avec des données réelles source_dir = self.versioned_store.prototypes_path / workflow_id source_dir.mkdir(exist_ok=True) # Créer un prototype réaliste qui sera versionné realistic_prototype = { "id": "error_test_prototype", "ui_type": "button", "semantic_role": "submit", "text_content": "Test Button", "confidence": 0.85, "usage_count": 10, "created_at": datetime.now().isoformat() } (source_dir / "test_prototype.json").write_text(json.dumps(realistic_prototype, indent=2)) # Créer une base de données réelle avec des données db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS target_elements ( id INTEGER PRIMARY KEY, workflow_id TEXT, element_id TEXT, confidence REAL ) """) conn.execute(""" INSERT INTO target_elements (workflow_id, element_id, confidence) VALUES (?, 'test_element', 0.85) """, (workflow_id,)) conn.commit() # Sauvegarder les permissions originales original_mode = self.temp_dir.stat().st_mode try: # Rendre le répertoire de base en lecture seule pour empêcher la création de nouvelles versions self.temp_dir.chmod(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH) # Essayer de créer un snapshot - devrait échouer à cause des permissions with pytest.raises(PermissionError): self.versioned_store.snapshot_version(workflow_id, 0.85) finally: # Restaurer les permissions originales pour le nettoyage try: self.temp_dir.chmod(original_mode) except: pass # Vérifier qu'aucune version partielle n'a été créée versions = self.versioned_store.list_versions(workflow_id) assert len(versions) == 0 # Vérifier que les données originales sont toujours intactes assert (source_dir / "test_prototype.json").exists() with open(source_dir / "test_prototype.json", 'r') as f: restored_data = json.load(f) assert restored_data['id'] == "error_test_prototype" assert restored_data['confidence'] == 0.85 # Vérifier que la base de données originale est intacte # (3 éléments de setup_method + 1 ajouté dans ce test = 4 au total) with sqlite3.connect(str(db_path)) as conn: cursor = conn.execute("SELECT COUNT(*) FROM target_elements WHERE workflow_id = ?", (workflow_id,)) count = cursor.fetchone()[0] assert count == 4 class TestVersionedStoreIntegration: """Tests d'intégration pour VersionedStore avec scénarios réalistes""" def setup_method(self): """Setup pour chaque test""" self.temp_dir = Path(tempfile.mkdtemp()) self.versioned_store = VersionedStore(base_path=self.temp_dir) def teardown_method(self): """Cleanup après chaque test""" if self.temp_dir.exists(): shutil.rmtree(self.temp_dir) def test_complete_workflow_versioning_cycle(self): """Test d'un cycle complet de versioning avec données réalistes""" workflow_id = "integration_test_workflow" # 1. Créer des données initiales réalistes prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id prototypes_dir.mkdir(parents=True, exist_ok=True) initial_prototype = { "id": "submit_button_v1", "ui_type": "button", "semantic_role": "submit", "text_content": "Envoyer", "visual_features": { "color": "#007bff", "size": {"width": 80, "height": 30} }, "embedding": [0.1, 0.2, 0.3] * 170, # 510 dimensions "confidence": 0.85, "usage_count": 5, "success_rate": 0.80 } (prototypes_dir / "submit_button.json").write_text(json.dumps(initial_prototype, indent=2)) # Créer une base de données initiale db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: conn.execute(""" CREATE TABLE target_elements ( id INTEGER PRIMARY KEY, workflow_id TEXT, element_id TEXT, confidence REAL, success_count INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.execute(""" INSERT INTO target_elements (workflow_id, element_id, confidence, success_count) VALUES (?, 'submit_btn', 0.85, 5) """, (workflow_id,)) conn.commit() # 2. Créer la première version version1 = self.versioned_store.snapshot_version(workflow_id, 0.80) assert version1 is not None # 3. Modifier les données de façon réaliste (amélioration du prototype) improved_prototype = { "id": "submit_button_v2", "ui_type": "button", "semantic_role": "submit", "text_content": "Envoyer", "visual_features": { "color": "#007bff", "size": {"width": 80, "height": 30} }, "embedding": [0.2, 0.3, 0.4] * 170, # Embedding amélioré "confidence": 0.92, # Confiance améliorée "usage_count": 12, # Plus d'utilisation "success_rate": 0.95 # Meilleur taux de succès } (prototypes_dir / "submit_button.json").write_text(json.dumps(improved_prototype, indent=2)) # Ajouter un nouveau prototype new_prototype = { "id": "cancel_button_v1", "ui_type": "button", "semantic_role": "cancel", "text_content": "Annuler", "confidence": 0.88 } (prototypes_dir / "cancel_button.json").write_text(json.dumps(new_prototype, indent=2)) # Mettre à jour la base de données with sqlite3.connect(str(db_path)) as conn: conn.execute(""" UPDATE target_elements SET confidence = 0.92, success_count = 12 WHERE element_id = 'submit_btn' """) conn.execute(""" INSERT INTO target_elements (workflow_id, element_id, confidence, success_count) VALUES (?, 'cancel_btn', 0.88, 3) """, (workflow_id,)) conn.commit() # 4. Créer une deuxième version version2 = self.versioned_store.snapshot_version(workflow_id, 0.95) assert version2 is not None # 5. Vérifier qu'on a 2 versions versions = self.versioned_store.list_versions(workflow_id) assert len(versions) == 2 assert versions[0].version_id == version2 # Plus récente en premier assert versions[1].version_id == version1 # 6. Simuler une dégradation (corruption des données) corrupted_prototype = { "id": "submit_button_corrupted", "ui_type": "unknown", "confidence": 0.30, # Très faible confiance "error": "corrupted_data" } (prototypes_dir / "submit_button.json").write_text(json.dumps(corrupted_prototype, indent=2)) # 7. Rollback vers la première version (données stables) success = self.versioned_store.rollback_to_previous(workflow_id, version1) assert success is True # 8. Vérifier que les données originales ont été restaurées with open(prototypes_dir / "submit_button.json", 'r') as f: restored_data = json.load(f) assert restored_data['id'] == "submit_button_v1" assert restored_data['confidence'] == 0.85 assert restored_data['usage_count'] == 5 assert restored_data['success_rate'] == 0.80 # Vérifier que le nouveau prototype a été supprimé assert not (prototypes_dir / "cancel_button.json").exists() # Vérifier la restauration de la base de données with sqlite3.connect(str(db_path)) as conn: cursor = conn.execute(""" SELECT confidence, success_count FROM target_elements WHERE element_id = 'submit_btn' """) result = cursor.fetchone() assert result[0] == 0.85 # Confiance originale assert result[1] == 5 # Success count original # Vérifier que le nouvel élément a été supprimé cursor = conn.execute(""" SELECT COUNT(*) FROM target_elements WHERE element_id = 'cancel_btn' """) count = cursor.fetchone()[0] assert count == 0 # 9. Mettre à jour les taux de succès avec des valeurs réalistes self.versioned_store.update_version_success_rate(workflow_id, version1, 0.82) self.versioned_store.update_version_success_rate(workflow_id, version2, 0.88) # 10. Vérifier les statistiques finales stats = self.versioned_store.get_version_stats(workflow_id) assert stats['total_versions'] == 2 assert stats['average_success_rate_before'] == 0.875 # (0.80 + 0.95) / 2 assert stats['average_success_rate_after'] == 0.85 # (0.82 + 0.88) / 2 # 11. Nettoyer les anciennes versions self.versioned_store.cleanup_old_versions(workflow_id, keep_count=1) final_versions = self.versioned_store.list_versions(workflow_id) assert len(final_versions) == 1 assert final_versions[0].version_id == version2 # Plus récente conservée def test_concurrent_versioning_operations(self): """Test de gestion des opérations de versioning concurrentes avec données réelles""" import threading import concurrent.futures workflow_id = "concurrent_test_workflow" # Créer des données initiales réalistes avec différents types de prototypes prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id prototypes_dir.mkdir(parents=True, exist_ok=True) # Créer plusieurs prototypes réalistes prototype_templates = [ {"id": "login_button", "ui_type": "button", "semantic_role": "submit", "text_content": "Login", "confidence": 0.90}, {"id": "username_field", "ui_type": "input", "semantic_role": "textbox", "placeholder": "Username", "confidence": 0.85}, {"id": "password_field", "ui_type": "input", "semantic_role": "password", "placeholder": "Password", "confidence": 0.88}, {"id": "cancel_button", "ui_type": "button", "semantic_role": "cancel", "text_content": "Cancel", "confidence": 0.82}, {"id": "remember_checkbox", "ui_type": "checkbox", "semantic_role": "checkbox", "text_content": "Remember me", "confidence": 0.75} ] for i, template in enumerate(prototype_templates): prototype_data = { **template, "usage_count": i + 5, "created_at": datetime.now().isoformat(), "embedding": [float(j + i) for j in range(100)] # Embedding réaliste } filename = f"{template['id']}.json" (prototypes_dir / filename).write_text(json.dumps(prototype_data, indent=2)) # Créer une base de données réelle avec des éléments correspondants db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: conn.execute(""" CREATE TABLE target_elements ( id INTEGER PRIMARY KEY, workflow_id TEXT, element_id TEXT, ui_type TEXT, confidence REAL, success_count INTEGER DEFAULT 0 ) """) # Insérer des éléments correspondant aux prototypes for i, template in enumerate(prototype_templates): conn.execute(""" INSERT INTO target_elements (workflow_id, element_id, ui_type, confidence, success_count) VALUES (?, ?, ?, ?, ?) """, (workflow_id, template['id'], template['ui_type'], template['confidence'], i + 3)) conn.commit() # Fonction pour créer des versions concurrentes avec modifications réalistes def create_version_with_modifications(version_num): try: # Modifier légèrement les données avant chaque version pour simuler l'évolution if version_num > 0: # Modifier un prototype existant prototype_to_modify = prototype_templates[version_num % len(prototype_templates)] modified_prototype = { **prototype_to_modify, "confidence": min(0.95, prototype_to_modify["confidence"] + 0.02 * version_num), "usage_count": prototype_to_modify.get("usage_count", 0) + version_num, "last_modified": datetime.now().isoformat(), "embedding": [float(j + version_num) for j in range(100)] } filename = f"{prototype_to_modify['id']}.json" (prototypes_dir / filename).write_text(json.dumps(modified_prototype, indent=2)) # Créer la version avec un taux de succès réaliste success_rate = 0.70 + (version_num * 0.03) version_id = self.versioned_store.snapshot_version(workflow_id, success_rate) return version_id except Exception as e: return f"error_{version_num}: {str(e)}" # Créer plusieurs versions en parallèle with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: futures = [executor.submit(create_version_with_modifications, i) for i in range(5)] results = [future.result() for future in concurrent.futures.as_completed(futures)] # Analyser les résultats successful_versions = [r for r in results if r.startswith("v")] error_results = [r for r in results if r.startswith("error_")] # Au moins quelques versions devraient avoir été créées avec succès assert len(successful_versions) >= 1, f"No successful versions created. Results: {results}" # Vérifier que les versions créées sont valides et contiennent des données réelles all_versions = self.versioned_store.list_versions(workflow_id) assert len(all_versions) >= len(successful_versions) # Vérifier l'intégrité des métadonnées et des données versionnées for version in all_versions: assert version.workflow_id == workflow_id assert 0.70 <= version.success_rate_before <= 0.85 # Vérifier que les prototypes ont été correctement versionnés version_prototypes_dir = self.versioned_store.prototypes_path / version.version_id if version_prototypes_dir.exists(): prototype_files = list(version_prototypes_dir.glob("*.json")) assert len(prototype_files) == len(prototype_templates) # Vérifier le contenu d'un prototype versionné sample_file = prototype_files[0] with open(sample_file, 'r') as f: prototype_data = json.load(f) assert "id" in prototype_data assert "confidence" in prototype_data assert isinstance(prototype_data["confidence"], float) assert 0.0 <= prototype_data["confidence"] <= 1.0 # Tester un rollback pour vérifier l'intégrité des données versionnées if all_versions: latest_version = all_versions[0] rollback_success = self.versioned_store.rollback_to_previous(workflow_id, latest_version.version_id) assert rollback_success is True # Vérifier que les données ont été correctement restaurées restored_files = list(prototypes_dir.glob("*.json")) assert len(restored_files) == len(prototype_templates) def test_large_data_versioning_performance(self): """Test de performance avec de gros volumes de données réalistes du système RPA""" workflow_id = "performance_test_workflow" # Créer un grand nombre de prototypes réalistes basés sur des éléments UI courants prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id prototypes_dir.mkdir(parents=True, exist_ok=True) # Templates d'éléments UI réalistes ui_templates = [ {"ui_type": "button", "semantic_roles": ["submit", "cancel", "save", "delete", "edit"], "texts": ["Valider", "Annuler", "Sauvegarder", "Supprimer", "Modifier"]}, {"ui_type": "input", "semantic_roles": ["textbox", "email", "password", "search"], "texts": ["", "email@example.com", "", "Rechercher..."]}, {"ui_type": "select", "semantic_roles": ["dropdown", "combobox"], "texts": ["Choisir une option", "Sélectionner"]}, {"ui_type": "checkbox", "semantic_roles": ["checkbox"], "texts": ["Accepter les conditions", "Se souvenir de moi"]}, {"ui_type": "link", "semantic_roles": ["link", "navigation"], "texts": ["En savoir plus", "Accueil", "Contact"]} ] # Créer 200 prototypes réalistes pour tester la performance start_time = time.time() for i in range(200): template = ui_templates[i % len(ui_templates)] role_index = i % len(template["semantic_roles"]) text_index = i % len(template["texts"]) # Générer des coordonnées réalistes pour les éléments UI x = 50 + (i % 20) * 40 # Répartir sur une grille y = 100 + (i // 20) * 35 width = 80 + (i % 5) * 20 height = 25 + (i % 3) * 10 prototype = { "id": f"{template['ui_type']}_prototype_{i:03d}", "ui_type": template["ui_type"], "semantic_role": template["semantic_roles"][role_index], "text_content": template["texts"][text_index], "visual_features": { "bbox": {"x": x, "y": y, "width": width, "height": height}, "color": f"#{(i * 123456) % 16777216:06x}", # Couleur pseudo-aléatoire "font_size": 12 + (i % 4) * 2 }, "embedding": [float((j + i) % 1000) / 1000.0 for j in range(512)], # 512 dimensions réalistes "confidence": 0.5 + (i % 50) / 100.0, "usage_count": i % 25, "success_rate": 0.6 + (i % 40) / 100.0, "metadata": { "created_at": datetime.now().isoformat(), "source": "performance_test", "batch_id": i // 20, "screen_resolution": "1920x1080", "application": f"TestApp_{i % 5}" } } filename = f"{template['ui_type']}_prototype_{i:03d}.json" (prototypes_dir / filename).write_text(json.dumps(prototype, indent=2)) creation_time = time.time() - start_time # Créer une base de données réaliste avec des éléments correspondants et des relations db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: # Schéma réaliste du système RPA conn.execute(""" CREATE TABLE target_elements ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, element_id TEXT NOT NULL, ui_type TEXT, semantic_role TEXT, text_content TEXT, bbox_x INTEGER, bbox_y INTEGER, bbox_width INTEGER, bbox_height INTEGER, confidence REAL CHECK(confidence >= 0.0 AND confidence <= 1.0), success_count INTEGER DEFAULT 0, failure_count INTEGER DEFAULT 0, last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Table pour les interactions utilisateur conn.execute(""" CREATE TABLE user_interactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, element_id TEXT NOT NULL, action_type TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, success BOOLEAN, execution_time_ms INTEGER ) """) # Insérer 2000 éléments avec des données réalistes elements_data = [] interactions_data = [] for i in range(2000): template = ui_templates[i % len(ui_templates)] role_index = i % len(template["semantic_roles"]) text_index = i % len(template["texts"]) # Coordonnées réalistes x = 50 + (i % 30) * 30 y = 100 + (i // 30) * 25 width = 80 + (i % 5) * 15 height = 25 + (i % 3) * 8 elements_data.append(( workflow_id, f"{template['ui_type']}_element_{i:04d}", template["ui_type"], template["semantic_roles"][role_index], template["texts"][text_index], x, y, width, height, 0.5 + (i % 50) / 100.0, # confidence i % 30, # success_count i % 5 # failure_count )) # Ajouter des interactions réalistes for j in range(i % 3 + 1): # 1-3 interactions par élément action_types = ["click", "type", "hover", "scroll"] action = action_types[j % len(action_types)] success = (i + j) % 4 != 0 # 75% de succès exec_time = 50 + (i + j) % 200 # 50-250ms interactions_data.append(( workflow_id, f"{template['ui_type']}_element_{i:04d}", action, success, exec_time )) # Insertion par batch pour de meilleures performances conn.executemany(""" INSERT INTO target_elements (workflow_id, element_id, ui_type, semantic_role, text_content, bbox_x, bbox_y, bbox_width, bbox_height, confidence, success_count, failure_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, elements_data) conn.executemany(""" INSERT INTO user_interactions (workflow_id, element_id, action_type, success, execution_time_ms) VALUES (?, ?, ?, ?, ?) """, interactions_data) conn.commit() # Mesurer le temps de création de snapshot avec des données réalistes snapshot_start = time.time() version_id = self.versioned_store.snapshot_version(workflow_id, 0.85) snapshot_time = time.time() - snapshot_start # Vérifier que le snapshot a été créé avec toutes les données assert version_id is not None # Vérifier que tous les prototypes ont été copiés versioned_prototypes_dir = self.versioned_store.prototypes_path / version_id versioned_files = list(versioned_prototypes_dir.glob("*.json")) assert len(versioned_files) == 200 # Vérifier la qualité des données versionnées sample_file = versioned_files[0] with open(sample_file, 'r') as f: sample_data = json.load(f) assert "visual_features" in sample_data assert "embedding" in sample_data assert len(sample_data["embedding"]) == 512 assert "metadata" in sample_data # Vérifier que la base de données a été sauvegardée avec toutes les tables snapshot_db = self.versioned_store.memory_snapshots_path / f"{workflow_id}_{version_id}.db" assert snapshot_db.exists() # Vérifier l'intégrité des données dans le snapshot with sqlite3.connect(str(snapshot_db)) as conn: # Vérifier les éléments cursor = conn.execute("SELECT COUNT(*) FROM target_elements WHERE workflow_id = ?", (workflow_id,)) elements_count = cursor.fetchone()[0] assert elements_count == 2000 # Vérifier les interactions cursor = conn.execute("SELECT COUNT(*) FROM user_interactions WHERE workflow_id = ?", (workflow_id,)) interactions_count = cursor.fetchone()[0] assert interactions_count > 2000 # Au moins autant que les éléments # Vérifier l'intégrité des données cursor = conn.execute(""" SELECT AVG(confidence), COUNT(DISTINCT ui_type), COUNT(DISTINCT semantic_role) FROM target_elements WHERE workflow_id = ? """, (workflow_id,)) avg_conf, ui_types, roles = cursor.fetchone() assert 0.5 <= avg_conf <= 1.0 assert ui_types == len(ui_templates) # Tous les types UI présents assert roles > 5 # Plusieurs rôles sémantiques # Mesurer le temps de rollback avec modification réaliste des données # Modifier quelques prototypes de façon réaliste for i in range(20): modified_prototype = { "id": f"modified_prototype_{i}", "ui_type": "button", "confidence": 0.1, # Dégradation réaliste "text_content": "Modified", "last_modified": datetime.now().isoformat(), "modification_reason": "performance_test_corruption" } (prototypes_dir / f"button_prototype_{i:03d}.json").write_text(json.dumps(modified_prototype)) # Modifier la base de données de façon réaliste with sqlite3.connect(str(db_path)) as conn: conn.execute(""" UPDATE target_elements SET confidence = 0.1, failure_count = failure_count + 10 WHERE element_id LIKE 'button_element_%' AND ROWID <= 20 """) conn.commit() rollback_start = time.time() success = self.versioned_store.rollback_to_previous(workflow_id, version_id) rollback_time = time.time() - rollback_start assert success is True # Vérifier que les données ont été correctement restaurées restored_file = prototypes_dir / "button_prototype_000.json" with open(restored_file, 'r') as f: restored_data = json.load(f) assert restored_data['id'] == "button_prototype_000" assert restored_data['confidence'] != 0.1 # Pas la valeur corrompue assert "visual_features" in restored_data # Données complètes restaurées # Vérifier la restauration de la base de données with sqlite3.connect(str(db_path)) as conn: cursor = conn.execute(""" SELECT confidence, failure_count FROM target_elements WHERE element_id = 'button_element_0000' """) result = cursor.fetchone() if result: confidence, failure_count = result assert confidence > 0.1 # Valeur originale restaurée assert failure_count < 10 # Pas la valeur corrompue # Afficher les métriques de performance print(f"\nPerformance metrics (200 prototypes, 2000+ DB records):") print(f" Data creation: {creation_time:.2f}s") print(f" Snapshot creation: {snapshot_time:.2f}s") print(f" Rollback: {rollback_time:.2f}s") print(f" Snapshot size: {snapshot_db.stat().st_size / 1024 / 1024:.2f} MB") # Vérifier que les opérations sont raisonnablement rapides pour des données réalistes assert snapshot_time < 15.0, f"Snapshot too slow for realistic data: {snapshot_time:.2f}s" assert rollback_time < 15.0, f"Rollback too slow for realistic data: {rollback_time:.2f}s" def test_database_schema_validation(self): """Test de validation du schéma de base de données avec des données réalistes""" workflow_id = "schema_validation_workflow" # Créer une base de données avec le schéma réel du système db_path = self.temp_dir / "target_memory.db" with sqlite3.connect(str(db_path)) as conn: # Schéma complet basé sur le système réel conn.execute(""" CREATE TABLE target_elements ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, element_id TEXT NOT NULL, ui_type TEXT, semantic_role TEXT, text_content TEXT, bbox_x INTEGER, bbox_y INTEGER, bbox_width INTEGER, bbox_height INTEGER, confidence REAL CHECK(confidence >= 0.0 AND confidence <= 1.0), last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, success_count INTEGER DEFAULT 0, failure_count INTEGER DEFAULT 0, embedding_vector BLOB, visual_hash TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(workflow_id, element_id) ) """) # Table pour les métriques de performance conn.execute(""" CREATE TABLE performance_metrics ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, element_id TEXT NOT NULL, action_type TEXT, execution_time_ms INTEGER, success BOOLEAN, error_message TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY(workflow_id, element_id) REFERENCES target_elements(workflow_id, element_id) ) """) # Insérer des données réalistes avec contraintes realistic_elements = [ (workflow_id, "login_form_username", "input", "textbox", "", 150, 200, 250, 30, 0.95, 25, 2), (workflow_id, "login_form_password", "input", "password", "", 150, 240, 250, 30, 0.93, 25, 1), (workflow_id, "login_submit_button", "button", "submit", "Se connecter", 200, 280, 150, 40, 0.98, 30, 0), (workflow_id, "navigation_menu", "nav", "navigation", "", 50, 50, 200, 500, 0.85, 15, 3), (workflow_id, "search_input", "input", "search", "Rechercher...", 300, 60, 200, 35, 0.90, 20, 1) ] conn.executemany(""" INSERT INTO target_elements (workflow_id, element_id, ui_type, semantic_role, text_content, bbox_x, bbox_y, bbox_width, bbox_height, confidence, success_count, failure_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, realistic_elements) # Insérer des métriques de performance performance_data = [ (workflow_id, "login_submit_button", "click", 150, True, None), (workflow_id, "login_submit_button", "click", 180, True, None), (workflow_id, "login_submit_button", "click", 2500, False, "Element not found"), (workflow_id, "search_input", "type", 50, True, None), (workflow_id, "search_input", "type", 45, True, None) ] conn.executemany(""" INSERT INTO performance_metrics (workflow_id, element_id, action_type, execution_time_ms, success, error_message) VALUES (?, ?, ?, ?, ?, ?) """, performance_data) conn.commit() # Créer un snapshot version_id = self.versioned_store.snapshot_version(workflow_id, 0.92) # Vérifier que le snapshot préserve l'intégrité des données snapshot_db = self.versioned_store.memory_snapshots_path / f"{workflow_id}_{version_id}.db" assert snapshot_db.exists() with sqlite3.connect(str(snapshot_db)) as conn: # Vérifier que toutes les tables existent cursor = conn.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' """) tables = [row[0] for row in cursor.fetchall()] assert "target_elements" in tables assert "performance_metrics" in tables # Vérifier l'intégrité des données cursor = conn.execute("SELECT COUNT(*) FROM target_elements WHERE workflow_id = ?", (workflow_id,)) elements_count = cursor.fetchone()[0] assert elements_count == 5 cursor = conn.execute("SELECT COUNT(*) FROM performance_metrics WHERE workflow_id = ?", (workflow_id,)) metrics_count = cursor.fetchone()[0] assert metrics_count == 5 # Vérifier les contraintes de données cursor = conn.execute(""" SELECT element_id, confidence, success_count, failure_count FROM target_elements WHERE workflow_id = ? AND element_id = 'login_submit_button' """, (workflow_id,)) result = cursor.fetchone() assert result is not None element_id, confidence, success_count, failure_count = result assert 0.0 <= confidence <= 1.0 assert success_count >= 0 assert failure_count >= 0 # Vérifier les relations entre tables cursor = conn.execute(""" SELECT COUNT(*) FROM performance_metrics pm JOIN target_elements te ON pm.workflow_id = te.workflow_id AND pm.element_id = te.element_id WHERE pm.workflow_id = ? """, (workflow_id,)) joined_count = cursor.fetchone()[0] assert joined_count == 5 # Toutes les métriques doivent avoir un élément correspondant # Tester le rollback avec validation de schéma # Modifier la base de données originale with sqlite3.connect(str(db_path)) as conn: conn.execute(""" UPDATE target_elements SET confidence = 0.50, success_count = 50 WHERE element_id = 'login_submit_button' """) conn.execute(""" DELETE FROM performance_metrics WHERE element_id = 'search_input' """) conn.commit() # Effectuer le rollback success = self.versioned_store.rollback_to_previous(workflow_id, version_id) assert success is True # Vérifier que les données ont été restaurées correctement with sqlite3.connect(str(db_path)) as conn: cursor = conn.execute(""" SELECT confidence, success_count FROM target_elements WHERE element_id = 'login_submit_button' """) result = cursor.fetchone() assert result[0] == 0.98 # Confiance originale assert result[1] == 30 # Success count original cursor = conn.execute(""" SELECT COUNT(*) FROM performance_metrics WHERE element_id = 'search_input' """) count = cursor.fetchone()[0] assert count == 2 # Métriques restaurées def test_real_system_integration_with_faiss_data(self): """Test d'intégration avec des données FAISS réalistes du système""" workflow_id = "faiss_integration_workflow" # Créer des données FAISS réalistes faiss_dir = self.temp_dir / "faiss_index" / f"workflow_{workflow_id}" faiss_dir.mkdir(parents=True, exist_ok=True) # Simuler un fichier FAISS binaire réaliste (structure simplifiée) # En réalité, FAISS génère des fichiers binaires complexes faiss_binary_data = bytearray() # Header FAISS simulé faiss_binary_data.extend(b'FAISS_INDEX_V1') # Magic number faiss_binary_data.extend((512).to_bytes(4, 'little')) # Dimensions faiss_binary_data.extend((1000).to_bytes(4, 'little')) # Nombre de vecteurs # Données de vecteurs simulées (512 dimensions * 1000 vecteurs * 4 bytes) import struct for i in range(1000): for j in range(512): # Valeurs réalistes d'embeddings normalisés value = (i + j) % 1000 / 1000.0 - 0.5 # Valeurs entre -0.5 et 0.5 faiss_binary_data.extend(struct.pack('f', value)) # Sauvegarder le fichier FAISS simulé (faiss_dir / "index.faiss").write_bytes(faiss_binary_data) # Métadonnées FAISS réalistes basées sur le système faiss_metadata = { "version": "1.7.4", "index_type": "IndexFlatIP", # Inner Product pour cosine similarity "metric": "METRIC_INNER_PRODUCT", "dimension": 512, "ntotal": 1000, "is_trained": True, "created_at": datetime.now().isoformat(), "last_updated": datetime.now().isoformat(), "embedding_model": "ViT-B-32", "preprocessing": { "normalization": "l2", "dimension_reduction": None }, "performance_stats": { "avg_search_time_ms": 2.5, "memory_usage_mb": 2.1, "last_rebuild": datetime.now().isoformat() } } (faiss_dir / "metadata.json").write_text(json.dumps(faiss_metadata, indent=2)) # Créer un fichier de mapping ID -> métadonnées (utilisé par le système) id_mapping = {} for i in range(1000): id_mapping[str(i)] = { "embedding_id": f"emb_{workflow_id}_{i:04d}", "source_type": "ui_element", "ui_type": ["button", "input", "select", "checkbox"][i % 4], "semantic_role": f"role_{i % 10}", "confidence": 0.5 + (i % 50) / 100.0, "created_at": datetime.now().isoformat() } (faiss_dir / "id_mapping.json").write_text(json.dumps(id_mapping, indent=2)) # Créer des prototypes correspondants prototypes_dir = self.temp_dir / "learning" / "prototypes" / workflow_id prototypes_dir.mkdir(parents=True, exist_ok=True) # Créer quelques prototypes qui correspondent aux embeddings FAISS for i in range(10): prototype = { "id": f"prototype_{i:03d}", "embedding_id": f"emb_{workflow_id}_{i:04d}", # Lien avec FAISS "ui_type": ["button", "input", "select", "checkbox"][i % 4], "semantic_role": f"role_{i % 10}", "visual_features": { "bbox": {"x": 100 + i * 50, "y": 200, "width": 120, "height": 30}, "color": f"#{(i * 123456) % 16777216:06x}", "text_content": f"Element {i}" }, "embedding": [float((j + i) % 1000) / 1000.0 - 0.5 for j in range(512)], "confidence": 0.5 + (i % 50) / 100.0, "usage_count": i * 3, "faiss_index": i, # Index dans FAISS "metadata": { "source": "real_system_integration", "faiss_synchronized": True } } (prototypes_dir / f"prototype_{i:03d}.json").write_text(json.dumps(prototype, indent=2)) # Créer un snapshot version_id = self.versioned_store.snapshot_version(workflow_id, 0.88) assert version_id is not None # Vérifier que les données FAISS ont été correctement versionnées faiss_version_path = self.versioned_store.faiss_path / f"workflow_{workflow_id}" / version_id assert faiss_version_path.exists() # Vérifier que tous les fichiers FAISS ont été copiés assert (faiss_version_path / "index.faiss").exists() assert (faiss_version_path / "metadata.json").exists() assert (faiss_version_path / "id_mapping.json").exists() # Vérifier l'intégrité des métadonnées FAISS versionnées with open(faiss_version_path / "metadata.json", 'r') as f: versioned_metadata = json.load(f) assert versioned_metadata["dimension"] == 512 assert versioned_metadata["ntotal"] == 1000 assert versioned_metadata["index_type"] == "IndexFlatIP" # Vérifier l'intégrité du mapping ID with open(faiss_version_path / "id_mapping.json", 'r') as f: versioned_mapping = json.load(f) assert len(versioned_mapping) == 1000 assert "emb_faiss_integration_workflow_0000" in versioned_mapping["0"]["embedding_id"] # Vérifier que le fichier binaire a été copié correctement versioned_binary = (faiss_version_path / "index.faiss").read_bytes() original_binary = (faiss_dir / "index.faiss").read_bytes() assert len(versioned_binary) == len(original_binary) assert versioned_binary[:20] == original_binary[:20] # Vérifier le header # Simuler une corruption des données FAISS # Corrompre le fichier binaire corrupted_data = b'CORRUPTED_DATA' * 1000 (faiss_dir / "index.faiss").write_bytes(corrupted_data) # Corrompre les métadonnées corrupted_metadata = {"error": "corrupted", "dimension": -1} (faiss_dir / "metadata.json").write_text(json.dumps(corrupted_metadata)) # Supprimer le mapping (faiss_dir / "id_mapping.json").unlink() # Effectuer un rollback success = self.versioned_store.rollback_to_previous(workflow_id, version_id) assert success is True # Vérifier que les données FAISS ont été restaurées assert (faiss_dir / "index.faiss").exists() assert (faiss_dir / "metadata.json").exists() assert (faiss_dir / "id_mapping.json").exists() # Vérifier l'intégrité des données restaurées restored_binary = (faiss_dir / "index.faiss").read_bytes() assert restored_binary[:20] == original_binary[:20] # Header restauré assert len(restored_binary) == len(original_binary) with open(faiss_dir / "metadata.json", 'r') as f: restored_metadata = json.load(f) assert restored_metadata["dimension"] == 512 assert restored_metadata["ntotal"] == 1000 assert "error" not in restored_metadata with open(faiss_dir / "id_mapping.json", 'r') as f: restored_mapping = json.load(f) assert len(restored_mapping) == 1000 # Vérifier que les prototypes sont toujours synchronisés for i in range(10): prototype_file = prototypes_dir / f"prototype_{i:03d}.json" assert prototype_file.exists() with open(prototype_file, 'r') as f: prototype_data = json.load(f) assert prototype_data["faiss_index"] == i assert prototype_data["metadata"]["faiss_synchronized"] is True if __name__ == "__main__": pytest.main([__file__])