172167f6c0
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>
431 lines
15 KiB
Python
431 lines
15 KiB
Python
"""
|
|
Tests du ShadowObserver — observation temps réel de Léa.
|
|
|
|
Vérifie que :
|
|
- L'observer démarre et s'arrête correctement
|
|
- Les événements sont segmentés en étapes logiques
|
|
- Les variables sont détectées pendant la frappe
|
|
- Les notifications sont émises avec le bon niveau
|
|
- La compréhension est accessible en temps réel
|
|
- Le callback de notification est appelé
|
|
- Les événements parasites (heartbeat) sont ignorés
|
|
"""
|
|
|
|
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 (
|
|
NiveauNotification,
|
|
NotificationShadow,
|
|
ShadowObserver,
|
|
UnderstoodStep,
|
|
get_shared_observer,
|
|
)
|
|
|
|
|
|
# =========================================================================
|
|
# Fixtures
|
|
# =========================================================================
|
|
|
|
|
|
def _evt_click(text="Rechercher", title="Explorateur", ts=100.0):
|
|
return {
|
|
"type": "mouse_click",
|
|
"pos": [400, 580],
|
|
"window": {"title": title, "app_name": title.split(" - ")[-1] if " - " in title else title},
|
|
"timestamp": ts,
|
|
"vision_info": {"text": text},
|
|
}
|
|
|
|
|
|
def _evt_type(text="bonjour", title="Bloc-notes", ts=101.0):
|
|
return {
|
|
"type": "text_input",
|
|
"text": text,
|
|
"window": {"title": title, "app_name": title},
|
|
"timestamp": ts,
|
|
}
|
|
|
|
|
|
def _evt_key(keys=None, title="Bloc-notes", ts=102.0):
|
|
return {
|
|
"type": "key_combo",
|
|
"keys": keys or ["enter"],
|
|
"window": {"title": title, "app_name": title},
|
|
"timestamp": ts,
|
|
}
|
|
|
|
|
|
def _evt_heartbeat(ts=103.0):
|
|
return {"type": "heartbeat", "timestamp": ts}
|
|
|
|
|
|
def _make_session_events():
|
|
"""Scénario typique : ouvrir Bloc-notes, taper du texte, sauvegarder."""
|
|
return [
|
|
_evt_click(text="Rechercher", title="Menu Démarrer", ts=100.0),
|
|
_evt_type(text="blocnote", title="Menu Démarrer", ts=101.0),
|
|
_evt_key(keys=["enter"], title="Menu Démarrer", ts=102.0),
|
|
_evt_heartbeat(ts=103.0), # Doit être ignoré
|
|
# Changement d'application → nouveau segment attendu
|
|
_evt_click(text="", title="Sans titre - Bloc-notes", ts=105.0),
|
|
_evt_type(text="Bonjour le monde", title="Sans titre - Bloc-notes", ts=106.0),
|
|
_evt_key(keys=["ctrl", "s"], title="Sans titre - Bloc-notes", ts=108.0),
|
|
]
|
|
|
|
|
|
# =========================================================================
|
|
# Tests de base
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverBase:
|
|
|
|
def test_start_et_stop(self):
|
|
obs = ShadowObserver()
|
|
obs.start("sess_test")
|
|
assert obs.has_session("sess_test")
|
|
obs.stop("sess_test")
|
|
assert obs.has_session("sess_test") # stop ne supprime pas
|
|
obs.reset("sess_test")
|
|
assert not obs.has_session("sess_test")
|
|
|
|
def test_auto_start_sur_premier_event(self):
|
|
"""observe_event() sans start() doit auto-démarrer la session."""
|
|
obs = ShadowObserver()
|
|
obs.observe_event("sess_auto", _evt_click())
|
|
assert obs.has_session("sess_auto")
|
|
steps = obs.get_understanding("sess_auto")
|
|
assert len(steps) >= 1
|
|
|
|
def test_heartbeat_ignore(self):
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_heartbeat())
|
|
steps = obs.get_understanding("s1")
|
|
assert len(steps) == 0 # Aucune étape créée par un heartbeat
|
|
|
|
def test_focus_change_ignore(self):
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", {"type": "focus_change", "timestamp": 100})
|
|
assert len(obs.get_understanding("s1")) == 0
|
|
|
|
|
|
# =========================================================================
|
|
# Segmentation
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverSegmentation:
|
|
|
|
def test_segmentation_par_changement_app(self):
|
|
"""Un changement d'application crée un nouveau segment."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(title="Firefox"))
|
|
obs.observe_event("s1", _evt_click(title="Firefox", ts=100.5))
|
|
obs.observe_event("s1", _evt_click(title="Bloc-notes", ts=101.0))
|
|
|
|
steps = obs.get_understanding("s1")
|
|
assert len(steps) >= 2 # Au moins 2 segments
|
|
|
|
def test_segmentation_par_pause_longue(self):
|
|
"""Une pause > 4s coupe le segment."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(title="App1", ts=100.0))
|
|
obs.observe_event("s1", _evt_click(title="App1", ts=100.5))
|
|
# Pause de 10 secondes
|
|
obs.observe_event("s1", _evt_click(title="App1", ts=110.5))
|
|
|
|
steps = obs.get_understanding("s1")
|
|
assert len(steps) >= 2
|
|
|
|
def test_segment_complet(self):
|
|
"""Scénario complet : Bloc-notes + texte + save."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
for evt in _make_session_events():
|
|
obs.observe_event("s1", evt)
|
|
obs.stop("s1")
|
|
|
|
steps = obs.get_understanding("s1")
|
|
assert len(steps) >= 2 # Au moins Menu Démarrer + Bloc-notes
|
|
|
|
# Au moins une étape doit mentionner le Bloc-notes
|
|
intents = " ".join(s["intent"].lower() for s in steps)
|
|
assert "bloc" in intents or "enregistr" in intents or "sauvegard" in intents or \
|
|
"écrir" in intents or "text" in intents.lower()
|
|
|
|
|
|
# =========================================================================
|
|
# Intention et raffinement
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverIntent:
|
|
|
|
def test_intent_recherche(self):
|
|
"""Clic + saisie + entrée → 'Rechercher X'."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(text="Champ recherche", title="Explorateur", ts=100.0))
|
|
obs.observe_event("s1", _evt_type(text="calculatrice", title="Explorateur", ts=100.5))
|
|
obs.observe_event("s1", _evt_key(keys=["enter"], title="Explorateur", ts=101.0))
|
|
|
|
current = obs.get_current_step("s1")
|
|
assert current is not None
|
|
assert "recherch" in current["intent"].lower() or "calculatrice" in current["intent"].lower()
|
|
|
|
def test_intent_ctrl_s(self):
|
|
"""Ctrl+S → 'Sauvegarder'."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(title="Bloc-notes", ts=100.0))
|
|
obs.observe_event("s1", _evt_key(keys=["ctrl", "s"], title="Bloc-notes", ts=100.5))
|
|
|
|
current = obs.get_current_step("s1")
|
|
assert current is not None
|
|
assert "sauvegard" in current["intent"].lower()
|
|
|
|
def test_intent_ecrire(self):
|
|
"""Saisie seule → 'Écrire'."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_type(text="Un texte libre", title="Bloc-notes"))
|
|
|
|
current = obs.get_current_step("s1")
|
|
assert current is not None
|
|
assert "écri" in current["intent"].lower() or "text" in current["intent"].lower()
|
|
|
|
def test_confidence_augmente_avec_contexte(self):
|
|
"""La confiance augmente quand le contexte devient clair."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(title="App"))
|
|
c1 = obs.get_current_step("s1")["confidence"]
|
|
|
|
obs.observe_event("s1", _evt_key(keys=["ctrl", "s"], title="App"))
|
|
c2 = obs.get_current_step("s1")["confidence"]
|
|
|
|
assert c2 >= c1
|
|
|
|
|
|
# =========================================================================
|
|
# Détection de variables
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverVariables:
|
|
|
|
def test_variable_detectee_lors_saisie(self):
|
|
"""Une saisie texte > 3 caractères crée une variable."""
|
|
obs = ShadowObserver()
|
|
notifs = []
|
|
obs._notify_callback = lambda n: notifs.append(n)
|
|
obs.start("s1")
|
|
|
|
obs.observe_event("s1", _evt_type(text="Jean Dupont", title="Formulaire"))
|
|
|
|
var_notifs = [n for n in notifs if n.niveau == NiveauNotification.VARIABLE]
|
|
assert len(var_notifs) == 1
|
|
assert "Jean Dupont" in var_notifs[0].message
|
|
assert var_notifs[0].data["variable_name"].startswith("texte_")
|
|
|
|
def test_variable_type_date(self):
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_type(text="15/03/2026", title="Formulaire"))
|
|
|
|
current = obs.get_current_step("s1")
|
|
assert len(current["variables_detectees"]) == 1
|
|
|
|
def test_variable_type_email(self):
|
|
obs = ShadowObserver()
|
|
notifs = []
|
|
obs._notify_callback = lambda n: notifs.append(n)
|
|
obs.start("s1")
|
|
|
|
obs.observe_event("s1", _evt_type(text="jean@example.com", title="Formulaire"))
|
|
|
|
var_notifs = [n for n in notifs if n.niveau == NiveauNotification.VARIABLE]
|
|
assert len(var_notifs) == 1
|
|
assert "e-mail" in var_notifs[0].message or "mail" in var_notifs[0].message
|
|
|
|
def test_texte_court_ignore(self):
|
|
"""Un texte de moins de 3 caractères n'est pas une variable."""
|
|
obs = ShadowObserver()
|
|
notifs = []
|
|
obs._notify_callback = lambda n: notifs.append(n)
|
|
obs.start("s1")
|
|
|
|
obs.observe_event("s1", _evt_type(text="ab", title="App"))
|
|
|
|
var_notifs = [n for n in notifs if n.niveau == NiveauNotification.VARIABLE]
|
|
assert len(var_notifs) == 0
|
|
|
|
|
|
# =========================================================================
|
|
# Notifications et callbacks
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverNotifications:
|
|
|
|
def test_notification_au_demarrage(self):
|
|
notifs = []
|
|
obs = ShadowObserver(notify_callback=lambda n: notifs.append(n))
|
|
obs.start("s1")
|
|
|
|
assert len(notifs) >= 1
|
|
assert notifs[0].niveau == NiveauNotification.INFO
|
|
assert "observe" in notifs[0].message.lower()
|
|
|
|
def test_notification_nouvelle_etape(self):
|
|
"""Un changement d'application émet une notification DECOUVERTE."""
|
|
notifs = []
|
|
obs = ShadowObserver(notify_callback=lambda n: notifs.append(n))
|
|
obs.start("s1")
|
|
|
|
obs.observe_event("s1", _evt_click(title="Firefox", ts=100.0))
|
|
obs.observe_event("s1", _evt_click(title="Bloc-notes", ts=101.0))
|
|
|
|
decouverts = [n for n in notifs if n.niveau == NiveauNotification.DECOUVERTE]
|
|
assert len(decouverts) >= 1
|
|
|
|
def test_notification_stop_resume(self):
|
|
"""Au stop, on émet un résumé du nombre d'étapes."""
|
|
notifs = []
|
|
obs = ShadowObserver(notify_callback=lambda n: notifs.append(n))
|
|
obs.start("s1")
|
|
for evt in _make_session_events():
|
|
obs.observe_event("s1", evt)
|
|
obs.stop("s1")
|
|
|
|
messages = [n.message.lower() for n in notifs]
|
|
assert any("étape" in m or "observ" in m for m in messages)
|
|
|
|
def test_notifications_since_ts(self):
|
|
"""get_notifications(since_ts=...) filtre correctement."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
time.sleep(0.01)
|
|
mid_ts = time.time()
|
|
time.sleep(0.01)
|
|
obs.observe_event("s1", _evt_click(title="Firefox"))
|
|
obs.observe_event("s1", _evt_click(title="Bloc-notes"))
|
|
|
|
recentes = obs.get_notifications("s1", since_ts=mid_ts)
|
|
toutes = obs.get_notifications("s1", since_ts=0)
|
|
assert len(recentes) < len(toutes)
|
|
|
|
def test_callback_erreur_ne_plante_pas(self):
|
|
"""Un callback qui lève ne doit pas faire planter l'observer."""
|
|
def bad_callback(notif):
|
|
raise RuntimeError("boom")
|
|
|
|
obs = ShadowObserver(notify_callback=bad_callback)
|
|
obs.start("s1") # Devrait émettre une notification (qui plante en callback)
|
|
# Si on arrive ici, c'est OK
|
|
obs.observe_event("s1", _evt_click())
|
|
|
|
|
|
# =========================================================================
|
|
# Compréhension et API publique
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverUnderstanding:
|
|
|
|
def test_get_understanding_format(self):
|
|
"""La structure retournée est bien celle attendue."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(title="App"))
|
|
|
|
steps = obs.get_understanding("s1")
|
|
assert isinstance(steps, list)
|
|
assert len(steps) >= 1
|
|
step = steps[0]
|
|
assert "step" in step
|
|
assert "intent" in step
|
|
assert "confidence" in step
|
|
assert isinstance(step["step"], int)
|
|
assert 0.0 <= step["confidence"] <= 1.0
|
|
|
|
def test_get_understanding_sans_session(self):
|
|
obs = ShadowObserver()
|
|
steps = obs.get_understanding("inexistante")
|
|
assert steps == []
|
|
|
|
def test_get_current_step(self):
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
assert obs.get_current_step("s1") is None
|
|
|
|
obs.observe_event("s1", _evt_click(title="App"))
|
|
current = obs.get_current_step("s1")
|
|
assert current is not None
|
|
assert current["step_index"] == 1
|
|
|
|
def test_get_steps_internal(self):
|
|
"""get_steps_internal retourne des UnderstoodStep copiés."""
|
|
obs = ShadowObserver()
|
|
obs.start("s1")
|
|
obs.observe_event("s1", _evt_click(title="App"))
|
|
|
|
internals = obs.get_steps_internal("s1")
|
|
assert len(internals) >= 1
|
|
assert isinstance(internals[0], UnderstoodStep)
|
|
|
|
# Mutation externe ne doit pas affecter l'observer
|
|
internals[0].intent = "HACKED"
|
|
again = obs.get_steps_internal("s1")
|
|
assert again[0].intent != "HACKED"
|
|
|
|
|
|
# =========================================================================
|
|
# Singleton partagé
|
|
# =========================================================================
|
|
|
|
|
|
class TestSharedObserver:
|
|
|
|
def test_singleton(self):
|
|
obs1 = get_shared_observer()
|
|
obs2 = get_shared_observer()
|
|
assert obs1 is obs2
|
|
|
|
|
|
# =========================================================================
|
|
# Performance (contrainte : observe_event doit être rapide)
|
|
# =========================================================================
|
|
|
|
|
|
class TestShadowObserverPerformance:
|
|
|
|
def test_observe_event_rapide(self):
|
|
"""observe_event() doit traiter 1000 events en moins de 500ms."""
|
|
obs = ShadowObserver()
|
|
obs.start("s_perf")
|
|
|
|
events = []
|
|
for i in range(1000):
|
|
events.append(_evt_click(title="App", ts=100.0 + i * 0.01))
|
|
|
|
start = time.time()
|
|
for evt in events:
|
|
obs.observe_event("s_perf", evt)
|
|
elapsed = time.time() - start
|
|
|
|
assert elapsed < 0.5, f"Trop lent : {elapsed:.2f}s pour 1000 events"
|