224 lines
8.2 KiB
Python
Executable File
224 lines
8.2 KiB
Python
Executable File
"""去重纯算法测试 — hamming_distance + histogram_similarity + VideoFingerprint."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from unittest.mock import MagicMock
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
# 模块级mock有副作用的依赖（纯算法测试不需要db/celery/cv2）
|
||
# 注意：必须在 import dedup 前全部 mock 完，避免链式导入触发db连接
|
||
|
||
# cv2（视频处理依赖，纯算法测试不需要）
|
||
sys.modules["cv2"] = MagicMock()
|
||
# 模块级mock worker_app.db（dedup模块import时会触发数据库初始化，纯算法测试不需要）
|
||
sys.modules["worker_app.db"] = MagicMock()
|
||
sys.modules["worker_app.db"].SessionLocal = MagicMock()
|
||
|
||
# celery 及其子模块
|
||
_mock_celery = MagicMock()
|
||
_mock_celery.Task = MagicMock
|
||
_mock_celery.Celery = MagicMock
|
||
sys.modules["celery"] = _mock_celery
|
||
|
||
# sqlalchemy 作为包结构 mock
|
||
_mock_sqla = MagicMock()
|
||
_mock_sqla.__path__ = []
|
||
_mock_sqla.__package__ = "sqlalchemy"
|
||
_mock_sqla_orm = MagicMock()
|
||
_mock_sqla_orm.__path__ = []
|
||
_mock_sqla_orm.Session = MagicMock
|
||
_mock_sqla_engine = MagicMock()
|
||
sys.modules["sqlalchemy"] = _mock_sqla
|
||
sys.modules["sqlalchemy.orm"] = _mock_sqla_orm
|
||
sys.modules["sqlalchemy.engine"] = _mock_sqla_engine
|
||
sys.modules["sqlalchemy.ext"] = MagicMock()
|
||
sys.modules["sqlalchemy.ext.declarative"] = MagicMock()
|
||
|
||
# worker_app 及其子模块（避免导入时触发数据库连接）
|
||
_mock_worker_app = MagicMock()
|
||
_mock_worker_app.__path__ = []
|
||
_mock_worker_db = MagicMock()
|
||
_mock_worker_db.SessionLocal = MagicMock()
|
||
_mock_worker_celery = MagicMock()
|
||
_mock_worker_celery.celery_app = MagicMock()
|
||
_mock_worker_core = MagicMock()
|
||
_mock_worker_core.__path__ = []
|
||
_mock_worker_config = MagicMock()
|
||
_mock_worker_config.get_settings = MagicMock(return_value=MagicMock())
|
||
sys.modules["worker_app"] = _mock_worker_app
|
||
sys.modules["worker_app.db"] = _mock_worker_db
|
||
sys.modules["worker_app.celery_app"] = _mock_worker_celery
|
||
sys.modules["worker_app.core"] = _mock_worker_core
|
||
sys.modules["worker_app.core.config"] = _mock_worker_config
|
||
|
||
# packages.adapters.sqlalchemy_impl（整个包mock掉）
|
||
_mock_sqla_impl = MagicMock()
|
||
_mock_sqla_impl.__path__ = []
|
||
sys.modules["packages.adapters.sqlalchemy_impl"] = _mock_sqla_impl
|
||
sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = MagicMock()
|
||
sys.modules["packages.adapters.sqlalchemy_impl.schema_guard"] = MagicMock()
|
||
|
||
# packages.shared
|
||
_mock_packages_shared = MagicMock()
|
||
_mock_packages_shared.__path__ = []
|
||
_mock_shared_config = MagicMock()
|
||
_mock_shared_config.get_shared_settings = MagicMock(return_value=MagicMock())
|
||
sys.modules["packages.shared"] = _mock_packages_shared
|
||
sys.modules["packages.shared.config"] = _mock_shared_config
|
||
sys.modules["packages.shared.storage"] = MagicMock()
|
||
|
||
from video_processing.dedup import ( # noqa: E402
|
||
VideoDeduplicator,
|
||
VideoFingerprint,
|
||
hamming_distance,
|
||
)
|
||
|
||
|
||
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
|