""" Tests de câblage complet V4 : - SurfaceClassifier + ExecutionCompiler : paramètres adaptés par surface - IRBuilder lit uia_snapshot depuis les événements - ExecutionCompiler crée une stratégie UIA quand dispo - execution_plan_runner propage uia_target dans target_spec - Pipeline E2E : RawTrace (avec UIA) → WorkflowIR → Plan → action runtime """ import sys from pathlib import Path from unittest.mock import patch import pytest _ROOT = str(Path(__file__).resolve().parents[2]) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) from core.workflow.workflow_ir import WorkflowIR, Step, Action from core.workflow.execution_plan import ExecutionPlan, ExecutionNode, ResolutionStrategy from core.workflow.execution_compiler import ExecutionCompiler from core.workflow.surface_classifier import SurfaceClassifier, SurfaceProfile, SurfaceType from core.workflow.ir_builder import IRBuilder from agent_v0.server_v1.execution_plan_runner import ( execution_node_to_action, execution_plan_to_actions, _strategy_to_target_spec, ) # ========================================================================= # ExecutionCompiler avec SurfaceProfile # ========================================================================= class TestCompilerWithSurfaceProfile: def test_profil_citrix_impose_timeouts_longs(self): """Profil Citrix → timeouts longs, retries 3x.""" ir = WorkflowIR.new("Test") ir.add_step("Clic", actions=[{"type": "click", "target": "Bouton", "anchor_hint": "OK"}]) profile = SurfaceProfile( surface_type=SurfaceType.CITRIX, timeout_click_ms=15000, max_retries=3, ocr_threshold=0.65, ) compiler = ExecutionCompiler() plan = compiler.compile(ir, surface_profile=profile) click_node = [n for n in plan.nodes if n.action_type == "click"][0] assert click_node.timeout_ms == 15000 assert click_node.max_retries == 3 def test_profil_web_impose_timeouts_courts(self): """Profil web → timeouts courts, 1 retry.""" ir = WorkflowIR.new("Test") ir.add_step("Clic", actions=[{"type": "click", "target": "X", "anchor_hint": "Login"}]) profile = SurfaceProfile( surface_type=SurfaceType.WEB_LOCAL, timeout_click_ms=5000, max_retries=1, ) compiler = ExecutionCompiler() plan = compiler.compile(ir, surface_profile=profile) click_node = [n for n in plan.nodes if n.action_type == "click"][0] assert click_node.timeout_ms == 5000 assert click_node.max_retries == 1 def test_sans_profil_utilise_defauts(self): """Sans surface_profile, comportement par défaut.""" ir = WorkflowIR.new("Test") ir.add_step("Clic", actions=[{"type": "click", "target": "X", "anchor_hint": "Y"}]) compiler = ExecutionCompiler() plan = compiler.compile(ir) click_node = [n for n in plan.nodes if n.action_type == "click"][0] assert click_node.timeout_ms == 10000 # Défaut assert click_node.max_retries == 2 # Défaut # ========================================================================= # Stratégie UIA dans la compilation # ========================================================================= class TestUiaStrategyCompilation: def _make_ir_with_uia(self): """Créer un WorkflowIR avec une action portant un uia_snapshot.""" ir = WorkflowIR.new("Test UIA") action = Action( type="click", target="Bloc-notes", anchor_hint="Enregistrer", ) # Simuler l'enrichissement avec UIA action._enrichment = { "by_text": "Enregistrer", "anchor_image_base64": "fake_crop_data", "vlm_description": "Le bouton Enregistrer du menu Fichier", "uia_snapshot": { "name": "Enregistrer", "control_type": "bouton", "automation_id": "btnSave", "parent_path": [ {"name": "Bloc-notes", "control_type": "fenêtre"}, {"name": "Fichier", "control_type": "menu"}, ], }, } step = Step(step_id="s1", intent="Sauvegarder", actions=[action]) ir.steps.append(step) return ir def test_uia_strategie_creee_si_surface_windows(self): """Sur Windows natif avec UIA dispo, la stratégie UIA est primaire.""" ir = self._make_ir_with_uia() profile = SurfaceProfile( surface_type=SurfaceType.WINDOWS_NATIVE, uia_available=True, ) compiler = ExecutionCompiler() plan = compiler.compile(ir, surface_profile=profile) click = [n for n in plan.nodes if n.action_type == "click"][0] assert click.strategy_primary is not None assert click.strategy_primary.method == "uia" assert click.strategy_primary.uia_name == "Enregistrer" assert click.strategy_primary.uia_control_type == "bouton" def test_uia_desactive_sur_citrix(self): """Sur Citrix, UIA est ignoré même si snapshot présent.""" ir = self._make_ir_with_uia() profile = SurfaceProfile( surface_type=SurfaceType.CITRIX, uia_available=False, ) compiler = ExecutionCompiler() plan = compiler.compile(ir, surface_profile=profile) click = [n for n in plan.nodes if n.action_type == "click"][0] assert click.strategy_primary.method != "uia" # OCR est la primaire (texte dispo) assert click.strategy_primary.method == "ocr" def test_uia_fallback_sur_ocr_si_uia_manquant(self): """Sans uia_snapshot, OCR primaire.""" ir = WorkflowIR.new("Test") action = Action( type="click", target="Fichier", anchor_hint="Fichier", ) action._enrichment = { "by_text": "Fichier", "vlm_description": "Menu Fichier", } step = Step(step_id="s1", intent="Ouvrir menu", actions=[action]) ir.steps.append(step) profile = SurfaceProfile( surface_type=SurfaceType.WINDOWS_NATIVE, uia_available=True, ) compiler = ExecutionCompiler() plan = compiler.compile(ir, surface_profile=profile) click = [n for n in plan.nodes if n.action_type == "click"][0] assert click.strategy_primary.method == "ocr" # ========================================================================= # IRBuilder lit uia_snapshot depuis les événements # ========================================================================= class TestIRBuilderLitUiaSnapshot: def test_ir_builder_propage_uia_snapshot(self): """Un event avec uia_snapshot → Action._enrichment contient uia_snapshot.""" events = [ { "event": { "type": "mouse_click", "pos": [500, 300], "window": {"title": "Bloc-notes"}, "timestamp": 100.0, "uia_snapshot": { "name": "Enregistrer", "control_type": "bouton", "automation_id": "btnSave", "parent_path": [{"name": "Fichier", "control_type": "menu"}], }, } } ] builder = IRBuilder(gemma4_port="99999") ir = builder.build(events, name="Test") # Parcourir les steps pour trouver le clic found_action = None for step in ir.steps: for action in step.actions: if action.type == "click": found_action = action break assert found_action is not None enrichment = getattr(found_action, "_enrichment", None) or {} assert "uia_snapshot" in enrichment assert enrichment["uia_snapshot"]["name"] == "Enregistrer" assert enrichment["uia_snapshot"]["control_type"] == "bouton" # ========================================================================= # execution_plan_runner propage uia_target dans target_spec # ========================================================================= class TestUiaTargetPropagation: def test_strategy_uia_produit_uia_target(self): """Une stratégie UIA primaire → target_spec contient uia_target.""" primary = ResolutionStrategy( method="uia", uia_name="Enregistrer", uia_control_type="bouton", uia_automation_id="btnSave", uia_parent_path=[{"name": "Fichier", "control_type": "menu"}], ) fallbacks = [ ResolutionStrategy(method="ocr", target_text="Enregistrer"), ResolutionStrategy(method="vlm", vlm_description="bouton Enregistrer"), ] spec = _strategy_to_target_spec(primary, fallbacks) assert "uia_target" in spec assert spec["uia_target"]["name"] == "Enregistrer" assert spec["uia_target"]["control_type"] == "bouton" assert spec["uia_target"]["automation_id"] == "btnSave" assert spec["resolve_order"][0] == "uia" assert "ocr" in spec["resolve_order"] assert "vlm" in spec["resolve_order"] def test_pas_de_uia_target_si_pas_de_stratégie(self): """Sans stratégie UIA → pas de uia_target.""" primary = ResolutionStrategy(method="ocr", target_text="test") spec = _strategy_to_target_spec(primary, []) assert "uia_target" not in spec assert "uia" not in spec.get("resolve_order", []) def test_execution_node_to_action_avec_uia(self): """Un ExecutionNode avec stratégie UIA produit une action complète.""" node = ExecutionNode( node_id="n1", action_type="click", intent="Cliquer Enregistrer", strategy_primary=ResolutionStrategy( method="uia", uia_name="Enregistrer", uia_control_type="bouton", ), strategy_fallbacks=[ ResolutionStrategy(method="ocr", target_text="Enregistrer"), ], ) action = execution_node_to_action(node) assert action is not None assert action["type"] == "click" assert "uia_target" in action["target_spec"] assert action["target_spec"]["uia_target"]["name"] == "Enregistrer" assert action["target_spec"]["resolve_order"] == ["uia", "ocr"] # ========================================================================= # Pipeline E2E : événement avec UIA → action runtime avec uia_target # ========================================================================= class TestPipelineE2EUia: def test_pipeline_complet_uia(self): """RawTrace (avec uia_snapshot) → WorkflowIR → Plan → action runtime.""" # Événements simulés d'un enregistrement sur Windows natif events = [ { "event": { "type": "mouse_click", "pos": [500, 300], "window": {"title": "Bloc-notes"}, "timestamp": 100.0, "uia_snapshot": { "name": "Enregistrer", "control_type": "bouton", "automation_id": "btnSave", "parent_path": [ {"name": "Bloc-notes", "control_type": "fenêtre"}, ], }, } } ] # Pipeline complet builder = IRBuilder(gemma4_port="99999") ir = builder.build(events, name="Test E2E UIA") profile = SurfaceProfile( surface_type=SurfaceType.WINDOWS_NATIVE, uia_available=True, timeout_click_ms=8000, max_retries=2, ) compiler = ExecutionCompiler() plan = compiler.compile(ir, surface_profile=profile) actions = execution_plan_to_actions(plan) # Vérifier que l'action finale a toutes les données UIA click_actions = [a for a in actions if a["type"] == "click"] assert len(click_actions) == 1 action = click_actions[0] assert "target_spec" in action spec = action["target_spec"] assert "resolve_order" in spec assert spec["resolve_order"][0] == "uia" assert "uia_target" in spec assert spec["uia_target"]["name"] == "Enregistrer" assert spec["uia_target"]["control_type"] == "bouton" assert action.get("timeout_ms") == 8000 assert action.get("max_retries") == 2