feat: SurfaceClassifier + UIAHelper — détection et wrapper Python

SurfaceClassifier — détecte le type d'application au runtime
- 4 surfaces : citrix / windows_native / web_local / unknown
- Paramètres adaptés par surface :
  * Citrix : OCR 0.65, timeouts 15s, retries 3x (compression JPEG tolérée)
  * Windows natif : OCR 0.75, timeouts 8s, UIA bonus si dispo
  * Web : OCR 0.80, timeouts 5s, paramètres rapides
  * Unknown : fallback sûr
- resolve_order() construit la chaîne selon les capacités disponibles
- Détection UIA via health check du helper Rust
- Détection CDP via localhost:9222

UIAHelper — wrapper Python pour lea_uia.exe
- Subprocess + JSON stdin/stdout
- 3 méthodes : query_at(x,y), find_by_name(name,...), capture_focused()
- Fallback silencieux (None) si helper absent, timeout, crash
- Singleton global get_shared_helper()
- Dataclass UiaElement avec center(), is_clickable(), path_signature()

29 nouveaux tests (détection 4 surfaces, dataclass, wrapper, mocks).
485 tests au total, 0 régression.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
2026-04-10 10:54:19 +02:00
parent f85d56ac05
commit ac9c207474
3 changed files with 968 additions and 0 deletions

View File

@@ -0,0 +1,337 @@
# core/workflow/surface_classifier.py
"""
SurfaceClassifier — détecte le type de surface applicative au moment de l'exécution.
4 types de surfaces reconnus :
- citrix : session Citrix/RDP/TSE (wfica32.exe, mstsc.exe, CDViewer.exe)
→ vision pure obligatoire, paramètres tolérants
- windows_native : application Windows native (notepad.exe, explorer.exe, DPI...)
→ vision + UIA bonus, paramètres standards
- web_local : navigateur local (chrome.exe, firefox.exe, msedge.exe)
→ vision + DOM/CDP bonus (si activé), paramètres rapides
- unknown : fallback → vision pure, paramètres par défaut
Le classifier s'exécute UNE SEULE FOIS au début d'une session ou d'un replay.
Son résultat détermine :
1. Quels helpers sont activés (UIA ? CDP ?)
2. Les paramètres de résolution (timeouts, seuils OCR)
3. La stratégie de recovery
Principe : la vision reste le fondement. Le classifier décide juste
des bonus à activer et des paramètres à tuner.
"""
import logging
import os
import platform
import subprocess
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
class SurfaceType(str, Enum):
"""Types de surfaces applicatives."""
CITRIX = "citrix"
WINDOWS_NATIVE = "windows_native"
WEB_LOCAL = "web_local"
UNKNOWN = "unknown"
# Processus connus par type de surface
_CITRIX_PROCESSES = {
"wfica32.exe", # Citrix Workspace (Windows 10+)
"cdviewer.exe", # Citrix Desktop Viewer
"cdviewer.exe",
"mstsc.exe", # Microsoft Remote Desktop
"vmware-vmx.exe", # VMware (cas RDS)
"xen.exe", # Citrix XenApp
"receiver.exe", # Citrix Receiver (ancien)
"selfservice.exe", # Citrix Self-Service Plug-in
}
_BROWSER_PROCESSES = {
"chrome.exe",
"msedge.exe",
"firefox.exe",
"brave.exe",
"opera.exe",
"vivaldi.exe",
}
# Processus système Windows qui ne sont PAS des surfaces applicatives
_SYSTEM_PROCESSES = {
"explorer.exe", # Shell Windows (cas spécial — on le compte comme natif)
"searchhost.exe", # Recherche Windows
"startmenuexperiencehost.exe",
"shellexperiencehost.exe",
"applicationframehost.exe",
}
@dataclass
class SurfaceProfile:
"""Profil complet d'une surface détectée."""
surface_type: SurfaceType
process_name: str = "" # Processus de la fenêtre active
window_title: str = "" # Titre de la fenêtre active
confidence: float = 1.0 # Confiance de la détection (0-1)
# Capacités disponibles
uia_available: bool = False # Le helper UIA peut être utilisé
cdp_available: bool = False # Chrome DevTools Protocol accessible
ocr_available: bool = True # OCR toujours dispo (docTR)
vlm_available: bool = True # VLM toujours dispo (qwen2.5vl)
# Paramètres adaptés à la surface
timeout_click_ms: int = 10000
timeout_resolve_ms: int = 5000
ocr_threshold: float = 0.75
template_threshold: float = 0.85
max_retries: int = 2
retry_delay_ms: int = 2000
# Métadonnées
detected_at: float = 0.0
details: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"surface_type": self.surface_type.value,
"process_name": self.process_name,
"window_title": self.window_title,
"confidence": round(self.confidence, 3),
"capabilities": {
"uia": self.uia_available,
"cdp": self.cdp_available,
"ocr": self.ocr_available,
"vlm": self.vlm_available,
},
"parameters": {
"timeout_click_ms": self.timeout_click_ms,
"timeout_resolve_ms": self.timeout_resolve_ms,
"ocr_threshold": self.ocr_threshold,
"template_threshold": self.template_threshold,
"max_retries": self.max_retries,
"retry_delay_ms": self.retry_delay_ms,
},
"details": self.details,
}
def resolve_order(self) -> List[str]:
"""Construire l'ordre de résolution selon la surface et les capacités."""
order = []
if self.uia_available and self.surface_type == SurfaceType.WINDOWS_NATIVE:
order.append("uia")
if self.cdp_available and self.surface_type == SurfaceType.WEB_LOCAL:
order.append("dom")
order.extend(["ocr", "template", "vlm"])
return order
class SurfaceClassifier:
"""Détecte la surface et configure les paramètres adaptés.
Usage :
classifier = SurfaceClassifier()
profile = classifier.classify(process="notepad.exe", title="Sans titre Bloc-notes")
if profile.uia_available:
# Utiliser lea_uia.exe
"""
def __init__(self, uia_helper_path: str = ""):
"""
Args:
uia_helper_path: Chemin vers lea_uia.exe (optionnel, auto-détection sinon)
"""
self._uia_helper_path = uia_helper_path or self._find_uia_helper()
def _find_uia_helper(self) -> str:
"""Trouver lea_uia.exe dans les emplacements standards."""
candidates = [
r"C:\Lea\helpers\lea_uia.exe",
r".\helpers\lea_uia.exe",
os.path.join(os.path.dirname(__file__), "..", "..", "agent_rust", "lea_uia",
"target", "x86_64-pc-windows-gnu", "release", "lea_uia.exe"),
]
for path in candidates:
if os.path.isfile(path):
return os.path.abspath(path)
return ""
def classify(
self,
process_name: str = "",
window_title: str = "",
screen_info: Optional[Dict] = None,
) -> SurfaceProfile:
"""Classifier une surface depuis le contexte fenêtre.
Args:
process_name: Nom du processus (ex: "notepad.exe")
window_title: Titre de la fenêtre active
screen_info: Infos écran (résolution, DPI, compression détectée)
"""
import time
process_lower = process_name.lower().strip()
title_lower = window_title.lower()
# Détection Citrix — priorité absolue
if process_lower in _CITRIX_PROCESSES:
return self._build_citrix_profile(process_name, window_title, time.time())
# Titre Citrix (ex: "Session Citrix", "Citrix Receiver")
if any(marker in title_lower for marker in ["citrix", "ica session", "rdp session"]):
return self._build_citrix_profile(process_name, window_title, time.time())
# Navigateur
if process_lower in _BROWSER_PROCESSES:
# Cas particulier : navigateur qui contient du Citrix embedded
if "citrix" in title_lower:
return self._build_citrix_profile(process_name, window_title, time.time())
return self._build_web_profile(process_name, window_title, time.time())
# Application Windows native
if process_lower.endswith(".exe") and process_lower not in _SYSTEM_PROCESSES:
return self._build_windows_profile(process_name, window_title, time.time())
# Shell Windows (explorer.exe) — compté comme natif
if process_lower == "explorer.exe":
return self._build_windows_profile(process_name, window_title, time.time())
# Unknown — fallback sûr
return self._build_unknown_profile(process_name, window_title, time.time())
def _build_citrix_profile(self, process: str, title: str, ts: float) -> SurfaceProfile:
"""Profil Citrix — vision pure, paramètres tolérants."""
return SurfaceProfile(
surface_type=SurfaceType.CITRIX,
process_name=process,
window_title=title,
confidence=0.95,
uia_available=False, # UIA n'est pas dispo dans Citrix
cdp_available=False,
ocr_available=True,
vlm_available=True,
# Citrix : compression JPEG, latence, retries agressifs
timeout_click_ms=15000,
timeout_resolve_ms=10000,
ocr_threshold=0.65, # Plus tolérant (compression)
template_threshold=0.75, # Plus tolérant
max_retries=3,
retry_delay_ms=3000,
detected_at=ts,
details={"reason": "citrix_process_or_title"},
)
def _build_windows_profile(self, process: str, title: str, ts: float) -> SurfaceProfile:
"""Profil Windows natif — vision + UIA bonus."""
uia_ok = self._check_uia_available()
return SurfaceProfile(
surface_type=SurfaceType.WINDOWS_NATIVE,
process_name=process,
window_title=title,
confidence=0.9,
uia_available=uia_ok,
cdp_available=False,
ocr_available=True,
vlm_available=True,
timeout_click_ms=8000,
timeout_resolve_ms=5000,
ocr_threshold=0.75,
template_threshold=0.85,
max_retries=2,
retry_delay_ms=2000,
detected_at=ts,
details={
"reason": "native_windows_process",
"uia_helper": self._uia_helper_path if uia_ok else "",
},
)
def _build_web_profile(self, process: str, title: str, ts: float) -> SurfaceProfile:
"""Profil web local — vision (+ CDP plus tard)."""
cdp_ok = self._check_cdp_available()
return SurfaceProfile(
surface_type=SurfaceType.WEB_LOCAL,
process_name=process,
window_title=title,
confidence=0.9,
uia_available=False, # UIA limité pour les navigateurs
cdp_available=cdp_ok,
ocr_available=True,
vlm_available=True,
# Web local : rapide, texte bien rendu
timeout_click_ms=5000,
timeout_resolve_ms=3000,
ocr_threshold=0.80,
template_threshold=0.88,
max_retries=1,
retry_delay_ms=1000,
detected_at=ts,
details={"reason": "browser_process"},
)
def _build_unknown_profile(self, process: str, title: str, ts: float) -> SurfaceProfile:
"""Profil inconnu — paramètres sûrs par défaut."""
return SurfaceProfile(
surface_type=SurfaceType.UNKNOWN,
process_name=process,
window_title=title,
confidence=0.5,
uia_available=False,
cdp_available=False,
ocr_available=True,
vlm_available=True,
timeout_click_ms=10000,
timeout_resolve_ms=5000,
ocr_threshold=0.70,
template_threshold=0.80,
max_retries=2,
retry_delay_ms=2000,
detected_at=ts,
details={"reason": "fallback"},
)
def _check_uia_available(self) -> bool:
"""Vérifier que lea_uia.exe est dispo et fonctionnel.
Sur Windows : appelle `lea_uia.exe health`.
Sur Linux : toujours False (stub).
"""
if platform.system() != "Windows":
return False
if not self._uia_helper_path or not os.path.isfile(self._uia_helper_path):
return False
try:
result = subprocess.run(
[self._uia_helper_path, "health"],
capture_output=True,
text=True,
timeout=5,
)
if result.returncode != 0:
return False
import json
data = json.loads(result.stdout.strip())
return data.get("status") == "ok"
except Exception as e:
logger.debug(f"UIA health check failed: {e}")
return False
def _check_cdp_available(self) -> bool:
"""Vérifier que Chrome DevTools Protocol est accessible.
Teste la présence d'un endpoint CDP sur localhost:9222.
"""
try:
import urllib.request
with urllib.request.urlopen(
"http://localhost:9222/json/version", timeout=1
) as resp:
return resp.status == 200
except Exception:
return False

278
core/workflow/uia_helper.py Normal file
View File

@@ -0,0 +1,278 @@
# core/workflow/uia_helper.py
"""
UIAHelper — Wrapper Python pour lea_uia.exe (helper Rust UI Automation).
Expose une API Python simple pour interroger UIA via le binaire Rust.
Communique via subprocess + stdin/stdout JSON.
Pourquoi un helper Rust ?
- 5-10x plus rapide que pywinauto (10-20ms vs 50-200ms)
- Binaire standalone ~500 Ko, aucune dépendance runtime
- Pas de problèmes de threading COM en Python
- Crash-safe (le crash du helper n'affecte pas l'agent Python)
Architecture :
Python executor
↓ subprocess.run
lea_uia.exe query --x 812 --y 436
↓ UIA API Windows
JSON response
↓ stdout
Python executor parse JSON
Si lea_uia.exe n'est pas disponible (Linux, binaire absent, crash) :
toutes les méthodes retournent None → fallback vision automatique.
"""
import json
import logging
import os
import platform
import subprocess
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
# Timeout par défaut pour les appels UIA (en secondes)
_DEFAULT_TIMEOUT = 5.0
@dataclass
class UiaElement:
"""Représentation Python d'un élément UIA."""
name: str = ""
control_type: str = ""
class_name: str = ""
automation_id: str = ""
bounding_rect: Tuple[int, int, int, int] = (0, 0, 0, 0)
is_enabled: bool = False
is_offscreen: bool = True
parent_path: List[Dict[str, str]] = field(default_factory=list)
process_name: str = ""
def center(self) -> Tuple[int, int]:
"""Retourner le centre du rectangle (pixels)."""
x1, y1, x2, y2 = self.bounding_rect
return ((x1 + x2) // 2, (y1 + y2) // 2)
def width(self) -> int:
return self.bounding_rect[2] - self.bounding_rect[0]
def height(self) -> int:
return self.bounding_rect[3] - self.bounding_rect[1]
def is_clickable(self) -> bool:
"""Peut-on cliquer dessus ?"""
return (
self.is_enabled
and not self.is_offscreen
and self.width() > 0
and self.height() > 0
)
def path_signature(self) -> str:
"""Signature du chemin parent (pour retrouver l'élément)."""
parts = [f"{p['control_type']}[{p['name']}]" for p in self.parent_path if p.get("name")]
parts.append(f"{self.control_type}[{self.name}]")
return " > ".join(parts)
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"control_type": self.control_type,
"class_name": self.class_name,
"automation_id": self.automation_id,
"bounding_rect": list(self.bounding_rect),
"is_enabled": self.is_enabled,
"is_offscreen": self.is_offscreen,
"parent_path": self.parent_path,
"process_name": self.process_name,
}
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "UiaElement":
rect = d.get("bounding_rect", [0, 0, 0, 0])
if isinstance(rect, list) and len(rect) >= 4:
rect = tuple(rect[:4])
else:
rect = (0, 0, 0, 0)
return cls(
name=d.get("name", ""),
control_type=d.get("control_type", ""),
class_name=d.get("class_name", ""),
automation_id=d.get("automation_id", ""),
bounding_rect=rect,
is_enabled=d.get("is_enabled", False),
is_offscreen=d.get("is_offscreen", True),
parent_path=d.get("parent_path", []),
process_name=d.get("process_name", ""),
)
class UIAHelper:
"""Wrapper Python pour lea_uia.exe."""
def __init__(self, helper_path: str = "", timeout: float = _DEFAULT_TIMEOUT):
self._helper_path = helper_path or self._find_helper()
self._timeout = timeout
self._available = self._check_available()
def _find_helper(self) -> str:
"""Trouver lea_uia.exe dans les emplacements standards."""
candidates = [
r"C:\Lea\helpers\lea_uia.exe",
os.path.join(os.path.dirname(__file__), "..", "..",
"agent_rust", "lea_uia", "target",
"x86_64-pc-windows-gnu", "release", "lea_uia.exe"),
"./helpers/lea_uia.exe",
"lea_uia.exe",
]
for path in candidates:
if os.path.isfile(path):
return os.path.abspath(path)
return ""
def _check_available(self) -> bool:
"""Vérifier que le helper est utilisable (Windows + binaire + health OK)."""
if platform.system() != "Windows":
logger.debug("UIAHelper: Linux/Mac — helper désactivé")
return False
if not self._helper_path:
logger.debug("UIAHelper: lea_uia.exe introuvable")
return False
if not os.path.isfile(self._helper_path):
logger.debug(f"UIAHelper: chemin invalide {self._helper_path}")
return False
return True
@property
def available(self) -> bool:
return self._available
@property
def helper_path(self) -> str:
return self._helper_path
def _run(self, args: List[str]) -> Optional[Dict[str, Any]]:
"""Exécuter lea_uia.exe avec les arguments et parser le JSON."""
if not self._available:
return None
try:
result = subprocess.run(
[self._helper_path] + args,
capture_output=True,
text=True,
timeout=self._timeout,
encoding="utf-8",
errors="replace",
)
if result.returncode != 0:
logger.debug(
f"UIAHelper: exit code {result.returncode}, "
f"stderr: {result.stderr[:200]}"
)
return None
output = result.stdout.strip()
if not output:
return None
return json.loads(output)
except subprocess.TimeoutExpired:
logger.debug(f"UIAHelper: timeout ({self._timeout}s) sur {args}")
return None
except json.JSONDecodeError as e:
logger.debug(f"UIAHelper: JSON invalide — {e}")
return None
except Exception as e:
logger.debug(f"UIAHelper: erreur {e}")
return None
def health(self) -> bool:
"""Vérifier que UIA répond."""
data = self._run(["health"])
return data is not None and data.get("status") == "ok"
def query_at(
self,
x: int,
y: int,
with_parents: bool = True,
) -> Optional[UiaElement]:
"""Récupérer l'élément UIA à une position écran.
Args:
x, y: Coordonnées pixel absolues
with_parents: Inclure la hiérarchie des parents
Returns:
UiaElement si trouvé, None sinon (pas d'élément ou UIA indispo)
"""
args = ["query", "--x", str(x), "--y", str(y)]
if not with_parents:
args.append("--with-parents=false")
data = self._run(args)
if not data or data.get("status") != "ok":
return None
elem_data = data.get("element")
if not elem_data:
return None
return UiaElement.from_dict(elem_data)
def find_by_name(
self,
name: str,
control_type: Optional[str] = None,
automation_id: Optional[str] = None,
window: Optional[str] = None,
timeout_ms: int = 2000,
) -> Optional[UiaElement]:
"""Rechercher un élément par son nom (+ filtres optionnels).
Args:
name: Nom exact de l'élément
control_type: Type de contrôle (Button, Edit, MenuItem...)
automation_id: ID d'automation
window: Restreindre à une fenêtre spécifique
timeout_ms: Timeout de recherche en millisecondes
"""
args = ["find", "--name", name, "--timeout-ms", str(timeout_ms)]
if control_type:
args.extend(["--control-type", control_type])
if automation_id:
args.extend(["--automation-id", automation_id])
if window:
args.extend(["--window", window])
data = self._run(args)
if not data or data.get("status") != "ok":
return None
elem_data = data.get("element")
if not elem_data:
return None
return UiaElement.from_dict(elem_data)
def capture_focused(self, max_depth: int = 3) -> Optional[UiaElement]:
"""Capturer l'élément ayant le focus + son contexte."""
data = self._run(["capture", "--max-depth", str(max_depth)])
if not data or data.get("status") != "ok":
return None
elem_data = data.get("element")
if not elem_data:
return None
return UiaElement.from_dict(elem_data)
# Instance globale partagée (singleton léger)
_SHARED_HELPER: Optional[UIAHelper] = None
def get_shared_helper() -> UIAHelper:
"""Retourner une instance partagée de UIAHelper."""
global _SHARED_HELPER
if _SHARED_HELPER is None:
_SHARED_HELPER = UIAHelper()
return _SHARED_HELPER

View File

@@ -0,0 +1,353 @@
"""
Tests du SurfaceClassifier et du UIAHelper.
Vérifie :
- Détection correcte des 4 types de surfaces (citrix, windows_native, web, unknown)
- Paramètres adaptés par surface (timeouts, seuils)
- Fallback gracieux si helper UIA absent
- Sérialisation des profils
- Wrapper UIAHelper avec mocks subprocess
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
from core.workflow.surface_classifier import (
SurfaceClassifier,
SurfaceProfile,
SurfaceType,
)
from core.workflow.uia_helper import UIAHelper, UiaElement
# =========================================================================
# SurfaceClassifier — détection par processus
# =========================================================================
class TestSurfaceClassifier:
"""Tests de détection des surfaces."""
def _classifier(self):
"""Classifier sans helper UIA (pour que les tests soient reproductibles)."""
return SurfaceClassifier(uia_helper_path="")
def test_detection_citrix_wfica(self):
"""wfica32.exe → Citrix."""
c = self._classifier()
profile = c.classify(process_name="wfica32.exe", window_title="Session Citrix")
assert profile.surface_type == SurfaceType.CITRIX
assert profile.uia_available is False
assert profile.ocr_threshold < 0.75 # Plus tolérant
assert profile.max_retries >= 3
def test_detection_citrix_mstsc(self):
"""mstsc.exe → Citrix (RDP)."""
c = self._classifier()
profile = c.classify(process_name="mstsc.exe", window_title="Remote Desktop")
assert profile.surface_type == SurfaceType.CITRIX
def test_detection_citrix_par_titre(self):
"""Titre 'Citrix' → Citrix même si process non listé."""
c = self._classifier()
profile = c.classify(process_name="chrome.exe", window_title="DxCare - Citrix Receiver")
assert profile.surface_type == SurfaceType.CITRIX
def test_detection_windows_natif_notepad(self):
"""notepad.exe → Windows natif."""
c = self._classifier()
profile = c.classify(process_name="notepad.exe", window_title="Sans titre Bloc-notes")
assert profile.surface_type == SurfaceType.WINDOWS_NATIVE
assert profile.ocr_threshold == 0.75
assert profile.timeout_click_ms == 8000
def test_detection_windows_natif_explorer(self):
"""explorer.exe → Windows natif (cas spécial)."""
c = self._classifier()
profile = c.classify(process_name="explorer.exe", window_title="Lea")
assert profile.surface_type == SurfaceType.WINDOWS_NATIVE
def test_detection_windows_natif_dxcare(self):
"""dxcare.exe (DPI hospitalier) → Windows natif."""
c = self._classifier()
profile = c.classify(process_name="dxcare.exe", window_title="DxCare - Dossier 12345")
assert profile.surface_type == SurfaceType.WINDOWS_NATIVE
def test_detection_web_chrome(self):
"""chrome.exe → Web local."""
c = self._classifier()
profile = c.classify(process_name="chrome.exe", window_title="Google - Google Chrome")
assert profile.surface_type == SurfaceType.WEB_LOCAL
assert profile.ocr_threshold == 0.80 # Plus strict (texte bien rendu)
assert profile.max_retries == 1 # Rapide
def test_detection_web_edge(self):
c = self._classifier()
profile = c.classify(process_name="msedge.exe", window_title="Edge")
assert profile.surface_type == SurfaceType.WEB_LOCAL
def test_detection_web_firefox(self):
c = self._classifier()
profile = c.classify(process_name="firefox.exe", window_title="Firefox")
assert profile.surface_type == SurfaceType.WEB_LOCAL
def test_detection_unknown_fallback(self):
"""Process non reconnu → unknown avec paramètres sûrs."""
c = self._classifier()
profile = c.classify(process_name="", window_title="")
assert profile.surface_type == SurfaceType.UNKNOWN
assert profile.confidence < 1.0
assert profile.ocr_available is True # OCR toujours dispo
def test_citrix_dans_navigateur(self):
"""Citrix embedded dans Chrome → Citrix."""
c = self._classifier()
profile = c.classify(
process_name="chrome.exe",
window_title="Citrix Workspace - DxCare",
)
assert profile.surface_type == SurfaceType.CITRIX
def test_resolve_order_par_surface(self):
"""Ordre de résolution cohérent avec la surface."""
c = self._classifier()
citrix = c.classify("wfica32.exe", "Session")
assert "uia" not in citrix.resolve_order()
assert "ocr" in citrix.resolve_order()
windows = c.classify("notepad.exe", "Bloc-notes")
# UIA pas dispo (helper path vide) donc absent
assert "ocr" in windows.resolve_order()
web = c.classify("chrome.exe", "Google")
assert "ocr" in web.resolve_order()
class TestSurfaceProfile:
"""Tests du dataclass SurfaceProfile."""
def test_to_dict_structure(self):
p = SurfaceProfile(
surface_type=SurfaceType.WINDOWS_NATIVE,
process_name="notepad.exe",
window_title="Test",
)
d = p.to_dict()
assert d["surface_type"] == "windows_native"
assert "capabilities" in d
assert "parameters" in d
assert d["capabilities"]["ocr"] is True
assert d["capabilities"]["uia"] is False # Par défaut
def test_resolve_order_construction(self):
"""L'ordre de résolution utilise les capacités dispo."""
p = SurfaceProfile(
surface_type=SurfaceType.WINDOWS_NATIVE,
uia_available=True,
)
order = p.resolve_order()
assert order[0] == "uia" # UIA en premier si dispo
assert "ocr" in order
assert "vlm" in order
def test_resolve_order_sans_uia(self):
p = SurfaceProfile(
surface_type=SurfaceType.CITRIX,
uia_available=False,
)
order = p.resolve_order()
assert "uia" not in order
assert order[0] == "ocr" # OCR en premier
# =========================================================================
# UIAHelper — wrapper Python
# =========================================================================
class TestUIAHelper:
"""Tests du wrapper UIAHelper."""
def test_initialization_sans_helper(self):
"""Sans helper trouvé, available = False."""
helper = UIAHelper(helper_path="/chemin/inexistant.exe")
assert helper.available is False
def test_query_retourne_none_si_indispo(self):
"""Si le helper n'est pas dispo, query retourne None."""
helper = UIAHelper(helper_path="/chemin/inexistant.exe")
result = helper.query_at(100, 200)
assert result is None
def test_find_retourne_none_si_indispo(self):
helper = UIAHelper(helper_path="/chemin/inexistant.exe")
result = helper.find_by_name("Enregistrer")
assert result is None
def test_health_retourne_false_si_indispo(self):
helper = UIAHelper(helper_path="/chemin/inexistant.exe")
assert helper.health() is False
@patch("core.workflow.uia_helper.os.path.isfile", return_value=True)
@patch("core.workflow.uia_helper.platform.system", return_value="Windows")
@patch("core.workflow.uia_helper.subprocess.run")
def test_query_success_mock(self, mock_run, mock_platform, mock_isfile):
"""Query avec mock subprocess retourne un UiaElement."""
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = json.dumps({
"status": "ok",
"element": {
"name": "Enregistrer",
"control_type": "bouton",
"class_name": "Button",
"automation_id": "btnSave",
"bounding_rect": [100, 200, 200, 250],
"is_enabled": True,
"is_offscreen": False,
"parent_path": [
{"name": "Bloc-notes", "control_type": "fenêtre"}
],
},
"elapsed_ms": 15,
})
mock_run.return_value = mock_result
helper = UIAHelper(helper_path="fake_lea_uia.exe")
element = helper.query_at(150, 225)
assert element is not None
assert element.name == "Enregistrer"
assert element.control_type == "bouton"
assert element.bounding_rect == (100, 200, 200, 250)
assert element.center() == (150, 225)
assert element.is_clickable() is True
assert len(element.parent_path) == 1
@patch("core.workflow.uia_helper.os.path.isfile", return_value=True)
@patch("core.workflow.uia_helper.platform.system", return_value="Windows")
@patch("core.workflow.uia_helper.subprocess.run")
def test_find_success_mock(self, mock_run, mock_platform, mock_isfile):
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = json.dumps({
"status": "ok",
"element": {
"name": "Fichier",
"control_type": "menu",
"bounding_rect": [0, 20, 50, 40],
"is_enabled": True,
"is_offscreen": False,
},
})
mock_run.return_value = mock_result
helper = UIAHelper(helper_path="fake.exe")
element = helper.find_by_name("Fichier", control_type="menu")
assert element is not None
assert element.name == "Fichier"
@patch("core.workflow.uia_helper.os.path.isfile", return_value=True)
@patch("core.workflow.uia_helper.platform.system", return_value="Windows")
@patch("core.workflow.uia_helper.subprocess.run")
def test_not_found(self, mock_run, mock_platform, mock_isfile):
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = json.dumps({
"status": "not_found",
"reason": "Pas d'élément",
"elapsed_ms": 5,
})
mock_run.return_value = mock_result
helper = UIAHelper(helper_path="fake.exe")
assert helper.query_at(999, 999) is None
@patch("core.workflow.uia_helper.os.path.isfile", return_value=True)
@patch("core.workflow.uia_helper.platform.system", return_value="Windows")
@patch("core.workflow.uia_helper.subprocess.run")
def test_timeout(self, mock_run, mock_platform, mock_isfile):
"""Un timeout subprocess ne fait pas crash le helper."""
import subprocess as sp
mock_run.side_effect = sp.TimeoutExpired("lea_uia", 5)
helper = UIAHelper(helper_path="fake.exe")
assert helper.query_at(100, 100) is None
@patch("core.workflow.uia_helper.os.path.isfile", return_value=True)
@patch("core.workflow.uia_helper.platform.system", return_value="Windows")
@patch("core.workflow.uia_helper.subprocess.run")
def test_json_invalide(self, mock_run, mock_platform, mock_isfile):
"""Une sortie non-JSON ne fait pas crash."""
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "pas du JSON"
mock_run.return_value = mock_result
helper = UIAHelper(helper_path="fake.exe")
assert helper.query_at(100, 100) is None
class TestUiaElement:
"""Tests du dataclass UiaElement."""
def test_from_dict_minimal(self):
e = UiaElement.from_dict({"name": "test"})
assert e.name == "test"
assert e.bounding_rect == (0, 0, 0, 0)
def test_center(self):
e = UiaElement(bounding_rect=(100, 200, 200, 300))
assert e.center() == (150, 250)
def test_is_clickable(self):
e = UiaElement(
bounding_rect=(100, 100, 200, 150),
is_enabled=True,
is_offscreen=False,
)
assert e.is_clickable() is True
e2 = UiaElement(
bounding_rect=(100, 100, 200, 150),
is_enabled=False,
is_offscreen=False,
)
assert e2.is_clickable() is False
def test_path_signature(self):
e = UiaElement(
name="Enregistrer",
control_type="bouton",
parent_path=[
{"name": "Bloc-notes", "control_type": "fenêtre"},
{"name": "Fichier", "control_type": "menu"},
],
)
sig = e.path_signature()
assert "Bloc-notes" in sig
assert "Enregistrer" in sig
assert " > " in sig
def test_roundtrip_dict(self):
original = UiaElement(
name="test",
control_type="bouton",
bounding_rect=(10, 20, 30, 40),
is_enabled=True,
is_offscreen=False,
)
d = original.to_dict()
copy = UiaElement.from_dict(d)
assert copy.name == original.name
assert copy.bounding_rect == original.bounding_rect
assert copy.is_enabled == original.is_enabled