"""去重纯算法测试 — 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)