merge: disparition Lea (watchdog)

tests/unit/test_agent_v1_session_watchdog.py

@@ -0,0 +1,248 @@
+"""Tests du watchdog de session interactive (résilience RDP/Citrix).
+
+Vérifie que :
+  - Le tray est ré-affiché à la reconnexion RDP (run_ui rappelé).
+  - Un seul tray tourne à la fois (invariant « un seul tray »).
+  - Les threads de fond de l'agent (heartbeat/replay) ne sont JAMAIS
+    relancés par le watchdog (il ne relance QUE l'UI).
+  - Un Quitter explicite arrête le watchdog (pas de résurrection du tray).
+  - Le détecteur de session Windows tombe en marche (True) hors Windows.
+
+Aucune vraie UI : run_ui et is_available sont des callables mockés.
+"""
+
+import sys
+import threading
+import time
+from unittest.mock import MagicMock
+
+# Mocker les libs GUI/Win32 avant tout import du module sous test.
+sys.modules.setdefault("pynput", MagicMock())
+sys.modules.setdefault("pynput.mouse", MagicMock())
+sys.modules.setdefault("pynput.keyboard", MagicMock())
+sys.modules.setdefault("pystray", MagicMock())
+
+from agent_v0.agent_v1.ui.session_watchdog import (  # noqa: E402
+    InteractiveSessionWatchdog,
+    is_interactive_desktop_available,
+)
+
+
+# ---------------------------------------------------------------------------
+# Détection de session
+# ---------------------------------------------------------------------------
+
+def test_detection_hors_windows_renvoie_true(monkeypatch):
+    """Hors Windows (dev/tests Linux) : bureau toujours 'disponible'."""
+    monkeypatch.setattr(
+        "agent_v0.agent_v1.ui.session_watchdog.platform.system",
+        lambda: "Linux",
+    )
+    assert is_interactive_desktop_available() is True
+
+
+# ---------------------------------------------------------------------------
+# Boucle du watchdog
+# ---------------------------------------------------------------------------
+
+def test_relance_ui_a_la_reconnexion():
+    """Session absente puis présente => le tray est (ré)affiché une fois dispo.
+
+    Scénario : la 1re sonde dit 'indisponible' (RDP déconnecté), la 2e dit
+    'disponible' (reconnexion) => run_ui doit être appelé exactement une fois,
+    puis le watchdog s'arrête.
+    """
+    availability = iter([False, True])
+    run_ui_calls = []
+
+    def _run_ui():
+        run_ui_calls.append(time.time())
+
+    # L'agent vit jusqu'à ce que le tray ait été affiché une fois.
+    state = {"alive": True}
+
+    def _is_running():
+        # Après le premier affichage du tray, l'agent s'arrête.
+        return state["alive"] and len(run_ui_calls) == 0
+
+    def _is_available():
+        return next(availability)
+
+    wd = InteractiveSessionWatchdog(
+        run_ui=_run_ui,
+        is_running=_is_running,
+        is_available=_is_available,
+        poll_interval_s=0.01,  # sonde très rapide pour le test
+    )
+    wd.run()
+
+    # Le tray a été (ré)affiché exactement une fois après la reconnexion.
+    assert len(run_ui_calls) == 1
+
+
+def test_reaffichage_apres_chaque_deconnexion():
+    """Deux cycles connexion→déconnexion => tray relancé à chaque reconnexion."""
+    run_ui_calls = []
+
+    # is_available toujours True ; le tray 'sort' immédiatement (déconnexion).
+    def _run_ui():
+        run_ui_calls.append(1)
+
+    def _is_running():
+        # Vivre pour 2 affichages de tray, puis arrêter.
+        return len(run_ui_calls) < 2
+
+    wd = InteractiveSessionWatchdog(
+        run_ui=_run_ui,
+        is_running=_is_running,
+        is_available=lambda: True,
+        poll_interval_s=0.01,
+    )
+    wd.run()
+
+    assert len(run_ui_calls) == 2
+
+
+def test_un_seul_tray_a_la_fois():
+    """L'invariant 'un seul tray' : run_ui n'est jamais réentrant en parallèle."""
+    concurrent = {"count": 0, "max": 0}
+    lock = threading.Lock()
+
+    def _run_ui():
+        with lock:
+            concurrent["count"] += 1
+            concurrent["max"] = max(concurrent["max"], concurrent["count"])
+        time.sleep(0.02)  # simule un tray qui tourne un peu
+        with lock:
+            concurrent["count"] -= 1
+
+    calls = {"n": 0}
+
+    def _is_running():
+        calls["n"] += 1
+        return calls["n"] <= 3  # 3 cycles de tray
+
+    wd = InteractiveSessionWatchdog(
+        run_ui=_run_ui,
+        is_running=_is_running,
+        is_available=lambda: True,
+        poll_interval_s=0.01,
+    )
+    wd.run()
+
+    # Jamais deux trays simultanés.
+    assert concurrent["max"] == 1
+
+
+def test_stop_reveille_le_watchdog_en_attente():
+    """stop() sort immédiatement la boucle quand la session est absente."""
+    run_ui_calls = []
+
+    wd = InteractiveSessionWatchdog(
+        run_ui=lambda: run_ui_calls.append(1),
+        is_running=lambda: True,
+        is_available=lambda: False,  # jamais de session => reste en attente
+        poll_interval_s=60,  # long : seul stop() peut débloquer
+    )
+
+    t = threading.Thread(target=wd.run)
+    t.start()
+    time.sleep(0.05)  # laisser entrer dans l'attente
+    wd.stop()
+    t.join(timeout=2)
+
+    assert not t.is_alive()  # le watchdog est bien sorti
+    assert run_ui_calls == []  # aucun tray (jamais de session dispo)
+
+
+def test_crash_du_tray_ne_tue_pas_le_watchdog():
+    """Une exception dans run_ui est absorbée ; le watchdog reste maître."""
+    calls = {"n": 0}
+
+    def _run_ui():
+        calls["n"] += 1
+        raise RuntimeError("tray HS")
+
+    def _is_running():
+        return calls["n"] < 2  # tolérer 2 crashs puis sortir
+
+    wd = InteractiveSessionWatchdog(
+        run_ui=_run_ui,
+        is_running=_is_running,
+        is_available=lambda: True,
+        poll_interval_s=0.01,
+    )
+    # Ne doit PAS lever : le crash est loggé, pas propagé.
+    wd.run()
+
+    assert calls["n"] == 2
+
+
+# ---------------------------------------------------------------------------
+# Intégration avec main._agent_should_live (Quitter vs déconnexion)
+# ---------------------------------------------------------------------------
+
+def test_tray_run_reentrant_ne_relance_pas_les_threads_de_fond(monkeypatch):
+    """SmartTrayV1.run() appelé 2x (reconnexion RDP) : threads de fond 1x seulement.
+
+    On vérifie que `_start_background_once` est idempotent : les threads
+    connexion/cache et l'accueil ne démarrent qu'au premier affichage, mais
+    une nouvelle icône pystray est recréée à chaque appel (ré-affichage).
+    """
+    import threading as _threading
+    from agent_v0.agent_v1.ui import smart_tray as smart_tray_mod
+
+    tray = smart_tray_mod.SmartTrayV1.__new__(smart_tray_mod.SmartTrayV1)
+    tray._bg_started = False
+    tray.machine_id = "poste_test"
+    tray.server_client = None  # pas de threads réseau => simplifie
+    tray.icon = None
+
+    greet_calls = {"n": 0}
+    hotkey_calls = {"n": 0}
+    icons_created = {"n": 0}
+
+    tray._notifier = MagicMock()
+    tray._notifier.greet.side_effect = lambda: greet_calls.__setitem__("n", greet_calls["n"] + 1)
+    monkeypatch.setattr(tray, "_start_hotkey", lambda: hotkey_calls.__setitem__("n", hotkey_calls["n"] + 1))
+    monkeypatch.setattr(tray, "_current_icon", lambda: object())
+    monkeypatch.setattr(tray, "_get_menu_items", lambda: [])
+
+    # Icône pystray factice : run() ne bloque pas (simule une sortie immédiate).
+    class _FakeIcon:
+        def __init__(self, *a, **k):
+            icons_created["n"] += 1
+
+        def run(self):
+            return None
+
+    monkeypatch.setattr(smart_tray_mod.pystray, "Icon", _FakeIcon)
+    monkeypatch.setattr(smart_tray_mod.pystray, "Menu", lambda *a, **k: None)
+
+    # Deux affichages successifs (déconnexion → reconnexion).
+    tray.run()
+    tray.run()
+
+    # Accueil + hotkey : une seule fois (one-shot).
+    assert greet_calls["n"] == 1
+    assert hotkey_calls["n"] == 1
+    # Mais une nouvelle icône à chaque affichage (le tray revient bien).
+    assert icons_created["n"] == 2
+
+
+def test_agent_should_live_distingue_quit_et_deconnexion():
+    """Quitter explicite arrête le watchdog ; une déconnexion RDP non."""
+    from types import SimpleNamespace
+    from agent_v0.agent_v1.main import _agent_should_live
+
+    # Agent actif, tray sans quit demandé => doit vivre (déconnexion RDP OK).
+    agent = SimpleNamespace(running=True, ui=SimpleNamespace(_quit_requested=False))
+    assert _agent_should_live(agent) is True
+
+    # Quitter explicite => ne doit plus vivre (pas de résurrection).
+    agent.ui._quit_requested = True
+    assert _agent_should_live(agent) is False
+
+    # agent.running=False => ne vit plus (arrêt global).
+    agent2 = SimpleNamespace(running=False, ui=SimpleNamespace(_quit_requested=False))
+    assert _agent_should_live(agent2) is False