cc673755f7
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>
262 lines
9.7 KiB
Python
262 lines
9.7 KiB
Python
"""
|
||
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
|