"""Tests pour le module screen_change_detector (pHash). Charge des screenshots réels de sessions live et vérifie que : - le calcul de pHash est rapide (<5ms par image) - les seuils SAME/MINOR/MAJOR sont cohérents - les heartbeats consécutifs sont classés SAME (même écran, ~5s d'intervalle) - les shots d'actions différentes ont une distance plus élevée """ import os import time import glob import pytest from PIL import Image from core.analytics.screen_change_detector import ( compute_phash, compare_screenshots, compare_hashes, ScreenChangeLevel, ) # Dossier de la session la plus riche en screenshots SESSION_DIR = os.path.join( os.path.dirname(__file__), "..", "..", "data", "training", "live_sessions", "sess_20260314T173236_c7de11", "shots", ) SESSION_DIR = os.path.normpath(SESSION_DIR) def _load_heartbeats(max_count: int = 10): """Charge les heartbeat screenshots (captures régulières toutes les ~5s).""" pattern = os.path.join(SESSION_DIR, "heartbeat_*.png") files = sorted(glob.glob(pattern))[:max_count] images = [] for f in files: img = Image.open(f) images.append((os.path.basename(f), img)) return images def _load_action_shots(max_count: int = 10): """Charge les shots d'actions (captures déclenchées par des événements utilisateur).""" pattern = os.path.join(SESSION_DIR, "shot_*_full.png") files = sorted(glob.glob(pattern))[:max_count] images = [] for f in files: img = Image.open(f) images.append((os.path.basename(f), img)) return images @pytest.fixture(scope="module") def heartbeats(): imgs = _load_heartbeats(10) if len(imgs) < 2: pytest.skip("Pas assez de heartbeats dans la session de test") return imgs @pytest.fixture(scope="module") def action_shots(): imgs = _load_action_shots(10) if len(imgs) < 2: pytest.skip("Pas assez de shots d'action dans la session de test") return imgs class TestPHashPerformance: """Vérifie que le calcul de pHash est rapide (<5ms par image).""" def test_phash_speed(self, heartbeats): """Le pHash doit être calculé en moins de 50ms par image (screenshots 2560x1600).""" times = [] for name, img in heartbeats: t0 = time.perf_counter() h = compute_phash(img) elapsed_ms = (time.perf_counter() - t0) * 1000 times.append(elapsed_ms) print(f" pHash({name}): {elapsed_ms:.2f}ms -> {h}") # Exclure le premier appel (chargement initial plus lent) warm_times = times[1:] if len(times) > 1 else times avg_ms = sum(warm_times) / len(warm_times) max_ms = max(warm_times) print(f"\n Moyenne (hors warmup): {avg_ms:.2f}ms | Max: {max_ms:.2f}ms | N={len(warm_times)}") # ~15ms par hash pour des screenshots 2560x1600, seuil large pour CI assert avg_ms < 50.0, f"pHash trop lent: {avg_ms:.2f}ms en moyenne (attendu <50ms)" def test_comparison_speed(self, heartbeats): """La comparaison de deux screenshots doit prendre moins de 100ms.""" if len(heartbeats) < 2: pytest.skip("Pas assez d'images") # Warmup _ = compute_phash(heartbeats[0][1]) t0 = time.perf_counter() distance, level = compare_screenshots(heartbeats[0][1], heartbeats[1][1]) elapsed_ms = (time.perf_counter() - t0) * 1000 print(f" compare_screenshots: {elapsed_ms:.2f}ms (distance={distance}, level={level.value})") assert elapsed_ms < 100.0, f"Comparaison trop lente: {elapsed_ms:.2f}ms" class TestHeartbeatConsistency: """Les heartbeats consécutifs (~5s) doivent être classés SAME ou MINOR.""" def test_consecutive_heartbeats_are_similar(self, heartbeats): """Les heartbeats consécutifs ne doivent pas être classés MAJOR.""" # Pré-calcul des hashes hashes = [] for name, img in heartbeats: hashes.append((name, compute_phash(img))) print("\n Comparaisons consécutives des heartbeats:") for i in range(len(hashes) - 1): name1, h1 = hashes[i] name2, h2 = hashes[i + 1] distance, level = compare_hashes(h1, h2) print(f" {name1} <-> {name2}: distance={distance}, level={level.value}") # Les heartbeats sont pris toutes les 5s environ sur le même écran # On s'attend a SAME ou MINOR (curseur, horloge, etc.) # Note : certains heartbeats peuvent capturer un changement d'écran # donc on ne peut pas garantir SAME pour tous, mais la majorité doit l'être class TestActionShotsDifferences: """Les shots d'actions différentes doivent montrer des changements.""" def test_action_shots_show_variation(self, action_shots): """Au moins certaines paires de shots d'action doivent montrer des changements.""" hashes = [] for name, img in action_shots: hashes.append((name, compute_phash(img))) print("\n Comparaisons des shots d'action:") distances = [] for i in range(len(hashes) - 1): name1, h1 = hashes[i] name2, h2 = hashes[i + 1] distance, level = compare_hashes(h1, h2) distances.append(distance) print(f" {name1} <-> {name2}: distance={distance}, level={level.value}") # On s'attend à ce que au moins certaines paires aient une distance > 0 max_distance = max(distances) if distances else 0 print(f"\n Distance max entre shots: {max_distance}") assert max_distance > 0, "Tous les shots d'action sont identiques, ce n'est pas normal" class TestThresholdCoherence: """Vérifie que les seuils SAME/MINOR/MAJOR sont cohérents.""" def test_same_image_is_same(self, heartbeats): """La même image comparée à elle-même doit donner distance=0, SAME.""" img = heartbeats[0][1] distance, level = compare_screenshots(img, img) assert distance == 0 assert level == ScreenChangeLevel.SAME def test_heartbeat_vs_action_shot(self, heartbeats, action_shots): """Un heartbeat vs un shot d'action lointain doit être MINOR ou MAJOR.""" # Prend le premier heartbeat et le dernier shot d'action _, img1 = heartbeats[0] _, img2 = action_shots[-1] distance, level = compare_screenshots(img1, img2) print(f" heartbeat[0] vs action_shot[-1]: distance={distance}, level={level.value}") # On vérifie juste que ça fonctionne sans erreur assert distance >= 0 assert isinstance(level, ScreenChangeLevel) def test_compare_hashes_matches_compare_screenshots(self, heartbeats): """compare_hashes doit donner le même résultat que compare_screenshots.""" if len(heartbeats) < 2: pytest.skip("Pas assez d'images") img1 = heartbeats[0][1] img2 = heartbeats[1][1] d1, l1 = compare_screenshots(img1, img2) h1 = compute_phash(img1) h2 = compute_phash(img2) d2, l2 = compare_hashes(h1, h2) assert d1 == d2 assert l1 == l2 class TestFullSessionSummary: """Résumé complet de la session pour validation humaine.""" def test_full_session_summary(self, heartbeats, action_shots): """Affiche un résumé complet des distances pour validation humaine.""" all_images = heartbeats + action_shots hashes = [(name, compute_phash(img)) for name, img in all_images] print("\n === RÉSUMÉ COMPLET DE LA SESSION ===") print(f" {len(heartbeats)} heartbeats + {len(action_shots)} shots d'action") same_count = 0 minor_count = 0 major_count = 0 total_comparisons = 0 for i in range(len(hashes) - 1): name1, h1 = hashes[i] name2, h2 = hashes[i + 1] distance, level = compare_hashes(h1, h2) total_comparisons += 1 if level == ScreenChangeLevel.SAME: same_count += 1 elif level == ScreenChangeLevel.MINOR: minor_count += 1 else: major_count += 1 print(f" Comparaisons consécutives: {total_comparisons}") print(f" SAME (<5): {same_count} ({100*same_count/max(total_comparisons,1):.0f}%)") print(f" MINOR (5-15): {minor_count} ({100*minor_count/max(total_comparisons,1):.0f}%)") print(f" MAJOR (>=15): {major_count} ({100*major_count/max(total_comparisons,1):.0f}%)")