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 TestAverageHistogramSimilarity:
    """_average_histogram_similarity 直方图相似度测试."""

    def test_identical_histograms(self):
        """完全相同的直方图相似度为1.0."""
        hist = [[0.5, 0.5, 0.0], [0.3, 0.4, 0.3]]
        sim = VideoDeduplicator._average_histogram_similarity(hist, hist)
        assert sim == pytest.approx(1.0)

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

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

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

    def test_orthogonal_histograms(self):
        """正交直方图相似度为0."""
        # [1, 0] 和 [0, 1] 正交
        sim = VideoDeduplicator._average_histogram_similarity([[1.0, 0.0]], [[0.0, 1.0]])
        assert sim == pytest.approx(0.0)

    def test_partial_similarity(self):
        """部分相似."""
        # [1, 1] 和 [1, 0] 的余弦相似度 = 1/√2 ≈ 0.707
        sim = VideoDeduplicator._average_histogram_similarity([[1.0, 1.0]], [[1.0, 0.0]])
        assert sim == pytest.approx(1.0 / (2**0.5), rel=0.01)

    def test_multiple_frames_best_match(self):
        """多帧时取最佳匹配."""
        # 第一帧完全不同，第二帧完全相同 → 平均 best = (0 + 1) / 2 = 0.5
        sim = VideoDeduplicator._average_histogram_similarity(
            [[1.0, 0.0], [0.0, 1.0]],
            [[0.0, 1.0]],  # 只有一帧，和第一帧0相似，和第二帧1相似
        )
        # 第一帧最佳匹配=0，第二帧最佳匹配=1，平均=0.5
        assert sim == pytest.approx(0.5)

    def test_zero_norm_histogram_skipped(self):
        """零范数直方图被跳过."""
        sim = VideoDeduplicator._average_histogram_similarity([[0.0, 0.0]], [[1.0, 1.0]])
        # 第一组的零范数被跳过，similarities为空，返回0
        assert sim == 0.0

    def test_different_length_histograms(self):
        """不同长度的直方图取最小长度对齐."""
        sim = VideoDeduplicator._average_histogram_similarity(
            [[1.0, 1.0, 0.0, 0.0]],  # 4维
            [[1.0, 1.0]],  # 2维
        )
        # 对齐到前2维，都是[1,1]，相似度1.0
        assert sim == pytest.approx(1.0)

    def test_similarity_in_zero_one_range(self):
        """相似度在[0, 1]范围内."""
        hist_a = [np.random.rand(96).tolist() for _ in range(5)]
        hist_b = [np.random.rand(96).tolist() for _ in range(5)]
        sim = VideoDeduplicator._average_histogram_similarity(hist_a, hist_b)
        assert 0.0 <= sim <= 1.0