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