Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
013fe071a256c6dabe1be35af07602ecc291dae2
rpa_vision_v3/tests/unit/test_effective_lru_cache.py
Dom cf495dd82f feat: chat unifié, GestureCatalog, Copilot, Léa UI, extraction données, vérification replay 
Refonte majeure du système Agent Chat et ajout de nombreux modules :

- Chat unifié : suppression du dual Workflows/Agent Libre, tout passe par /api/chat
  avec résolution en 3 niveaux (workflow → geste → "montre-moi")
- GestureCatalog : 38 raccourcis clavier universels Windows avec matching sémantique,
  substitution automatique dans les replays, et endpoint /api/gestures
- Mode Copilot : exécution pas-à-pas des workflows avec validation humaine via WebSocket
  (approve/skip/abort) avant chaque action
- Léa UI (agent_v0/lea_ui/) : interface PyQt5 pour Windows avec overlay transparent
  pour feedback visuel pendant le replay
- Data Extraction (core/extraction/) : moteur d'extraction visuelle de données
  (OCR + VLM → SQLite), avec schémas YAML et export CSV/Excel
- ReplayVerifier (agent_v0/server_v1/) : vérification post-action par comparaison
  de screenshots, avec logique de retry (max 3)
- IntentParser durci : meilleur fallback regex, type GREETING, patterns améliorés
- Dashboard : nouvelles pages gestures, streaming, extractions
- Tests : 63 tests GestureCatalog, 47 tests extraction, corrections tests existants
- Dépréciation : /api/agent/plan et /api/agent/execute retournent HTTP 410,
  suppression du code hardcodé _plan_to_replay_actions

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-15 10:02:09 +01:00

597 lines
20 KiB
Python
Raw Blame History

"""
Tests unitaires pour EffectiveLRUCache
Tests pour l'exigence 6.1: Implémenter EffectiveLRUCache
- Limites de taille ET de mémoire effectives
- Éviction basée sur l'utilisation mémoire réelle
"""
import pytest
import time
import numpy as np
from datetime import datetime, timedelta
from unittest.mock import patch, MagicMock
from core.execution.memory_cache import (
EffectiveLRUCache,
MemoryManager,
MemoryEstimator,
get_memory_manager,
shutdown_memory_manager
)
class TestMemoryEstimator:
"""Tests pour l'estimateur de mémoire."""
def test_estimate_none(self):
"""Test estimation pour None."""
assert MemoryEstimator.estimate_size(None) == 0
def test_estimate_numpy_array(self):
"""Test estimation pour numpy array."""
arr = np.zeros((100, 100), dtype=np.float32)
expected_size = 100 * 100 * 4 # 4 bytes per float32
assert MemoryEstimator.estimate_size(arr) == expected_size
def test_estimate_string(self):
"""Test estimation pour string."""
text = "Hello World"
size = MemoryEstimator.estimate_size(text)
assert size > 0
assert isinstance(size, int)
def test_estimate_list(self):
"""Test estimation pour liste."""
data = [1, 2, 3, "test", [4, 5]]
size = MemoryEstimator.estimate_size(data)
assert size > 0
assert isinstance(size, int)
def test_estimate_dict(self):
"""Test estimation pour dictionnaire."""
data = {"key1": "value1", "key2": [1, 2, 3]}
size = MemoryEstimator.estimate_size(data)
assert size > 0
assert isinstance(size, int)
class TestEffectiveLRUCache:
"""Tests pour EffectiveLRUCache."""
def setup_method(self):
"""Setup pour chaque test."""
self.cache = EffectiveLRUCache(
max_size=5,
max_memory_mb=1.0, # 1MB
enable_monitoring=False # Désactiver pour les tests
)
def teardown_method(self):
"""Cleanup après chaque test."""
if hasattr(self, 'cache'):
try:
self.cache.stop_monitoring()
except Exception:
pass # Ignorer les erreurs de cleanup
def test_basic_operations(self):
"""Test opérations de base."""
# Put et get
assert self.cache.put("key1", "value1")
assert self.cache.get("key1") == "value1"
# Miss
assert self.cache.get("nonexistent") is None
# Contains
assert "key1" in self.cache
assert "nonexistent" not in self.cache
# Length
assert len(self.cache) == 1
def test_lru_eviction_by_size(self):
"""Test éviction LRU par taille."""
# Remplir le cache
for i in range(5):
assert self.cache.put(f"key{i}", f"value{i}")
assert len(self.cache) == 5
# Ajouter un 6ème élément doit évict le premier
assert self.cache.put("key5", "value5")
assert len(self.cache) == 5
assert self.cache.get("key0") is None # Évicté
assert self.cache.get("key1") == "value1" # Toujours là
def test_lru_eviction_by_memory(self):
"""Test éviction LRU par mémoire."""
# Créer plusieurs objets moyens qui ensemble dépassent la limite
medium_arrays = []
for i in range(4):
arr = np.zeros((128, 128), dtype=np.float32) # ~64KB chacun
medium_arrays.append(arr)
assert self.cache.put(f"medium{i}", arr)
# À ce point, on a ~256KB dans le cache
assert len(self.cache) == 4
# Ajouter un gros objet qui doit évict plusieurs petits
big_array = np.zeros((512, 512), dtype=np.float32) # ~1MB
assert self.cache.put("big", big_array)
# Le gros objet doit avoir évicté les petits pour faire de la place
assert self.cache.get("big") is not None
# Vérifier qu'au moins quelques objets ont été évictés
remaining_mediums = sum(1 for i in range(4) if self.cache.get(f"medium{i}") is not None)
assert remaining_mediums < 4 # Au moins un a été évicté
def test_memory_limit_rejection(self):
"""Test rejet d'objets trop gros."""
# Créer un objet plus gros que la limite du cache
huge_array = np.zeros((1024, 1024), dtype=np.float32) # ~4MB > 1MB limit
# Doit être rejeté
assert not self.cache.put("huge", huge_array)
assert self.cache.get("huge") is None
assert len(self.cache) == 0
def test_update_existing_key(self):
"""Test mise à jour d'une clé existante."""
# Ajouter une valeur
assert self.cache.put("key1", "value1")
assert self.cache.get("key1") == "value1"
# Mettre à jour
assert self.cache.put("key1", "new_value")
assert self.cache.get("key1") == "new_value"
assert len(self.cache) == 1
def test_lru_order(self):
"""Test ordre LRU."""
# Ajouter des éléments
for i in range(3):
self.cache.put(f"key{i}", f"value{i}")
# Accéder à key0 pour le rendre récent
self.cache.get("key0")
# Ajouter plus d'éléments pour déclencher éviction
for i in range(3, 6):
self.cache.put(f"key{i}", f"value{i}")
# key0 doit toujours être là (récemment accédé)
assert self.cache.get("key0") == "value0"
# key1 doit avoir été évicté (plus ancien)
assert self.cache.get("key1") is None
def test_remove(self):
"""Test suppression d'éléments."""
# Ajouter des éléments
self.cache.put("key1", "value1")
self.cache.put("key2", "value2")
# Supprimer
assert self.cache.remove("key1")
assert self.cache.get("key1") is None
assert self.cache.get("key2") == "value2"
assert len(self.cache) == 1
# Supprimer inexistant
assert not self.cache.remove("nonexistent")
def test_clear(self):
"""Test vidage du cache."""
# Ajouter des éléments
for i in range(3):
self.cache.put(f"key{i}", f"value{i}")
assert len(self.cache) == 3
# Vider
self.cache.clear()
assert len(self.cache) == 0
for i in range(3):
assert self.cache.get(f"key{i}") is None
def test_cleanup_old_entries(self):
"""Test nettoyage des entrées anciennes."""
# Ajouter des éléments
self.cache.put("old1", "value1")
self.cache.put("old2", "value2")
# Simuler le passage du temps
old_time = datetime.now() - timedelta(hours=2)
self.cache._access_times["old1"] = old_time
self.cache._access_times["old2"] = old_time
# Ajouter un élément récent
self.cache.put("recent", "value_recent")
# Nettoyer les entrées de plus d'1 heure
cleaned = self.cache.cleanup_old_entries(max_age_hours=1.0)
assert cleaned == 2
assert self.cache.get("old1") is None
assert self.cache.get("old2") is None
assert self.cache.get("recent") == "value_recent"
def test_memory_usage_stats(self):
"""Test statistiques d'utilisation mémoire."""
# Ajouter quelques éléments
self.cache.put("key1", "value1")
self.cache.put("key2", np.zeros(100, dtype=np.float32))
usage = self.cache.get_memory_usage()
assert usage['current_bytes'] > 0
assert usage['current_mb'] > 0
assert usage['max_bytes'] == 1024 * 1024 # 1MB
assert usage['max_mb'] == 1.0
assert 0 <= usage['usage_percent'] <= 100
assert usage['items_count'] == 2
assert usage['max_items'] == 5
assert usage['avg_item_size'] > 0
def test_comprehensive_stats(self):
"""Test statistiques complètes."""
# Générer quelques hits et misses
self.cache.put("key1", "value1")
self.cache.get("key1") # hit
self.cache.get("nonexistent") # miss
stats = self.cache.get_stats()
assert stats['hits'] == 1
assert stats['misses'] == 1
assert stats['total_requests'] == 2
assert stats['hit_rate'] == 0.5
assert stats['evictions'] == 0
assert stats['memory_evictions'] == 0
assert stats['size'] == 1
assert stats['max_size'] == 5
# Vérifier que les stats mémoire sont incluses
assert 'current_bytes' in stats
assert 'current_mb' in stats
class TestMemoryManager:
"""Tests pour MemoryManager."""
def setup_method(self):
"""Setup pour chaque test."""
# Désactiver le monitoring pour les tests pour éviter les interférences
self.manager = MemoryManager(
max_memory_mb=100,
cleanup_threshold=0.8,
check_interval=60.0, # Intervalle long pour éviter les interférences
enable_monitoring=False # Désactiver pour les tests
)
def teardown_method(self):
"""Cleanup après chaque test."""
if hasattr(self, 'manager'):
try:
self.manager.shutdown()
except Exception:
pass # Ignorer les erreurs de cleanup
def test_resource_registration(self):
"""Test enregistrement de ressources."""
resource = {"data": "test"}
cleanup_func = MagicMock()
# Enregistrer
self.manager.register_resource(
"test_resource",
resource,
cleanup_func,
{"type": "test"}
)
# Vérifier
assert "test_resource" in self.manager.resource_registry
assert self.manager.resource_registry["test_resource"]["resource"] == resource
assert self.manager.resource_registry["test_resource"]["metadata"]["type"] == "test"
assert "test_resource" in self.manager.cleanup_functions
def test_resource_unregistration(self):
"""Test désenregistrement de ressources."""
resource = {"data": "test"}
# Enregistrer puis désenregistrer
self.manager.register_resource("test_resource", resource)
assert self.manager.unregister_resource("test_resource")
# Vérifier suppression
assert "test_resource" not in self.manager.resource_registry
# Désenregistrer inexistant
assert not self.manager.unregister_resource("nonexistent")
@patch('psutil.Process')
def test_memory_usage(self, mock_process):
"""Test mesure d'utilisation mémoire."""
# Mock psutil
mock_memory_info = MagicMock()
mock_memory_info.rss = 100 * 1024 * 1024 # 100MB
mock_process.return_value.memory_info.return_value = mock_memory_info
usage = self.manager.get_memory_usage()
assert usage == 100.0
@patch('psutil.Process')
def test_cleanup_triggered(self, mock_process):
"""Test déclenchement du nettoyage."""
# Mock mémoire élevée
mock_memory_info = MagicMock()
mock_memory_info.rss = 90 * 1024 * 1024 # 90MB > 80MB threshold
mock_process.return_value.memory_info.return_value = mock_memory_info
# Enregistrer une ressource ancienne
cleanup_func = MagicMock()
self.manager.register_resource("old_resource", {"data": "test"}, cleanup_func)
# Simuler ancienneté
old_time = datetime.now() - timedelta(hours=2)
self.manager.resource_registry["old_resource"]["last_accessed"] = old_time
# Déclencher nettoyage
stats = self.manager.cleanup_if_needed()
assert stats['cleanup_triggered']
assert stats['resources_cleaned'] == 1
cleanup_func.assert_called_once()
def test_stats(self):
"""Test statistiques du gestionnaire."""
# Compter les ressources déjà enregistrées (ex: gpu_resource_manager)
baseline = len(self.manager.resource_registry)
# Enregistrer quelques ressources
for i in range(3):
self.manager.register_resource(f"resource{i}", {"data": i})
stats = self.manager.get_stats()
assert stats['max_memory_mb'] == 100
assert stats['registered_resources'] == baseline + 3
assert stats['cleanup_threshold'] == 0.8
assert stats['check_interval'] == 60.0 # Corrigé: était 1.0
assert not stats['running'] or not self.manager.enable_monitoring # Monitoring désactivé
@pytest.mark.slow
def test_gpu_resource_management(self):
"""Test gestion des ressources GPU."""
# Créer un manager avec gestion GPU activée
manager = MemoryManager(
max_memory_mb=100,
enable_monitoring=False,
enable_gpu_management=True
)
try:
# Enregistrer une ressource GPU
def cleanup_gpu_model(resource_id):
# Simuler le nettoyage d'un modèle GPU
pass
manager.register_gpu_resource(
"test_model",
"model",
cleanup_gpu_model,
{"size_mb": 500}
)
# Vérifier l'enregistrement
assert "test_model" in manager._gpu_resources
assert "gpu_test_model" in manager.resource_registry
# Obtenir les stats GPU
gpu_usage = manager.get_gpu_memory_usage()
assert isinstance(gpu_usage, dict)
# Nettoyer les ressources GPU
cleaned = manager.cleanup_gpu_resources(max_age_hours=0.0) # Force cleanup
assert cleaned >= 0 # Peut être 0 si pas de GPU ou pas de ressources anciennes
# Désenregistrer
assert manager.unregister_gpu_resource("test_model")
assert "test_model" not in manager._gpu_resources
finally:
manager.shutdown()
def test_gpu_management_disabled(self):
"""Test comportement quand gestion GPU désactivée."""
manager = MemoryManager(
enable_monitoring=False,
enable_gpu_management=False
)
try:
# Tenter d'enregistrer une ressource GPU
manager.register_gpu_resource("test", "model")
# Ne doit pas être enregistrée
assert "test" not in manager._gpu_resources
# Stats GPU doivent indiquer que c'est désactivé
gpu_usage = manager.get_gpu_memory_usage()
assert not gpu_usage['available']
assert 'disabled' in gpu_usage['reason']
finally:
manager.shutdown()
"""Test arrêt propre."""
# Enregistrer des ressources avec cleanup
cleanup_funcs = []
for i in range(3):
cleanup_func = MagicMock()
cleanup_funcs.append(cleanup_func)
self.manager.register_resource(f"resource{i}", {"data": i}, cleanup_func)
# Arrêter
self.manager.shutdown()
# Vérifier que tous les cleanups ont été appelés
for cleanup_func in cleanup_funcs:
cleanup_func.assert_called_once()
# Vérifier état
assert not self.manager._running
assert len(self.manager.resource_registry) == 0
assert len(self.manager.cleanup_functions) == 0
@pytest.mark.slow
def test_gpu_resource_management_global(self):
"""Test gestion des ressources GPU."""
# Créer un manager avec gestion GPU activée
manager = MemoryManager(
max_memory_mb=100,
enable_monitoring=False,
enable_gpu_management=True
)
try:
# Enregistrer une ressource GPU
def cleanup_gpu_model(resource_id):
# Simuler le nettoyage d'un modèle GPU
pass
manager.register_gpu_resource(
"test_model",
"model",
cleanup_gpu_model,
{"size_mb": 500}
)
# Vérifier l'enregistrement
if manager.enable_gpu_management: # Peut être désactivé si pas de GPU
assert "test_model" in manager._gpu_resources
assert "gpu_test_model" in manager.resource_registry
# Obtenir les stats GPU
gpu_usage = manager.get_gpu_memory_usage()
assert isinstance(gpu_usage, dict)
# Nettoyer les ressources GPU
cleaned = manager.cleanup_gpu_resources(max_age_hours=0.0) # Force cleanup
assert cleaned >= 0 # Peut être 0 si pas de GPU ou pas de ressources anciennes
# Désenregistrer
result = manager.unregister_gpu_resource("test_model")
# Peut être False si GPU management désactivé
finally:
manager.shutdown()
def test_gpu_management_disabled(self):
"""Test comportement quand gestion GPU désactivée."""
manager = MemoryManager(
enable_monitoring=False,
enable_gpu_management=False
)
try:
# Tenter d'enregistrer une ressource GPU
manager.register_gpu_resource("test", "model")
# Ne doit pas être enregistrée
assert "test" not in manager._gpu_resources
# Stats GPU doivent indiquer que c'est désactivé
gpu_usage = manager.get_gpu_memory_usage()
assert not gpu_usage['available']
assert 'disabled' in gpu_usage['reason']
finally:
manager.shutdown()
class TestGlobalMemoryManager:
"""Tests pour le gestionnaire global."""
def teardown_method(self):
"""Cleanup après chaque test."""
try:
shutdown_memory_manager()
except Exception:
pass # Ignorer les erreurs de cleanup
def test_singleton_behavior(self):
"""Test comportement singleton."""
manager1 = get_memory_manager(enable_monitoring=False, enable_gpu_management=False)
manager2 = get_memory_manager(enable_monitoring=False, enable_gpu_management=False)
assert manager1 is manager2
def test_shutdown_global(self):
"""Test arrêt du gestionnaire global."""
manager = get_memory_manager(enable_monitoring=False, enable_gpu_management=False)
assert manager is not None
shutdown_memory_manager()
# Nouveau gestionnaire après shutdown
new_manager = get_memory_manager(enable_monitoring=False, enable_gpu_management=False)
assert new_manager is not manager
class TestIntegration:
"""Tests d'intégration entre les composants."""
def setup_method(self):
"""Setup pour chaque test."""
self.cache = EffectiveLRUCache(
max_size=10,
max_memory_mb=2.0,
enable_monitoring=False
)
# Désactiver le monitoring et GPU pour les tests
self.manager = get_memory_manager(enable_monitoring=False, enable_gpu_management=False)
def teardown_method(self):
"""Cleanup après chaque test."""
if hasattr(self, 'cache'):
try:
self.cache.stop_monitoring()
except Exception:
pass
try:
shutdown_memory_manager()
except Exception:
pass
def test_cache_with_memory_manager(self):
"""Test intégration cache avec gestionnaire mémoire."""
# Enregistrer le cache dans le gestionnaire
self.manager.register_resource(
"test_cache",
self.cache,
lambda cache: cache.clear()
)
# Ajouter des données au cache
for i in range(5):
self.cache.put(f"key{i}", np.zeros(100, dtype=np.float32))
assert len(self.cache) == 5
# Simuler nettoyage
old_time = datetime.now() - timedelta(hours=2)
self.manager.resource_registry["test_cache"]["last_accessed"] = old_time
# Forcer nettoyage
cleaned = self.manager._cleanup_old_resources(max_age_hours=1.0)
assert cleaned == 1
assert len(self.cache) == 0 # Cache vidé par cleanup
if __name__ == "__main__":
pytest.main([__file__])
Reference in New Issue View Git Blame Copy Permalink