feat: WorkflowIR — représentation intermédiaire du savoir-faire
Format canonique entre RawTrace (capture) et ExecutionPlan (exécution). C'est ce que Léa a COMPRIS en observant l'utilisateur. - WorkflowIR : steps, variables, intentions, pré/postconditions - IRBuilder : transforme les événements bruts en WorkflowIR via gemma4 - Générique : fonctionne pour TIM, compta, RH, stocks — le domaine est une couche par-dessus - Versionné, sérialisable JSON, save/load - Détection automatique des variables (texte saisi → substituable) - 18 tests (format, sérialisation, builder, segmentation, variables) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
261
tests/unit/test_workflow_ir.py
Normal file
261
tests/unit/test_workflow_ir.py
Normal file
@@ -0,0 +1,261 @@
|
||||
"""
|
||||
Tests du WorkflowIR et de l'IRBuilder.
|
||||
|
||||
Vérifie que :
|
||||
- Le format WorkflowIR est correct (sérialisation, désérialisation, versioning)
|
||||
- L'IRBuilder segmente et comprend les traces brutes
|
||||
- Les variables sont détectées et substituables
|
||||
- Le tout fonctionne sans gemma4 (fallback gracieux)
|
||||
"""
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
_ROOT = str(Path(__file__).resolve().parents[2])
|
||||
if _ROOT not in sys.path:
|
||||
sys.path.insert(0, _ROOT)
|
||||
|
||||
from core.workflow.workflow_ir import WorkflowIR, Step, Action, Variable
|
||||
from core.workflow.ir_builder import IRBuilder
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# WorkflowIR — format et sérialisation
|
||||
# =========================================================================
|
||||
|
||||
|
||||
class TestWorkflowIR:
|
||||
|
||||
def test_creation_vide(self):
|
||||
ir = WorkflowIR.new("Test workflow")
|
||||
assert ir.workflow_id.startswith("wf_")
|
||||
assert ir.version == 1
|
||||
assert ir.name == "Test workflow"
|
||||
assert ir.steps == []
|
||||
assert ir.variables == []
|
||||
|
||||
def test_ajout_etapes(self):
|
||||
ir = WorkflowIR.new("Test")
|
||||
ir.add_step("Ouvrir l'application", actions=[
|
||||
{"type": "click", "target": "icône app"},
|
||||
{"type": "wait", "duration_ms": 2000},
|
||||
])
|
||||
ir.add_step("Saisir les données", actions=[
|
||||
{"type": "type", "text": "bonjour"},
|
||||
])
|
||||
assert len(ir.steps) == 2
|
||||
assert ir.steps[0].intent == "Ouvrir l'application"
|
||||
assert len(ir.steps[0].actions) == 2
|
||||
assert ir.steps[0].actions[0].type == "click"
|
||||
|
||||
def test_ajout_variables(self):
|
||||
ir = WorkflowIR.new("Test")
|
||||
ir.add_variable("patient", description="Nom du patient", source="screen")
|
||||
ir.add_variable("code", description="Code à saisir", default="A00.0")
|
||||
assert len(ir.variables) == 2
|
||||
assert ir.variables[0].name == "patient"
|
||||
assert ir.variables[1].default == "A00.0"
|
||||
|
||||
def test_serialisation_json(self):
|
||||
ir = WorkflowIR.new("Mon workflow", domain="tim_codage")
|
||||
ir.add_step("Étape 1")
|
||||
ir.add_variable("var1", description="Une variable")
|
||||
|
||||
json_str = ir.to_json()
|
||||
data = json.loads(json_str)
|
||||
|
||||
assert data["name"] == "Mon workflow"
|
||||
assert data["domain"] == "tim_codage"
|
||||
assert len(data["steps"]) == 1
|
||||
assert len(data["variables"]) == 1
|
||||
|
||||
def test_deserialisation_json(self):
|
||||
ir = WorkflowIR.new("Test roundtrip")
|
||||
ir.add_step("Ouvrir", actions=[{"type": "click", "target": "bouton"}])
|
||||
ir.add_variable("v1", description="test")
|
||||
|
||||
json_str = ir.to_json()
|
||||
ir2 = WorkflowIR.from_json(json_str)
|
||||
|
||||
assert ir2.name == "Test roundtrip"
|
||||
assert len(ir2.steps) == 1
|
||||
assert ir2.steps[0].intent == "Ouvrir"
|
||||
assert ir2.steps[0].actions[0].type == "click"
|
||||
assert len(ir2.variables) == 1
|
||||
|
||||
def test_save_et_load(self):
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
try:
|
||||
ir = WorkflowIR.new("Save test")
|
||||
ir.add_step("Étape 1")
|
||||
path = ir.save(tmpdir)
|
||||
|
||||
assert path.is_file()
|
||||
|
||||
ir2 = WorkflowIR.load(str(path))
|
||||
assert ir2.name == "Save test"
|
||||
assert len(ir2.steps) == 1
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
|
||||
def test_increment_version(self):
|
||||
ir = WorkflowIR.new("Versionning")
|
||||
assert ir.version == 1
|
||||
|
||||
ir2 = ir.increment_version()
|
||||
assert ir2.version == 2
|
||||
assert ir.version == 1 # Original inchangé
|
||||
assert ir2.name == "Versionning"
|
||||
|
||||
def test_domaine_generique(self):
|
||||
"""Le WorkflowIR est générique — pas lié à un métier."""
|
||||
for domain in ["tim_codage", "comptabilite", "rh_paie", "stocks", "generic"]:
|
||||
ir = WorkflowIR.new("Test", domain=domain)
|
||||
assert ir.domain == domain
|
||||
|
||||
def test_etape_optionnelle(self):
|
||||
ir = WorkflowIR.new("Test")
|
||||
ir.add_step("Vérification facultative", is_optional=True)
|
||||
assert ir.steps[0].is_optional is True
|
||||
|
||||
def test_etape_boucle(self):
|
||||
ir = WorkflowIR.new("Test")
|
||||
ir.add_step("Traiter chaque dossier", is_loop=True, loop_variable="dossier")
|
||||
assert ir.steps[0].is_loop is True
|
||||
assert ir.steps[0].loop_variable == "dossier"
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# IRBuilder — construction depuis RawTrace
|
||||
# =========================================================================
|
||||
|
||||
|
||||
class TestIRBuilder:
|
||||
|
||||
def _make_events(self):
|
||||
"""Créer des événements bruts simulés (comme live_events.jsonl)."""
|
||||
return [
|
||||
{"event": {"type": "mouse_click", "pos": [400, 580], "window": {"title": "Lea : Explorateur"}, "timestamp": 100.0, "vision_info": {"text": "Rechercher"}}},
|
||||
{"event": {"type": "text_input", "text": "blocnote", "window": {"title": "Rechercher"}, "timestamp": 102.0}},
|
||||
{"event": {"type": "key_combo", "keys": ["enter"], "window": {"title": "Rechercher"}, "timestamp": 103.0}},
|
||||
{"event": {"type": "heartbeat", "timestamp": 104.0}}, # Parasite — doit être filtré
|
||||
{"event": {"type": "mouse_click", "pos": [300, 200], "window": {"title": "Rechercher"}, "timestamp": 105.0, "vision_info": {"text": "Bloc-notes"}}},
|
||||
{"event": {"type": "mouse_click", "pos": [500, 300], "window": {"title": "Sans titre – Bloc-notes"}, "timestamp": 112.0, "vision_info": {"text": ""}}},
|
||||
{"event": {"type": "text_input", "text": "Bonjour le monde", "window": {"title": "*Sans titre – Bloc-notes"}, "timestamp": 113.0}},
|
||||
{"event": {"type": "key_combo", "keys": ["ctrl", "s"], "window": {"title": "*Sans titre – Bloc-notes"}, "timestamp": 115.0}},
|
||||
]
|
||||
|
||||
def test_builder_sans_gemma4(self):
|
||||
"""Le builder fonctionne même sans gemma4 (fallback gracieux)."""
|
||||
builder = IRBuilder(gemma4_port="99999") # Port invalide
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, session_id="test_sess", domain="generic", name="Test")
|
||||
|
||||
assert ir.name == "Test"
|
||||
assert ir.learned_from == "test_sess"
|
||||
assert len(ir.steps) >= 1
|
||||
assert len(ir.applications) >= 1
|
||||
|
||||
def test_filtre_heartbeat(self):
|
||||
"""Les heartbeat sont filtrés."""
|
||||
builder = IRBuilder(gemma4_port="99999")
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Test")
|
||||
|
||||
# Vérifier qu'aucune action n'est de type heartbeat
|
||||
for step in ir.steps:
|
||||
for action in step.actions:
|
||||
assert action.type != "heartbeat"
|
||||
|
||||
def test_detection_applications(self):
|
||||
"""Les applications utilisées sont détectées."""
|
||||
builder = IRBuilder(gemma4_port="99999")
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Test")
|
||||
|
||||
assert "Bloc-notes" in ir.applications or "Explorateur" in ir.applications
|
||||
|
||||
def test_detection_variables(self):
|
||||
"""Le texte saisi est détecté comme variable."""
|
||||
builder = IRBuilder(gemma4_port="99999")
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Test")
|
||||
|
||||
# Le texte "blocnote" et "Bonjour le monde" doivent être des variables
|
||||
assert len(ir.variables) >= 1
|
||||
var_defaults = [v.default for v in ir.variables]
|
||||
assert any("blocnote" in d or "Bonjour" in d for d in var_defaults)
|
||||
|
||||
def test_segmentation_par_application(self):
|
||||
"""Les événements sont segmentés par changement d'application."""
|
||||
builder = IRBuilder(gemma4_port="99999")
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Test")
|
||||
|
||||
# Au moins 2 étapes (Explorateur → Bloc-notes)
|
||||
assert len(ir.steps) >= 2
|
||||
|
||||
def test_actions_dans_les_etapes(self):
|
||||
"""Chaque étape contient les bonnes actions."""
|
||||
builder = IRBuilder(gemma4_port="99999")
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Test")
|
||||
|
||||
all_actions = []
|
||||
for step in ir.steps:
|
||||
all_actions.extend(step.actions)
|
||||
|
||||
types = [a.type for a in all_actions]
|
||||
assert "click" in types
|
||||
assert "type" in types
|
||||
assert "key_combo" in types
|
||||
|
||||
def test_workflow_ir_complet_roundtrip(self):
|
||||
"""Build → JSON → reload → même contenu."""
|
||||
builder = IRBuilder(gemma4_port="99999")
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Roundtrip test", domain="compta")
|
||||
json_str = ir.to_json()
|
||||
ir2 = WorkflowIR.from_json(json_str)
|
||||
|
||||
assert ir2.name == "Roundtrip test"
|
||||
assert ir2.domain == "compta"
|
||||
assert len(ir2.steps) == len(ir.steps)
|
||||
assert len(ir2.variables) == len(ir.variables)
|
||||
|
||||
@patch("requests.post")
|
||||
def test_builder_avec_gemma4_mock(self, mock_post):
|
||||
"""Avec gemma4, le builder enrichit les intentions."""
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.ok = True
|
||||
mock_resp.json.return_value = {
|
||||
"message": {"content": (
|
||||
"INTENTION: Rechercher et ouvrir le Bloc-notes\n"
|
||||
"AVANT: L'explorateur de fichiers est ouvert\n"
|
||||
"APRÈS: Le Bloc-notes est ouvert et actif"
|
||||
)}
|
||||
}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
builder = IRBuilder()
|
||||
events = self._make_events()
|
||||
|
||||
ir = builder.build(events, name="Test gemma4")
|
||||
|
||||
# Au moins une étape doit avoir une intention enrichie
|
||||
intents = [s.intent for s in ir.steps]
|
||||
has_enriched = any("Bloc-notes" in i or "Rechercher" in i for i in intents)
|
||||
assert has_enriched or len(ir.steps) >= 1 # Fallback acceptable
|
||||
Reference in New Issue
Block a user