394 lines
13 KiB
Python
Executable File
394 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Test complet du Mode Assisté - Suggestions en temps réel
|
||
Valide l'intégration complète : SuggestionManager + Orchestrator + GUI
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent / "geniusia2"))
|
||
|
||
from core.orchestrator import Orchestrator
|
||
from core.learning_manager import LearningManager
|
||
from core.embeddings_manager import EmbeddingsManager
|
||
from core.suggestion_manager import SuggestionManager
|
||
from core.task_replay import TaskReplayEngine
|
||
from core.utils.vision_utils import VisionUtils
|
||
from core.utils.input_utils import InputUtils
|
||
from core.logger import Logger
|
||
from core.config import get_config
|
||
from core.utils.image_utils import capture_screen
|
||
import numpy as np
|
||
|
||
|
||
def print_section(title):
|
||
"""Affiche un titre de section."""
|
||
print(f"\n{'='*60}")
|
||
print(f" {title}")
|
||
print(f"{'='*60}\n")
|
||
|
||
|
||
async def test_suggestion_manager():
|
||
"""Test 1: SuggestionManager seul."""
|
||
print_section("TEST 1: SuggestionManager")
|
||
|
||
try:
|
||
# Initialiser les composants
|
||
config = get_config()
|
||
logger = Logger()
|
||
embeddings_manager = EmbeddingsManager(logger=logger)
|
||
learning_manager = LearningManager(
|
||
embeddings_manager,
|
||
logger,
|
||
config,
|
||
profiles_path="geniusia2/data/user_profiles"
|
||
)
|
||
|
||
suggestion_manager = SuggestionManager(
|
||
learning_manager,
|
||
embeddings_manager,
|
||
logger,
|
||
config
|
||
)
|
||
|
||
print(f"✅ SuggestionManager initialisé")
|
||
print(f" Seuil de similarité: {suggestion_manager.similarity_threshold}")
|
||
print(f" Timeout: {suggestion_manager.suggestion_timeout}s")
|
||
|
||
# Créer un contexte de test
|
||
screenshot = capture_screen()
|
||
embedding = np.random.rand(512).astype(np.float32)
|
||
|
||
context = {
|
||
"screenshot": screenshot,
|
||
"window": "Test Window",
|
||
"embedding": embedding
|
||
}
|
||
|
||
print(f"\n🔍 Recherche de suggestion...")
|
||
suggestion = suggestion_manager.find_suggestion(context)
|
||
|
||
if suggestion:
|
||
print(f"✅ Suggestion trouvée:")
|
||
print(f" Tâche: {suggestion['task_name']}")
|
||
print(f" Type: {suggestion['action_type']}")
|
||
print(f" Confiance: {suggestion['confidence']:.2%}")
|
||
print(f" Similarité: {suggestion['similarity']:.2%}")
|
||
return True
|
||
else:
|
||
print(f"ℹ️ Aucune suggestion trouvée")
|
||
print(f" (Normal si pas de tâches similaires dans l'index)")
|
||
return True # Pas une erreur
|
||
|
||
except Exception as e:
|
||
print(f"❌ Erreur: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_orchestrator_integration():
|
||
"""Test 2: Intégration dans l'Orchestrator."""
|
||
print_section("TEST 2: Intégration Orchestrator")
|
||
|
||
try:
|
||
# Créer l'orchestrator
|
||
config = get_config()
|
||
logger = Logger()
|
||
|
||
embeddings_manager = EmbeddingsManager(logger=logger)
|
||
learning_manager = LearningManager(
|
||
embeddings_manager,
|
||
logger,
|
||
config,
|
||
profiles_path="geniusia2/data/user_profiles"
|
||
)
|
||
vision_utils = VisionUtils(config)
|
||
input_utils = InputUtils(logger, config)
|
||
|
||
# Importer LLMManager
|
||
from core.llm_manager import LLMManager
|
||
llm_manager = LLMManager(config, logger)
|
||
|
||
orchestrator = Orchestrator(
|
||
learning_manager=learning_manager,
|
||
vision_utils=vision_utils,
|
||
llm_manager=llm_manager,
|
||
input_utils=input_utils,
|
||
logger=logger,
|
||
config=config
|
||
)
|
||
|
||
print(f"✅ Orchestrator créé avec Mode Assisté")
|
||
|
||
# Vérifier les composants
|
||
checks = {
|
||
"SuggestionManager": hasattr(orchestrator, 'suggestion_manager'),
|
||
"TaskReplayEngine": hasattr(orchestrator, 'replay_engine'),
|
||
"check_for_suggestions()": hasattr(orchestrator, 'check_for_suggestions'),
|
||
"accept_current_suggestion()": hasattr(orchestrator, 'accept_current_suggestion'),
|
||
"reject_current_suggestion()": hasattr(orchestrator, 'reject_current_suggestion'),
|
||
"_on_suggestion_created()": hasattr(orchestrator, '_on_suggestion_created'),
|
||
"_on_suggestion_accepted()": hasattr(orchestrator, '_on_suggestion_accepted'),
|
||
"_on_suggestion_rejected()": hasattr(orchestrator, '_on_suggestion_rejected'),
|
||
"_execute_suggestion()": hasattr(orchestrator, '_execute_suggestion'),
|
||
}
|
||
|
||
all_ok = True
|
||
for name, exists in checks.items():
|
||
status = "✅" if exists else "❌"
|
||
print(f" {name:30} : {status}")
|
||
if not exists:
|
||
all_ok = False
|
||
|
||
if all_ok:
|
||
print(f"\n🔍 Test de vérification de suggestions...")
|
||
|
||
# Tester la méthode check_for_suggestions
|
||
orchestrator.check_for_suggestions()
|
||
|
||
# Vérifier s'il y a une suggestion
|
||
current = orchestrator.suggestion_manager.get_current_suggestion()
|
||
|
||
if current:
|
||
print(f"✅ Suggestion créée automatiquement:")
|
||
print(f" Tâche: {current['task_name']}")
|
||
print(f" Confiance: {current['confidence']:.2%}")
|
||
else:
|
||
print(f"ℹ️ Aucune suggestion créée (normal si pas de correspondance)")
|
||
|
||
return all_ok
|
||
|
||
except Exception as e:
|
||
print(f"❌ Erreur: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_gui_integration():
|
||
"""Test 3: Intégration GUI."""
|
||
print_section("TEST 3: Intégration GUI")
|
||
|
||
try:
|
||
# Importer la GUI
|
||
from gui.minimal_gui import MinimalGUI
|
||
from gui.suggestion_overlay import SuggestionOverlay
|
||
|
||
print(f"✅ Imports GUI réussis")
|
||
print(f" MinimalGUI: Disponible")
|
||
print(f" SuggestionOverlay: Disponible")
|
||
|
||
# Vérifier les méthodes de MinimalGUI
|
||
gui_methods = [
|
||
'show_suggestion',
|
||
'hide_suggestion',
|
||
'show_execution_result',
|
||
'keyPressEvent'
|
||
]
|
||
|
||
all_ok = True
|
||
for method in gui_methods:
|
||
has_method = hasattr(MinimalGUI, method)
|
||
status = "✅" if has_method else "❌"
|
||
print(f" {method}():{'':20} {status}")
|
||
if not has_method:
|
||
all_ok = False
|
||
|
||
# Vérifier SuggestionOverlay
|
||
overlay_attrs = ['feedback_received', 'init_ui', 'setup_shortcuts']
|
||
print(f"\n SuggestionOverlay:")
|
||
for attr in overlay_attrs:
|
||
has_attr = hasattr(SuggestionOverlay, attr)
|
||
status = "✅" if has_attr else "❌"
|
||
print(f" {attr}:{'':20} {status}")
|
||
if not has_attr:
|
||
all_ok = False
|
||
|
||
return all_ok
|
||
|
||
except Exception as e:
|
||
print(f"❌ Erreur: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_callbacks():
|
||
"""Test 4: Callbacks et événements."""
|
||
print_section("TEST 4: Callbacks et Événements")
|
||
|
||
try:
|
||
config = get_config()
|
||
logger = Logger()
|
||
|
||
embeddings_manager = EmbeddingsManager(logger=logger)
|
||
learning_manager = LearningManager(
|
||
embeddings_manager,
|
||
logger,
|
||
config,
|
||
profiles_path="geniusia2/data/user_profiles"
|
||
)
|
||
|
||
suggestion_manager = SuggestionManager(
|
||
learning_manager,
|
||
embeddings_manager,
|
||
logger,
|
||
config
|
||
)
|
||
|
||
# Tester les callbacks
|
||
callbacks_called = {
|
||
'created': False,
|
||
'accepted': False,
|
||
'rejected': False,
|
||
'timeout': False
|
||
}
|
||
|
||
def on_created(suggestion):
|
||
callbacks_called['created'] = True
|
||
print(f" ✅ Callback 'created' appelé")
|
||
|
||
def on_accepted(suggestion):
|
||
callbacks_called['accepted'] = True
|
||
print(f" ✅ Callback 'accepted' appelé")
|
||
|
||
def on_rejected(suggestion):
|
||
callbacks_called['rejected'] = True
|
||
print(f" ✅ Callback 'rejected' appelé")
|
||
|
||
def on_timeout(suggestion):
|
||
callbacks_called['timeout'] = True
|
||
print(f" ✅ Callback 'timeout' appelé")
|
||
|
||
# Connecter les callbacks
|
||
suggestion_manager.on_suggestion_created = on_created
|
||
suggestion_manager.on_suggestion_accepted = on_accepted
|
||
suggestion_manager.on_suggestion_rejected = on_rejected
|
||
suggestion_manager.on_suggestion_timeout = on_timeout
|
||
|
||
print(f"✅ Callbacks connectés")
|
||
|
||
# Créer une suggestion factice
|
||
fake_suggestion = {
|
||
'task_id': 'test_123',
|
||
'task_name': 'Test Task',
|
||
'action_type': 'click',
|
||
'confidence': 0.85,
|
||
'similarity': 0.90,
|
||
'created_at': 0
|
||
}
|
||
|
||
print(f"\n🧪 Test des callbacks...")
|
||
|
||
# Tester created
|
||
suggestion_manager.current_suggestion = fake_suggestion
|
||
if suggestion_manager.on_suggestion_created:
|
||
suggestion_manager.on_suggestion_created(fake_suggestion)
|
||
|
||
# Tester accepted
|
||
if suggestion_manager.on_suggestion_accepted:
|
||
suggestion_manager.on_suggestion_accepted(fake_suggestion)
|
||
|
||
# Tester rejected
|
||
if suggestion_manager.on_suggestion_rejected:
|
||
suggestion_manager.on_suggestion_rejected(fake_suggestion)
|
||
|
||
# Tester timeout
|
||
if suggestion_manager.on_suggestion_timeout:
|
||
suggestion_manager.on_suggestion_timeout(fake_suggestion)
|
||
|
||
# Vérifier les résultats
|
||
print(f"\n📊 Résultats:")
|
||
all_ok = all(callbacks_called.values())
|
||
for name, called in callbacks_called.items():
|
||
status = "✅" if called else "❌"
|
||
print(f" {name:15} : {status}")
|
||
|
||
return all_ok
|
||
|
||
except Exception as e:
|
||
print(f"❌ Erreur: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def main():
|
||
"""Fonction principale."""
|
||
print("\n" + "="*60)
|
||
print(" 🧪 TEST COMPLET DU MODE ASSISTÉ")
|
||
print("="*60)
|
||
|
||
results = []
|
||
|
||
# Test 1: SuggestionManager
|
||
try:
|
||
result1 = await test_suggestion_manager()
|
||
results.append(("SuggestionManager", result1))
|
||
except Exception as e:
|
||
print(f"❌ Erreur SuggestionManager: {e}")
|
||
results.append(("SuggestionManager", False))
|
||
|
||
# Test 2: Orchestrator
|
||
try:
|
||
result2 = await test_orchestrator_integration()
|
||
results.append(("Orchestrator", result2))
|
||
except Exception as e:
|
||
print(f"❌ Erreur Orchestrator: {e}")
|
||
results.append(("Orchestrator", False))
|
||
|
||
# Test 3: GUI
|
||
try:
|
||
result3 = await test_gui_integration()
|
||
results.append(("GUI", result3))
|
||
except Exception as e:
|
||
print(f"❌ Erreur GUI: {e}")
|
||
results.append(("GUI", False))
|
||
|
||
# Test 4: Callbacks
|
||
try:
|
||
result4 = await test_callbacks()
|
||
results.append(("Callbacks", result4))
|
||
except Exception as e:
|
||
print(f"❌ Erreur Callbacks: {e}")
|
||
results.append(("Callbacks", False))
|
||
|
||
# Résumé
|
||
print_section("📊 RÉSUMÉ DES TESTS")
|
||
|
||
for test_name, success in results:
|
||
status = "✅ PASS" if success else "❌ FAIL"
|
||
print(f" {test_name:25} : {status}")
|
||
|
||
total_success = sum(1 for _, success in results if success)
|
||
total_tests = len(results)
|
||
|
||
print(f"\n🎯 Résultat: {total_success}/{total_tests} tests réussis")
|
||
|
||
if total_success == total_tests:
|
||
print("\n" + "="*60)
|
||
print(" 🎉 TOUS LES TESTS SONT PASSÉS !")
|
||
print("="*60)
|
||
print("\n💡 Le Mode Assisté est prêt à être utilisé.")
|
||
print("\n📝 Prochaines étapes:")
|
||
print(" 1. cd geniusia2 && ./run.sh")
|
||
print(" 2. Effectuer des actions répétitives (3x)")
|
||
print(" 3. Refaire une action similaire")
|
||
print(" 4. Vérifier qu'une suggestion apparaît")
|
||
print(" 5. Tester Entrée (accepter) et Échap (refuser)")
|
||
else:
|
||
print("\n" + "="*60)
|
||
print(" ⚠️ CERTAINS TESTS ONT ÉCHOUÉ")
|
||
print("="*60)
|
||
print("\n🔍 Vérifiez les erreurs ci-dessus.")
|
||
|
||
return total_success == total_tests
|
||
|
||
|
||
if __name__ == "__main__":
|
||
success = asyncio.run(main())
|
||
sys.exit(0 if success else 1)
|