Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
bb4ed2a75d6c8c1eb33407c1a1a3d1fb4ccb4650
rpa_vision_v3/tests/unit/test_shadow_validator.py
Dom 172167f6c0 feat: Léa apprentissage — mode Shadow amélioré (observation + validation) 
Aspect 3/4 Léa : Léa montre ce qu'elle comprend pendant l'enregistrement.

ShadowObserver (observation temps réel) :
- Segmentation incrémentale en UnderstoodStep (changement app, pause, Ctrl+S)
- Détection de variables pendant la saisie (typage : date, email, code, texte)
- Notifications 4 niveaux : INFO, DECOUVERTE, QUESTION, VARIABLE
- Heartbeat périodique, hook gemma4 optionnel (asynchrone)
- Thread-safe (RLock), singleton partagé
- Performance : 1000 events en < 500ms

ShadowValidator (feedback utilisateur) :
- 6 actions : validate, correct, undo, cancel, merge_next, split
- Reconstruit un WorkflowIR propre avec variables substituées
- Historique complet des feedbacks

5 endpoints REST /api/v1/shadow/* :
- start, stop, feedback, understanding, build

Hook non-bloquant dans stream_event() (try/except, no-op si inactif).
Mode optionnel : pas d'impact tant que shadow/start n'est pas appelé.

54 tests (26 observer + 28 validator), 0 régression.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-10 09:04:37 +02:00

532 lines
17 KiB
Python
Raw Blame History

"""
Tests du ShadowValidator — feedback utilisateur et reconstruction WorkflowIR.
Vérifie que :
- Les feedbacks (validate/correct/undo/merge_next/split/cancel) sont appliqués
- Le WorkflowIR final est bien reconstruit à partir des étapes corrigées
- Les variables sont détectées dans les actions finales
- L'historique des feedbacks est conservé
- Les erreurs (index invalide, action inconnue) sont gérées proprement
"""
import sys
import time
from pathlib import Path
from unittest.mock import MagicMock
import pytest
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
from core.workflow.shadow_observer import ShadowObserver, UnderstoodStep
from core.workflow.shadow_validator import FeedbackResult, ShadowValidator
from core.workflow.workflow_ir import WorkflowIR
# =========================================================================
# Fixtures
# =========================================================================
def _make_step(step_index=1, intent="Ouvrir Firefox", app="Firefox",
events=None, **kwargs):
return UnderstoodStep(
step_index=step_index,
intent=intent,
app_name=app,
window_title=app,
events=events or [
{"type": "mouse_click", "window": {"title": app},
"vision_info": {"text": "bouton"}, "timestamp": 100.0}
],
started_at=100.0,
ended_at=101.0,
**kwargs,
)
def _make_type_step(texte="Bonjour", app="Bloc-notes", step_index=1):
return UnderstoodStep(
step_index=step_index,
intent=f"Écrire « {texte} »",
app_name=app,
window_title=app,
events=[
{"type": "text_input", "text": texte,
"window": {"title": app}, "timestamp": 100.0}
],
started_at=100.0,
)
def _make_3_steps():
return [
_make_step(1, "Ouvrir le Bloc-notes", "Bloc-notes"),
_make_type_step("Bonjour le monde", step_index=2),
_make_step(3, "Sauvegarder", "Bloc-notes", events=[
{"type": "key_combo", "keys": ["ctrl", "s"],
"window": {"title": "Bloc-notes"}, "timestamp": 103.0}
]),
]
# =========================================================================
# Initialisation
# =========================================================================
class TestShadowValidatorBase:
def test_creation(self):
v = ShadowValidator()
assert v.steps == []
assert v.history == []
assert not v.is_cancelled
def test_set_steps(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
assert len(v.steps) == 3
assert v.steps[0].intent == "Ouvrir le Bloc-notes"
def test_clone_protege_mutation(self):
"""set_steps clone les étapes pour éviter les mutations externes."""
v = ShadowValidator()
steps = _make_3_steps()
v.set_steps(steps)
steps[0].intent = "HACKED"
assert v.steps[0].intent == "Ouvrir le Bloc-notes"
# =========================================================================
# validate
# =========================================================================
class TestValidatorValidate:
def test_validate_etape(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "validate", "step_index": 1})
assert result.ok is True
assert v.steps[0].validated is True
assert v.steps[0].confidence >= 0.95
assert v.steps[0].intent_provisoire is False
def test_validate_index_invalide(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "validate", "step_index": 99})
assert result.ok is False
assert "invalide" in result.message.lower()
def test_validate_toutes_les_etapes(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
for i in range(1, 4):
v.apply_feedback({"action": "validate", "step_index": i})
assert all(s.validated for s in v.steps)
# =========================================================================
# correct
# =========================================================================
class TestValidatorCorrect:
def test_correct_intent(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({
"action": "correct",
"step_index": 1,
"new_intent": "Démarrer la rédaction d'un email",
})
assert result.ok is True
assert v.steps[0].intent == "Démarrer la rédaction d'un email"
assert v.steps[0].corrected is True
assert v.steps[0].validated is True # Corriger = valider implicitement
assert "old_intent" in result.data
def test_correct_intent_vide(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({
"action": "correct",
"step_index": 1,
"new_intent": "",
})
assert result.ok is False
assert v.steps[0].corrected is False
# =========================================================================
# undo
# =========================================================================
class TestValidatorUndo:
def test_undo_etape(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "undo", "step_index": 2})
assert result.ok is True
assert v.steps[1].cancelled is True
def test_undo_exclut_etape_du_workflow(self):
"""Une étape undo ne doit pas apparaître dans le WorkflowIR final."""
v = ShadowValidator()
v.set_steps(_make_3_steps())
v.apply_feedback({"action": "undo", "step_index": 2})
ir = v.build_workflow_ir(session_id="s1", name="Test")
assert ir is not None
assert len(ir.steps) == 2 # 3 - 1 = 2
# =========================================================================
# merge_next
# =========================================================================
class TestValidatorMergeNext:
def test_merge_deux_etapes(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "merge_next", "step_index": 1})
assert result.ok is True
assert len(v.steps) == 2 # 3 - 1 = 2
def test_merge_conserve_les_events(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
total_events_before = sum(len(s.events) for s in v.steps)
v.apply_feedback({"action": "merge_next", "step_index": 2})
total_events_after = sum(len(s.events) for s in v.steps)
assert total_events_before == total_events_after
def test_merge_derniere_etape_echoue(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "merge_next", "step_index": 3})
assert result.ok is False
# =========================================================================
# split
# =========================================================================
class TestValidatorSplit:
def test_split_en_deux(self):
v = ShadowValidator()
multi_events_step = _make_step(
1, "Étape composite",
events=[
{"type": "mouse_click", "window": {"title": "App"},
"timestamp": 100.0, "vision_info": {}},
{"type": "text_input", "text": "partie 1",
"window": {"title": "App"}, "timestamp": 101.0},
{"type": "text_input", "text": "partie 2",
"window": {"title": "App"}, "timestamp": 102.0},
],
)
v.set_steps([multi_events_step])
result = v.apply_feedback({
"action": "split",
"step_index": 1,
"at_event_index": 2,
})
assert result.ok is True
assert len(v.steps) == 2
assert len(v.steps[0].events) == 2
assert len(v.steps[1].events) == 1
def test_split_index_invalide(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({
"action": "split",
"step_index": 1,
"at_event_index": 99,
})
assert result.ok is False
# =========================================================================
# cancel
# =========================================================================
class TestValidatorCancel:
def test_cancel_workflow(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "cancel"})
assert result.ok is True
assert v.is_cancelled
def test_cancel_build_retourne_none(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
v.apply_feedback({"action": "cancel"})
ir = v.build_workflow_ir(session_id="s1", name="Test")
assert ir is None
# =========================================================================
# Action inconnue
# =========================================================================
class TestValidatorUnknownAction:
def test_action_inconnue(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
result = v.apply_feedback({"action": "do_magic"})
assert result.ok is False
assert "inconnue" in result.message.lower()
# =========================================================================
# Construction du WorkflowIR
# =========================================================================
class TestValidatorBuild:
def test_build_workflow_ir(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
ir = v.build_workflow_ir(
session_id="sess_test",
name="Mon workflow",
domain="generic",
)
assert ir is not None
assert isinstance(ir, WorkflowIR)
assert ir.name == "Mon workflow"
assert ir.learned_from == "sess_test"
assert len(ir.steps) == 3
def test_build_with_variables(self):
"""Les textes saisis deviennent des variables dans le WorkflowIR."""
v = ShadowValidator()
v.set_steps([
_make_type_step("Jean Dupont", step_index=1),
_make_type_step("jean@example.com", app="Email", step_index=2),
])
ir = v.build_workflow_ir(session_id="s1", name="Test")
assert len(ir.variables) == 2
# Les actions de type type doivent référencer les variables
for step in ir.steps:
for action in step.actions:
if action.type == "type":
assert action.variable is True
assert action.text.startswith("{")
def test_build_respecte_corrections(self):
"""Les intentions corrigées se retrouvent dans le WorkflowIR."""
v = ShadowValidator()
v.set_steps(_make_3_steps())
v.apply_feedback({
"action": "correct",
"step_index": 1,
"new_intent": "Lancer l'application de prise de notes",
})
ir = v.build_workflow_ir(session_id="s1", name="Test")
assert ir.steps[0].intent == "Lancer l'application de prise de notes"
def test_build_exclut_etapes_annulees(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
v.apply_feedback({"action": "undo", "step_index": 2})
ir = v.build_workflow_ir(session_id="s1", name="Test")
assert len(ir.steps) == 2
def test_build_require_all_validated(self):
"""Avec require_all_validated, erreur si une étape n'est pas validée."""
v = ShadowValidator()
v.set_steps(_make_3_steps())
v.apply_feedback({"action": "validate", "step_index": 1})
# Étapes 2 et 3 pas validées
with pytest.raises(ValueError):
v.build_workflow_ir(
session_id="s1", name="Test", require_all_validated=True
)
def test_build_applications_detectees(self):
v = ShadowValidator()
v.set_steps([
_make_step(1, "Ouvrir Firefox", "Firefox"),
_make_step(2, "Écrire", "Bloc-notes"),
])
ir = v.build_workflow_ir(session_id="s1", name="Test")
assert "Firefox" in ir.applications
assert "Bloc-notes" in ir.applications
# =========================================================================
# Historique
# =========================================================================
class TestValidatorHistory:
def test_historique_feedbacks(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
v.apply_feedback({"action": "validate", "step_index": 1})
v.apply_feedback({
"action": "correct", "step_index": 2,
"new_intent": "Écrire le texte"
})
v.apply_feedback({"action": "undo", "step_index": 3})
history = v.history
assert len(history) == 3
assert history[0].action == "validate"
assert history[1].action == "correct"
assert history[2].action == "undo"
def test_apply_feedbacks_batch(self):
v = ShadowValidator()
v.set_steps(_make_3_steps())
results = v.apply_feedbacks([
{"action": "validate", "step_index": 1},
{"action": "validate", "step_index": 2},
{"action": "undo", "step_index": 3},
])
assert len(results) == 3
assert all(r.ok for r in results)
# =========================================================================
# Intégration ShadowObserver + ShadowValidator
# =========================================================================
class TestShadowObserverValidatorIntegration:
def test_observer_vers_validator(self):
"""Flow complet : Observer → Validator → WorkflowIR."""
obs = ShadowObserver()
obs.start("sess_flow")
# Simuler des événements
obs.observe_event("sess_flow", {
"type": "mouse_click",
"window": {"title": "Menu Démarrer", "app_name": "Menu Démarrer"},
"vision_info": {"text": "Rechercher"},
"timestamp": 100.0,
})
obs.observe_event("sess_flow", {
"type": "text_input",
"text": "blocnote",
"window": {"title": "Menu Démarrer", "app_name": "Menu Démarrer"},
"timestamp": 100.5,
})
obs.observe_event("sess_flow", {
"type": "key_combo",
"keys": ["enter"],
"window": {"title": "Menu Démarrer", "app_name": "Menu Démarrer"},
"timestamp": 101.0,
})
# Changement d'application
obs.observe_event("sess_flow", {
"type": "text_input",
"text": "Hello world",
"window": {"title": "Sans titre - Bloc-notes", "app_name": "Bloc-notes"},
"timestamp": 105.0,
})
obs.stop("sess_flow")
# Récupérer les étapes
internals = obs.get_steps_internal("sess_flow")
assert len(internals) >= 2
# Passer au validator
validator = ShadowValidator()
validator.set_steps(internals)
# Valider la première étape, corriger la seconde
validator.apply_feedback({"action": "validate", "step_index": 1})
validator.apply_feedback({
"action": "correct",
"step_index": 2,
"new_intent": "Écrire un texte de démonstration",
})
# Construire le WorkflowIR
ir = validator.build_workflow_ir(
session_id="sess_flow",
name="Flow de test",
domain="generic",
)
assert ir is not None
assert len(ir.steps) >= 2
assert ir.steps[1].intent == "Écrire un texte de démonstration"
assert len(ir.variables) >= 1 # Au moins "blocnote" ou "Hello world"
def test_undo_puis_build(self):
obs = ShadowObserver()
obs.start("sess_undo")
for i in range(3):
obs.observe_event("sess_undo", {
"type": "mouse_click",
"window": {"title": f"App{i}"},
"vision_info": {"text": "bouton"},
"timestamp": 100.0 + i * 6.0, # > 4s pour créer des segments
})
obs.stop("sess_undo")
validator = ShadowValidator()
validator.set_steps(obs.get_steps_internal("sess_undo"))
nb_before = len(validator.steps)
assert nb_before >= 2
validator.apply_feedback({"action": "undo", "step_index": 1})
ir = validator.build_workflow_ir(session_id="sess_undo", name="Test")
assert len(ir.steps) == nb_before - 1
Reference in New Issue View Git Blame Copy Permalink