merge(anchors): matérialisation des ancres à l'import R1 (fix « Ancre requise ») — 2026-07-03
Helper partagé materialize_anchor_from_b64 (généralisé à tous les needsAnchor) réutilisé par la route HTTP + le bridge R1 (dé-duplication) ; source canonique anchor_action_types.NEEDS_ANCHOR ; backfill idempotent tools/backfill_anchors_r1.py (--dry-run par défaut, NON appliqué). Revue adversariale MERGE-READY, 2 nits levés, 29 tests unit verts. Déploiement clinique + backfill --apply = décision Dom, NON inclus. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
220
tools/backfill_anchors_r1.py
Normal file
220
tools/backfill_anchors_r1.py
Normal file
@@ -0,0 +1,220 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Backfill des ancres visuelles pour les workflows importés par le pont R1.
|
||||
|
||||
Contexte (diagnostic + plan 2026-07-03) : 62 workflows `source='learned_import'`
|
||||
sont déjà en DB VWB avec le crop `_anchor_image_base64` resté dans
|
||||
`Step.parameters_json` et `anchor_id = NULL` → le VWB affiche « Ancre requise »
|
||||
alors que l'image est là (verdict H1 : ~69 % des clics ont le crop en base).
|
||||
Corriger le flux futur (helper R1 + route HTTP) ne répare PAS l'existant.
|
||||
|
||||
Ce script matérialise l'ancre pour chaque step réparable, via le MÊME helper
|
||||
partagé `materialize_anchor_from_b64` (zéro logique dupliquée).
|
||||
|
||||
Sécurité / garde-fous :
|
||||
* `--dry-run` par défaut : n'écrit RIEN. Affiche le chemin DB résolu, le
|
||||
nombre de steps réparables (par workflow), l'estimation du nombre de PNG à
|
||||
créer et l'espace disque.
|
||||
* Filtre dur : `source='learned_import'` ET `action_type ∈ NEEDS_ANCHOR` ET
|
||||
`anchor_id IS NULL` ET `_anchor_image_base64` présent dans parameters_json.
|
||||
Les workflows `source='manual'` ne sont JAMAIS touchés.
|
||||
* Idempotent : un step déjà pourvu d'`anchor_id` est ignoré ; une ré-exécution
|
||||
d'`--apply` est un no-op (le helper retire aussi la clé b64 des params).
|
||||
* Cible DB résolue depuis la config runtime VWB (env `DATABASE_URL` + logique
|
||||
`instance_path` de Flask), PAS un chemin codé en dur (réserve QG R1).
|
||||
|
||||
⚠️ Ce script est manuel, hors flux runtime. Il NE fait PAS de backup — l'appelant
|
||||
doit sauvegarder la DB avant `--apply` (copie horodatée), cf. plan §3.3.
|
||||
Ne JAMAIS lancer `--apply` sans backup ni sans avoir validé le `--dry-run`.
|
||||
|
||||
Usage :
|
||||
python tools/backfill_anchors_r1.py # dry-run (défaut)
|
||||
python tools/backfill_anchors_r1.py --dry-run # explicite
|
||||
python tools/backfill_anchors_r1.py --db /chemin/workflows.db --dry-run
|
||||
python tools/backfill_anchors_r1.py --apply # écrit (backup requis avant)
|
||||
|
||||
Auteur : Dom, Claude — 2026-07-03
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# --- Chemins : racine projet + backend VWB (pour db.models, services.*) ---
|
||||
_ROOT = Path(__file__).resolve().parent.parent
|
||||
_BACKEND = _ROOT / "visual_workflow_builder" / "backend"
|
||||
for p in (str(_ROOT), str(_BACKEND)):
|
||||
if p not in sys.path:
|
||||
sys.path.insert(0, p)
|
||||
|
||||
|
||||
def resolve_vwb_db_path(explicit: str | None = None) -> Path:
|
||||
"""Résout le chemin du fichier SQLite VWB depuis la config runtime.
|
||||
|
||||
Ordre de priorité :
|
||||
1. `--db` explicite (échappatoire manuelle) ;
|
||||
2. `DATABASE_URL` env (même variable que `app.py`) si sqlite:/// ;
|
||||
3. défaut `app.py` = `sqlite:///workflows.db`, résolu relativement à
|
||||
`Flask.instance_path` du backend VWB — EXACTEMENT comme l'app au runtime.
|
||||
|
||||
On instancie une app Flask *sonde* légère (root_path = backend) au lieu
|
||||
d'importer `app.py` (lourd : socketio, blueprints). Flask calcule alors le
|
||||
même `instance_path` que l'app réelle → même résolution de la DB.
|
||||
"""
|
||||
if explicit:
|
||||
return Path(explicit).resolve()
|
||||
|
||||
uri = os.getenv("DATABASE_URL", "sqlite:///workflows.db")
|
||||
if not uri.startswith("sqlite:"):
|
||||
raise SystemExit(
|
||||
f"DATABASE_URL non-SQLite ({uri!r}) : backfill SQLite uniquement. "
|
||||
"Passez --db pour un chemin explicite."
|
||||
)
|
||||
|
||||
# sqlite:////abs/path (4 slashes) → chemin absolu ; sqlite:///rel → relatif.
|
||||
tail = uri[len("sqlite://"):]
|
||||
if tail.startswith("//"): # //// → chemin absolu
|
||||
return Path(tail[1:]).resolve()
|
||||
|
||||
rel = tail.lstrip("/") # ex. "workflows.db"
|
||||
from flask import Flask
|
||||
|
||||
probe = Flask("vwb_backfill_probe", root_path=str(_BACKEND))
|
||||
return (Path(probe.instance_path) / rel).resolve()
|
||||
|
||||
|
||||
def _iter_repairable_steps(session, NEEDS_ANCHOR):
|
||||
"""Génère les tuples (workflow, step, params) réparables.
|
||||
|
||||
Réparable = source learned_import, action_type ∈ NEEDS_ANCHOR,
|
||||
anchor_id NULL, et `_anchor_image_base64` présent (non vide) dans params.
|
||||
"""
|
||||
from db.models import Workflow, Step
|
||||
|
||||
q = (
|
||||
session.query(Workflow, Step)
|
||||
.join(Step, Step.workflow_id == Workflow.id)
|
||||
.filter(Workflow.source == "learned_import")
|
||||
.filter(Step.anchor_id.is_(None))
|
||||
.filter(Step.action_type.in_(tuple(NEEDS_ANCHOR)))
|
||||
.filter(Step.parameters_json.like("%_anchor_image_base64%"))
|
||||
)
|
||||
for wf, step in q.all():
|
||||
params = step.parameters # dict décodé
|
||||
b64 = params.get("_anchor_image_base64")
|
||||
if b64:
|
||||
yield wf, step, params
|
||||
|
||||
|
||||
def _estimate_png_bytes(b64: str) -> int:
|
||||
"""Taille approximative du PNG décodé (le fichier écrit ≈ crop optimisé)."""
|
||||
try:
|
||||
clean = b64.split(",", 1)[1] if "," in b64 else b64
|
||||
return len(base64.b64decode(clean))
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
|
||||
def run(db_path: Path, apply: bool) -> int:
|
||||
from flask import Flask
|
||||
from db.models import db
|
||||
from services.anchor_action_types import NEEDS_ANCHOR
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
|
||||
if not db_path.exists():
|
||||
raise SystemExit(f"DB introuvable : {db_path}")
|
||||
|
||||
app = Flask("vwb_backfill")
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
db.init_app(app)
|
||||
|
||||
mode = "APPLY (écriture)" if apply else "DRY-RUN (lecture seule)"
|
||||
print(f"=== Backfill ancres R1 — {mode} ===")
|
||||
print(f"DB résolue : {db_path}")
|
||||
print(f"action_type ∈ NEEDS_ANCHOR ({len(NEEDS_ANCHOR)}) : "
|
||||
f"{', '.join(sorted(NEEDS_ANCHOR))}")
|
||||
print()
|
||||
|
||||
with app.app_context():
|
||||
session = db.session
|
||||
per_workflow: dict = {}
|
||||
total_repairable = 0
|
||||
total_est_bytes = 0
|
||||
|
||||
for wf, step, params in _iter_repairable_steps(session, NEEDS_ANCHOR):
|
||||
total_repairable += 1
|
||||
total_est_bytes += _estimate_png_bytes(params["_anchor_image_base64"])
|
||||
bucket = per_workflow.setdefault(
|
||||
wf.id, {"name": wf.name, "count": 0}
|
||||
)
|
||||
bucket["count"] += 1
|
||||
|
||||
if apply:
|
||||
# Le helper retire la clé b64 de params, crée VisualAnchor +
|
||||
# fichier, pose step.anchor_id. On réécrit params (b64 retiré).
|
||||
anchor_id = materialize_anchor_from_b64(
|
||||
step, params, db_session=session, source="learned_import"
|
||||
)
|
||||
if anchor_id:
|
||||
step.parameters = params # persiste params sans le b64
|
||||
session.add(step)
|
||||
|
||||
if apply:
|
||||
session.commit()
|
||||
|
||||
# --- Rapport ---
|
||||
print(f"Workflows learned_import concernés : {len(per_workflow)}")
|
||||
for wf_id, info in sorted(per_workflow.items(),
|
||||
key=lambda kv: -kv[1]["count"]):
|
||||
print(f" - {wf_id} ({info['name'][:48]}) : "
|
||||
f"{info['count']} step(s) réparable(s)")
|
||||
print()
|
||||
print(f"Total steps réparables : {total_repairable}")
|
||||
|
||||
if apply:
|
||||
print(f"→ {total_repairable} ancres matérialisées (VisualAnchor + PNG).")
|
||||
print("Ré-exécuter ce script = no-op (idempotent).")
|
||||
else:
|
||||
est_png_mb = total_est_bytes / (1024 * 1024)
|
||||
# Thumbnails JPEG q80 ≈ 30 % du PNG en plus (majoration prudente).
|
||||
est_total_mb = est_png_mb * 1.3
|
||||
print(f"→ PNG à créer : {total_repairable}")
|
||||
print(f"→ Espace disque estimé : ~{est_png_mb:.2f} Mo (PNG) "
|
||||
f"+ thumbnails ≈ ~{est_total_mb:.2f} Mo total")
|
||||
print()
|
||||
print("DRY-RUN : aucune écriture. Vérifiez le chemin DB ci-dessus,")
|
||||
print("SAUVEGARDEZ la DB, puis relancez avec --apply.")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Backfill idempotent des ancres visuelles des workflows R1."
|
||||
)
|
||||
group = parser.add_mutually_exclusive_group()
|
||||
group.add_argument(
|
||||
"--dry-run", action="store_true", default=True,
|
||||
help="Lecture seule (défaut) : rapport sans écriture.",
|
||||
)
|
||||
group.add_argument(
|
||||
"--apply", action="store_true", default=False,
|
||||
help="Écrit les ancres (backup DB requis AVANT).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--db", default=None,
|
||||
help="Chemin explicite du fichier SQLite VWB (sinon résolu depuis config).",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
apply = bool(args.apply)
|
||||
db_path = resolve_vwb_db_path(args.db)
|
||||
return run(db_path, apply=apply)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -27,7 +27,7 @@ from flask import jsonify, request
|
||||
|
||||
from . import api_v3_bp
|
||||
from .workflow import generate_id
|
||||
from db.models import db, Workflow, Step, VisualAnchor
|
||||
from db.models import db, Workflow, Step
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -328,56 +328,16 @@ def import_learned_workflow(workflow_id: str):
|
||||
)
|
||||
params = dict(step_data.get("parameters", {}))
|
||||
|
||||
# Extraire et sauvegarder le screenshot d'ancre si présent
|
||||
anchor_b64 = params.pop("_anchor_image_base64", None)
|
||||
# NE PAS supprimer _anchor_bbox : on le conserve dans params pour que le frontend puisse lire x_pct/y_pct
|
||||
# et afficher la zone ciblée, au lieu de le jeter et de créer une bbox factice.
|
||||
if anchor_b64:
|
||||
try:
|
||||
from services.anchor_image_service import (
|
||||
save_anchor_image, generate_anchor_id
|
||||
# Matérialiser l'ancre visuelle (VisualAnchor + fichier PNG + anchor_id)
|
||||
# via le helper partagé — MÊME source de vérité que l'import R1
|
||||
# (`import_core_workflow_to_db`). Le helper retire `_anchor_image_base64`
|
||||
# de params (pas de double stockage) et ne commit pas.
|
||||
# NE PAS supprimer _anchor_bbox : conservé dans params pour que le
|
||||
# frontend lise x_pct/y_pct et affiche la zone ciblée.
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
materialize_anchor_from_b64(
|
||||
step, params, db_session=db.session, workflow_id=wf_id
|
||||
)
|
||||
from PIL import Image
|
||||
from io import BytesIO
|
||||
import base64 as b64mod
|
||||
|
||||
if ',' in anchor_b64:
|
||||
anchor_b64 = anchor_b64.split(',', 1)[1]
|
||||
img_data = b64mod.b64decode(anchor_b64)
|
||||
img = Image.open(BytesIO(img_data))
|
||||
# Fallback sécurisé pour le service de crop si _anchor_bbox n'a pas le format attendu,
|
||||
# mais les données x_pct/y_pct restent intactes dans params pour le frontend.
|
||||
bbox = {
|
||||
"x": 0, "y": 0,
|
||||
"width": img.width, "height": img.height
|
||||
}
|
||||
anchor_id = generate_anchor_id()
|
||||
result = save_anchor_image(
|
||||
anchor_id=anchor_id,
|
||||
image_base64=anchor_b64,
|
||||
bounding_box=bbox,
|
||||
metadata={"source": "learned_import", "workflow_id": wf_id}
|
||||
)
|
||||
if result.get("success"):
|
||||
from services.anchor_image_service import (
|
||||
get_original_path, get_thumbnail_path
|
||||
)
|
||||
va = VisualAnchor(
|
||||
id=anchor_id,
|
||||
image_path=str(get_original_path(anchor_id) or ""),
|
||||
thumbnail_path=str(get_thumbnail_path(anchor_id) or ""),
|
||||
bbox_x=0, bbox_y=0,
|
||||
bbox_width=img.width, bbox_height=img.height,
|
||||
description=step_data.get("label", ""),
|
||||
capture_method="learned_import",
|
||||
)
|
||||
db.session.add(va)
|
||||
step.anchor_id = anchor_id
|
||||
logger.info("Ancre sauvegardée: %s pour step %s",
|
||||
anchor_id, step.id)
|
||||
except Exception as e:
|
||||
logger.warning("Échec sauvegarde ancre pour step %s: %s",
|
||||
step_data.get("order"), e)
|
||||
|
||||
step.parameters = params
|
||||
db.session.add(step)
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
"""
|
||||
Source canonique (backend) de l'ensemble des `action_type` qui requièrent une
|
||||
ancre visuelle — miroir Python du catalogue frontend `needsAnchor: true`.
|
||||
|
||||
⚠️ SOURCE DE VÉRITÉ PARTAGÉE — synchroniser avec le frontend :
|
||||
`visual_workflow_builder/frontend_v4/src/types.ts` (champ `needsAnchor` de
|
||||
`ACTION_DEFINITIONS`). Toute action ajoutée là avec `needsAnchor: true` doit être
|
||||
ajoutée ici, et réciproquement. Le test
|
||||
`tests/unit/test_needs_anchor_canonical.py` verrouille la cohérence (garde-fou
|
||||
anti-divergence exigé par Dom : pas de liste en dur qui divergera).
|
||||
|
||||
Ce set est utilisé par :
|
||||
- `anchor_image_service.materialize_anchor_from_b64` (indirectement via l'appelant) ;
|
||||
- `tools/backfill_anchors_r1.py` (filtre `action_type ∈ NEEDS_ANCHOR`).
|
||||
|
||||
Auteur : Dom, Claude — 2026-07-03
|
||||
"""
|
||||
|
||||
# Miroir de `types.ts` : toutes les entrées `needsAnchor: true` de ACTION_DEFINITIONS.
|
||||
NEEDS_ANCHOR: frozenset = frozenset(
|
||||
{
|
||||
"click_anchor",
|
||||
"double_click_anchor",
|
||||
"right_click_anchor",
|
||||
"hover_anchor",
|
||||
"drag_drop_anchor",
|
||||
"scroll_to_anchor",
|
||||
"focus_anchor",
|
||||
"wait_for_anchor",
|
||||
"extract_table",
|
||||
"visual_condition",
|
||||
"loop_visual",
|
||||
"ai_ocr",
|
||||
"ai_extract",
|
||||
"verify_element_exists",
|
||||
"verify_text_content",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def needs_anchor(action_type: str) -> bool:
|
||||
"""True si `action_type` requiert une ancre visuelle (miroir `needsAnchor`)."""
|
||||
return action_type in NEEDS_ANCHOR
|
||||
@@ -17,6 +17,7 @@ Structure sur disque :
|
||||
import os
|
||||
import json
|
||||
import base64
|
||||
import logging
|
||||
import uuid
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
@@ -25,6 +26,8 @@ from typing import Optional, Dict, Any, Tuple
|
||||
from PIL import Image
|
||||
from io import BytesIO
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Configuration
|
||||
DATA_DIR = Path(__file__).parent.parent / 'data' / 'anchor_images'
|
||||
@@ -188,6 +191,112 @@ def save_anchor_image(
|
||||
raise ValueError(f"Erreur lors de la sauvegarde de l'image: {str(e)}")
|
||||
|
||||
|
||||
def materialize_anchor_from_b64(
|
||||
step,
|
||||
params: Dict[str, Any],
|
||||
*,
|
||||
db_session,
|
||||
source: str = "learned_import",
|
||||
workflow_id: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
"""Matérialise une ancre visuelle à partir de `_anchor_image_base64`.
|
||||
|
||||
Helper GÉNÉRIQUE (mandat Dom) : appelable pour TOUT `action_type` qui
|
||||
requiert une ancre (`needsAnchor` : clic, clic droit, double-clic, survol,
|
||||
scroll, wait…). Ne présuppose PAS `click_anchor`.
|
||||
|
||||
Comportement, si `params` contient une clé `_anchor_image_base64` non vide :
|
||||
1. retire la clé de `params` (`pop`) → le crop volumineux n'est PAS
|
||||
re-stocké en double dans `parameters_json`. Le `pop` a lieu AVANT le
|
||||
`try` : un crop CORROMPU (b64 invalide, PNG au magic cassé) est
|
||||
inexploitable, donc retiré quand même (pas de re-tentative possible,
|
||||
éviter de traîner une donnée morte volumineuse dans les params) ;
|
||||
2. décode le b64 → PIL.Image (pour connaître les dimensions du crop) ;
|
||||
3. `save_anchor_image(...)` écrit `original.png` + `thumbnail.jpg` sur disque ;
|
||||
4. crée une ligne `VisualAnchor` (bbox = plein crop) + `db_session.add(...)` ;
|
||||
5. pose `step.anchor_id = anchor_id`.
|
||||
|
||||
Ne commit PAS : l'appelant (import R1, route HTTP, backfill) détient la
|
||||
transaction et committe.
|
||||
|
||||
Tolérant : sur crop invalide (b64 corrompu, décodage impossible), log un
|
||||
avertissement et retourne None sans lever — l'import du reste du workflow
|
||||
ne doit jamais casser à cause d'une ancre. Dans ce cas `step.anchor_id`
|
||||
reste NULL (« Ancre requise » légitime, pas un contournement du 100% vision).
|
||||
|
||||
Args:
|
||||
step: instance `Step` (SQLAlchemy) sur laquelle poser `anchor_id`.
|
||||
params: dict des paramètres du step (mutable — la clé b64 est retirée).
|
||||
db_session: session SQLAlchemy (transaction gérée par l'appelant).
|
||||
source: `capture_method` / marqueur de provenance (défaut learned_import).
|
||||
workflow_id: id du workflow VWB parent — écrit dans le sidecar
|
||||
`metadata.json` (champ informatif, parité avec la route d'origine).
|
||||
|
||||
Returns:
|
||||
L'`anchor_id` créé, ou None si aucun crop exploitable.
|
||||
"""
|
||||
anchor_b64 = params.pop("_anchor_image_base64", None)
|
||||
if not anchor_b64:
|
||||
return None
|
||||
|
||||
# Import paresseux : évite un import circulaire db.models ↔ services au load.
|
||||
from db.models import VisualAnchor
|
||||
|
||||
try:
|
||||
clean_b64 = anchor_b64.split(",", 1)[1] if "," in anchor_b64 else anchor_b64
|
||||
img = Image.open(BytesIO(base64.b64decode(clean_b64)))
|
||||
width, height = img.width, img.height
|
||||
|
||||
anchor_id = generate_anchor_id()
|
||||
# bbox plein crop : le b64 est DÉJÀ le crop de la zone cible (cf.
|
||||
# stream_processor crop 80×80), pas un screenshot plein écran.
|
||||
bbox = {"x": 0, "y": 0, "width": width, "height": height}
|
||||
anchor_meta = {"source": source}
|
||||
if workflow_id:
|
||||
anchor_meta["workflow_id"] = workflow_id
|
||||
result = save_anchor_image(
|
||||
anchor_id=anchor_id,
|
||||
image_base64=clean_b64,
|
||||
bounding_box=bbox,
|
||||
metadata=anchor_meta,
|
||||
)
|
||||
if not result.get("success"):
|
||||
logger.warning(
|
||||
"Matérialisation ancre : save_anchor_image a échoué (step %s)",
|
||||
getattr(step, "id", "?"),
|
||||
)
|
||||
return None
|
||||
|
||||
va = VisualAnchor(
|
||||
id=anchor_id,
|
||||
image_path=str(get_original_path(anchor_id) or ""),
|
||||
thumbnail_path=str(get_thumbnail_path(anchor_id) or ""),
|
||||
bbox_x=0,
|
||||
bbox_y=0,
|
||||
bbox_width=width,
|
||||
bbox_height=height,
|
||||
description=getattr(step, "label", "") or "",
|
||||
capture_method=source,
|
||||
)
|
||||
db_session.add(va)
|
||||
step.anchor_id = anchor_id
|
||||
logger.info(
|
||||
"Ancre matérialisée : %s pour step %s (%s)",
|
||||
anchor_id,
|
||||
getattr(step, "id", "?"),
|
||||
getattr(step, "action_type", "?"),
|
||||
)
|
||||
return anchor_id
|
||||
|
||||
except Exception as e: # tolérant : ne jamais casser l'import à cause d'une ancre
|
||||
logger.warning(
|
||||
"Échec matérialisation ancre pour step %s : %s",
|
||||
getattr(step, "id", "?"),
|
||||
e,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def get_thumbnail_path(anchor_id: str) -> Optional[Path]:
|
||||
"""
|
||||
Obtenir le chemin du fichier miniature.
|
||||
|
||||
@@ -446,9 +446,16 @@ def import_core_workflow_to_db(
|
||||
label=step_data.get("label", step_data["action_type"]),
|
||||
)
|
||||
params = dict(step_data.get("parameters", {}))
|
||||
# L'image d'ancre (_anchor_image_base64) est laissée dans params : la
|
||||
# persistance d'ancre (VisualAnchor + fichier) reste pilotée par la route
|
||||
# HTTP existante. Cette unité se concentre sur l'idempotence Workflow/Step.
|
||||
# Matérialiser l'ancre visuelle si un crop `_anchor_image_base64` est
|
||||
# présent : crée le VisualAnchor + fichier PNG et pose step.anchor_id
|
||||
# (sinon anchor_id reste NULL → « Ancre requise » légitime). Helper
|
||||
# partagé avec la route HTTP `import_learned_workflow` (source unique).
|
||||
# `materialize_anchor_from_b64` retire la clé b64 de params (pas de
|
||||
# double stockage) et ne commit pas (transaction gérée ci-dessous).
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
materialize_anchor_from_b64(
|
||||
step, params, db_session=db_session, workflow_id=wf_id
|
||||
)
|
||||
step.parameters = params
|
||||
db_session.add(step)
|
||||
|
||||
|
||||
@@ -0,0 +1,449 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test TDD — matérialisation des ancres visuelles à l'import R1.
|
||||
|
||||
Contexte (diagnostic 2026-07-03, verdict H1) : le pont R1
|
||||
`import_core_workflow_to_db` importe les steps clic mais LAISSE le crop
|
||||
`_anchor_image_base64` dans `Step.parameters` sans créer le `VisualAnchor`
|
||||
ni poser `Step.anchor_id` → le VWB affiche « Ancre requise » alors que
|
||||
l'image est là. La route HTTP `import_learned_workflow` fait déjà ce travail
|
||||
mais en duplique la logique inline.
|
||||
|
||||
Mandat de généralisation (Dom) : le fix doit être GÉNÉRAL — un helper unique
|
||||
`materialize_anchor_from_b64` couvrant TOUS les `action_type ∈ needsAnchor`
|
||||
(clic, clic droit, double-clic, survol, scroll…), réutilisé par R1 ET par la
|
||||
route HTTP. Jamais une rustine `click_anchor` seul.
|
||||
|
||||
Ces tests sont volontairement isolés du chemin live (app Flask minimale +
|
||||
SQLite en mémoire, même pattern que `test_import_core_workflow_to_db.py`).
|
||||
Les fichiers PNG sont écrits dans un `DATA_DIR` temporaire (monkeypatch).
|
||||
"""
|
||||
|
||||
import base64
|
||||
import sys
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from PIL import Image
|
||||
|
||||
# --- Chemins : racine projet (pour core.*) + backend (pour db.models, services.*) ---
|
||||
_BACKEND = Path(__file__).resolve().parent.parent.parent # .../visual_workflow_builder/backend
|
||||
_ROOT = _BACKEND.parent.parent # .../rpa_vision_v3
|
||||
for p in (str(_ROOT), str(_BACKEND)):
|
||||
if p not in sys.path:
|
||||
sys.path.insert(0, p)
|
||||
|
||||
from db.models import db, Workflow, Step, VisualAnchor # noqa: E402
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures : DB en mémoire + DATA_DIR temporaire pour les PNG d'ancre
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def db_app():
|
||||
"""App Flask minimale liée à une SQLite en mémoire, schéma créé."""
|
||||
app = Flask("test_anchor_materialization")
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
db.init_app(app)
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
yield app
|
||||
db.session.remove()
|
||||
db.drop_all()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anchor_data_dir(tmp_path, monkeypatch):
|
||||
"""Redirige le stockage des images d'ancre vers un répertoire temporaire."""
|
||||
import services.anchor_image_service as ais
|
||||
|
||||
tmp_anchor_dir = tmp_path / "anchor_images"
|
||||
monkeypatch.setattr(ais, "DATA_DIR", tmp_anchor_dir)
|
||||
return tmp_anchor_dir
|
||||
|
||||
|
||||
def _png_b64(width: int = 12, height: int = 8, color=(200, 30, 30)) -> str:
|
||||
"""Génère un petit PNG valide encodé base64 (crop d'ancre factice)."""
|
||||
img = Image.new("RGB", (width, height), color)
|
||||
buf = BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
return base64.b64encode(buf.getvalue()).decode("ascii")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. Le helper matérialise l'ancre pour PLUSIEURS types (pas que clic)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"action_type",
|
||||
[
|
||||
"click_anchor",
|
||||
"double_click_anchor",
|
||||
"right_click_anchor",
|
||||
"hover_anchor",
|
||||
"scroll_to_anchor",
|
||||
"wait_for_anchor",
|
||||
],
|
||||
)
|
||||
def test_materialize_anchor_for_all_needs_anchor_types(
|
||||
db_app, anchor_data_dir, action_type
|
||||
):
|
||||
"""Le helper crée un VisualAnchor + fichier + pose anchor_id pour TOUS
|
||||
les action_type qui requièrent une ancre — pas seulement click_anchor."""
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_x", name="wf", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(
|
||||
id="step_x",
|
||||
workflow_id="wf_x",
|
||||
action_type=action_type,
|
||||
order=0,
|
||||
label="cible",
|
||||
)
|
||||
db.session.add(step)
|
||||
|
||||
params = {"_anchor_image_base64": _png_b64(), "x_pct": 0.5, "y_pct": 0.4}
|
||||
anchor_id = materialize_anchor_from_b64(
|
||||
step, params, db_session=db.session
|
||||
)
|
||||
db.session.commit()
|
||||
|
||||
assert anchor_id, f"{action_type} : le helper doit retourner un anchor_id"
|
||||
assert step.anchor_id == anchor_id, "anchor_id posé sur le step"
|
||||
|
||||
va = VisualAnchor.query.get(anchor_id)
|
||||
assert va is not None, "ligne VisualAnchor créée"
|
||||
assert va.image_path and Path(va.image_path).exists(), "fichier PNG écrit"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. Le b64 est retiré des params après matérialisation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_b64_removed_from_params_after_materialization(db_app, anchor_data_dir):
|
||||
"""Après matérialisation, `_anchor_image_base64` n'est plus dans params
|
||||
(évite le double stockage du crop volumineux en DB)."""
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_y", name="wf", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(id="step_y", workflow_id="wf_y",
|
||||
action_type="click_anchor", order=0)
|
||||
db.session.add(step)
|
||||
|
||||
params = {"_anchor_image_base64": _png_b64(), "target_text": "OK"}
|
||||
materialize_anchor_from_b64(step, params, db_session=db.session)
|
||||
|
||||
assert "_anchor_image_base64" not in params, "b64 retiré de params"
|
||||
assert params.get("target_text") == "OK", "autres params préservés"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. Pas de crop → anchor_id reste NULL (« Ancre requise » légitime)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_no_crop_leaves_anchor_id_null(db_app, anchor_data_dir):
|
||||
"""Sans crop b64, le helper ne crée rien et retourne None : anchor_id
|
||||
reste NULL. Comportement correct — pas un contournement du 100% vision."""
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_z", name="wf", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(id="step_z", workflow_id="wf_z",
|
||||
action_type="click_anchor", order=0)
|
||||
db.session.add(step)
|
||||
|
||||
params = {"target_text": "sans crop"}
|
||||
anchor_id = materialize_anchor_from_b64(
|
||||
step, params, db_session=db.session
|
||||
)
|
||||
|
||||
assert anchor_id is None
|
||||
assert step.anchor_id is None
|
||||
assert VisualAnchor.query.count() == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Import R1 : matérialise l'ancre pour un step clic (le bug d'aujourd'hui)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_import_r1_materializes_anchor_for_click_step(db_app, anchor_data_dir):
|
||||
"""Un core workflow avec un edge mouse_click portant anchor_image_base64
|
||||
→ import_core_workflow_to_db pose anchor_id + crée le VisualAnchor + fichier.
|
||||
C'EST le test qui échoue aujourd'hui (R1 laisse anchor_id NULL)."""
|
||||
from services.learned_workflow_bridge import import_core_workflow_to_db
|
||||
|
||||
core = {
|
||||
"workflow_id": "wf_sess_click_anchor_001",
|
||||
"name": "Léa Clic Ancré",
|
||||
"entry_nodes": ["n1"],
|
||||
"nodes": [{"node_id": "n1", "name": "A"}, {"node_id": "n2", "name": "B"}],
|
||||
"edges": [
|
||||
{
|
||||
"edge_id": "e1",
|
||||
"from_node": "n1",
|
||||
"to_node": "n2",
|
||||
"action": {
|
||||
"type": "mouse_click",
|
||||
"target": {
|
||||
"by_text": "Valider",
|
||||
"by_role": "button",
|
||||
"anchor_image_base64": _png_b64(),
|
||||
},
|
||||
"parameters": {"button": "left"},
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
with db_app.app_context():
|
||||
result = import_core_workflow_to_db(
|
||||
core,
|
||||
machine_id="DESKTOP-TEST_windows",
|
||||
source_session_id="sess_click_anchor_001",
|
||||
db_session=db.session,
|
||||
)
|
||||
assert result["created"] is True
|
||||
wf_id = result["workflow_id"]
|
||||
|
||||
click_steps = [
|
||||
s for s in Step.query.filter_by(workflow_id=wf_id).all()
|
||||
if s.action_type in ("click_anchor", "double_click_anchor",
|
||||
"right_click_anchor")
|
||||
]
|
||||
assert click_steps, "au moins un step clic importé"
|
||||
click = click_steps[0]
|
||||
assert click.anchor_id is not None, "R1 doit poser anchor_id (bug corrigé)"
|
||||
|
||||
va = VisualAnchor.query.get(click.anchor_id)
|
||||
assert va is not None and Path(va.image_path).exists()
|
||||
# b64 retiré des params persistés
|
||||
assert "_anchor_image_base64" not in click.parameters
|
||||
|
||||
|
||||
def test_import_r1_click_without_crop_stays_anchor_null(db_app, anchor_data_dir):
|
||||
"""Un edge mouse_click SANS crop → anchor_id reste NULL (légitime)."""
|
||||
from services.learned_workflow_bridge import import_core_workflow_to_db
|
||||
|
||||
core = {
|
||||
"workflow_id": "wf_sess_noimg_002",
|
||||
"name": "Léa Clic Sans Image",
|
||||
"entry_nodes": ["n1"],
|
||||
"nodes": [{"node_id": "n1", "name": "A"}, {"node_id": "n2", "name": "B"}],
|
||||
"edges": [
|
||||
{
|
||||
"edge_id": "e1",
|
||||
"from_node": "n1",
|
||||
"to_node": "n2",
|
||||
"action": {
|
||||
"type": "mouse_click",
|
||||
"target": {"by_text": "Fichier", "by_role": "menu"},
|
||||
"parameters": {"button": "left"},
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
with db_app.app_context():
|
||||
result = import_core_workflow_to_db(
|
||||
core,
|
||||
machine_id="DESKTOP-TEST_windows",
|
||||
source_session_id="sess_noimg_002",
|
||||
db_session=db.session,
|
||||
)
|
||||
wf_id = result["workflow_id"]
|
||||
steps = Step.query.filter_by(workflow_id=wf_id).all()
|
||||
click = [s for s in steps if s.action_type == "click_anchor"][0]
|
||||
assert click.anchor_id is None
|
||||
assert VisualAnchor.query.count() == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. Ré-import même signature = idempotent (une seule ancre)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_reimport_idempotent_preserves_single_anchor(db_app, anchor_data_dir):
|
||||
"""Ré-importer 2× le même core (create-or-skip) → 1 seul workflow ET
|
||||
pas d'ancre dupliquée."""
|
||||
from services.learned_workflow_bridge import import_core_workflow_to_db
|
||||
|
||||
def _core():
|
||||
return {
|
||||
"workflow_id": "wf_sess_idem_003",
|
||||
"name": "Léa Idempotent",
|
||||
"entry_nodes": ["n1"],
|
||||
"nodes": [{"node_id": "n1", "name": "A"},
|
||||
{"node_id": "n2", "name": "B"}],
|
||||
"edges": [
|
||||
{
|
||||
"edge_id": "e1",
|
||||
"from_node": "n1",
|
||||
"to_node": "n2",
|
||||
"action": {
|
||||
"type": "mouse_click",
|
||||
"target": {
|
||||
"by_text": "Ouvrir",
|
||||
"anchor_image_base64": _png_b64(),
|
||||
},
|
||||
"parameters": {"button": "left"},
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
with db_app.app_context():
|
||||
import_core_workflow_to_db(
|
||||
_core(), machine_id="M", source_session_id="s1",
|
||||
db_session=db.session,
|
||||
)
|
||||
anchors_after_first = VisualAnchor.query.count()
|
||||
|
||||
second = import_core_workflow_to_db(
|
||||
_core(), machine_id="M", source_session_id="s1_rerun",
|
||||
db_session=db.session,
|
||||
)
|
||||
|
||||
assert Workflow.query.count() == 1, "pas de doublon workflow"
|
||||
assert second["created"] is False
|
||||
assert VisualAnchor.query.count() == anchors_after_first, \
|
||||
"pas d'ancre dupliquée au ré-import (skip)"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. Non-régression route HTTP : la route et R1 partagent le MÊME helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_http_route_delegates_to_shared_helper(db_app, anchor_data_dir):
|
||||
"""La route `import_learned_workflow` a été refactorée pour appeler le
|
||||
helper partagé (dé-duplication réelle). On vérifie ici que le chemin
|
||||
per-step de la route matérialise l'ancre à l'identique — sans booter l'app
|
||||
Flask complète. On reproduit la boucle de la route (build Step + helper),
|
||||
qui doit produire un anchor_id + VisualAnchor comme l'import R1."""
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
from api_v3 import learned_workflows as route_mod
|
||||
|
||||
# Garde-fou dé-duplication : la route n'a plus de logique d'ancre inline.
|
||||
src = Path(route_mod.__file__).read_text(encoding="utf-8")
|
||||
assert "materialize_anchor_from_b64" in src
|
||||
assert "save_anchor_image(" not in src, "logique inline non supprimée"
|
||||
assert "VisualAnchor(" not in src, "construction VisualAnchor inline restante"
|
||||
|
||||
# Chemin per-step de la route (identique à R1 via le helper partagé).
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_http", name="via route", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(id="step_http", workflow_id="wf_http",
|
||||
action_type="right_click_anchor", order=0, label="Menu")
|
||||
db.session.add(step)
|
||||
params = {"_anchor_image_base64": _png_b64(), "target_role": "menu"}
|
||||
materialize_anchor_from_b64(step, params, db_session=db.session)
|
||||
step.parameters = params
|
||||
db.session.commit()
|
||||
|
||||
assert step.anchor_id is not None
|
||||
assert VisualAnchor.query.get(step.anchor_id) is not None
|
||||
assert "_anchor_image_base64" not in step.parameters
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 7. Crop CORROMPU : contrat figé (pas d'exception, anchor_id NULL, clé retirée)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"corrupt_b64",
|
||||
[
|
||||
"!!! pas du base64 valide !!!", # base64 invalide
|
||||
"Zm9vYmFy", # b64 valide mais pas une image
|
||||
base64.b64encode(b"\x89PNG\x0d\x0a broken").decode("ascii"), # magic PNG cassé
|
||||
],
|
||||
ids=["b64_invalide", "b64_non_image", "png_magic_casse"],
|
||||
)
|
||||
def test_corrupt_crop_is_tolerated(db_app, anchor_data_dir, corrupt_b64):
|
||||
"""FIGE le contrat : un `_anchor_image_base64` corrompu est inexploitable.
|
||||
Le helper ne lève PAS, retourne None, laisse `anchor_id` NULL, ne crée
|
||||
aucun VisualAnchor — ET retire quand même la clé b64 de params (pop avant
|
||||
le try : un crop mort ne doit pas traîner, volumineux, dans les params)."""
|
||||
from services.anchor_image_service import materialize_anchor_from_b64
|
||||
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_corrupt", name="wf", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(id="step_corrupt", workflow_id="wf_corrupt",
|
||||
action_type="click_anchor", order=0, label="cible")
|
||||
db.session.add(step)
|
||||
|
||||
params = {"_anchor_image_base64": corrupt_b64, "target_text": "OK"}
|
||||
# Ne doit PAS lever malgré le crop corrompu.
|
||||
anchor_id = materialize_anchor_from_b64(
|
||||
step, params, db_session=db.session
|
||||
)
|
||||
|
||||
assert anchor_id is None, "crop corrompu → pas d'anchor_id"
|
||||
assert step.anchor_id is None, "anchor_id reste NULL"
|
||||
assert VisualAnchor.query.count() == 0, "aucun VisualAnchor créé"
|
||||
assert "_anchor_image_base64" not in params, \
|
||||
"clé b64 retirée même si le crop est inexploitable (pop avant try)"
|
||||
assert params.get("target_text") == "OK", "autres params préservés"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 8. workflow_id présent dans le sidecar metadata.json (parité route d'origine)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_workflow_id_written_to_sidecar_metadata(db_app, anchor_data_dir):
|
||||
"""Le `workflow_id` passé au helper est écrit dans le sidecar metadata.json
|
||||
(champ informatif — parité avec la route HTTP d'origine)."""
|
||||
from services.anchor_image_service import (
|
||||
materialize_anchor_from_b64,
|
||||
get_anchor_metadata,
|
||||
)
|
||||
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_meta", name="wf", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(id="step_meta", workflow_id="wf_meta",
|
||||
action_type="click_anchor", order=0)
|
||||
db.session.add(step)
|
||||
|
||||
params = {"_anchor_image_base64": _png_b64()}
|
||||
anchor_id = materialize_anchor_from_b64(
|
||||
step, params, db_session=db.session, workflow_id="wf_meta"
|
||||
)
|
||||
assert anchor_id is not None
|
||||
|
||||
meta = get_anchor_metadata(anchor_id)
|
||||
assert meta is not None, "metadata.json écrit"
|
||||
extra = meta.get("extra") or {}
|
||||
assert extra.get("workflow_id") == "wf_meta", \
|
||||
"workflow_id présent dans le sidecar (parité route d'origine)"
|
||||
assert extra.get("source") == "learned_import"
|
||||
|
||||
|
||||
def test_workflow_id_absent_when_not_provided(db_app, anchor_data_dir):
|
||||
"""Sans workflow_id fourni, la clé n'est pas écrite (pas de None parasite)."""
|
||||
from services.anchor_image_service import (
|
||||
materialize_anchor_from_b64,
|
||||
get_anchor_metadata,
|
||||
)
|
||||
|
||||
with db_app.app_context():
|
||||
wf = Workflow(id="wf_nom", name="wf", source="learned_import")
|
||||
db.session.add(wf)
|
||||
step = Step(id="step_nom", workflow_id="wf_nom",
|
||||
action_type="click_anchor", order=0)
|
||||
db.session.add(step)
|
||||
|
||||
anchor_id = materialize_anchor_from_b64(
|
||||
step, {"_anchor_image_base64": _png_b64()}, db_session=db.session
|
||||
)
|
||||
meta = get_anchor_metadata(anchor_id)
|
||||
extra = meta.get("extra") or {}
|
||||
assert "workflow_id" not in extra, "pas de clé workflow_id parasite"
|
||||
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test TDD — script de backfill des ancres R1 (`tools/backfill_anchors_r1.py`).
|
||||
|
||||
Vérifie :
|
||||
* matérialisation d'un step réparable (learned_import, click_anchor, anchor_id
|
||||
NULL, crop présent) → VisualAnchor + anchor_id posé + b64 retiré des params ;
|
||||
* idempotence : 2e `--apply` = no-op (aucune ancre supplémentaire) ;
|
||||
* filtre dur : un workflow `source='manual'` avec crop n'est JAMAIS touché ;
|
||||
* un step d'un action_type HORS NEEDS_ANCHOR (type_text) n'est pas matérialisé.
|
||||
|
||||
On utilise une DB SQLite sur FICHIER temporaire (pas :memory:) pour exercer la
|
||||
vraie résolution de chemin et un ré-attachement propre entre exécutions.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import importlib
|
||||
import sys
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from PIL import Image
|
||||
|
||||
_BACKEND = Path(__file__).resolve().parent.parent.parent
|
||||
_ROOT = _BACKEND.parent.parent
|
||||
for p in (str(_ROOT), str(_BACKEND)):
|
||||
if p not in sys.path:
|
||||
sys.path.insert(0, p)
|
||||
|
||||
from db.models import db, Workflow, Step, VisualAnchor # noqa: E402
|
||||
|
||||
|
||||
def _png_b64(w=10, h=6, color=(10, 120, 200)) -> str:
|
||||
img = Image.new("RGB", (w, h), color)
|
||||
buf = BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
return base64.b64encode(buf.getvalue()).decode("ascii")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def backfill_module(tmp_path, monkeypatch):
|
||||
"""Charge le module backfill et redirige le stockage PNG en temp."""
|
||||
import importlib.util
|
||||
|
||||
script = _ROOT / "tools" / "backfill_anchors_r1.py"
|
||||
spec = importlib.util.spec_from_file_location("backfill_anchors_r1", script)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
|
||||
import services.anchor_image_service as ais
|
||||
monkeypatch.setattr(ais, "DATA_DIR", tmp_path / "anchor_images")
|
||||
return mod
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def seeded_db(tmp_path):
|
||||
"""Crée une DB SQLite fichier avec 1 step réparable + 1 manual + 1 type_text."""
|
||||
db_path = tmp_path / "workflows.db"
|
||||
app = Flask("seed")
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
db.init_app(app)
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
|
||||
# Réparable : learned_import + click_anchor + anchor_id NULL + crop.
|
||||
wf = Workflow(id="wf_learned", name="Léa parcours",
|
||||
source="learned_import")
|
||||
db.session.add(wf)
|
||||
s_click = Step(id="s_click", workflow_id="wf_learned",
|
||||
action_type="click_anchor", order=0, label="Bouton")
|
||||
s_click.parameters = {"_anchor_image_base64": _png_b64(), "x_pct": 0.3}
|
||||
db.session.add(s_click)
|
||||
|
||||
# HORS NEEDS_ANCHOR : type_text avec un crop (bruit) → non réparable.
|
||||
s_text = Step(id="s_text", workflow_id="wf_learned",
|
||||
action_type="type_text", order=1)
|
||||
s_text.parameters = {"_anchor_image_base64": _png_b64(), "text": "x"}
|
||||
db.session.add(s_text)
|
||||
|
||||
# Manual avec crop → intouchable.
|
||||
wf_m = Workflow(id="wf_manual", name="Démo manuelle", source="manual")
|
||||
db.session.add(wf_m)
|
||||
s_m = Step(id="s_manual", workflow_id="wf_manual",
|
||||
action_type="click_anchor", order=0)
|
||||
s_m.parameters = {"_anchor_image_base64": _png_b64()}
|
||||
db.session.add(s_m)
|
||||
|
||||
db.session.commit()
|
||||
db.session.remove()
|
||||
|
||||
# Détacher le db de cette app pour permettre au backfill de le ré-attacher.
|
||||
return db_path
|
||||
|
||||
|
||||
def _count_anchors(db_path):
|
||||
app = Flask("count")
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
db.init_app(app)
|
||||
with app.app_context():
|
||||
n = VisualAnchor.query.count()
|
||||
click = Step.query.get("s_click")
|
||||
manual = Step.query.get("s_manual")
|
||||
text = Step.query.get("s_text")
|
||||
info = {
|
||||
"anchors": n,
|
||||
"click_anchor_id": click.anchor_id,
|
||||
"manual_anchor_id": manual.anchor_id,
|
||||
"text_anchor_id": text.anchor_id,
|
||||
"click_has_b64": "_anchor_image_base64" in click.parameters,
|
||||
}
|
||||
db.session.remove()
|
||||
return info
|
||||
|
||||
|
||||
def test_backfill_apply_materializes_repairable_only(backfill_module, seeded_db):
|
||||
"""--apply matérialise le step réparable, laisse manual et type_text intacts."""
|
||||
backfill_module.run(seeded_db, apply=True)
|
||||
|
||||
info = _count_anchors(seeded_db)
|
||||
assert info["anchors"] == 1, "une seule ancre créée (le click réparable)"
|
||||
assert info["click_anchor_id"] is not None, "anchor_id posé sur le click"
|
||||
assert info["manual_anchor_id"] is None, "workflow manual jamais touché"
|
||||
assert info["text_anchor_id"] is None, "type_text (hors NEEDS_ANCHOR) ignoré"
|
||||
assert info["click_has_b64"] is False, "b64 retiré des params après backfill"
|
||||
|
||||
|
||||
def test_backfill_is_idempotent(backfill_module, seeded_db):
|
||||
"""2e --apply = no-op : aucune ancre supplémentaire, anchor_id inchangé."""
|
||||
backfill_module.run(seeded_db, apply=True)
|
||||
first = _count_anchors(seeded_db)
|
||||
|
||||
backfill_module.run(seeded_db, apply=True)
|
||||
second = _count_anchors(seeded_db)
|
||||
|
||||
assert second["anchors"] == first["anchors"] == 1, "pas d'ancre en double"
|
||||
assert second["click_anchor_id"] == first["click_anchor_id"], \
|
||||
"anchor_id inchangé au 2e passage"
|
||||
|
||||
|
||||
def test_backfill_dry_run_writes_nothing(backfill_module, seeded_db):
|
||||
"""--dry-run (apply=False) ne crée aucune ancre ni ne modifie les params."""
|
||||
backfill_module.run(seeded_db, apply=False)
|
||||
|
||||
info = _count_anchors(seeded_db)
|
||||
assert info["anchors"] == 0, "dry-run n'écrit aucune ancre"
|
||||
assert info["click_anchor_id"] is None
|
||||
assert info["click_has_b64"] is True, "params intacts en dry-run"
|
||||
|
||||
|
||||
def test_resolve_db_path_uses_config_not_hardcoded(backfill_module, monkeypatch):
|
||||
"""Réserve R1 : la cible DB vient de la config runtime (env DATABASE_URL /
|
||||
instance_path Flask), pas d'un chemin codé en dur."""
|
||||
# Défaut : résolution via instance_path du backend VWB.
|
||||
default_path = backfill_module.resolve_vwb_db_path(None)
|
||||
assert default_path.name == "workflows.db"
|
||||
assert "instance" in str(default_path), \
|
||||
"défaut résolu via Flask instance_path du backend"
|
||||
|
||||
# Override explicite honoré.
|
||||
explicit = backfill_module.resolve_vwb_db_path("/tmp/custom_vwb.db")
|
||||
assert str(explicit) == "/tmp/custom_vwb.db"
|
||||
|
||||
# DATABASE_URL env sqlite absolu honoré.
|
||||
monkeypatch.setenv("DATABASE_URL", "sqlite:////var/data/wf.db")
|
||||
env_path = backfill_module.resolve_vwb_db_path(None)
|
||||
assert str(env_path) == "/var/data/wf.db"
|
||||
@@ -0,0 +1,56 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Garde-fou anti-divergence : l'ensemble Python `NEEDS_ANCHOR`
|
||||
(`services/anchor_action_types.py`) doit rester synchronisé avec la source
|
||||
canonique frontend `frontend_v4/src/types.ts` (entrées `needsAnchor: true`).
|
||||
|
||||
Exigence Dom : le set `needsAnchor` vient d'une source canonique, pas d'une
|
||||
liste en dur qui divergera. Ce test lit `types.ts` et compare.
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_BACKEND = Path(__file__).resolve().parent.parent.parent
|
||||
_VWB = _BACKEND.parent # .../visual_workflow_builder
|
||||
if str(_BACKEND) not in sys.path:
|
||||
sys.path.insert(0, str(_BACKEND))
|
||||
|
||||
from services.anchor_action_types import NEEDS_ANCHOR # noqa: E402
|
||||
|
||||
_TYPES_TS = _VWB / "frontend_v4" / "src" / "types.ts"
|
||||
|
||||
# Chaque entrée de ACTION_DEFINITIONS commence par `{ type: '<action>'`.
|
||||
# On segmente le fichier sur ce marqueur puis on lit le `needsAnchor` de CHAQUE
|
||||
# segment AVANT tout tableau `params:` imbriqué (dont les `type: 'string'` /
|
||||
# `type: 'number'` ne doivent PAS être confondus avec un action_type).
|
||||
_ACTION_START_RE = re.compile(r"\{\s*type:\s*'(?P<type>[a-z_]+)'")
|
||||
_NEEDS_ANCHOR_RE = re.compile(r"needsAnchor:\s*(true|false)")
|
||||
|
||||
|
||||
def _frontend_needs_anchor() -> set:
|
||||
text = _TYPES_TS.read_text(encoding="utf-8")
|
||||
starts = list(_ACTION_START_RE.finditer(text))
|
||||
result = set()
|
||||
for idx, m in enumerate(starts):
|
||||
action_type = m.group("type")
|
||||
seg_start = m.end()
|
||||
seg_end = starts[idx + 1].start() if idx + 1 < len(starts) else len(text)
|
||||
segment = text[seg_start:seg_end]
|
||||
# Le needsAnchor de l'entrée est le PREMIER de son segment (avant params).
|
||||
flag = _NEEDS_ANCHOR_RE.search(segment)
|
||||
if flag and flag.group(1) == "true":
|
||||
result.add(action_type)
|
||||
return result
|
||||
|
||||
|
||||
def test_needs_anchor_matches_frontend_types_ts():
|
||||
"""Le set Python doit être identique aux entrées needsAnchor:true du front."""
|
||||
front = _frontend_needs_anchor()
|
||||
assert front, "aucune entrée needsAnchor:true parsée depuis types.ts"
|
||||
assert set(NEEDS_ANCHOR) == front, (
|
||||
"Divergence NEEDS_ANCHOR (Python) vs types.ts (front) :\n"
|
||||
f" manquantes côté Python : {sorted(front - set(NEEDS_ANCHOR))}\n"
|
||||
f" en trop côté Python : {sorted(set(NEEDS_ANCHOR) - front)}"
|
||||
)
|
||||
Reference in New Issue
Block a user