xiaoxia-saas/tests/unit/test_in_memory_asset_repository.py

"""InMemoryAssetRepository 单测 — 素材仓储内存实现."""

from __future__ import annotations

import pytest

from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository
from packages.domain import Asset, AssetStatus

# ── Fixtures ───────────────────────────────────────────────────────────────


@pytest.fixture
def repo():
    return InMemoryAssetRepository()


@pytest.fixture
def sample_asset():
    return Asset.create(
        project_id="proj-1",
        library_id="lib-1",
        name="test.mp4",
        storage_key="assets/test.mp4",
        mime_type="video/mp4",
        file_size=1024,
        file_hash="hash-abc",
    )


@pytest.fixture
def sample_assets(repo):
    """创建几个测试素材."""
    assets = []
    for i in range(5):
        asset = Asset.create(
            project_id="proj-1",
            library_id="lib-1",
            name=f"video_{i}.mp4",
            storage_key=f"assets/video_{i}.mp4",
            mime_type="video/mp4",
            file_size=1000 + i,
            file_hash=f"hash-{i}",
        )
        assets.append(repo.create(asset))
    return assets


# ── CRUD 基本操作 ──────────────────────────────────────────────────────────


class TestAssetRepoCRUD:
    """基本 CRUD 操作."""

    def test_create_and_get(self, repo, sample_asset):
        created = repo.create(sample_asset)
        assert created.id == sample_asset.id

        fetched = repo.get(sample_asset.id)
        assert fetched is not None
        assert fetched.id == sample_asset.id
        assert fetched.name == "test.mp4"

    def test_get_not_found(self, repo):
        assert repo.get("nonexistent") is None

    def test_find_by_id_alias(self, repo, sample_asset):
        repo.create(sample_asset)
        assert repo.find_by_id(sample_asset.id).id == sample_asset.id

    def test_update(self, repo, sample_asset):
        repo.create(sample_asset)
        sample_asset.name = "renamed.mp4"
        updated = repo.update(sample_asset)
        assert updated.name == "renamed.mp4"

        fetched = repo.get(sample_asset.id)
        assert fetched.name == "renamed.mp4"

    def test_delete_existing(self, repo, sample_asset):
        repo.create(sample_asset)
        result = repo.delete(sample_asset.id)
        assert result is True
        assert repo.get(sample_asset.id) is None

    def test_delete_nonexistent(self, repo):
        result = repo.delete("nonexistent")
        assert result is False


# ── 查询方法 ───────────────────────────────────────────────────────────────


class TestAssetRepoQueries:
    """查询类方法."""

    def test_list_by_project(self, repo, sample_assets):
        result = repo.list_by_project("proj-1")
        assert len(result) == 5

    def test_list_by_project_empty(self, repo):
        result = repo.list_by_project("nonexistent")
        assert result == []

    def test_list_by_library(self, repo, sample_assets):
        result = repo.list_by_library("lib-1")
        assert len(result) == 5

    def test_find_by_library_alias(self, repo, sample_assets):
        result = repo.find_by_library("lib-1")
        assert len(result) == 5

    def test_find_by_library_and_file_type_video(self, repo):
        video = Asset.create(
            project_id="p1",
            library_id="lib-1",
            name="v.mp4",
            storage_key="v.mp4",
            mime_type="video/mp4",
        )
        audio = Asset.create(
            project_id="p1",
            library_id="lib-1",
            name="a.mp3",
            storage_key="a.mp3",
            mime_type="audio/mp3",
        )
        image = Asset.create(
            project_id="p1",
            library_id="lib-1",
            name="i.jpg",
            storage_key="i.jpg",
            mime_type="image/jpeg",
        )
        repo.create(video)
        repo.create(audio)
        repo.create(image)

        videos = repo.find_by_library_and_file_type("lib-1", "video")
        assert len(videos) == 1
        assert videos[0].id == video.id

        audios = repo.find_by_library_and_file_type("lib-1", "audio")
        assert len(audios) == 1
        assert audios[0].id == audio.id

    def test_find_by_project_with_pagination(self, repo, sample_assets):
        result = repo.find_by_project("proj-1", skip=0, limit=3)
        assert len(result) == 3

        result2 = repo.find_by_project("proj-1", skip=3, limit=10)
        assert len(result2) == 2

    def test_find_by_tag_ids_single_tag(self, repo):
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.tag_ids = ["tag1", "tag2"]
        a2 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a2.mp4",
            storage_key="a2.mp4",
            mime_type="video/mp4",
        )
        a2.tag_ids = ["tag1"]
        a3 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a3.mp4",
            storage_key="a3.mp4",
            mime_type="video/mp4",
        )
        a3.tag_ids = ["tag3"]
        repo.create(a1)
        repo.create(a2)
        repo.create(a3)

        result = repo.find_by_tag_ids(["tag1"])
        assert len(result) == 2

    def test_find_by_tag_ids_multiple_tags_all_match(self, repo):
        """必须包含所有指定标签（AND 逻辑）."""
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.tag_ids = ["tag1", "tag2"]
        a2 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a2.mp4",
            storage_key="a2.mp4",
            mime_type="video/mp4",
        )
        a2.tag_ids = ["tag1"]
        repo.create(a1)
        repo.create(a2)

        result = repo.find_by_tag_ids(["tag1", "tag2"])
        assert len(result) == 1
        assert result[0].id == a1.id

    def test_find_by_tag_ids_empty_list(self, repo, sample_assets):
        result = repo.find_by_tag_ids([])
        assert result == []

    def test_find_by_library_and_file_hash(self, repo, sample_asset):
        repo.create(sample_asset)
        result = repo.find_by_library_and_file_hash("lib-1", "hash-abc")
        assert result is not None
        assert result.id == sample_asset.id

    def test_find_by_library_and_file_hash_not_found(self, repo):
        result = repo.find_by_library_and_file_hash("lib-1", "nonexistent")
        assert result is None

    def test_find_by_library_and_file_hash_empty_hash(self, repo, sample_asset):
        repo.create(sample_asset)
        result = repo.find_by_library_and_file_hash("lib-1", "")
        assert result is None


# ── 批量操作 ───────────────────────────────────────────────────────────────


class TestAssetRepoBatchOperations:
    """批量操作方法."""

    def test_batch_delete_marks_deleted(self, repo):
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a2 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a2.mp4",
            storage_key="a2.mp4",
            mime_type="video/mp4",
        )
        repo.create(a1)
        repo.create(a2)

        count = repo.batch_delete([a1.id, a2.id])
        assert count == 2

        # 状态变为 deleted
        assert repo.get(a1.id).status == AssetStatus.DELETED
        assert repo.get(a2.id).status == AssetStatus.DELETED

    def test_batch_delete_skip_already_deleted(self, repo):
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.status = AssetStatus.DELETED
        repo.create(a1)
        a2 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a2.mp4",
            storage_key="a2.mp4",
            mime_type="video/mp4",
        )
        repo.create(a2)

        count = repo.batch_delete([a1.id, a2.id])
        assert count == 1  # 只有a2被标记

    def test_batch_delete_nonexistent(self, repo):
        count = repo.batch_delete(["nonexistent"])
        assert count == 0

    def test_batch_update_metadata(self, repo):
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.metadata = {"key1": "val1"}
        a2 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a2.mp4",
            storage_key="a2.mp4",
            mime_type="video/mp4",
        )
        repo.create(a1)
        repo.create(a2)

        count = repo.batch_update_metadata([a1.id, a2.id], {"key2": "val2"})
        assert count == 2

        # 合并而非覆盖
        assert repo.get(a1.id).metadata["key1"] == "val1"
        assert repo.get(a1.id).metadata["key2"] == "val2"
        assert repo.get(a2.id).metadata["key2"] == "val2"

    def test_batch_add_tags(self, repo):
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.tag_ids = ["existing"]
        repo.create(a1)

        count = repo.batch_add_tags([a1.id], ["tag1", "tag2"])
        assert count == 1

        tags = repo.get(a1.id).tag_ids
        assert "existing" in tags
        assert "tag1" in tags
        assert "tag2" in tags

    def test_batch_add_tags_dedup(self, repo):
        """添加已存在的标签不会重复."""
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.tag_ids = ["tag1"]
        repo.create(a1)

        before_count = len(a1.tag_ids)
        repo.batch_add_tags([a1.id], ["tag1", "tag1"])
        # 没有变化，count 应该是0？不对，tag_ids去重后还是["tag1"]，但原先是["tag1"]
        # 添加tag1时发现已存在，changed=False，所以count=0
        assert repo.get(a1.id).tag_ids.count("tag1") == 1

    def test_batch_replace_tags(self, repo):
        a1 = Asset.create(
            project_id="p1",
            library_id="l1",
            name="a1.mp4",
            storage_key="a1.mp4",
            mime_type="video/mp4",
        )
        a1.tag_ids = ["old1", "old2"]
        repo.create(a1)

        count = repo.batch_replace_tags([a1.id], ["new1", "new2"])
        assert count == 1

        tags = repo.get(a1.id).tag_ids
        assert tags == ["new1", "new2"]