#!/usr/bin/env python3 """ Tests de Propriété pour VisualEmbeddingManager - RPA Vision V3 Tests basés sur les propriétés pour valider le comportement universel du gestionnaire d'embeddings visuels. Feature: visual-rpa-properties-enhancement Property 7: Génération de Signatures Visuelles Uniques Auteur: Assistant IA Date: 2026-01-07 """ import pytest import numpy as np from hypothesis import given, strategies as st, settings, assume from PIL import Image import asyncio import tempfile import os from unittest.mock import Mock, AsyncMock from core.visual.visual_embedding_manager import ( VisualEmbeddingManager, EmbeddingCacheEntry, MatchResult, SimilarityMetrics ) from core.embedding.fusion_engine import FusionEngine from core.models import BBox # Configuration Hypothesis MAX_EXAMPLES = 100 DEADLINE = 30000 # 30 secondes class TestVisualEmbeddingManagerProperties: """Tests de propriété pour VisualEmbeddingManager""" @pytest.fixture def mock_fusion_engine(self): """Mock du FusionEngine pour les tests""" engine = Mock(spec=FusionEngine) engine.generate_embedding = AsyncMock() return engine @pytest.fixture def embedding_manager(self, mock_fusion_engine): """Instance de VisualEmbeddingManager pour les tests""" return VisualEmbeddingManager( fusion_engine=mock_fusion_engine, cache_size=100 ) @pytest.fixture def sample_image(self): """Crée une image de test""" image = Image.new('RGB', (100, 100), color='white') return image @pytest.fixture def sample_embedding(self): """Crée un embedding de test normalisé""" embedding = np.random.randn(256).astype(np.float32) # Normaliser l'embedding norm = np.linalg.norm(embedding) if norm > 0: embedding = embedding / norm return embedding @given( width=st.integers(min_value=10, max_value=500), height=st.integers(min_value=10, max_value=500), color_r=st.integers(min_value=0, max_value=255), color_g=st.integers(min_value=0, max_value=255), color_b=st.integers(min_value=0, max_value=255) ) @settings(max_examples=MAX_EXAMPLES, deadline=DEADLINE) def test_property_embedding_generation_consistency( self, embedding_manager, mock_fusion_engine, width, height, color_r, color_g, color_b ): """ Propriété 7: Génération de Signatures Visuelles Uniques Pour toute image donnée, la génération d'embedding doit être: 1. Déterministe (même image = même embedding) 2. Normalisée (norme L2 = 1.0) 3. Cohérente avec le cache """ # Créer une image avec les paramètres donnés image = Image.new('RGB', (width, height), color=(color_r, color_g, color_b)) # Mock du FusionEngine pour retourner un embedding déterministe expected_embedding = np.random.randn(256).astype(np.float32) expected_embedding = expected_embedding / np.linalg.norm(expected_embedding) mock_fusion_engine.generate_embedding.return_value = expected_embedding async def test_consistency(): # Première génération embedding1 = await embedding_manager.generate_embedding(image, use_cache=True) # Deuxième génération (devrait utiliser le cache) embedding2 = await embedding_manager.generate_embedding(image, use_cache=True) # Propriété 1: Déterminisme np.testing.assert_array_equal(embedding1, embedding2) # Propriété 2: Normalisation norm1 = np.linalg.norm(embedding1) norm2 = np.linalg.norm(embedding2) assert abs(norm1 - 1.0) < 1e-6, f"Embedding non normalisé: norme = {norm1}" assert abs(norm2 - 1.0) < 1e-6, f"Embedding non normalisé: norme = {norm2}" # Propriété 3: Cohérence du cache # La deuxième génération ne devrait pas appeler le FusionEngine assert mock_fusion_engine.generate_embedding.call_count == 1 # Vérifier que l'embedding est en cache signature = embedding_manager._generate_image_signature(image) cached_embedding = embedding_manager.get_cached_embedding(signature) assert cached_embedding is not None np.testing.assert_array_equal(cached_embedding, embedding1) # Exécuter le test asynchrone asyncio.run(test_consistency()) @given( embedding_dim=st.integers(min_value=64, max_value=512), num_embeddings=st.integers(min_value=2, max_value=10) ) @settings(max_examples=MAX_EXAMPLES, deadline=DEADLINE) def test_property_similarity_calculation_properties( self, embedding_manager, embedding_dim, num_embeddings ): """ Propriété: Calcul de Similarité Cohérent Pour tout ensemble d'embeddings: 1. Similarité avec soi-même = 1.0 2. Similarité symétrique: sim(A,B) = sim(B,A) 3. Similarité dans [0,1] 4. Triangle inequality respectée (approximativement) """ # Créer des embeddings aléatoires normalisés embeddings = [] for _ in range(num_embeddings): emb = np.random.randn(embedding_dim).astype(np.float32) emb = emb / np.linalg.norm(emb) embeddings.append(emb) async def test_similarity_properties(): for i, emb_a in enumerate(embeddings): # Propriété 1: Similarité avec soi-même self_sim = await embedding_manager.compare_embeddings(emb_a, emb_a) assert abs(self_sim - 1.0) < 1e-6, f"Similarité avec soi-même != 1.0: {self_sim}" for j, emb_b in enumerate(embeddings): if i != j: # Propriété 2: Symétrie sim_ab = await embedding_manager.compare_embeddings(emb_a, emb_b) sim_ba = await embedding_manager.compare_embeddings(emb_b, emb_a) assert abs(sim_ab - sim_ba) < 1e-6, f"Similarité non symétrique: {sim_ab} != {sim_ba}" # Propriété 3: Bornes [0,1] assert 0.0 <= sim_ab <= 1.0, f"Similarité hors bornes: {sim_ab}" # Propriété 4: Triangle inequality (approximative pour similarité cosinus) for k, emb_c in enumerate(embeddings): if k != i and k != j: sim_ac = await embedding_manager.compare_embeddings(emb_a, emb_c) sim_bc = await embedding_manager.compare_embeddings(emb_b, emb_c) # Pour la similarité cosinus, on vérifie une inégalité relaxée # car la métrique de distance cosinus ne respecte pas strictement # l'inégalité triangulaire distance_ab = 1 - sim_ab distance_ac = 1 - sim_ac distance_bc = 1 - sim_bc # Vérification relaxée (tolérance pour les erreurs numériques) tolerance = 0.1 assert distance_ab <= distance_ac + distance_bc + tolerance asyncio.run(test_similarity_properties()) @given( num_candidates=st.integers(min_value=1, max_value=20), min_confidence=st.floats(min_value=0.1, max_value=0.9) ) @settings(max_examples=MAX_EXAMPLES, deadline=DEADLINE) def test_property_best_match_selection( self, embedding_manager, sample_embedding, num_candidates, min_confidence ): """ Propriété: Sélection de la Meilleure Correspondance Pour toute recherche de correspondance: 1. Si aucun candidat >= min_confidence, retourner None 2. Si candidats valides, retourner celui avec la plus haute confiance 3. Le résultat doit avoir une confiance >= min_confidence 4. Les alternatives doivent être triées par confiance décroissante """ # Créer des candidats avec des similarités variées candidates = [] expected_confidences = [] for i in range(num_candidates): # Créer un embedding candidat candidate_emb = np.random.randn(256).astype(np.float32) candidate_emb = candidate_emb / np.linalg.norm(candidate_emb) # Calculer une similarité déterministe basée sur l'index # pour avoir un contrôle sur les résultats confidence = 0.5 + (i / num_candidates) * 0.5 # Entre 0.5 et 1.0 expected_confidences.append(confidence) signature = f"candidate_{i}" candidates.append((signature, candidate_emb)) # Mock de la méthode compare_embeddings pour retourner des valeurs contrôlées original_compare = embedding_manager.compare_embeddings async def mock_compare(emb1, emb2): # Trouver l'index du candidat basé sur l'embedding for i, (_, candidate_emb) in enumerate(candidates): if np.array_equal(emb2, candidate_emb): return expected_confidences[i] return 0.5 # Valeur par défaut embedding_manager.compare_embeddings = mock_compare try: async def test_best_match(): result = await embedding_manager.find_best_match( sample_embedding, candidates, min_confidence ) # Trouver les candidats valides valid_candidates = [ (sig, conf) for (sig, _), conf in zip(candidates, expected_confidences) if conf >= min_confidence ] if not valid_candidates: # Propriété 1: Aucun candidat valide assert result is None, "Devrait retourner None si aucun candidat valide" else: # Propriété 2: Retourner le meilleur candidat assert result is not None, "Devrait retourner un résultat si candidats valides" # Propriété 3: Confiance >= min_confidence assert result.confidence >= min_confidence, \ f"Confiance {result.confidence} < min_confidence {min_confidence}" # Propriété 4: Meilleure confiance max_confidence = max(conf for _, conf in valid_candidates) assert abs(result.confidence - max_confidence) < 1e-6, \ f"Pas la meilleure confiance: {result.confidence} != {max_confidence}" asyncio.run(test_best_match()) finally: # Restaurer la méthode originale embedding_manager.compare_embeddings = original_compare @given( cache_size=st.integers(min_value=1, max_value=50), num_operations=st.integers(min_value=1, max_value=100) ) @settings(max_examples=MAX_EXAMPLES, deadline=DEADLINE) def test_property_cache_behavior( self, mock_fusion_engine, cache_size, num_operations ): """ Propriété: Comportement du Cache Pour toute séquence d'opérations de cache: 1. Le cache ne dépasse jamais la taille maximale 2. Les éléments récemment accédés sont préservés (LRU) 3. Les statistiques de cache sont cohérentes 4. La récupération d'un élément en cache ne génère pas d'embedding """ manager = VisualEmbeddingManager( fusion_engine=mock_fusion_engine, cache_size=cache_size ) # Créer des embeddings de test test_embeddings = {} for i in range(num_operations): emb = np.random.randn(256).astype(np.float32) emb = emb / np.linalg.norm(emb) test_embeddings[f"sig_{i}"] = emb # Simuler des opérations de cache for i, (signature, embedding) in enumerate(test_embeddings.items()): manager.cache_embedding(signature, embedding) # Propriété 1: Taille du cache stats = manager.get_cache_stats() assert stats['cache_size'] <= cache_size, \ f"Cache dépasse la taille max: {stats['cache_size']} > {cache_size}" # Vérifier que l'élément est bien en cache cached = manager.get_cached_embedding(signature) if stats['cache_size'] < cache_size or i < cache_size: # Si le cache n'est pas plein ou si c'est un des premiers éléments assert cached is not None, f"Élément {signature} devrait être en cache" np.testing.assert_array_equal(cached, embedding) # Propriété 3: Cohérence des statistiques final_stats = manager.get_cache_stats() assert final_stats['cache_size'] >= 0 assert final_stats['cache_size'] <= cache_size assert final_stats['total_generations'] >= 0 @given( batch_size=st.integers(min_value=1, max_value=20), use_cache=st.booleans() ) @settings(max_examples=MAX_EXAMPLES, deadline=DEADLINE) def test_property_batch_processing_consistency( self, embedding_manager, mock_fusion_engine, batch_size, use_cache ): """ Propriété: Cohérence du Traitement par Lots Pour tout traitement par lots: 1. Le nombre de résultats = nombre d'entrées valides 2. Chaque résultat correspond à son entrée 3. Les embeddings sont normalisés 4. Le cache est utilisé de manière cohérente """ # Créer des images de test images = [] for i in range(batch_size): size = 50 + i * 10 # Tailles variées color = (i * 30 % 256, (i * 50) % 256, (i * 70) % 256) image = Image.new('RGB', (size, size), color=color) signature = f"batch_image_{i}" images.append((signature, image, None)) # (signature, image, bounding_box) # Mock pour retourner des embeddings déterministes def mock_generate_embedding(image, bounding_box=None): # Créer un embedding basé sur les propriétés de l'image width, height = image.size seed = width * height np.random.seed(seed % 2**32) emb = np.random.randn(256).astype(np.float32) emb = emb / np.linalg.norm(emb) return emb mock_fusion_engine.generate_embedding.side_effect = mock_generate_embedding async def test_batch_consistency(): results = await embedding_manager.batch_generate_embeddings( images, use_cache=use_cache ) # Propriété 1: Nombre de résultats assert len(results) <= len(images), \ f"Trop de résultats: {len(results)} > {len(images)}" # Propriété 2: Correspondance des résultats for signature, image, _ in images: if signature in results: embedding = results[signature] # Propriété 3: Normalisation norm = np.linalg.norm(embedding) assert abs(norm - 1.0) < 1e-6, \ f"Embedding non normalisé pour {signature}: norme = {norm}" # Vérifier la cohérence avec la génération individuelle expected_emb = mock_generate_embedding(image) np.testing.assert_array_almost_equal( embedding, expected_emb, decimal=6, err_msg=f"Embedding incohérent pour {signature}" ) asyncio.run(test_batch_consistency()) def test_property_detailed_similarity_metrics(self, embedding_manager, sample_embedding): """ Propriété: Métriques de Similarité Détaillées Pour toute paire d'embeddings: 1. Les métriques sont dans les bonnes plages 2. Le score combiné est cohérent avec les métriques individuelles 3. Les métriques sont stables (pas de NaN/Inf) """ # Créer un deuxième embedding embedding2 = np.random.randn(256).astype(np.float32) embedding2 = embedding2 / np.linalg.norm(embedding2) async def test_detailed_metrics(): metrics = await embedding_manager.compare_embeddings( sample_embedding, embedding2, detailed_metrics=True ) assert isinstance(metrics, SimilarityMetrics) # Propriété 1: Plages des métriques assert 0.0 <= metrics.cosine_similarity <= 1.0, \ f"Similarité cosinus hors plage: {metrics.cosine_similarity}" assert metrics.euclidean_distance >= 0.0, \ f"Distance euclidienne négative: {metrics.euclidean_distance}" assert 0.0 <= metrics.normalized_correlation <= 1.0, \ f"Corrélation normalisée hors plage: {metrics.normalized_correlation}" assert 0.0 <= metrics.combined_score <= 1.0, \ f"Score combiné hors plage: {metrics.combined_score}" # Propriété 2: Stabilité (pas de NaN/Inf) assert not np.isnan(metrics.cosine_similarity) assert not np.isnan(metrics.euclidean_distance) assert not np.isnan(metrics.normalized_correlation) assert not np.isnan(metrics.combined_score) assert not np.isinf(metrics.cosine_similarity) assert not np.isinf(metrics.euclidean_distance) assert not np.isinf(metrics.normalized_correlation) assert not np.isinf(metrics.combined_score) # Propriété 3: Cohérence du score combiné # Le score combiné devrait être influencé par les métriques individuelles expected_range_min = min( metrics.cosine_similarity, metrics.normalized_correlation ) * 0.5 expected_range_max = max( metrics.cosine_similarity, metrics.normalized_correlation ) assert expected_range_min <= metrics.combined_score <= expected_range_max, \ f"Score combiné incohérent: {metrics.combined_score} pas dans [{expected_range_min}, {expected_range_max}]" asyncio.run(test_detailed_metrics()) if __name__ == "__main__": pytest.main([__file__, "-v", "--tb=short"])