cf495dd82f
Refonte majeure du système Agent Chat et ajout de nombreux modules : - Chat unifié : suppression du dual Workflows/Agent Libre, tout passe par /api/chat avec résolution en 3 niveaux (workflow → geste → "montre-moi") - GestureCatalog : 38 raccourcis clavier universels Windows avec matching sémantique, substitution automatique dans les replays, et endpoint /api/gestures - Mode Copilot : exécution pas-à-pas des workflows avec validation humaine via WebSocket (approve/skip/abort) avant chaque action - Léa UI (agent_v0/lea_ui/) : interface PyQt5 pour Windows avec overlay transparent pour feedback visuel pendant le replay - Data Extraction (core/extraction/) : moteur d'extraction visuelle de données (OCR + VLM → SQLite), avec schémas YAML et export CSV/Excel - ReplayVerifier (agent_v0/server_v1/) : vérification post-action par comparaison de screenshots, avec logique de retry (max 3) - IntentParser durci : meilleur fallback regex, type GREETING, patterns améliorés - Dashboard : nouvelles pages gestures, streaming, extractions - Tests : 63 tests GestureCatalog, 47 tests extraction, corrections tests existants - Dépréciation : /api/agent/plan et /api/agent/execute retournent HTTP 410, suppression du code hardcodé _plan_to_replay_actions Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
402 lines
15 KiB
Python
402 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Tests Unitaires Registry Actions VWB
|
||
|
||
Auteur : Dom, Alice, Kiro - 09 janvier 2026
|
||
|
||
Ce script teste le registry des actions VisionOnly pour le Visual Workflow Builder.
|
||
|
||
Tests :
|
||
- Création et initialisation du registry
|
||
- Enregistrement d'actions
|
||
- Recherche et récupération d'actions
|
||
- Création d'instances d'actions
|
||
- Auto-découverte des actions
|
||
- Thread-safety du registry
|
||
"""
|
||
|
||
import sys
|
||
import unittest
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Dict, Any, Optional
|
||
|
||
# Ajouter le répertoire racine au path
|
||
ROOT_DIR = Path(__file__).parent.parent.parent
|
||
sys.path.insert(0, str(ROOT_DIR))
|
||
sys.path.insert(0, str(ROOT_DIR / "visual_workflow_builder" / "backend"))
|
||
|
||
try:
|
||
# Import avec chemin absolu
|
||
sys.path.insert(0, str(ROOT_DIR / "visual_workflow_builder" / "backend"))
|
||
|
||
from actions.registry import VWBActionRegistry, get_global_registry, vwb_action
|
||
from actions.base_action import BaseVWBAction, VWBActionResult, VWBActionStatus
|
||
|
||
# Essayer d'importer les actions spécifiques
|
||
try:
|
||
from actions.vision_ui.click_anchor import VWBClickAnchorAction
|
||
from actions.vision_ui.type_text import VWBTypeTextAction
|
||
from actions.vision_ui.wait_for_anchor import VWBWaitForAnchorAction
|
||
SPECIFIC_ACTIONS_OK = True
|
||
except ImportError:
|
||
SPECIFIC_ACTIONS_OK = False
|
||
print("⚠️ Actions spécifiques non disponibles")
|
||
|
||
IMPORTS_OK = True
|
||
print("✅ Imports du registry réussis")
|
||
except ImportError as e:
|
||
print(f"⚠️ Imports non disponibles: {e}")
|
||
IMPORTS_OK = False
|
||
BaseVWBAction = None
|
||
VWBActionResult = None
|
||
VWBActionStatus = None
|
||
|
||
|
||
if IMPORTS_OK and BaseVWBAction is not None:
|
||
class MockVWBAction(BaseVWBAction):
|
||
"""Action mock pour les tests."""
|
||
|
||
def __init__(self, action_id: str, parameters: Optional[Dict[str, Any]] = None, **kwargs):
|
||
super().__init__(action_id, parameters or {})
|
||
self.executed = False
|
||
|
||
def _execute_impl(self, step_id: str, workflow_id: Optional[str] = None,
|
||
user_id: Optional[str] = None) -> VWBActionResult:
|
||
"""Implémentation mock de l'exécution."""
|
||
self.executed = True
|
||
|
||
result = VWBActionResult(
|
||
action_id=self.action_id,
|
||
step_id=step_id,
|
||
status=VWBActionStatus.SUCCESS,
|
||
workflow_id=workflow_id,
|
||
user_id=user_id
|
||
)
|
||
result.output_data = {"mock": True, "executed": True}
|
||
return result
|
||
|
||
def validate_parameters(self) -> list:
|
||
"""Validation mock."""
|
||
return []
|
||
else:
|
||
MockVWBAction = None
|
||
|
||
|
||
@unittest.skipUnless(IMPORTS_OK, "Imports VWB non disponibles")
|
||
class TestVWBActionRegistry(unittest.TestCase):
|
||
"""Tests pour le registry des actions VWB."""
|
||
|
||
def setUp(self):
|
||
"""Préparation des tests."""
|
||
self.registry = VWBActionRegistry()
|
||
|
||
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, VWBActionRegistry)
|
||
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(
|
||
MockVWBAction,
|
||
"mock_action",
|
||
"test",
|
||
{"description": "Action de test"}
|
||
)
|
||
|
||
self.assertTrue(success)
|
||
self.assertIn("mock_action", self.registry.list_actions())
|
||
self.assertIn("test", self.registry.list_categories())
|
||
|
||
# Vérifier les métadonnées
|
||
metadata = self.registry.get_action_metadata("mock_action")
|
||
self.assertIsNotNone(metadata)
|
||
self.assertEqual(metadata["category"], "test")
|
||
self.assertEqual(metadata["class_name"], "MockVWBAction")
|
||
|
||
def test_register_duplicate_action(self):
|
||
"""Test de l'enregistrement d'actions dupliquées."""
|
||
# Premier enregistrement
|
||
success1 = self.registry.register_action(MockVWBAction, "duplicate_test")
|
||
self.assertTrue(success1)
|
||
|
||
# Tentative de duplication
|
||
success2 = self.registry.register_action(MockVWBAction, "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(MockVWBAction, "test_get_class")
|
||
|
||
# Récupérer la classe
|
||
action_class = self.registry.get_action_class("test_get_class")
|
||
self.assertEqual(action_class, MockVWBAction)
|
||
|
||
# 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(MockVWBAction, "test_create")
|
||
|
||
# Créer une instance
|
||
instance = self.registry.create_action(
|
||
"test_create",
|
||
{"param1": "value1"}
|
||
)
|
||
|
||
self.assertIsNotNone(instance)
|
||
self.assertIsInstance(instance, MockVWBAction)
|
||
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(MockVWBAction, "action1", "category1")
|
||
self.registry.register_action(MockVWBAction, "action2", "category1")
|
||
self.registry.register_action(MockVWBAction, "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(MockVWBAction, "click_button", "ui")
|
||
self.registry.register_action(MockVWBAction, "type_text", "ui")
|
||
self.registry.register_action(MockVWBAction, "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(MockVWBAction, "action1", "cat1")
|
||
self.registry.register_action(MockVWBAction, "action2", "cat1")
|
||
self.registry.register_action(MockVWBAction, "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_auto_discover_actions(self):
|
||
"""Test de la découverte automatique d'actions."""
|
||
# Note: Ce test dépend de la structure des fichiers
|
||
# Il peut échouer si les actions VWB ne sont pas disponibles
|
||
|
||
try:
|
||
discovered_count = self.registry.auto_discover_actions()
|
||
|
||
# Vérifier qu'au moins quelques actions ont été découvertes
|
||
self.assertGreaterEqual(discovered_count, 0)
|
||
|
||
# Vérifier que le registry est marqué comme initialisé
|
||
stats = self.registry.get_registry_stats()
|
||
self.assertTrue(stats["initialized"])
|
||
|
||
print(f"✅ Découverte automatique : {discovered_count} actions trouvées")
|
||
|
||
except Exception as e:
|
||
# La découverte peut échouer si les modules ne sont pas disponibles
|
||
print(f"⚠️ Découverte automatique échouée : {e}")
|
||
self.skipTest("Découverte automatique non disponible")
|
||
|
||
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(
|
||
MockVWBAction,
|
||
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_decorator_registration(self):
|
||
"""Test de l'enregistrement via décorateur."""
|
||
|
||
@vwb_action("decorated_action", "decorator_test", {"decorated": True})
|
||
class DecoratedAction(BaseVWBAction):
|
||
def _execute_impl(self, step_id: str, workflow_id: Optional[str] = None,
|
||
user_id: Optional[str] = None) -> VWBActionResult:
|
||
result = VWBActionResult(
|
||
action_id=self.action_id,
|
||
step_id=step_id,
|
||
status=VWBActionStatus.SUCCESS,
|
||
workflow_id=workflow_id,
|
||
user_id=user_id
|
||
)
|
||
return result
|
||
|
||
def validate_parameters(self) -> list:
|
||
return []
|
||
|
||
# Vérifier que l'action a été enregistrée automatiquement
|
||
global_registry = get_global_registry()
|
||
self.assertIn("decorated_action", global_registry.list_actions())
|
||
|
||
# Vérifier les métadonnées
|
||
metadata = global_registry.get_action_metadata("decorated_action")
|
||
self.assertEqual(metadata["category"], "decorator_test")
|
||
self.assertTrue(metadata["metadata"]["decorated"])
|
||
|
||
|
||
@unittest.skipUnless(IMPORTS_OK, "Imports VWB non disponibles")
|
||
class TestGlobalRegistry(unittest.TestCase):
|
||
"""Tests pour le registry global."""
|
||
|
||
def test_global_registry_singleton(self):
|
||
"""Test du pattern singleton pour le registry global."""
|
||
registry1 = get_global_registry()
|
||
registry2 = get_global_registry()
|
||
|
||
# Vérifier que c'est la même instance
|
||
self.assertIs(registry1, registry2)
|
||
|
||
def test_global_registry_auto_discovery(self):
|
||
"""Test de la découverte automatique au premier accès."""
|
||
registry = get_global_registry()
|
||
|
||
# Le registry global devrait avoir découvert des actions automatiquement
|
||
stats = registry.get_registry_stats()
|
||
|
||
print(f"📊 Registry global - Actions: {stats['total_actions']}, Catégories: {len(stats['categories'])}")
|
||
|
||
# Afficher les actions découvertes
|
||
actions = registry.list_actions()
|
||
if actions:
|
||
print(f"🔍 Actions découvertes: {', '.join(actions[:5])}{'...' if len(actions) > 5 else ''}")
|
||
|
||
|
||
def run_tests():
|
||
"""Exécute tous les tests."""
|
||
print("=" * 60)
|
||
print(" TESTS UNITAIRES REGISTRY ACTIONS VWB")
|
||
print("=" * 60)
|
||
print("Auteur : Dom, Alice, Kiro - 09 janvier 2026")
|
||
print("")
|
||
|
||
if not IMPORTS_OK:
|
||
print("❌ Imports non disponibles - tests ignorés")
|
||
return False
|
||
|
||
# Créer la suite de tests
|
||
loader = unittest.TestLoader()
|
||
suite = unittest.TestSuite()
|
||
|
||
# Ajouter les tests
|
||
suite.addTests(loader.loadTestsFromTestCase(TestVWBActionRegistry))
|
||
suite.addTests(loader.loadTestsFromTestCase(TestGlobalRegistry))
|
||
|
||
# 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) |