Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b808e48b1f6d308d4ac1ce5e04b6b04696f385ab
rpa_vision_v3/tests/unit/test_v4_wiring.py
Dom 332366b58c feat: câblage complet V4 — stratégie UIA + surface profile 
Pipeline V4 câblé de bout en bout :
  RawTrace (avec uia_snapshot) → IRBuilder → Action._enrichment
  WorkflowIR → ExecutionCompiler (avec SurfaceProfile) → ExecutionPlan
  ExecutionPlan → runner → target_spec (avec uia_target + resolve_order)

ResolutionStrategy étendu :
- Champs UIA : uia_name, uia_control_type, uia_automation_id, uia_parent_path
- Champs DOM : dom_selector, dom_xpath, dom_url_pattern (préparation web)

ExecutionCompiler.compile(surface_profile=...) :
- Timeouts/retries tirés du profil (citrix=15s/3x, web=5s/1x, natif=8s/2x)
- UIA primaire seulement si surface=WINDOWS_NATIVE et uia_available
- Citrix ignore UIA même si snapshot présent (UIA ne marche pas dans Citrix)

IRBuilder lit evt['uia_snapshot'] et le stocke dans action._enrichment
(à remplir par l'agent Windows pendant l'enregistrement via lea_uia.exe)

execution_plan_runner propage uia_target et dom_target dans target_spec
pour que l'agent Windows puisse les consommer au runtime.

11 tests de câblage E2E :
- Profils (Citrix/web/natif) imposent bien les timeouts
- Stratégie UIA créée quand snapshot+surface OK
- Stratégie UIA bloquée sur Citrix
- IRBuilder propage uia_snapshot
- Runner produit target_spec avec uia_target + resolve_order=['uia', 'ocr', 'vlm']

496 tests au total, 0 régression.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-10 11:02:51 +02:00

350 lines
13 KiB
Python
Raw Blame History

"""
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
Reference in New Issue View Git Blame Copy Permalink