""" Tests de l'ExecutionCompiler et de l'ExecutionPlan. Vérifie que : - Le compilateur produit un plan déterministe depuis un WorkflowIR - Les stratégies de résolution sont correctement compilées (OCR > template > VLM) - Les timeouts, retries et recovery sont définis - Le plan est sérialisable et versionné """ import json import shutil import sys import tempfile from pathlib import Path 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 from core.workflow.execution_plan import ExecutionPlan, ExecutionNode, ResolutionStrategy, SuccessCondition from core.workflow.execution_compiler import ExecutionCompiler # ========================================================================= # ExecutionPlan — format et sérialisation # ========================================================================= class TestExecutionPlan: def test_serialisation_roundtrip(self): plan = ExecutionPlan( plan_id="plan_test", workflow_id="wf_123", version=1, ) plan.nodes.append(ExecutionNode( node_id="n1", action_type="click", intent="Cliquer sur Enregistrer", strategy_primary=ResolutionStrategy(method="ocr", target_text="Enregistrer"), strategy_fallbacks=[ResolutionStrategy(method="vlm", vlm_description="bouton Enregistrer")], success_condition=SuccessCondition(method="title_match", expected_title="Fichier sauvegardé"), )) json_str = plan.to_json() plan2 = ExecutionPlan.from_json(json_str) assert plan2.plan_id == "plan_test" assert len(plan2.nodes) == 1 assert plan2.nodes[0].strategy_primary.method == "ocr" assert len(plan2.nodes[0].strategy_fallbacks) == 1 assert plan2.nodes[0].success_condition.method == "title_match" def test_save_load(self): tmpdir = tempfile.mkdtemp() try: plan = ExecutionPlan(plan_id="plan_save", workflow_id="wf_1") plan.nodes.append(ExecutionNode(node_id="n1", action_type="click")) path = plan.save(tmpdir) plan2 = ExecutionPlan.load(str(path)) assert plan2.plan_id == "plan_save" assert len(plan2.nodes) == 1 finally: shutil.rmtree(tmpdir) def test_node_avec_variable(self): node = ExecutionNode( node_id="n1", action_type="type", text="{patient}", variable_name="patient", ) d = node.to_dict() assert d["variable_name"] == "patient" assert d["text"] == "{patient}" def test_node_timeout_et_retry(self): node = ExecutionNode( node_id="n1", action_type="click", timeout_ms=5000, max_retries=3, recovery_action="undo", ) assert node.timeout_ms == 5000 assert node.max_retries == 3 assert node.recovery_action == "undo" # ========================================================================= # ExecutionCompiler — compilation WorkflowIR → ExecutionPlan # ========================================================================= class TestExecutionCompiler: def _make_ir(self): """Créer un WorkflowIR de test.""" ir = WorkflowIR.new("Test workflow", domain="generic") ir.add_step( "Ouvrir le fichier", actions=[ {"type": "click", "target": "bouton Ouvrir", "anchor_hint": "Ouvrir"}, {"type": "wait", "duration_ms": 2000}, ], precondition="L'application est ouverte", postcondition="La fenêtre Ouvrir est affichée", ) ir.add_step( "Saisir le nom", actions=[ {"type": "type", "text": "{nom_fichier}", "variable": True}, {"type": "key_combo", "keys": ["enter"]}, ], ) ir.add_variable("nom_fichier", description="Nom du fichier", default="rapport.pdf") return ir def test_compilation_basique(self): compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) assert plan.workflow_id == ir.workflow_id assert plan.total_nodes == 4 # click + wait + type + key_combo assert plan.domain == "generic" def test_click_a_strategie_resolution(self): """Un clic doit avoir une stratégie primaire et des fallbacks.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) click_nodes = [n for n in plan.nodes if n.action_type == "click"] assert len(click_nodes) == 1 click = click_nodes[0] assert click.strategy_primary is not None assert click.strategy_primary.method in ("ocr", "template", "vlm") assert len(click.strategy_fallbacks) >= 1 def test_ocr_est_prioritaire(self): """Quand du texte est disponible, OCR est la stratégie primaire.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) click = [n for n in plan.nodes if n.action_type == "click"][0] assert click.strategy_primary.method == "ocr" assert click.strategy_primary.target_text == "Ouvrir" def test_vlm_est_fallback(self): """Le VLM est toujours en dernier fallback (exception handler).""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) click = [n for n in plan.nodes if n.action_type == "click"][0] vlm_fallbacks = [f for f in click.strategy_fallbacks if f.method == "vlm"] assert len(vlm_fallbacks) >= 1 def test_type_a_variable(self): """Une action type avec variable a le bon variable_name.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) type_nodes = [n for n in plan.nodes if n.action_type == "type"] assert len(type_nodes) == 1 assert type_nodes[0].variable_name == "nom_fichier" def test_wait_a_duration(self): """Un wait a la bonne durée.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) wait_nodes = [n for n in plan.nodes if n.action_type == "wait"] assert len(wait_nodes) == 1 assert wait_nodes[0].duration_ms == 2000 def test_click_a_recovery(self): """Un clic a une politique de recovery.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) click = [n for n in plan.nodes if n.action_type == "click"][0] assert click.recovery_action in ("escape", "undo", "close", "none") assert click.max_retries >= 1 def test_postcondition_devient_success_condition(self): """La postcondition du WorkflowIR devient la condition de succès du nœud.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) click = [n for n in plan.nodes if n.action_type == "click"][0] assert click.success_condition is not None assert "Ouvrir" in click.success_condition.description def test_statistiques_compilation(self): """Les statistiques de compilation sont correctes.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) assert plan.total_nodes == 4 assert plan.nodes_with_ocr >= 1 assert plan.estimated_duration_s > 0 def test_variables_dans_le_plan(self): """Les variables du WorkflowIR sont dans le plan avec leurs valeurs par défaut.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) assert "nom_fichier" in plan.variables assert plan.variables["nom_fichier"] == "rapport.pdf" def test_params_override_defaults(self): """Les params passés au compile écrasent les valeurs par défaut.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir, params={"nom_fichier": "facture_mars.pdf"}) assert plan.variables["nom_fichier"] == "facture_mars.pdf" def test_plan_json_roundtrip(self): """Compiler → JSON → recharger → même plan.""" compiler = ExecutionCompiler() ir = self._make_ir() plan = compiler.compile(ir) json_str = plan.to_json() plan2 = ExecutionPlan.from_json(json_str) assert plan2.total_nodes == plan.total_nodes assert plan2.workflow_id == plan.workflow_id assert len(plan2.nodes) == len(plan.nodes) def test_compilation_workflow_vide(self): """Un workflow vide produit un plan vide.""" compiler = ExecutionCompiler() ir = WorkflowIR.new("Vide") plan = compiler.compile(ir) assert plan.total_nodes == 0 assert plan.nodes == [] def test_plusieurs_domaines(self): """Le compilateur fonctionne pour différents domaines.""" compiler = ExecutionCompiler() for domain in ["tim_codage", "comptabilite", "rh_paie", "generic"]: ir = WorkflowIR.new("Test", domain=domain) ir.add_step("Action", actions=[{"type": "click", "target": "bouton"}]) plan = compiler.compile(ir) assert plan.domain == domain