Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(p1): persist workflows and semantic learning artifacts

Browse Source
This commit is contained in:
Dom
2026-06-02 16:20:38 +02:00
parent 7a1a5cb6fd
commit 86b3c8f7e7
21 changed files with 3816 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
tests/unit/test_clip_embedder_device_fix.py Normal file
View File

@@ -0,0 +1,108 @@
"""Tests de non-régression pour le fix UnboundLocalError sur 'torch'.
Cas couvert : appel `CLIPEmbedder(device="cpu")` explicite — le if `device is
None` n'était pas pris, donc l'import local `torch` n'était pas exécuté, mais
Python avait quand même noté `torch` comme local au scope `__init__`, faisant
planter `with torch.no_grad():` plus bas en UnboundLocalError.
Référence : inbox_codex/2026-05-25_1235_..._enquete-feedbackbus-5004.md
Fix : core/embedding/clip_embedder.py l. 60-77 (import local supprimé).
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
@pytest.mark.unit
def test_clip_embedder_init_no_local_torch_shadow():
"""Le source de CLIPEmbedder.__init__ ne contient plus 'import torch' à
l'intérieur du `if device is None:` (qui shadowait le torch module-level)."""
import inspect
from core.embedding import clip_embedder
src = inspect.getsource(clip_embedder.CLIPEmbedder.__init__)
# Tolérance : on accepte qu'un commentaire mentionne `import torch`,
# mais pas une vraie ligne d'instruction.
code_lines = [
line for line in src.splitlines()
if line.strip() and not line.strip().startswith("#")
]
code_only = "\n".join(code_lines)
# On ne doit plus avoir un import torch indenté au-delà du module-level.
# (l'import existe au top du fichier l. 8, pas dans __init__).
assert " import torch" not in code_only, (
"import torch local trouvé dans __init__ — il faut utiliser le torch "
"du scope module (l. 8 du fichier) pour éviter UnboundLocalError "
"quand l'appelant passe device='cpu'."
)
@pytest.mark.unit
def test_clip_embedder_module_imports_torch():
"""Le module clip_embedder doit avoir `import torch` au scope module
pour que les autres méthodes (embed_image, embed_text) puissent l'utiliser."""
import core.embedding.clip_embedder as ce
assert hasattr(ce, "torch"), (
"Le module clip_embedder doit exposer `torch` au scope module."
)
@pytest.mark.unit
def test_clip_embedder_handles_device_cpu_without_unbound_local(monkeypatch):
"""Reproduit le cas qui plantait : on appelle l'init avec device='cpu'.
Avant fix : UnboundLocalError sur `torch` au moment de `torch.no_grad()`.
Après fix : l'init doit échouer proprement sur l'absence éventuelle de
open_clip ou de poids, mais PAS sur UnboundLocalError.
On mocke open_clip et torch.no_grad pour ne pas charger un vrai modèle.
"""
import types
from core.embedding import clip_embedder
# Mock open_clip pour éviter le download
fake_open_clip = types.SimpleNamespace(
create_model_and_transforms=lambda *a, **kw: (
types.SimpleNamespace(
eval=lambda: None,
encode_image=lambda x: type("T", (), {"shape": (1, 512)})(),
),
None,
lambda img: img,
),
get_tokenizer=lambda name: lambda t: None,
)
monkeypatch.setattr(clip_embedder, "open_clip", fake_open_clip)
# Mock torch.no_grad et torch.zeros pour court-circuiter le dummy embed
class _FakeCtx:
def __enter__(self): return self
def __exit__(self, *a): return False
fake_zeros = lambda *args, **kwargs: type("Z", (), {"to": lambda self, d: self})()
monkeypatch.setattr(clip_embedder.torch, "no_grad", lambda: _FakeCtx())
monkeypatch.setattr(clip_embedder.torch, "zeros", fake_zeros)
# Appel direct avec device="cpu" — ne doit PAS lever UnboundLocalError.
# Peut échouer pour autre raison (ex. encode_image), on isole uniquement
# le bug torch unbound.
try:
embedder = clip_embedder.CLIPEmbedder(device="cpu")
except RuntimeError as e:
msg = str(e)
assert "cannot access local variable 'torch'" not in msg, (
f"UnboundLocalError torch toujours présent : {msg}"
)
# Autre erreur acceptée (mock incomplet)
pytest.skip(f"Mock incomplet, mais bug torch absent : {msg}")
except UnboundLocalError as e:
pytest.fail(f"Bug torch toujours présent : {e}")
# Si on arrive ici, init a réussi sans bug torch

234
tests/unit/test_competence_persist.py Normal file
View File

@@ -0,0 +1,234 @@
"""Tests unit pour core.competences.persist (helpers /persist endpoint).
Specs : docs/POC/SPECS_ENDPOINT_PERSIST_2026-06-01.md
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
import yaml
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
from core.competences import persist as P # noqa: E402
# ---------------------------------------------------------------------------
# slugify
# ---------------------------------------------------------------------------
class TestSlugify:
def test_slug_generation_normal(self):
assert P.slugify("Saisir Texte Word") == "saisir_texte_word"
def test_slug_generation_with_accents(self):
assert P.slugify("Créer Compte Patient") == "creer_compte_patient"
def test_slug_generation_too_short(self):
with pytest.raises(ValueError):
P.slugify("ab")
def test_slug_generation_empty(self):
with pytest.raises(ValueError):
P.slugify("")
def test_slug_max_80_chars(self):
long_name = "a" * 200
slug = P.slugify(long_name)
assert len(slug) <= 80
def test_slug_strips_special_chars(self):
# Cas tordu : "tab" est interdit ('\t'), donc on injecte du bruit
assert P.slugify("hello!! world??") == "hello_world"
def test_slug_starts_with_letter(self):
slug = P.slugify("123 abc def")
assert slug.startswith("c_") # prefix auto pour commencer par lettre
# ---------------------------------------------------------------------------
# PII detection
# ---------------------------------------------------------------------------
class TestPiiDetection:
def test_pii_email_detected(self):
matches = P.detect_pii({"intent": "envoyer mail a john.doe@example.com"})
assert matches # au moins un pattern
def test_pii_phone_detected(self):
matches = P.detect_pii({"steps": [{"value": "tel 01 23 45 67 89"}]})
assert matches
def test_no_pii_clean_payload(self):
clean = {"steps": [{"kind": "click", "target": "Bouton Valider"}]}
assert P.detect_pii(clean) == []
def test_pii_recursive_in_nested_list(self):
nested = {"a": {"b": [{"c": "email: x@y.fr"}]}}
assert P.detect_pii(nested)
# ---------------------------------------------------------------------------
# Atomic write
# ---------------------------------------------------------------------------
class TestAtomicWrite:
def test_atomic_write_then_rename(self, tmp_path):
target = tmp_path / "demo.yaml"
data = {"id": "demo", "name": "Demo"}
result = P.atomic_write_yaml(target, data, persist_id="pid-1")
assert result == target
assert target.exists()
# Pas de .tmp residuel
leftovers = list(tmp_path.glob(".*.tmp.*"))
assert leftovers == []
loaded = yaml.safe_load(target.read_text(encoding="utf-8"))
assert loaded["id"] == "demo"
def test_atomic_write_cleans_tmp_on_failure(self, tmp_path, monkeypatch):
target = tmp_path / "demo.yaml"
# Forcer un echec sur os.rename
import os as _os
original_rename = _os.rename
def boom(*a, **k):
raise OSError("disk full simulated")
monkeypatch.setattr(_os, "rename", boom)
with pytest.raises(OSError):
P.atomic_write_yaml(target, {"id": "demo"}, persist_id="pid-2")
monkeypatch.setattr(_os, "rename", original_rename)
# Le .tmp doit avoir ete nettoye
leftovers = list(tmp_path.glob(".*.tmp.*"))
assert leftovers == []
# ---------------------------------------------------------------------------
# Audit append
# ---------------------------------------------------------------------------
class TestAuditAppend:
def test_audit_append_monotonic_ids(self, tmp_path):
audit = tmp_path / "persist_audit.jsonl"
id1 = P.audit_append({"persist_id": "p1", "competence_id": "c1"}, audit_path=audit)
id2 = P.audit_append({"persist_id": "p2", "competence_id": "c2"}, audit_path=audit)
assert id1 == 1
assert id2 == 2
def test_audit_append_includes_timestamp(self, tmp_path):
audit = tmp_path / "audit.jsonl"
P.audit_append({"persist_id": "p1", "competence_id": "c1"}, audit_path=audit)
lines = audit.read_text(encoding="utf-8").strip().splitlines()
record = json.loads(lines[0])
assert "timestamp" in record
assert record["audit_entry_id"] == 1
def test_find_existing_audit_entry(self, tmp_path):
audit = tmp_path / "audit.jsonl"
P.audit_append(
{"persist_id": "p-uniq", "competence_id": "c1"},
audit_path=audit,
)
found = P.find_existing_audit_entry("p-uniq", audit_path=audit)
assert found is not None
assert found["competence_id"] == "c1"
assert P.find_existing_audit_entry("p-not-here", audit_path=audit) is None
# ---------------------------------------------------------------------------
# YAML schema build + validate
# ---------------------------------------------------------------------------
class TestBuildYaml:
def test_yaml_schema_required_fields_present(self):
body = P.build_competence_yaml(
slug="demo_test",
name="Demo Test",
workflow_ir={"steps": [{"kind": "click"}], "preconditions": []},
parameters=[{"name": "x", "type": "string", "required": True}],
intent_fr="faire demo",
learning_state="candidate",
session_id="sess1",
machine_id="machine1",
)
missing = P.validate_yaml_schema(body)
assert missing == [], f"champs manquants : {missing}"
def test_payload_stable_forced_to_candidate_via_helper(self):
# Le forcage stable -> candidate est fait dans le handler, mais on
# peut au moins verifier que build accepte le learning_state passe.
body = P.build_competence_yaml(
slug="demo_test_2",
name="Demo 2",
workflow_ir={"steps": [{"kind": "click"}]},
parameters=None,
intent_fr="demo",
learning_state="candidate",
session_id=None,
machine_id=None,
)
assert body["learning_state"] == "candidate"
# ---------------------------------------------------------------------------
# Cross-state collision
# ---------------------------------------------------------------------------
class TestCrossStateCollision:
def test_no_collision_returns_none(self, tmp_path):
root = tmp_path / "competences"
(root / "candidate").mkdir(parents=True)
assert P.detect_cross_state_collision("xyz", competences_root=root) is None
def test_collision_in_candidate_returns_dirname(self, tmp_path):
root = tmp_path / "competences"
(root / "candidate").mkdir(parents=True)
(root / "candidate" / "xyz.yaml").write_text("id: xyz\n", encoding="utf-8")
assert P.detect_cross_state_collision("xyz", competences_root=root) == "candidate"
def test_collision_in_stable_returns_dirname(self, tmp_path):
root = tmp_path / "competences"
(root / "stable").mkdir(parents=True)
(root / "stable" / "abc.yaml").write_text("id: abc\n", encoding="utf-8")
assert P.detect_cross_state_collision("abc", competences_root=root) == "stable"
# ---------------------------------------------------------------------------
# Rate limiter
# ---------------------------------------------------------------------------
class TestRateLimiter:
def test_below_limit_allowed(self):
lim = P.PersistRateLimiter(max_per_minute=3)
for _ in range(3):
allowed, _ = lim.allow("m1")
assert allowed
def test_above_limit_blocked(self):
lim = P.PersistRateLimiter(max_per_minute=2)
lim.allow("m1")
lim.allow("m1")
allowed, retry = lim.allow("m1")
assert not allowed
assert retry >= 1
def test_per_machine_isolation(self):
lim = P.PersistRateLimiter(max_per_minute=1)
a1, _ = lim.allow("m1")
a2, _ = lim.allow("m2")
assert a1 and a2

450
tests/unit/test_phase25_semantic.py Normal file
View File

@@ -0,0 +1,450 @@
"""Tests unitaires pour ``core.semantic.phase25_analyzer``.
Specs : ``docs/POC/SPECS_PHASE_25_SEMANTIQUE_2026-06-01.md``.
Couverture obligatoire :
- Hash perceptuel + grouping (Hamming threshold).
- Cap 10 écrans -> too_complex.
- Fallback OCR-seul si OmniParser KO (mock exception).
- Génération .semantic.yaml valide avec ``degraded`` correctement positionné.
- Validation session_id / slug (anti path-traversal).
"""
from __future__ import annotations
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
import yaml
from PIL import Image, ImageDraw
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
from core.semantic import phase25_analyzer as P # noqa: E402
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_image(size=(256, 256), color=(255, 255, 255), text=None):
img = Image.new("RGB", size, color=color)
if text:
draw = ImageDraw.Draw(img)
draw.text((10, 10), text, fill=(0, 0, 0))
return img
@pytest.fixture
def fake_omniparser_ok():
"""Wrapper OmniParser qui retourne des éléments factices sans erreur."""
w = P._OmniParserSafeWrapper.__new__(P._OmniParserSafeWrapper)
w._adapter = MagicMock()
w._available = True
w._import_error = None
def _fake_detect(image):
return [
{"label": "Valider", "bbox": [10, 20, 100, 50], "confidence": 0.9, "element_type": "button"},
{"label": "Nom patient", "bbox": [120, 20, 300, 60], "confidence": 0.85, "element_type": "input"},
{"label": "MOREL Catherine", "bbox": [120, 80, 300, 100], "confidence": 0.7, "element_type": "text"},
]
w._adapter.detect.side_effect = _fake_detect
return w
@pytest.fixture
def fake_omniparser_raising():
"""Wrapper OmniParser disponible qui lève une exception à chaque detect."""
w = P._OmniParserSafeWrapper.__new__(P._OmniParserSafeWrapper)
w._adapter = MagicMock()
w._available = True
w._import_error = None
w._adapter.detect.side_effect = RuntimeError("OmniParser corrupted weights")
return w
@pytest.fixture
def fake_omniparser_unavailable():
"""Wrapper OmniParser indisponible (adapter absent)."""
w = P._OmniParserSafeWrapper.__new__(P._OmniParserSafeWrapper)
w._adapter = None
w._available = False
w._import_error = "ImportError: No module named 'OmniParser'"
return w
# ---------------------------------------------------------------------------
# Tests : validation session_id / slug
# ---------------------------------------------------------------------------
class TestValidation:
def test_session_id_valid(self):
assert P._validate_session_id("abc-123_XYZ") == "abc-123_XYZ"
def test_session_id_empty_raises(self):
with pytest.raises(ValueError):
P._validate_session_id("")
def test_session_id_path_traversal_raises(self):
with pytest.raises(ValueError):
P._validate_session_id("../etc/passwd")
def test_session_id_with_slash_raises(self):
with pytest.raises(ValueError):
P._validate_session_id("abc/def")
def test_session_id_type_raises(self):
with pytest.raises(ValueError):
P._validate_session_id(None)
def test_slug_valid(self):
assert P._validate_slug("facturation_urgence") == "facturation_urgence"
def test_slug_too_short(self):
with pytest.raises(ValueError):
P._validate_slug("ab")
def test_slug_starts_with_digit(self):
with pytest.raises(ValueError):
P._validate_slug("123_abc")
# ---------------------------------------------------------------------------
# Tests : phash et grouping
# ---------------------------------------------------------------------------
class TestPerceptualHash:
def test_compute_phash_returns_str(self):
img = _make_image()
h = P.compute_phash(img)
assert isinstance(h, str) and len(h) > 0
def test_identical_images_same_phash(self):
img1 = _make_image(color=(255, 255, 255))
img2 = _make_image(color=(255, 255, 255))
assert P.compute_phash(img1) == P.compute_phash(img2)
def _noise_image(self, seed: int):
"""Image avec un motif différent par seed (forme + position)."""
import random
rng = random.Random(seed)
img = _make_image(color=(255, 255, 255))
d = ImageDraw.Draw(img)
for _ in range(40):
x = rng.randint(0, 240)
y = rng.randint(0, 240)
w = rng.randint(20, 60)
h = rng.randint(20, 60)
col = (rng.randint(0, 255), rng.randint(0, 255), rng.randint(0, 255))
d.rectangle([x, y, x + w, y + h], fill=col)
return img
def test_different_images_different_phash(self):
img1 = self._noise_image(seed=1)
img2 = self._noise_image(seed=999)
h1 = P.compute_phash(img1)
h2 = P.compute_phash(img2)
if h1.startswith("md5:") or h2.startswith("md5:"):
assert h1 != h2
else:
# Bruits différents -> distance largement > seuil.
assert P._hamming_distance(h1, h2) > P.PHASH_HAMMING_THRESHOLD
def test_identify_distinct_screens_groups_identicals(self):
img_a1 = self._noise_image(seed=42)
img_a2 = self._noise_image(seed=42) # même seed = même image = même phash
img_b = self._noise_image(seed=1337)
frames = [(0, img_a1), (1, img_a2), (5, img_b)]
reps = P.identify_distinct_screens(frames)
indexes = [r[0] for r in reps]
assert 0 in indexes
assert 5 in indexes
assert 1 not in indexes # regroupé avec idx 0
assert len(reps) == 2
def test_identify_distinct_screens_empty(self):
assert P.identify_distinct_screens([]) == []
# ---------------------------------------------------------------------------
# Tests : analyze_screen avec OmniParser OK
# ---------------------------------------------------------------------------
class TestAnalyzeScreenOmniParserOK:
def test_nominal_run(self, tmp_path, monkeypatch, fake_omniparser_ok):
# Rediriger le cache vers tmp
monkeypatch.setattr(P, "OMNIPARSER_CACHE_ROOT", tmp_path / "cache")
analyzer = P.Phase25Analyzer(session_id="sess1", omniparser=fake_omniparser_ok)
img = _make_image()
result = analyzer.analyze_screen(
frame_index=42, image=img, phash="deadbeef", screenshot_path=None,
)
assert result.index == 42
assert result.screen_id == "screen_042"
assert result.degraded is False
# Structure : 1 button + 1 field + 1 text_block (cf. fake_detect).
assert len(result.structure.buttons) == 1
assert result.structure.buttons[0]["label"] == "Valider"
assert len(result.structure.forms) == 1
assert len(result.structure.text_blocks) == 1
def test_cache_hit_skips_omniparser(self, tmp_path, monkeypatch, fake_omniparser_ok):
monkeypatch.setattr(P, "OMNIPARSER_CACHE_ROOT", tmp_path / "cache")
analyzer = P.Phase25Analyzer(session_id="sess1", omniparser=fake_omniparser_ok)
img = _make_image()
# 1er appel : remplit le cache.
analyzer.analyze_screen(frame_index=7, image=img, phash="aa")
call_count_1 = fake_omniparser_ok._adapter.detect.call_count
# 2e appel : doit lire depuis le cache, pas re-appeler OmniParser.
analyzer.analyze_screen(frame_index=7, image=img, phash="aa")
call_count_2 = fake_omniparser_ok._adapter.detect.call_count
assert call_count_2 == call_count_1
# ---------------------------------------------------------------------------
# Tests : fallback OCR-seul
# ---------------------------------------------------------------------------
class TestFallbackOCR:
def test_omniparser_raises_falls_back_degraded(
self, tmp_path, monkeypatch, fake_omniparser_raising
):
monkeypatch.setattr(P, "OMNIPARSER_CACHE_ROOT", tmp_path / "cache")
monkeypatch.setattr(P, "LOGS_DIR", tmp_path / "logs")
monkeypatch.setattr(P, "OMNIPARSER_ERROR_LOG", tmp_path / "logs" / "omniparser_errors.log")
# Stub docTR : retourne 2 text_blocks.
monkeypatch.setattr(
P, "_detect_via_doctr",
lambda image, screenshot_path: [
{"label": "Champ A", "text": "Champ A", "bbox": [0, 0, 50, 20], "confidence": 0.6},
{"label": "Champ B", "text": "Champ B", "bbox": [60, 0, 110, 20], "confidence": 0.6},
],
)
analyzer = P.Phase25Analyzer(
session_id="sessFB", omniparser=fake_omniparser_raising
)
img = _make_image()
result = analyzer.analyze_screen(frame_index=3, image=img, phash="zz")
assert result.degraded is True
assert result.degraded_reason and "omniparser_exception" in result.degraded_reason
# Fallback docTR doit avoir produit 2 text_blocks.
assert len(result.structure.text_blocks) == 2
# Le log d'erreur doit avoir été écrit.
assert (tmp_path / "logs" / "omniparser_errors.log").exists()
def test_omniparser_unavailable_uses_doctr(
self, tmp_path, monkeypatch, fake_omniparser_unavailable
):
monkeypatch.setattr(P, "OMNIPARSER_CACHE_ROOT", tmp_path / "cache")
monkeypatch.setattr(
P, "_detect_via_doctr",
lambda image, screenshot_path: [
{"label": "Hello", "text": "Hello", "bbox": [0, 0, 30, 10], "confidence": 0.6},
],
)
analyzer = P.Phase25Analyzer(
session_id="sessUNAV", omniparser=fake_omniparser_unavailable
)
img = _make_image()
result = analyzer.analyze_screen(frame_index=1, image=img, phash="aa")
assert result.degraded is True
assert "omniparser_unavailable" in (result.degraded_reason or "")
assert len(result.structure.text_blocks) == 1
# ---------------------------------------------------------------------------
# Tests : healthcheck
# ---------------------------------------------------------------------------
class TestHealthcheck:
def test_healthcheck_ok(self, fake_omniparser_ok):
analyzer = P.Phase25Analyzer(session_id="hc1", omniparser=fake_omniparser_ok)
assert analyzer.healthcheck() is True
assert analyzer._healthcheck_reason is None
def test_healthcheck_unavailable(self, fake_omniparser_unavailable):
analyzer = P.Phase25Analyzer(
session_id="hc2", omniparser=fake_omniparser_unavailable
)
assert analyzer.healthcheck() is False
assert analyzer._healthcheck_reason is not None
def test_healthcheck_raises_logs(self, tmp_path, monkeypatch, fake_omniparser_raising):
monkeypatch.setattr(P, "LOGS_DIR", tmp_path / "logs")
monkeypatch.setattr(P, "OMNIPARSER_ERROR_LOG", tmp_path / "logs" / "omniparser_errors.log")
analyzer = P.Phase25Analyzer(
session_id="hc3", omniparser=fake_omniparser_raising
)
assert analyzer.healthcheck() is False
assert (tmp_path / "logs" / "omniparser_errors.log").exists()
# ---------------------------------------------------------------------------
# Tests : pipeline analyze_frames + cap too_complex
# ---------------------------------------------------------------------------
class TestAnalyzeFrames:
def test_pipeline_groups_and_analyzes(self, tmp_path, monkeypatch, fake_omniparser_ok):
monkeypatch.setattr(P, "OMNIPARSER_CACHE_ROOT", tmp_path / "cache")
analyzer = P.Phase25Analyzer(session_id="pipeline1", omniparser=fake_omniparser_ok)
# 4 frames : 2 blancs (groupés) + 2 noirs (groupés).
frames = [
(0, _make_image(color=(255, 255, 255))),
(1, _make_image(color=(255, 255, 255))),
(2, _make_image(color=(0, 0, 0))),
(3, _make_image(color=(0, 0, 0))),
]
result = analyzer.analyze_frames(frames=frames, run_healthcheck=True)
assert result.too_complex is False
# Au plus 2 représentants après grouping.
assert len(result.screens) <= 2
assert result.omniparser_available is True
def test_too_complex_caps_at_max(self, tmp_path, monkeypatch, fake_omniparser_ok):
monkeypatch.setattr(P, "OMNIPARSER_CACHE_ROOT", tmp_path / "cache")
analyzer = P.Phase25Analyzer(
session_id="pipeline2",
omniparser=fake_omniparser_ok,
max_screens=3, # cap volontairement bas pour le test
)
# 5 frames "visuellement distinctes" avec couleurs très différentes.
frames = []
colors = [(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0), (0, 255, 255)]
for i, c in enumerate(colors):
img = _make_image(size=(256, 256), color=c)
# Ajouter du bruit pour que phash diffère bien.
draw = ImageDraw.Draw(img)
draw.rectangle([i * 20, i * 20, i * 20 + 50, i * 20 + 50], fill=(128, 128, 128))
frames.append((i, img))
result = analyzer.analyze_frames(frames=frames, run_healthcheck=True)
# Le cap doit s'appliquer.
assert len(result.screens) <= 3
if len(result.screens) == 3:
# too_complex doit refléter le fait qu'on a tronqué.
# (vrai uniquement si phash a vu > 3 représentants).
assert result.too_complex in (True, False)
# ---------------------------------------------------------------------------
# Tests : write_semantic_yaml
# ---------------------------------------------------------------------------
class TestWriteSemanticYaml:
def test_writes_valid_yaml(self, tmp_path, fake_omniparser_ok):
analyzer = P.Phase25Analyzer(session_id="yaml1", omniparser=fake_omniparser_ok)
result = P.Phase25Result(
session_id="yaml1",
generated_at="2026-06-01T18:30:00Z",
omniparser_available=True,
degraded=False,
too_complex=False,
screens=[
P.ScreenAnalysis(
index=42,
phash="abc123",
screen_id="screen_042",
screenshot_path="/tmp/shot.png",
structure=P.SemanticStructure(
buttons=[{"label": "OK", "bbox": [0, 0, 10, 10], "confidence": 0.9}],
),
),
],
)
target = analyzer.write_semantic_yaml(
result, slug="ma_competence", target_dir=tmp_path,
)
assert target.exists()
data = yaml.safe_load(target.read_text(encoding="utf-8"))
assert data["competence_id"] == "ma_competence"
assert data["semantic_version"] == 1
assert data["degraded"] is False
assert len(data["screens"]) == 1
assert data["screens"][0]["structure"]["buttons"][0]["label"] == "OK"
def test_degraded_yaml_is_valid(self, tmp_path, fake_omniparser_raising):
analyzer = P.Phase25Analyzer(session_id="yaml2", omniparser=fake_omniparser_raising)
result = P.Phase25Result(
session_id="yaml2",
generated_at="2026-06-01T18:30:00Z",
omniparser_available=False,
degraded=True,
too_complex=False,
screens=[
P.ScreenAnalysis(
index=0,
phash="00",
screen_id="screen_000",
screenshot_path=None,
structure=P.SemanticStructure(),
degraded=True,
degraded_reason="omniparser_exception: RuntimeError",
),
],
)
target = analyzer.write_semantic_yaml(result, slug="fallback_comp", target_dir=tmp_path)
data = yaml.safe_load(target.read_text(encoding="utf-8"))
assert data["degraded"] is True
assert data["screens"][0]["degraded"] is True
assert "omniparser_exception" in data["screens"][0]["degraded_reason"]
def test_invalid_slug_raises(self, tmp_path, fake_omniparser_ok):
analyzer = P.Phase25Analyzer(session_id="yaml3", omniparser=fake_omniparser_ok)
result = P.Phase25Result(
session_id="yaml3", generated_at="x", omniparser_available=True,
degraded=False, too_complex=False, screens=[],
)
with pytest.raises(ValueError):
analyzer.write_semantic_yaml(result, slug="../etc/passwd", target_dir=tmp_path)
def test_forbidden_target_dir(self, tmp_path, fake_omniparser_ok):
analyzer = P.Phase25Analyzer(session_id="yaml4", omniparser=fake_omniparser_ok)
result = P.Phase25Result(
session_id="yaml4", generated_at="x", omniparser_available=True,
degraded=False, too_complex=False, screens=[],
)
# Anti écriture dans supervised/stable.
forbidden = tmp_path / "supervised"
forbidden.mkdir()
with pytest.raises(ValueError):
analyzer.write_semantic_yaml(result, slug="abc_def", target_dir=forbidden)
# ---------------------------------------------------------------------------
# Tests : contrat snapshots (elements aplatis)
# ---------------------------------------------------------------------------
class TestSnapshotContract:
def test_screen_to_dict_includes_elements(self, fake_omniparser_ok):
s = P.ScreenAnalysis(
index=1,
phash="aa",
screen_id="screen_001",
screenshot_path="/tmp/s.png",
structure=P.SemanticStructure(
buttons=[{"label": "Valider", "bbox": [0, 0, 50, 20], "confidence": 0.9}],
forms=[{"label": "Nom", "bbox": [60, 0, 200, 20], "confidence": 0.8}],
text_blocks=[{"label": "Hello", "text": "Hello", "bbox": [0, 30, 100, 50], "confidence": 0.6}],
),
window_title="Easily Assure",
)
d = s.to_dict()
assert "elements" in d
assert any(e["kind"] == "button" and e["label"] == "Valider" for e in d["elements"])
assert any(e["kind"] == "field" and e["label"] == "Nom" for e in d["elements"])
assert any(e["kind"] == "text_block" for e in d["elements"])
assert d["window_title"] == "Easily Assure"

22
tests/unit/test_replay_critic.py
View File

@@ -346,6 +346,28 @@ class TestMergeResults:
class TestEnrichActionsWithIntentions:
@patch("requests.post")
@patch("requests.get")
def test_enrichissement_desactive_par_flag(
self,
mock_get,
mock_post,
monkeypatch,
tmp_path,
):
"""Le flag demo evite tout appel Ollama pendant le build replay."""
from agent_v0.server_v1.stream_processor import _enrich_actions_with_intentions
monkeypatch.setenv("RPA_SKIP_INTENTION_ENRICHMENT", "1")
actions = [
{"type": "click", "action_id": "act_001", "target_spec": {"by_text": "OK"}},
]
_enrich_actions_with_intentions(actions, tmp_path)
assert "intention" not in actions[0]
mock_get.assert_not_called()
mock_post.assert_not_called()
@patch("requests.post")
@patch("requests.get")

69
tests/unit/test_replay_memory.py
View File

@@ -68,6 +68,10 @@ def test_memory_lookup_keeps_learned_visual_coords_with_window_capture(monkeypat
target_spec={
"by_text": "Enregistrer",
"by_role": "yolo",
"context_hints": {
"expected_window_before": "*test – Bloc-notes",
"interaction": "toolbar_save_button",
},
"window_capture": {
"click_relative": [860, 634],
"window_size": [1920, 1116],
@@ -81,6 +85,71 @@ def test_memory_lookup_keeps_learned_visual_coords_with_window_capture(monkeypat
assert result["y_pct"] == 0.578125
def test_memory_lookup_skips_window_transition_even_if_record_exists(monkeypatch):
fp = SimpleNamespace(
bbox=(0.5, 0.8, 0.0, 0.0),
etype="grounding_vlm",
confidence=0.85,
)
monkeypatch.setattr(replay_memory, "get_memory_store", lambda: _DummyStore(fp))
result = replay_memory.memory_lookup(
window_title="*test – Bloc-notes",
target_spec={
"by_text": "Enregistrer",
"by_role": "button",
"context_hints": {
"expected_window_before": "*test – Bloc-notes",
"expected_window_after": "Enregistrer sous",
"requires_window_transition": True,
},
},
)
assert result is None
def test_memory_lookup_rejects_generic_button_without_context(monkeypatch):
fp = SimpleNamespace(
bbox=(0.5, 0.8, 0.0, 0.0),
etype="grounding_vlm",
confidence=0.85,
)
monkeypatch.setattr(replay_memory, "get_memory_store", lambda: _DummyStore(fp))
result = replay_memory.memory_lookup(
window_title="*test – Bloc-notes",
target_spec={"by_text": "Enregistrer", "by_role": "button"},
)
assert result is None
def test_memory_lookup_allows_generic_button_with_context(monkeypatch):
fp = SimpleNamespace(
bbox=(0.5, 0.8, 0.0, 0.0),
etype="grounding_vlm",
confidence=0.85,
)
monkeypatch.setattr(replay_memory, "get_memory_store", lambda: _DummyStore(fp))
result = replay_memory.memory_lookup(
window_title="Enregistrer sous",
target_spec={
"by_text": "Enregistrer",
"by_role": "button",
"window_title": "Enregistrer sous",
"context_hints": {
"expected_window_before": "Enregistrer sous",
"interaction": "save_dialog_primary_button",
},
},
)
assert result is not None
assert result["method"] == "memory_grounding_vlm"
def test_target_spec_hash_distinguishes_same_text_with_different_spatial_hints(tmp_path):
store = TargetMemoryStore(base_path=str(tmp_path / "learning"))