xiaoxia-saas/tests/unit/test_inmemory_small_repos.py

"""InMemory 小型仓储模块单元测试（asset_library/tag/project/ingest_job/classification_job）."""

from datetime import datetime, timezone

import pytest

from packages.adapters.in_memory.asset_library_repository import InMemoryAssetLibraryRepository
from packages.adapters.in_memory.classification_job_repository import InMemoryClassificationJobRepository
from packages.adapters.in_memory.ingest_job_repository import InMemoryIngestJobRepository
from packages.adapters.in_memory.project_repository import InMemoryProjectRepository
from packages.adapters.in_memory.tag_repository import InMemoryTagRepository
from packages.domain.classification import (
    AssetLibraryKind,
    ClassificationJob,
    ClassificationJobStatus,
    IngestJobStatus,
)
from packages.domain.entities import AssetLibrary, IngestJob, Project, User
from packages.domain.tag import Tag

# ==================== AssetLibrary ====================


class TestInMemoryAssetLibraryRepository:
    @pytest.fixture
    def repo(self):
        return InMemoryAssetLibraryRepository()

    @pytest.fixture
    def lib_video(self):
        return AssetLibrary.create(project_id="p1", name="视频库", kind=AssetLibraryKind.VIDEO)

    @pytest.fixture
    def lib_image(self):
        return AssetLibrary.create(project_id="p1", name="图片库", kind=AssetLibraryKind.IMAGE)

    @pytest.fixture
    def lib_other_project(self):
        return AssetLibrary.create(project_id="p2", name="其他项目库", kind=AssetLibraryKind.VIDEO)

    def test_create_and_get(self, repo, lib_video):
        result = repo.create(lib_video)
        assert result.id == lib_video.id
        assert result.name == "视频库"

        fetched = repo.get(lib_video.id)
        assert fetched is not None
        assert fetched.id == lib_video.id

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

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

    def test_find_by_project(self, repo, lib_video, lib_image, lib_other_project):
        repo.create(lib_video)
        repo.create(lib_image)
        repo.create(lib_other_project)

        p1_libs = repo.find_by_project("p1")
        assert len(p1_libs) == 2

        p2_libs = repo.find_by_project("p2")
        assert len(p2_libs) == 1
        assert p2_libs[0].id == lib_other_project.id

    def test_find_by_project_with_kind_filter(self, repo, lib_video, lib_image):
        repo.create(lib_video)
        repo.create(lib_image)

        video_libs = repo.find_by_project("p1", kind=AssetLibraryKind.VIDEO)
        assert len(video_libs) == 1
        assert video_libs[0].kind == AssetLibraryKind.VIDEO

        image_libs = repo.find_by_project("p1", kind=AssetLibraryKind.IMAGE)
        assert len(image_libs) == 1

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

    def test_update(self, repo, lib_video):
        repo.create(lib_video)
        lib_video.name = "新名称"
        result = repo.update(lib_video)
        assert result.name == "新名称"
        assert repo.get(lib_video.id).name == "新名称"

    def test_delete(self, repo, lib_video):
        repo.create(lib_video)
        assert repo.delete(lib_video.id) is True
        assert repo.get(lib_video.id) is None

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

    def test_increment_asset_count(self, repo, lib_video):
        repo.create(lib_video)
        repo.increment_asset_count(lib_video.id, size_delta=1024)

        lib = repo.get(lib_video.id)
        assert lib.asset_count == 1
        assert lib.total_size == 1024

        repo.increment_asset_count(lib_video.id, size_delta=512)
        lib = repo.get(lib_video.id)
        assert lib.asset_count == 2
        assert lib.total_size == 1536

    def test_increment_asset_count_nonexistent(self, repo):
        # 不报错，静默忽略
        repo.increment_asset_count("nonexistent", size_delta=100)

    def test_decrement_asset_count(self, repo, lib_video):
        repo.create(lib_video)
        repo.increment_asset_count(lib_video.id, size_delta=1024)
        repo.increment_asset_count(lib_video.id, size_delta=512)

        repo.decrement_asset_count(lib_video.id, size_delta=512)
        lib = repo.get(lib_video.id)
        assert lib.asset_count == 1
        assert lib.total_size == 1024

    def test_decrement_asset_count_not_below_zero(self, repo, lib_video):
        repo.create(lib_video)
        repo.decrement_asset_count(lib_video.id, size_delta=9999)
        lib = repo.get(lib_video.id)
        assert lib.asset_count == 0
        assert lib.total_size == 0

    def test_decrement_asset_count_nonexistent(self, repo):
        repo.decrement_asset_count("nonexistent", size_delta=100)


# ==================== Tag ====================


class TestInMemoryTagRepository:
    @pytest.fixture
    def repo(self):
        return InMemoryTagRepository()

    @pytest.fixture
    def tag1(self):
        return Tag.create(user_id="u1", name="风景")

    @pytest.fixture
    def tag2(self):
        return Tag.create(user_id="u1", name="人物")

    @pytest.fixture
    def tag_other_user(self):
        return Tag.create(user_id="u2", name="风景")

    def test_create_and_get(self, repo, tag1):
        result = repo.create(tag1)
        assert result.id == tag1.id
        assert result.name == "风景"

        fetched = repo.get(tag1.id)
        assert fetched is not None
        assert fetched.id == tag1.id

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

    def test_find_by_name(self, repo, tag1, tag_other_user):
        repo.create(tag1)
        repo.create(tag_other_user)

        # 同用户同名
        found = repo.find_by_name("u1", "风景")
        assert found is not None
        assert found.id == tag1.id

        # 不同用户同名不冲突
        found2 = repo.find_by_name("u2", "风景")
        assert found2 is not None
        assert found2.id == tag_other_user.id

    def test_find_by_name_not_found(self, repo, tag1):
        repo.create(tag1)
        assert repo.find_by_name("u1", "不存在") is None
        assert repo.find_by_name("u2", "风景") is None

    def test_list_by_user(self, repo, tag1, tag2, tag_other_user):
        repo.create(tag1)
        repo.create(tag2)
        repo.create(tag_other_user)

        u1_tags = repo.list_by_user("u1")
        assert len(u1_tags) == 2

        u2_tags = repo.list_by_user("u2")
        assert len(u2_tags) == 1
        assert u2_tags[0].id == tag_other_user.id

    def test_list_by_user_pagination(self, repo):
        for i in range(5):
            repo.create(Tag.create(user_id="u1", name=f"tag-{i}"))

        page1 = repo.list_by_user("u1", limit=2)
        assert len(page1) == 2

        page2 = repo.list_by_user("u1", skip=2, limit=2)
        assert len(page2) == 2

    def test_list_by_user_sorted_by_created_at_desc(self, repo):
        t1 = Tag.create(user_id="u1", name="old")
        t2 = Tag.create(user_id="u1", name="new")
        repo.create(t1)
        repo.create(t2)

        tags = repo.list_by_user("u1")
        # 新创建的排前面
        assert tags[0].id == t2.id
        assert tags[1].id == t1.id

    def test_count_by_user(self, repo, tag1, tag2, tag_other_user):
        repo.create(tag1)
        repo.create(tag2)
        repo.create(tag_other_user)

        assert repo.count_by_user("u1") == 2
        assert repo.count_by_user("u2") == 1
        assert repo.count_by_user("u3") == 0

    def test_delete(self, repo, tag1):
        repo.create(tag1)
        assert repo.delete(tag1.id) is True
        assert repo.get(tag1.id) is None

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


# ==================== Project ====================


class TestInMemoryProjectRepository:
    @pytest.fixture
    def repo(self):
        return InMemoryProjectRepository()

    @pytest.fixture
    def project1(self):
        return Project(id="proj-1", owner_user_id="u1", name="项目一", shared_users=[])

    @pytest.fixture
    def project2(self):
        return Project(id="proj-2", owner_user_id="u1", name="项目二", shared_users=["u2"])

    @pytest.fixture
    def project_other(self):
        return Project(id="proj-3", owner_user_id="u3", name="他人项目", shared_users=["u2"])

    def test_save_and_find_by_id(self, repo, project1):
        result = repo.save(project1)
        assert result.id == "proj-1"

        found = repo.find_by_id("proj-1")
        assert found is not None
        assert found.name == "项目一"

    def test_find_by_id_not_found(self, repo):
        assert repo.find_by_id("nonexistent") is None

    def test_find_by_owner_user_id(self, repo, project1, project2, project_other):
        repo.save(project1)
        repo.save(project2)
        repo.save(project_other)

        u1_projects = repo.find_by_owner_user_id("u1")
        assert len(u1_projects) == 2

        u3_projects = repo.find_by_owner_user_id("u3")
        assert len(u3_projects) == 1

    def test_find_accessible_projects_owner(self, repo, project1, project_other):
        repo.save(project1)
        repo.save(project_other)

        # u1 可以访问自己的项目
        accessible = repo.find_accessible_projects("u1")
        assert len(accessible) == 1
        assert accessible[0].id == "proj-1"

    def test_find_accessible_projects_shared(self, repo, project2, project_other):
        repo.save(project2)
        repo.save(project_other)

        # u2 被两个项目共享
        accessible = repo.find_accessible_projects("u2")
        assert len(accessible) == 2
        ids = {p.id for p in accessible}
        assert ids == {"proj-2", "proj-3"}

    def test_find_accessible_projects_none(self, repo, project1):
        repo.save(project1)
        assert repo.find_accessible_projects("nobody") == []

    def test_count_by_owner(self, repo, project1, project2, project_other):
        repo.save(project1)
        repo.save(project2)
        repo.save(project_other)

        assert repo.count_by_owner("u1") == 2
        assert repo.count_by_owner("u3") == 1
        assert repo.count_by_owner("nobody") == 0

    def test_delete(self, repo, project1):
        repo.save(project1)
        assert repo.delete("proj-1") is True
        assert repo.find_by_id("proj-1") is None

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


# ==================== IngestJob ====================


class TestInMemoryIngestJobRepository:
    @pytest.fixture
    def repo(self):
        return InMemoryIngestJobRepository()

    @pytest.fixture
    def job(self):
        return IngestJob.create(project_id="p1", library_id="l1", storage_key="key1", file_hash="hash1")

    def test_create_and_get(self, repo, job):
        result = repo.create(job)
        assert result.id == job.id
        assert result.status == IngestJobStatus.PENDING

        fetched = repo.get(job.id)
        assert fetched is not None
        assert fetched.storage_key == "key1"

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

    def test_update(self, repo, job):
        repo.create(job)
        job.status = IngestJobStatus.PROCESSING
        job.error_message = ""
        result = repo.update(job)
        assert result.status == IngestJobStatus.PROCESSING

        fetched = repo.get(job.id)
        assert fetched.status == IngestJobStatus.PROCESSING

    def test_update_with_result(self, repo, job):
        repo.create(job)
        job.status = IngestJobStatus.COMPLETED
        job.result_asset_id = "asset-123"
        repo.update(job)

        fetched = repo.get(job.id)
        assert fetched.status == IngestJobStatus.COMPLETED
        assert fetched.result_asset_id == "asset-123"


# ==================== ClassificationJob ====================


class TestInMemoryClassificationJobRepository:
    @pytest.fixture
    def repo(self):
        return InMemoryClassificationJobRepository()

    @pytest.fixture
    def job(self):
        return ClassificationJob.create(project_id="p1", asset_id="a1")

    def test_create_and_get(self, repo, job):
        result = repo.create(job)
        assert result.id == job.id
        assert result.status == ClassificationJobStatus.PENDING
        assert result.confidence == 0.0

        fetched = repo.get(job.id)
        assert fetched is not None
        assert fetched.asset_id == "a1"

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

    def test_update_status_and_result(self, repo, job):
        repo.create(job)
        job.status = ClassificationJobStatus.COMPLETED
        job.classification = "video"
        job.confidence = 0.95
        result = repo.update(job)

        assert result.status == ClassificationJobStatus.COMPLETED
        assert result.classification == "video"
        assert result.confidence == 0.95

    def test_update_failed(self, repo, job):
        repo.create(job)
        job.status = ClassificationJobStatus.FAILED
        job.error_message = "something went wrong"
        repo.update(job)

        fetched = repo.get(job.id)
        assert fetched.status == ClassificationJobStatus.FAILED
        assert fetched.error_message == "something went wrong"


class TestUserRepositoryUniqueness:
    """唯一性约束测试 - 模拟数据库唯一索引冲突."""

    @pytest.fixture
    def repo(self):
        from packages.adapters.in_memory.user_repository import InMemoryUserRepository

        return InMemoryUserRepository()

    @pytest.fixture
    def user1(self):
        return User(
            id="user-1",
            email="user1@example.com",
            display_name="User One",
            username="user1",
            phone="13800000001",
            wechat_openid="wx-openid-1",
            wechat_unionid="wx-unionid-1",
            created_at=datetime.now(timezone.utc),
        )

    @pytest.fixture
    def user2(self):
        return User(
            id="user-2",
            email="user2@example.com",
            display_name="User Two",
            username="user2",
            phone="13800000002",
            wechat_openid="wx-openid-2",
            wechat_unionid="wx-unionid-2",
            created_at=datetime.now(timezone.utc),
        )

    def test_duplicate_email_raises(self, repo, user1, user2):
        repo.save(user1)
        user2.email = "User1@example.com"  # 大小写不同，应视为冲突
        with pytest.raises(ValueError, match="Email already in use"):
            repo.save(user2)

    def test_duplicate_username_raises(self, repo, user1, user2):
        repo.save(user1)
        user2.username = "USER1"  # 大小写不同，应视为冲突
        with pytest.raises(ValueError, match="Username already in use"):
            repo.save(user2)

    def test_duplicate_phone_raises(self, repo, user1, user2):
        repo.save(user1)
        user2.phone = "13800000001"
        with pytest.raises(ValueError, match="Phone already in use"):
            repo.save(user2)

    def test_duplicate_wechat_openid_raises(self, repo, user1, user2):
        repo.save(user1)
        user2.wechat_openid = "wx-openid-1"
        with pytest.raises(ValueError, match="openid already in use"):
            repo.save(user2)

    def test_duplicate_wechat_unionid_raises(self, repo, user1, user2):
        repo.save(user1)
        user2.wechat_unionid = "wx-unionid-1"
        with pytest.raises(ValueError, match="unionid already in use"):
            repo.save(user2)

    def test_same_user_update_email_ok(self, repo, user1):
        """同一用户更新自己的邮箱不视为冲突."""
        repo.save(user1)
        user1.email = "newemail@example.com"
        repo.save(user1)  # 不应抛异常

        found = repo.find_by_email("newemail@example.com")
        assert found is not None
        assert found.id == "user-1"
        assert repo.find_by_email("user1@example.com") is None

    def test_duplicate_email_fails_cleanly(self, repo, user1, user2):
        """唯一性冲突时，用户数据不应被部分写入."""
        repo.save(user1)
        user2.email = "user1@example.com"

        with pytest.raises(ValueError):
            repo.save(user2)

        # user2 不应该被保存
        assert repo.find_by_id("user-2") is None
        # user1 仍然完好
        assert repo.find_by_id("user-1") is not None
        assert repo.find_by_email("user1@example.com").id == "user-1"