xiaoxia-saas/tests/unit/test_fingerprint_chunks.py

"""分片指纹存储单元测试 — Issue #1657.

覆盖：
- 分片策略：60秒视频 → 30片，120秒视频 → 24片
- VideoFingerprint.to_chunk_models() 输出正确
- _save_fingerprint_chunks 幂等性（已有数据跳过）
- to_dict() 向后兼容
"""

from __future__ import annotations

import sys
from unittest.mock import MagicMock


def _mock_module(**attrs):
    """Create a mock module with __spec__ to avoid AttributeError."""
    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 ──
_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.adapters.sqlalchemy_impl.models",
        "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()


# Mock VideoFingerprintChunkModel with class-level column attributes
class _FakeChunkModel:
    video_id = MagicMock()
    project_id = MagicMock()
    user_id = MagicMock()
    start_time_ms = MagicMock()
    end_time_ms = MagicMock()
    phash_binary = MagicMock()
    color_histogram = MagicMock()
    frame_count = MagicMock()
    created_at = MagicMock()

    def __init__(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)


sys.modules["packages.adapters.sqlalchemy_impl.models"] = _mock_module(
    VideoFingerprintChunkModel=_FakeChunkModel,
)
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
    FingerprintChunk,
    VideoFingerprint,
    _save_fingerprint_chunks,
    compute_chunk_interval,
)

# ── Restore sys.modules immediately after import ──
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 TestChunkInterval:
    """测试分片间隔策略。"""

    def test_short_video_interval(self):
        """短视频（≤60秒）每 2 秒一个分片。"""
        assert compute_chunk_interval(0) == 2
        assert compute_chunk_interval(30) == 2
        assert compute_chunk_interval(60) == 2

    def test_long_video_interval(self):
        """长视频（>60秒）每 5 秒一个分片。"""
        assert compute_chunk_interval(61) == 5
        assert compute_chunk_interval(120) == 5
        assert compute_chunk_interval(300) == 5

    def test_chunk_count_60s_video(self):
        """60秒视频 → 30 片（60/2=30）。"""
        duration = 60
        interval = compute_chunk_interval(duration)
        expected_chunks = int(duration / interval)
        assert expected_chunks == 30

    def test_chunk_count_120s_video(self):
        """120秒视频 → 24 片（120/5=24）。"""
        duration = 120
        interval = compute_chunk_interval(duration)
        expected_chunks = int(duration / interval)
        assert expected_chunks == 24


class TestVideoFingerprintToChunkModels:
    """测试 VideoFingerprint.to_chunk_models() 输出。"""

    def test_to_chunk_models_output(self):
        """to_chunk_models 返回正确的 Model 列表。"""
        fp = VideoFingerprint(
            md5="abc123",
            keyframe_phashes=["a1b2", "c3d4"],
            color_histograms=[[0.1] * 96, [0.2] * 96],
            duration=10.0,
            resolution=(1920, 1080),
            chunks=[
                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
                FingerprintChunk(start_time_ms=2000, end_time_ms=4000, phash_binary="c3d4", color_histogram=[0.2] * 96),
            ],
        )

        models = fp.to_chunk_models(video_id="v1", project_id="p1", user_id="u1")

        assert len(models) == 2
        assert models[0].video_id == "v1"
        assert models[0].project_id == "p1"
        assert models[0].user_id == "u1"
        assert models[0].start_time_ms == 0
        assert models[0].end_time_ms == 2000
        assert models[0].phash_binary == "a1b2"
        assert models[1].start_time_ms == 2000
        assert models[1].end_time_ms == 4000
        assert models[1].phash_binary == "c3d4"

    def test_to_chunk_models_empty_chunks(self):
        """空 chunks 列表返回空 Model 列表。"""
        fp = VideoFingerprint(
            md5="abc",
            keyframe_phashes=[],
            color_histograms=[],
            duration=0,
            resolution=(0, 0),
            chunks=[],
        )

        models = fp.to_chunk_models(video_id="v1", project_id="p1")
        assert models == []


class TestSaveFingerprintChunksIdempotent:
    """测试 _save_fingerprint_chunks 幂等性。"""

    def test_save_skips_existing(self):
        """已有分片数据时跳过写入。"""
        fp = VideoFingerprint(
            md5="abc",
            keyframe_phashes=["a1b2"],
            color_histograms=[[0.1] * 96],
            duration=5.0,
            resolution=(1920, 1080),
            chunks=[
                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
            ],
        )

        session = MagicMock()
        # Mock: 已有 1 条分片数据
        session.query.return_value.filter.return_value.count.return_value = 1

        _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)

        # bulk_save_objects 不应被调用
        session.bulk_save_objects.assert_not_called()

    def test_save_writes_new(self):
        """无分片数据时写入。"""
        fp = VideoFingerprint(
            md5="abc",
            keyframe_phashes=["a1b2"],
            color_histograms=[[0.1] * 96],
            duration=5.0,
            resolution=(1920, 1080),
            chunks=[
                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
            ],
        )

        session = MagicMock()
        # Mock: 无分片数据
        session.query.return_value.filter.return_value.count.return_value = 0

        _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)

        # bulk_save_objects 应被调用一次
        session.bulk_save_objects.assert_called_once()
        saved_models = session.bulk_save_objects.call_args[0][0]
        assert len(saved_models) == 1
        assert saved_models[0].video_id == "v1"
        assert saved_models[0].phash_binary == "a1b2"

    def test_save_skips_no_chunks(self):
        """指纹无 chunks 时跳过。"""
        fp = VideoFingerprint(
            md5="abc",
            keyframe_phashes=[],
            color_histograms=[],
            duration=0,
            resolution=(0, 0),
            chunks=[],
        )

        session = MagicMock()
        session.query.return_value.filter.return_value.count.return_value = 0

        _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)

        # bulk_save_objects 不应被调用
        session.bulk_save_objects.assert_not_called()


class TestFingerprintToDictBackwardCompat:
    """测试 to_dict() 向后兼容性。"""

    def test_to_dict_includes_chunks(self):
        """to_dict() 包含 chunks 字段。"""
        fp = VideoFingerprint(
            md5="abc123",
            keyframe_phashes=["a1b2"],
            color_histograms=[[0.1] * 96],
            duration=5.0,
            resolution=(1920, 1080),
            chunks=[
                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
            ],
        )

        d = fp.to_dict()

        assert "chunks" in d
        assert len(d["chunks"]) == 1
        assert d["chunks"][0]["start_time_ms"] == 0
        assert d["chunks"][0]["end_time_ms"] == 2000
        assert d["chunks"][0]["phash_binary"] == "a1b2"

    def test_to_dict_preserves_legacy_fields(self):
        """to_dict() 保留 keyframe_phashes 和 color_histograms 字段。"""
        fp = VideoFingerprint(
            md5="abc",
            keyframe_phashes=["a1b2", "c3d4"],
            color_histograms=[[0.1] * 96, [0.2] * 96],
            duration=10.0,
            resolution=(1920, 1080),
        )

        d = fp.to_dict()

        assert "keyframe_phashes" in d
        assert "color_histograms" in d
        assert len(d["keyframe_phashes"]) == 2
        assert len(d["color_histograms"]) == 2