feat(anchors): matérialisation ancre partagée à l'import R1 (fix « Ancre requise ») — 2026-07-03
Les steps needsAnchor des workflows importés par le pont R1 (import_core_workflow_to_db) affichaient « Ancre requise » car R1 laissait _anchor_image_base64 dans params sans créer le VisualAnchor ni poser anchor_id. La route HTTP import_learned_workflow faisait ce travail mais en dupliquait ~50 lignes inline. Fix GÉNÉRAL et réutilisable (mandat Dom), pas une rustine click_anchor : - helper unique services/anchor_image_service.materialize_anchor_from_b64 : crée VisualAnchor + fichier PNG + pose step.anchor_id + retire le b64 de params ; ne commit pas (transaction gérée par l'appelant) ; tolérant (crop invalide → anchor_id NULL, pas de crash) ; générique pour TOUS les action_type needsAnchor. - branchement dans import_core_workflow_to_db (flux R1) ; - route HTTP refactorée pour réutiliser le même helper (dé-duplication réelle, -48 lignes) → source unique de vérité. - source canonique needsAnchor côté backend : services/anchor_action_types.py (miroir de frontend_v4/src/types.ts, garde-fou anti-divergence testé). - script tools/backfill_anchors_r1.py (--dry-run par défaut / --apply) : filtre source=learned_import ET action_type ∈ NEEDS_ANCHOR ET anchor_id NULL ET crop présent ; DB résolue depuis la config runtime VWB (pas en dur) ; dry-run affiche le chemin DB résolu + estime PNG/disque ; idempotent. NON exécuté en --apply (backup + arbitrage Dom requis avant, jamais la DB clinique DGX). TDD : tests rouges d'abord puis verts. 20 tests neufs (matérialisation 6 types d'ancre, b64 retiré, sans-crop→NULL, R1 clic, ré-import idempotent, dé-dup route, backfill apply/idempotent/dry-run/filtre manual+hors-needsAnchor, canonique). 24 tests unit VERTS, 0 régression (test_import_core_workflow_to_db intact). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
@@ -0,0 +1,353 @@
|
||||
#!/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
|
||||
@@ -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