File size: 3,628 Bytes
23d337e 892fa81 23d337e 892fa81 23d337e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Provider tests for DuplicateDetectorProvider."""
from __future__ import annotations
import cv2
import numpy as np
import pytest
from config.settings import Settings
from pipeline.feature_extraction import PipelineOutput
from providers.forensics.duplicate_detector import DuplicateDetectorProvider
@pytest.fixture
def duplicate_provider():
return DuplicateDetectorProvider(settings=Settings(environment="test", db_path=":memory:"))
def _make_pipeline(image: np.ndarray, original_bytes: bytes | None = None) -> PipelineOutput:
return PipelineOutput(
image=image, image_hash="h", width=image.shape[1], height=image.shape[0],
source="bytes", original_bytes=original_bytes, original_format=".jpg",
)
class TestDuplicateDetectorProvider:
def test_name(self, duplicate_provider):
assert duplicate_provider.name == "duplicate_detector"
def test_capability(self, duplicate_provider):
from models.providers import ProviderCapability
assert duplicate_provider.capability == ProviderCapability.FORENSICS
def test_is_available(self, duplicate_provider):
assert duplicate_provider.is_available() is True
def test_first_image_not_duplicate(self, duplicate_provider, sample_image_bytes):
img = cv2.imdecode(np.frombuffer(sample_image_bytes, np.uint8), cv2.IMREAD_COLOR)
po = _make_pipeline(img, original_bytes=sample_image_bytes)
result = duplicate_provider.execute(po)
assert result.success is True
assert result.normalized["is_duplicate"] is False
assert result.normalized["duplicate_of"] is None
# Hashes should be present — 64 bits = 16 hex chars
assert "phash" in result.normalized["details"]
assert "dhash" in result.normalized["details"]
assert len(result.normalized["details"]["phash"]) == 16 # 64 bits as hex
def test_identical_image_detected_as_duplicate(self, duplicate_provider, sample_image_bytes):
img = cv2.imdecode(np.frombuffer(sample_image_bytes, np.uint8), cv2.IMREAD_COLOR)
po1 = _make_pipeline(img, original_bytes=sample_image_bytes)
po2 = _make_pipeline(img.copy(), original_bytes=sample_image_bytes)
# First call registers
duplicate_provider.execute(po1)
# Second call should detect duplicate
result = duplicate_provider.execute(po2)
assert result.normalized["is_duplicate"] is True
def test_different_image_not_duplicate(self, duplicate_provider,
sample_image_bytes, sample_face_image_bytes):
img1 = cv2.imdecode(np.frombuffer(sample_image_bytes, np.uint8), cv2.IMREAD_COLOR)
img2 = cv2.imdecode(np.frombuffer(sample_face_image_bytes, np.uint8), cv2.IMREAD_COLOR)
po1 = _make_pipeline(img1, original_bytes=sample_image_bytes)
po2 = _make_pipeline(img2, original_bytes=sample_face_image_bytes)
duplicate_provider.execute(po1)
result = duplicate_provider.execute(po2)
assert result.normalized["is_duplicate"] is False
def test_phash_stable_for_same_image(self, duplicate_provider, sample_image_bytes):
img = cv2.imdecode(np.frombuffer(sample_image_bytes, np.uint8), cv2.IMREAD_COLOR)
po1 = _make_pipeline(img, original_bytes=sample_image_bytes)
po2 = _make_pipeline(img.copy(), original_bytes=sample_image_bytes)
r1 = duplicate_provider.execute(po1)
r2 = duplicate_provider.execute(po2)
# phash should be the same for identical images
assert r1.normalized["details"]["phash"] == r2.normalized["details"]["phash"]
|