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

feat(vwb): pont R1 import idempotent core→DB par signature trajectoire

Browse Source 
Add import_core_workflow_to_db() — create-or-update par signature de
trajectoire (décision produit Dom 23/06). Les workflows source='manual'
sont exclus du filtre de fusion. Inclut test TDD idempotent (ré-import
2× → toujours 1 seul workflow).
This commit is contained in:
Dom
2026-07-02 13:01:33 +02:00
parent 9a8242add5
commit ebed4d7546
3 changed files with 395 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
visual_workflow_builder/backend/tests/unit/__init__.py Normal file
View File

226
visual_workflow_builder/backend/tests/unit/test_import_core_workflow_to_db.py Normal file
View File

@@ -0,0 +1,226 @@
#!/usr/bin/env python3
"""
Test TDD — pont R1 : `import_core_workflow_to_db` IDEMPOTENT.
Objectif chantier R1 : une session auto-apprise (workflow core JSON) doit pouvoir
être (ré)importée en DB VWB **sans créer de doublon**. La fusion se fait par
**signature de trajectoire** (cf. `core.execution.trajectory_signature`) — décision
produit Dom 23/06 : create-or-update, pas create-only.
Coeur du test (b) : ré-importer le MÊME core_dict 2× → toujours UN seul workflow.
Ce module est volontairement isolé du chemin live :
- il ne démarre PAS l'app Flask complète (`app.py`) ;
- il lie le `db` partagé (`db.models.db`) à une SQLite **en mémoire** via une
app Flask minimale, même pattern que `tests/conftest.py` mais sans dépendances
lourdes (pas de socketio, pas de blueprints).
"""
import sys
from pathlib import Path
import pytest
from flask import Flask
# --- 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 # noqa: E402
# ---------------------------------------------------------------------------
# Fixtures DB en mémoire (app Flask minimale, db partagé)
# ---------------------------------------------------------------------------
@pytest.fixture
def db_app():
"""App Flask minimale liée à une SQLite en mémoire, schéma créé."""
app = Flask("test_import_core_workflow")
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()
# ---------------------------------------------------------------------------
# Fixtures de workflows core (format JSON appris par Léa)
# ---------------------------------------------------------------------------
def _core_workflow_bloc_notes() -> dict:
"""Workflow core minimal : ouvrir Bloc-notes et saisir du texte."""
return {
"workflow_id": "wf_sess_bloc_notes_001",
"name": "Léa Bloc-notes",
"entry_nodes": ["n1"],
"nodes": [
{"node_id": "n1", "name": "Bureau"},
{"node_id": "n2", "name": "Bloc-notes ouvert"},
],
"edges": [
{
"edge_id": "e1",
"from_node": "n1",
"to_node": "n2",
"action": {
"type": "mouse_click",
"target": {"by_text": "Bloc-notes", "by_role": "ocr"},
"parameters": {"button": "left"},
},
},
{
"edge_id": "e2",
"from_node": "n2",
"to_node": "n2",
"action": {
"type": "text_input",
"target": {"by_text": "zone de saisie"},
"parameters": {"text": "bonjour"},
},
},
],
}
def _core_workflow_calculatrice() -> dict:
"""Workflow core d'une trajectoire DIFFÉRENTE (calculatrice)."""
return {
"workflow_id": "wf_sess_calc_002",
"name": "Léa Calculatrice",
"entry_nodes": ["n1"],
"nodes": [
{"node_id": "n1", "name": "Bureau"},
{"node_id": "n2", "name": "Calculatrice"},
],
"edges": [
{
"edge_id": "e1",
"from_node": "n1",
"to_node": "n2",
"action": {
"type": "mouse_click",
"target": {"by_text": "Calculatrice", "by_role": "ocr"},
"parameters": {"button": "left"},
},
},
],
}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_import_creates_workflow_with_steps(db_app):
"""(a) Un core_dict → 1 workflow VWB créé, avec ses steps."""
from services.learned_workflow_bridge import import_core_workflow_to_db
with db_app.app_context():
result = import_core_workflow_to_db(
_core_workflow_bloc_notes(),
machine_id="DESKTOP-TEST_windows",
source_session_id="sess_bloc_notes_001",
db_session=db.session,
)
assert result["created"] is True
wf_id = result["workflow_id"]
assert wf_id
wf = Workflow.query.get(wf_id)
assert wf is not None
assert wf.source == "learned_import"
assert wf.review_status == "pending_review"
steps = Step.query.filter_by(workflow_id=wf_id).all()
assert len(steps) >= 1, "le workflow importé doit avoir au moins une étape"
def test_reimport_same_workflow_is_idempotent(db_app):
"""(b) COEUR — ré-importer le MÊME core_dict 2× → toujours 1 seul workflow."""
from services.learned_workflow_bridge import import_core_workflow_to_db
with db_app.app_context():
first = import_core_workflow_to_db(
_core_workflow_bloc_notes(),
machine_id="DESKTOP-TEST_windows",
source_session_id="sess_bloc_notes_001",
db_session=db.session,
)
second = import_core_workflow_to_db(
_core_workflow_bloc_notes(),
machine_id="DESKTOP-TEST_windows",
source_session_id="sess_bloc_notes_001_rerun",
db_session=db.session,
)
# UN seul workflow en DB malgré deux imports
assert Workflow.query.count() == 1, "ré-import du même parcours = pas de doublon"
# Le second pointe vers le même workflow, marqué non-créé
assert first["workflow_id"] == second["workflow_id"]
assert second["created"] is False
def test_different_trajectories_create_two_workflows(db_app):
"""(c) Deux trajectoires différentes → 2 workflows distincts."""
from services.learned_workflow_bridge import import_core_workflow_to_db
with db_app.app_context():
r1 = import_core_workflow_to_db(
_core_workflow_bloc_notes(),
machine_id="DESKTOP-TEST_windows",
source_session_id="sess_a",
db_session=db.session,
)
r2 = import_core_workflow_to_db(
_core_workflow_calculatrice(),
machine_id="DESKTOP-TEST_windows",
source_session_id="sess_b",
db_session=db.session,
)
assert Workflow.query.count() == 2
assert r1["workflow_id"] != r2["workflow_id"]
def test_manual_workflow_is_never_touched(db_app):
"""(d) Un workflow source='manual' préexistant n'est jamais modifié.
Même si, par construction, il partageait la signature d'un parcours importé,
la fonction ne doit cibler QUE les workflows source='learned_import'.
"""
from services.learned_workflow_bridge import import_core_workflow_to_db
with db_app.app_context():
# Workflow manuel préexistant (démo Urgence_aiva) — intouchable
manual = Workflow(
id="wf_manual_demo",
name="Urgence_aiva_demo",
description="Démo manuelle critique",
source="manual",
review_status="approved",
)
db.session.add(manual)
db.session.commit()
manual_name_before = manual.name
manual_review_before = manual.review_status
import_core_workflow_to_db(
_core_workflow_bloc_notes(),
machine_id="DESKTOP-TEST_windows",
source_session_id="sess_x",
db_session=db.session,
)
manual_after = Workflow.query.get("wf_manual_demo")
assert manual_after.name == manual_name_before
assert manual_after.review_status == manual_review_before
assert manual_after.source == "manual"