xiaoxia-saas/tests/unit/test_inmemory_asset_repository.py

"""InMemoryAssetRepository 单元测试."""

import pytest

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


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


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


@pytest.fixture
def asset2() -> Asset:
    return Asset.create(
        project_id="proj-1",
        library_id="lib-1",
        name="test2.jpg",
        storage_key="storage/key2",
        mime_type="image/jpeg",
        file_size=512,
        file_hash="hash-def",
    )


@pytest.fixture
def asset_other_project() -> Asset:
    return Asset.create(
        project_id="proj-2",
        library_id="lib-2",
        name="other.mp3",
        storage_key="storage/key3",
        mime_type="audio/mpeg",
        file_size=256,
        file_hash="hash-ghi",
    )


class TestCreateAndGet:
    def test_create_returns_asset(self, repo, sample_asset):
        result = repo.create(sample_asset)
        assert result.id == sample_asset.id
        assert result.name == "test.mp4"

    def test_get_existing_asset(self, repo, sample_asset):
        repo.create(sample_asset)
        result = repo.get(sample_asset.id)
        assert result is not None
        assert result.id == sample_asset.id

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

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


class TestListByProject:
    def test_list_by_project_filters_correctly(self, repo, sample_asset, asset2, asset_other_project):
        repo.create(sample_asset)
        repo.create(asset2)
        repo.create(asset_other_project)

        proj1 = repo.list_by_project("proj-1")
        assert len(proj1) == 2
        assert all(a.project_id == "proj-1" for a in proj1)

        proj2 = repo.list_by_project("proj-2")
        assert len(proj2) == 1
        assert proj2[0].id == asset_other_project.id

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


class TestListByLibrary:
    def test_list_by_library_filters_correctly(self, repo, sample_asset, asset2, asset_other_project):
        repo.create(sample_asset)
        repo.create(asset2)
        repo.create(asset_other_project)

        lib1 = repo.list_by_library("lib-1")
        assert len(lib1) == 2

        lib2 = repo.list_by_library("lib-2")
        assert len(lib2) == 1
        assert lib2[0].id == asset_other_project.id

    def test_find_by_library_is_alias(self, repo, sample_asset):
        repo.create(sample_asset)
        assert repo.find_by_library("lib-1") == repo.list_by_library("lib-1")

    def test_list_by_library_empty(self, repo):
        assert repo.list_by_library("nonexistent") == []


class TestFindByLibraryAndFileType:
    def test_filter_by_video(self, repo, sample_asset, asset2, asset_other_project):
        repo.create(sample_asset)
        repo.create(asset2)
        repo.create(asset_other_project)

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

    def test_filter_by_image(self, repo, sample_asset, asset2):
        repo.create(sample_asset)
        repo.create(asset2)

        images = repo.find_by_library_and_file_type("lib-1", "image")
        assert len(images) == 1
        assert images[0].mime_type.startswith("image/")

    def test_filter_by_audio(self, repo, sample_asset, asset_other_project):
        repo.create(sample_asset)
        repo.create(asset_other_project)

        audio = repo.find_by_library_and_file_type("lib-2", "audio")
        assert len(audio) == 1

    def test_empty_result(self, repo, sample_asset):
        repo.create(sample_asset)
        assert repo.find_by_library_and_file_type("lib-1", "audio") == []


class TestUpdate:
    def test_update_existing_asset(self, repo, sample_asset):
        repo.create(sample_asset)
        sample_asset.name = "updated.mp4"
        sample_asset.file_size = 2048

        result = repo.update(sample_asset)
        assert result.name == "updated.mp4"
        assert result.file_size == 2048

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

    def test_update_nonexistent_creates(self, repo, sample_asset):
        """update 直接覆盖，不存在则相当于 create."""
        result = repo.update(sample_asset)
        assert result.id == sample_asset.id
        assert repo.get(sample_asset.id) is not None


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

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


class TestBatchDelete:
    def test_batch_delete_soft_delete(self, repo, sample_asset, asset2):
        repo.create(sample_asset)
        repo.create(asset2)

        count = repo.batch_delete([sample_asset.id, asset2.id])
        assert count == 2

        a1 = repo.get(sample_asset.id)
        a2 = repo.get(asset2.id)
        assert a1.status == AssetStatus.DELETED
        assert a2.status == AssetStatus.DELETED
        assert a1.updated_at is not None
        assert a2.updated_at is not None

    def test_batch_delete_skip_already_deleted(self, repo, sample_asset):
        repo.create(sample_asset)
        sample_asset.status = AssetStatus.DELETED
        repo.update(sample_asset)

        count = repo.batch_delete([sample_asset.id])
        assert count == 0

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

    def test_batch_delete_partial(self, repo, sample_asset):
        repo.create(sample_asset)
        count = repo.batch_delete([sample_asset.id, "nonexistent"])
        assert count == 1


class TestBatchUpdateMetadata:
    def test_batch_update_metadata_merge(self, repo, sample_asset, asset2):
        sample_asset.metadata = {"key1": "val1"}
        repo.create(sample_asset)
        repo.create(asset2)

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

        a1 = repo.get(sample_asset.id)
        a2 = repo.get(asset2.id)
        assert a1.metadata == {"key1": "val1", "key2": "val2"}
        assert a2.metadata == {"key2": "val2"}

    def test_batch_update_metadata_overwrite_existing_key(self, repo, sample_asset):
        sample_asset.metadata = {"key1": "old"}
        repo.create(sample_asset)

        count = repo.batch_update_metadata([sample_asset.id], {"key1": "new"})
        assert count == 1
        assert repo.get(sample_asset.id).metadata["key1"] == "new"

    def test_batch_update_metadata_nonexistent(self, repo):
        count = repo.batch_update_metadata(["nonexistent"], {"key": "val"})
        assert count == 0


class TestBatchAddTags:
    def test_batch_add_tags_new_tags(self, repo, sample_asset, asset2):
        repo.create(sample_asset)
        repo.create(asset2)

        count = repo.batch_add_tags([sample_asset.id, asset2.id], ["tag1", "tag2"])
        assert count == 2

        a1 = repo.get(sample_asset.id)
        a2 = repo.get(asset2.id)
        assert set(a1.tag_ids) == {"tag1", "tag2"}
        assert set(a2.tag_ids) == {"tag1", "tag2"}

    def test_batch_add_tags_dedup(self, repo, sample_asset):
        sample_asset.tag_ids = ["tag1"]
        repo.create(sample_asset)

        count = repo.batch_add_tags([sample_asset.id], ["tag1", "tag2"])
        assert count == 1  # tag1已存在，但tag2新增，所以有变化

        tags = repo.get(sample_asset.id).tag_ids
        assert tags.count("tag1") == 1
        assert "tag2" in tags

    def test_batch_add_tags_no_change_when_all_exist(self, repo, sample_asset):
        sample_asset.tag_ids = ["tag1", "tag2"]
        repo.create(sample_asset)

        count = repo.batch_add_tags([sample_asset.id], ["tag1", "tag2"])
        assert count == 0  # 没有变化

    def test_batch_add_tags_nonexistent_assets(self, repo):
        count = repo.batch_add_tags(["nonexistent"], ["tag1"])
        assert count == 0


class TestBatchReplaceTags:
    def test_batch_replace_tags_override(self, repo, sample_asset):
        sample_asset.tag_ids = ["old1", "old2"]
        repo.create(sample_asset)

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

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

    def test_batch_replace_tags_empty(self, repo, sample_asset):
        sample_asset.tag_ids = ["tag1"]
        repo.create(sample_asset)

        count = repo.batch_replace_tags([sample_asset.id], [])
        assert count == 1
        assert repo.get(sample_asset.id).tag_ids == []

    def test_batch_replace_tags_nonexistent(self, repo):
        count = repo.batch_replace_tags(["nonexistent"], ["tag1"])
        assert count == 0


class TestFindByProjectPagination:
    @pytest.fixture
    def five_assets(self, repo):
        assets = []
        for i in range(5):
            a = Asset.create(
                project_id="proj-paged",
                library_id="lib-paged",
                name=f"asset-{i}.mp4",
                storage_key=f"key-{i}",
                mime_type="video/mp4",
            )
            repo.create(a)
            assets.append(a)
        return assets

    def test_find_by_project_default_pagination(self, repo, five_assets):
        result = repo.find_by_project("proj-paged")
        assert len(result) == 5

    def test_find_by_project_skip(self, repo, five_assets):
        result = repo.find_by_project("proj-paged", skip=2)
        assert len(result) == 3

    def test_find_by_project_limit(self, repo, five_assets):
        result = repo.find_by_project("proj-paged", limit=2)
        assert len(result) == 2

    def test_find_by_project_skip_and_limit(self, repo, five_assets):
        result = repo.find_by_project("proj-paged", skip=1, limit=2)
        assert len(result) == 2

    def test_find_by_project_skip_past_end(self, repo, five_assets):
        result = repo.find_by_project("proj-paged", skip=10)
        assert result == []

    def test_find_by_project_empty(self, repo):
        assert repo.find_by_project("nonexistent") == []


class TestFindByTagIds:
    def test_find_by_tag_ids_match_all(self, repo, sample_asset, asset2):
        sample_asset.tag_ids = ["tag1", "tag2", "tag3"]
        asset2.tag_ids = ["tag1", "tag2"]
        repo.create(sample_asset)
        repo.create(asset2)

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

    def test_find_by_tag_ids_subset_match(self, repo, sample_asset, asset2):
        sample_asset.tag_ids = ["tag1", "tag2"]
        asset2.tag_ids = ["tag1"]
        repo.create(sample_asset)
        repo.create(asset2)

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

    def test_find_by_tag_ids_empty_tag_list(self, repo, sample_asset):
        sample_asset.tag_ids = ["tag1"]
        repo.create(sample_asset)
        assert repo.find_by_tag_ids([]) == []

    def test_find_by_tag_ids_no_match(self, repo, sample_asset):
        sample_asset.tag_ids = ["tag1"]
        repo.create(sample_asset)
        assert repo.find_by_tag_ids(["tag999"]) == []

    def test_find_by_tag_ids_pagination(self, repo):
        for i in range(5):
            a = Asset.create(
                project_id="p1",
                library_id="l1",
                name=f"a{i}.mp4",
                storage_key=f"k{i}",
                mime_type="video/mp4",
            )
            a.tag_ids = ["shared-tag"]
            repo.create(a)

        result = repo.find_by_tag_ids(["shared-tag"], skip=1, limit=2)
        assert len(result) == 2


class TestFindByLibraryAndFileHash:
    def test_find_by_hash_match(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_hash_wrong_library(self, repo, sample_asset):
        repo.create(sample_asset)
        result = repo.find_by_library_and_file_hash("lib-2", "hash-abc")
        assert result is None

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

    def test_find_by_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