xiaoxia-saas/tests/unit/test_1970_atom_clip_resolver.py

"""#1970 原子片段 resolver 单元测试：DB 加载 + 内存兜底."""

from __future__ import annotations

from packages.domain.asset_atom_clip import AssetAtomClip
from packages.domain.atom_clip_resolver import (
    flatten_candidates,
    load_atom_clips_for_assets,
)


def _atom(asset_id: str, idx: int, start: float, end: float) -> AssetAtomClip:
    return AssetAtomClip(
        id=f"{asset_id}-clip-{idx}",
        asset_id=asset_id,
        start_time=start,
        end_time=end,
        duration=round(end - start, 3),
        clip_index=idx,
    )


class FakeAtomRepo:
    def __init__(self, by_asset):
        self._by_asset = by_asset

    def find_candidates_for_selection(self, asset_ids, *, limit=0):
        out = []
        for aid in asset_ids:
            out.extend(self._by_asset.get(aid, []))
        return out

    def find_by_asset(self, asset_id):
        return list(self._by_asset.get(asset_id, []))


class _Asset:
    def __init__(self, duration):
        self.duration = duration


class FakeAssetRepo:
    def __init__(self, durations):
        self._durations = durations

    def get(self, asset_id):
        d = self._durations.get(asset_id)
        return _Asset(d) if d is not None else None


class TestLoadAtomClips:
    def test_persisted_clips_loaded_sorted(self):
        clips = [_atom("a", 1, 4.5, 9.0), _atom("a", 0, 0.0, 4.5)]
        repo = FakeAtomRepo({"a": clips})
        result = load_atom_clips_for_assets(["a"], atom_clip_repo=repo)
        assert [c.clip_index for c in result["a"]] == [0, 1]

    def test_dedup_asset_ids_preserves_order(self):
        repo = FakeAtomRepo({"a": [_atom("a", 0, 0, 4)], "b": [_atom("b", 0, 0, 4)]})
        result = load_atom_clips_for_assets(["a", "b", "a"], atom_clip_repo=repo)
        assert list(result.keys()) == ["a", "b"]

    def test_fallback_when_no_persisted_clips(self):
        """老素材没有 atom_clips 时，内存按 3-6 秒均匀切片，标记 is_fallback。"""
        atom_repo = FakeAtomRepo({})
        asset_repo = FakeAssetRepo({"old": 20.0})
        result = load_atom_clips_for_assets(["old"], atom_clip_repo=atom_repo, asset_repo=asset_repo)
        assert "old" in result
        clips = result["old"]
        assert clips
        assert all(c.is_fallback for c in clips)
        assert abs(clips[-1].end_time - 20.0) < 0.01

    def test_missing_duration_skipped(self):
        atom_repo = FakeAtomRepo({})
        asset_repo = FakeAssetRepo({})
        result = load_atom_clips_for_assets(["ghost"], atom_clip_repo=atom_repo, asset_repo=asset_repo)
        assert result == {}

    def test_no_asset_repo_skips_empty_assets(self):
        atom_repo = FakeAtomRepo({})
        result = load_atom_clips_for_assets(["a"], atom_clip_repo=atom_repo, asset_repo=None)
        assert result == {}

    def test_mixed_persisted_and_fallback(self):
        atom_repo = FakeAtomRepo({"new": [_atom("new", 0, 0, 5)]})
        asset_repo = FakeAssetRepo({"new": 5.0, "old": 10.0})
        result = load_atom_clips_for_assets(["new", "old"], atom_clip_repo=atom_repo, asset_repo=asset_repo)
        assert not result["new"][0].is_fallback
        assert all(c.is_fallback for c in result["old"])

    def test_repo_exception_falls_back(self):
        class BrokenRepo(FakeAtomRepo):
            def find_candidates_for_selection(self, asset_ids, *, limit=0):
                raise RuntimeError("db down")

        asset_repo = FakeAssetRepo({"a": 9.0})
        result = load_atom_clips_for_assets(["a"], atom_clip_repo=BrokenRepo({}), asset_repo=asset_repo)
        assert result["a"]
        assert all(c.is_fallback for c in result["a"])

    def test_empty_input(self):
        assert load_atom_clips_for_assets([], atom_clip_repo=FakeAtomRepo({})) == {}


class TestFlatten:
    def test_flatten_order(self):
        clips = flatten_candidates({"a": [_atom("a", 0, 0, 4)], "b": [_atom("b", 0, 0, 4), _atom("b", 1, 4, 8)]})
        assert len(clips) == 3
        assert clips[0].asset_id == "a"