xiaoxia-saas/tests/unit/test_dedup_pure.py

"""去重纯算法测试 — hamming_distance + histogram_similarity + VideoFingerprint."""

from __future__ import annotations

import sys
from unittest.mock import MagicMock

import numpy as np
import pytest


def _mock_module(**attrs):
    """Create a mock module with __spec__ to avoid AttributeError: __spec__."""
    m = MagicMock()
    m.__spec__ = None
    for k, v in attrs.items():
        setattr(m, k, v)
    return m


# ── Module-level setup: mock deps, import dedup, then restore sys.modules ──
# This pattern ensures:
# 1. dedup is imported with mocks active (no db/celery/cv2 side effects)
# 2. sys.modules is restored immediately so other test files are not polluted
# 3. dedup objects are kept in module namespace for tests to use

_SAVED_MODULES_KEYS = set(sys.modules.keys())
_SAVED_MODULES_VALUES = {
    k: sys.modules.get(k)
    for k in [
        "cv2",
        "celery",
        "sqlalchemy",
        "sqlalchemy.orm",
        "sqlalchemy.engine",
        "sqlalchemy.ext",
        "sqlalchemy.ext.declarative",
        "worker_app.db",
        "worker_app.celery_app",
        "worker_app.core.config",
        "packages.adapters.sqlalchemy_impl.session",
        "packages.adapters.sqlalchemy_impl.generated_video_repository",
        "packages.shared.config",
        "packages.shared.storage",
    ]
}

# Set up mocks
sys.modules["cv2"] = _mock_module()

_mock_celery = MagicMock()
_mock_celery.Task = MagicMock
_mock_celery.Celery = MagicMock
_mock_celery.__spec__ = None
sys.modules["celery"] = _mock_celery

_mock_sqla = MagicMock()
_mock_sqla.__path__ = []
_mock_sqla.__spec__ = None
sys.modules["sqlalchemy"] = _mock_sqla

_mock_sqla_orm = MagicMock()
_mock_sqla_orm.__path__ = []
_mock_sqla_orm.__spec__ = None
_mock_sqla_orm.Session = MagicMock
sys.modules["sqlalchemy.orm"] = _mock_sqla_orm
sys.modules["sqlalchemy.engine"] = _mock_module()
sys.modules["sqlalchemy.ext"] = _mock_module()
sys.modules["sqlalchemy.ext.declarative"] = _mock_module()

sys.modules["worker_app.db"] = _mock_module(SessionLocal=MagicMock())
sys.modules["worker_app.celery_app"] = _mock_module(celery_app=MagicMock())
sys.modules["worker_app.core.config"] = _mock_module(get_settings=MagicMock(return_value=MagicMock()))

sys.modules["packages.adapters.sqlalchemy_impl.session"] = _mock_module(
    Base=MagicMock(),
    build_engine=MagicMock(),
    build_session_factory=MagicMock(),
    ensure_database_exists=MagicMock(),
    initialize_database=MagicMock(),
)
sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = _mock_module()
sys.modules["packages.shared.config"] = _mock_module(get_shared_settings=MagicMock(return_value=MagicMock()))
sys.modules["packages.shared.storage"] = _mock_module()

# Import dedup while mocks are active
from video_processing.dedup import (  # noqa: E402
    VideoDeduplicator,
    VideoFingerprint,
    hamming_distance,
)

# ── Restore sys.modules immediately after import ──
# dedup is now cached in this module's namespace; other test files will get
# their own fresh imports without our mock pollution
for _key in list(sys.modules.keys()):
    if _key not in _SAVED_MODULES_KEYS:
        del sys.modules[_key]
for _key, _value in _SAVED_MODULES_VALUES.items():
    if _value is not None:
        sys.modules[_key] = _value
    elif _key in sys.modules:
        del sys.modules[_key]
del _SAVED_MODULES_KEYS, _SAVED_MODULES_VALUES, _key, _value


class TestHammingDistance:
    """hamming_distance 汉明距离计算测试."""

    def test_identical_hashes_zero(self):
        """相同哈希距离为0."""
        assert hamming_distance("ff", "ff") == 0
        assert hamming_distance("00", "00") == 0

    def test_all_different(self):
        """全不同的8bit哈希距离为8."""
        assert hamming_distance("00", "ff") == 8

    def test_single_bit_diff(self):
        """1个bit不同."""
        # 0x01 = 00000001, 0x00 = 00000000 → 1 bit不同
        assert hamming_distance("01", "00") == 1

    def test_four_bits_diff(self):
        """4个bit不同."""
        # 0x0F = 00001111, 0xF0 = 11110000 → 8 bits都不同
        assert hamming_distance("0f", "f0") == 8

    def test_longer_hashes(self):
        """更长的哈希（如64-bit pHash）."""
        # 两个完全不同的64-bit哈希
        assert hamming_distance("0000000000000000", "ffffffffffffffff") == 64

    def test_partial_difference(self):
        """部分bit不同."""
        # a = 1010, 5 = 0101 → 4 bits不同（每个hex digit）
        assert hamming_distance("aa", "55") == 8

    def test_case_insensitive(self):
        """十六进制不区分大小写."""
        assert hamming_distance("FF", "ff") == 0
        assert hamming_distance("AbC123", "aBc123") == 0

    def test_different_length_hashes(self):
        """不同长度的哈希（短的前补零）."""
        # "ff" = 0xff = 255, "0ff" = 0x0ff = 255
        # int("ff", 16) = 255, int("0ff", 16) = 255
        assert hamming_distance("ff", "0ff") == 0


class TestVideoFingerprint:
    """VideoFingerprint 数据结构测试."""

    def test_to_dict_contains_all_fields(self):
        """to_dict返回完整字典."""
        fp = VideoFingerprint(
            md5="abc123",
            keyframe_phashes=["hash1", "hash2"],
            color_histograms=[[0.1, 0.2], [0.3, 0.4]],
            duration=30.5,
            resolution=(1920, 1080),
        )
        d = fp.to_dict()
        assert d["md5"] == "abc123"
        assert d["keyframe_phashes"] == ["hash1", "hash2"]
        assert d["duration"] == 30.5
        assert d["resolution"] == [1920, 1080]
        assert "color_histograms" in d

    def test_empty_phashes(self):
        """空关键帧列表."""
        fp = VideoFingerprint(
            md5="test",
            keyframe_phashes=[],
            color_histograms=[],
            duration=0.0,
            resolution=(0, 0),
        )
        d = fp.to_dict()
        assert d["keyframe_phashes"] == []
        assert d["color_histograms"] == []


class TestBhattacharyyaCoefficient:
    """_bhattacharyya_coefficient Bhattacharyya 系数测试."""

    def test_identical_histograms(self):
        """完全相同的直方图系数为1.0（#1702：按 Σ 归一，概率分布语义）。"""
        hist = [0.5, 0.5, 0.0, 0.0]  # Σ=1 的概率分布
        bc = VideoDeduplicator._bhattacharyya_coefficient(hist, hist)
        assert bc == pytest.approx(1.0)
        # 非归一化输入也归一到 1.0（三通道拼接 Σ=3 的等价情形）
        hist3 = [0.5, 0.5, 0.0, 0.3]
        bc3 = VideoDeduplicator._bhattacharyya_coefficient(hist3, hist3)
        assert bc3 == pytest.approx(1.0)

    def test_zero_histograms(self):
        """全零直方图系数为0."""
        bc = VideoDeduplicator._bhattacharyya_coefficient([0.0, 0.0], [0.0, 0.0])
        assert bc == 0.0

    def test_orthogonal_histograms(self):
        """正交直方图（无重叠）系数为0."""
        bc = VideoDeduplicator._bhattacharyya_coefficient([1.0, 0.0], [0.0, 1.0])
        assert bc == pytest.approx(0.0)

    def test_different_lengths(self):
        """不同长度直方图取最小长度对齐，并按各自总量归一（#1702 概率分布语义）。"""
        # 对齐到前 2 维：coeff = 2，norm = √(Σa·Σb) = √(2·2) = 2 → 1.0
        bc = VideoDeduplicator._bhattacharyya_coefficient([1.0, 1.0, 0.0, 0.0], [1.0, 1.0])
        assert bc == pytest.approx(1.0)

    def test_known_value(self):
        """已知值验证."""
        # [0.25, 0.25, 0.25, 0.25] vs [0.25, 0.25, 0.25, 0.25]
        # BC = 4 * √(0.25 * 0.25) = 4 * 0.25 = 1.0
        hist = [0.25, 0.25, 0.25, 0.25]
        bc = VideoDeduplicator._bhattacharyya_coefficient(hist, hist)
        assert bc == pytest.approx(1.0)


class TestComputeHistogramSimilarity:
    """_compute_histogram_similarity 多帧直方图相似度测试."""

    def test_identical_histogram_groups(self):
        """完全相同的两组直方图."""
        hist = [[0.5, 0.5], [0.3, 0.4]]
        sim = VideoDeduplicator._compute_histogram_similarity(hist, hist)
        # Each hist finds best match = itself
        assert sim > 0.0

    def test_empty_first(self):
        """第一组为空返回0."""
        assert VideoDeduplicator._compute_histogram_similarity([], [[0.5]]) == 0.0

    def test_empty_second(self):
        """第二组为空返回0."""
        assert VideoDeduplicator._compute_histogram_similarity([[0.5]], []) == 0.0

    def test_both_empty(self):
        """两组都为空返回0."""
        assert VideoDeduplicator._compute_histogram_similarity([], []) == 0.0

    def test_best_match_selection(self):
        """多帧时取最佳匹配."""
        # ha[0] 与 hb[0] 正交，与 hb[1] 完全相同
        a = [[1.0, 0.0]]
        b = [[0.0, 1.0], [1.0, 0.0]]
        sim = VideoDeduplicator._compute_histogram_similarity(a, b)
        # Best match for [1,0]: max(BC([1,0],[0,1]), BC([1,0],[1,0])) = max(0, 1) = 1
        assert sim == pytest.approx(1.0)