#!/usr/bin/env python3 """ Tests Unitaires Registry Actions VWB (Version Simplifiée) Auteur : Dom, Alice, Kiro - 09 janvier 2026 Ce script teste le registry des actions VisionOnly pour le Visual Workflow Builder avec une approche simplifiée qui évite les problèmes d'imports relatifs. Tests : - Création et initialisation du registry - Enregistrement d'actions mock - Recherche et récupération d'actions - Thread-safety du registry """ import sys import unittest import threading import time from pathlib import Path from typing import Dict, Any, Optional from datetime import datetime from enum import Enum # Ajouter le répertoire racine au path ROOT_DIR = Path(__file__).parent.parent.parent sys.path.insert(0, str(ROOT_DIR)) class MockActionStatus(Enum): """Status mock pour les tests.""" SUCCESS = "success" FAILURE = "failure" RUNNING = "running" class MockActionResult: """Résultat d'action mock pour les tests.""" def __init__(self, action_id: str, step_id: str, status: MockActionStatus): self.action_id = action_id self.step_id = step_id self.status = status self.start_time = datetime.now() self.end_time = datetime.now() self.execution_time_ms = 100.0 self.output_data = {} self.evidence_list = [] self.error = None self.retry_count = 0 self.workflow_id = None self.user_id = None self.session_id = None def is_success(self) -> bool: return self.status == MockActionStatus.SUCCESS class MockBaseAction: """Classe de base mock pour les actions.""" def __init__(self, action_id: str, parameters: Optional[Dict[str, Any]] = None): self.action_id = action_id self.parameters = parameters or {} self.executed = False def execute(self, step_id: str, workflow_id: Optional[str] = None, user_id: Optional[str] = None) -> MockActionResult: """Exécute l'action mock.""" self.executed = True return MockActionResult(self.action_id, step_id, MockActionStatus.SUCCESS) def validate_parameters(self) -> list: """Valide les paramètres.""" return [] class MockClickAction(MockBaseAction): """Action de clic mock.""" pass class MockTypeAction(MockBaseAction): """Action de saisie mock.""" pass class SimpleVWBActionRegistry: """Registry simplifié pour les tests.""" def __init__(self): """Initialise le registry.""" self._actions: Dict[str, type] = {} self._categories: Dict[str, set] = {} self._metadata: Dict[str, Dict[str, Any]] = {} self._lock = threading.RLock() self._initialized = False print("📋 Registry Actions VWB simplifié initialisé") def register_action(self, action_class: type, action_id: Optional[str] = None, category: str = "default", metadata: Optional[Dict[str, Any]] = None) -> bool: """Enregistre une action dans le registry.""" with self._lock: try: # Générer l'ID si non fourni if action_id is None: action_id = action_class.__name__.lower() # Vérifier l'unicité de l'ID if action_id in self._actions: print(f"⚠️ Action '{action_id}' déjà enregistrée") return False # Enregistrer l'action self._actions[action_id] = action_class # Gérer les catégories if category not in self._categories: self._categories[category] = set() self._categories[category].add(action_id) # Stocker les métadonnées self._metadata[action_id] = { 'class_name': action_class.__name__, 'module': getattr(action_class, '__module__', 'unknown'), 'category': category, 'registered_at': datetime.now().isoformat(), 'metadata': metadata or {} } print(f"✅ Action '{action_id}' enregistrée (catégorie: {category})") return True except Exception as e: print(f"❌ Erreur enregistrement action '{action_id}': {e}") return False def get_action_class(self, action_id: str) -> Optional[type]: """Récupère la classe d'une action par son ID.""" with self._lock: return self._actions.get(action_id) def create_action(self, action_id: str, parameters: Optional[Dict[str, Any]] = None, **kwargs) -> Optional[MockBaseAction]: """Crée une instance d'action.""" with self._lock: action_class = self._actions.get(action_id) if action_class is None: print(f"⚠️ Action '{action_id}' non trouvée dans le registry") return None try: # Créer l'instance instance = action_class( f"{action_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}", parameters or {} ) print(f"✅ Instance d'action '{action_id}' créée") return instance except Exception as e: print(f"❌ Erreur création instance '{action_id}': {e}") return None def list_actions(self, category: Optional[str] = None) -> list: """Liste les actions disponibles.""" with self._lock: if category is None: return list(self._actions.keys()) else: return list(self._categories.get(category, set())) def list_categories(self) -> list: """Liste les catégories disponibles.""" with self._lock: return list(self._categories.keys()) def get_action_metadata(self, action_id: str) -> Optional[Dict[str, Any]]: """Récupère les métadonnées d'une action.""" with self._lock: return self._metadata.get(action_id) def search_actions(self, query: str, category: Optional[str] = None) -> list: """Recherche des actions par nom.""" with self._lock: query_lower = query.lower() results = [] actions_to_search = self.list_actions(category) for action_id in actions_to_search: if query_lower in action_id.lower(): results.append(action_id) return results def get_registry_stats(self) -> Dict[str, Any]: """Obtient les statistiques du registry.""" with self._lock: return { 'total_actions': len(self._actions), 'categories': { cat: len(actions) for cat, actions in self._categories.items() }, 'initialized': self._initialized, 'last_update': datetime.now().isoformat() } def clear(self): """Vide le registry.""" with self._lock: self._actions.clear() self._categories.clear() self._metadata.clear() self._initialized = False print("🗑️ Registry vidé") class TestSimpleVWBActionRegistry(unittest.TestCase): """Tests pour le registry simplifié des actions VWB.""" def setUp(self): """Préparation des tests.""" self.registry = SimpleVWBActionRegistry() def tearDown(self): """Nettoyage après tests.""" self.registry.clear() def test_registry_initialization(self): """Test de l'initialisation du registry.""" self.assertIsInstance(self.registry, SimpleVWBActionRegistry) self.assertEqual(len(self.registry.list_actions()), 0) self.assertEqual(len(self.registry.list_categories()), 0) def test_register_action(self): """Test de l'enregistrement d'actions.""" # Enregistrer une action mock success = self.registry.register_action( MockClickAction, "mock_click", "test", {"description": "Action de test"} ) self.assertTrue(success) self.assertIn("mock_click", self.registry.list_actions()) self.assertIn("test", self.registry.list_categories()) # Vérifier les métadonnées metadata = self.registry.get_action_metadata("mock_click") self.assertIsNotNone(metadata) self.assertEqual(metadata["category"], "test") self.assertEqual(metadata["class_name"], "MockClickAction") def test_register_duplicate_action(self): """Test de l'enregistrement d'actions dupliquées.""" # Premier enregistrement success1 = self.registry.register_action(MockClickAction, "duplicate_test") self.assertTrue(success1) # Tentative de duplication success2 = self.registry.register_action(MockClickAction, "duplicate_test") self.assertFalse(success2) # Vérifier qu'il n'y a qu'une seule action actions = self.registry.list_actions() self.assertEqual(actions.count("duplicate_test"), 1) def test_get_action_class(self): """Test de récupération de classe d'action.""" # Enregistrer une action self.registry.register_action(MockClickAction, "test_get_class") # Récupérer la classe action_class = self.registry.get_action_class("test_get_class") self.assertEqual(action_class, MockClickAction) # Test avec action inexistante non_existent = self.registry.get_action_class("non_existent") self.assertIsNone(non_existent) def test_create_action_instance(self): """Test de création d'instances d'actions.""" # Enregistrer une action self.registry.register_action(MockClickAction, "test_create") # Créer une instance instance = self.registry.create_action( "test_create", {"param1": "value1"} ) self.assertIsNotNone(instance) self.assertIsInstance(instance, MockClickAction) self.assertEqual(instance.parameters["param1"], "value1") # Test avec action inexistante non_existent = self.registry.create_action("non_existent") self.assertIsNone(non_existent) def test_list_actions_by_category(self): """Test de listage d'actions par catégorie.""" # Enregistrer des actions dans différentes catégories self.registry.register_action(MockClickAction, "action1", "category1") self.registry.register_action(MockTypeAction, "action2", "category1") self.registry.register_action(MockClickAction, "action3", "category2") # Tester le listage par catégorie cat1_actions = self.registry.list_actions("category1") self.assertEqual(len(cat1_actions), 2) self.assertIn("action1", cat1_actions) self.assertIn("action2", cat1_actions) cat2_actions = self.registry.list_actions("category2") self.assertEqual(len(cat2_actions), 1) self.assertIn("action3", cat2_actions) # Tester le listage de toutes les actions all_actions = self.registry.list_actions() self.assertEqual(len(all_actions), 3) def test_search_actions(self): """Test de recherche d'actions.""" # Enregistrer des actions avec des noms différents self.registry.register_action(MockClickAction, "click_button", "ui") self.registry.register_action(MockTypeAction, "type_text", "ui") self.registry.register_action(MockClickAction, "wait_element", "control") # Recherche par terme click_results = self.registry.search_actions("click") self.assertIn("click_button", click_results) self.assertEqual(len(click_results), 1) # Recherche par catégorie ui_results = self.registry.search_actions("type", "ui") self.assertIn("type_text", ui_results) self.assertEqual(len(ui_results), 1) # Recherche sans résultat no_results = self.registry.search_actions("nonexistent") self.assertEqual(len(no_results), 0) def test_registry_stats(self): """Test des statistiques du registry.""" # Registry vide stats = self.registry.get_registry_stats() self.assertEqual(stats["total_actions"], 0) self.assertEqual(len(stats["categories"]), 0) # Ajouter des actions self.registry.register_action(MockClickAction, "action1", "cat1") self.registry.register_action(MockTypeAction, "action2", "cat1") self.registry.register_action(MockClickAction, "action3", "cat2") # Vérifier les statistiques stats = self.registry.get_registry_stats() self.assertEqual(stats["total_actions"], 3) self.assertEqual(stats["categories"]["cat1"], 2) self.assertEqual(stats["categories"]["cat2"], 1) def test_thread_safety(self): """Test de la thread-safety du registry.""" results = [] errors = [] def register_actions(thread_id: int): """Fonction pour enregistrer des actions dans un thread.""" try: for i in range(5): action_id = f"thread_{thread_id}_action_{i}" success = self.registry.register_action( MockClickAction, action_id, f"thread_{thread_id}" ) results.append((thread_id, action_id, success)) time.sleep(0.001) # Petite pause pour simuler du travail except Exception as e: errors.append((thread_id, str(e))) # Créer et lancer plusieurs threads threads = [] for i in range(3): thread = threading.Thread(target=register_actions, args=(i,)) threads.append(thread) thread.start() # Attendre la fin de tous les threads for thread in threads: thread.join() # Vérifier les résultats self.assertEqual(len(errors), 0, f"Erreurs dans les threads : {errors}") self.assertEqual(len(results), 15) # 3 threads × 5 actions # Vérifier que toutes les actions ont été enregistrées all_actions = self.registry.list_actions() self.assertEqual(len(all_actions), 15) print(f"✅ Thread-safety validée : {len(results)} enregistrements réussis") def test_action_execution(self): """Test de l'exécution d'actions via le registry.""" # Enregistrer une action self.registry.register_action(MockClickAction, "executable_action") # Créer et exécuter l'action instance = self.registry.create_action("executable_action") self.assertIsNotNone(instance) result = instance.execute("test_step", "test_workflow", "test_user") self.assertIsNotNone(result) self.assertTrue(result.is_success()) self.assertTrue(instance.executed) def run_tests(): """Exécute tous les tests.""" print("=" * 60) print(" TESTS UNITAIRES REGISTRY ACTIONS VWB (SIMPLIFIÉ)") print("=" * 60) print("Auteur : Dom, Alice, Kiro - 09 janvier 2026") print("") # Créer la suite de tests loader = unittest.TestLoader() suite = unittest.TestSuite() # Ajouter les tests suite.addTests(loader.loadTestsFromTestCase(TestSimpleVWBActionRegistry)) # Exécuter les tests runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) # Résumé print("") print("=" * 60) print(" RÉSUMÉ DES TESTS") print("=" * 60) print(f"📊 Tests exécutés : {result.testsRun}") print(f"✅ Tests réussis : {result.testsRun - len(result.failures) - len(result.errors)}") print(f"❌ Tests échoués : {len(result.failures)}") print(f"💥 Erreurs : {len(result.errors)}") if result.failures: print("\n❌ ÉCHECS :") for test, traceback in result.failures: print(f" - {test}: {traceback.split('AssertionError: ')[-1].split('\\n')[0]}") if result.errors: print("\n💥 ERREURS :") for test, traceback in result.errors: print(f" - {test}: {traceback.split('\\n')[-2]}") success_rate = (result.testsRun - len(result.failures) - len(result.errors)) / result.testsRun * 100 print(f"\n📈 Taux de succès : {success_rate:.1f}%") return len(result.failures) == 0 and len(result.errors) == 0 if __name__ == '__main__': success = run_tests() sys.exit(0 if success else 1)