a27b74cf22
- Frontend v4 accessible sur réseau local (192.168.1.40) - Ports ouverts: 3002 (frontend), 5001 (backend), 5004 (dashboard) - Ollama GPU fonctionnel - Self-healing interactif - Dashboard confiance Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
445 lines
19 KiB
Python
445 lines
19 KiB
Python
#!/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"]) |