tests/providers/test_duplicate_detector.py

"""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"]