xiaoxia-saas/tests/unit/test_in_memory_repositories.py

"""
In-Memory 项目仓储 + 小仓储测试.
"""

from __future__ import annotations

from datetime import datetime

import pytest

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.domain import ClassificationJob, IngestJob, Project

# ── Project Repository ──────────────────────────────────────────────────────


@pytest.fixture
def project_repo():
    return InMemoryProjectRepository()


@pytest.fixture
def make_project():
    counter = 0

    def _make(owner_id: str = "user_1", name: str = "测试项目"):
        nonlocal counter
        counter += 1
        return Project(
            id=f"proj_{counter}",
            owner_user_id=owner_id,
            name=name,
        )

    return _make


class TestInMemoryProjectRepository:
    """项目仓储."""

    def test_save_and_find_by_id(self, project_repo, make_project):
        proj = make_project()
        project_repo.save(proj)
        found = project_repo.find_by_id(proj.id)
        assert found is not None
        assert found.id == proj.id
        assert found.name == "测试项目"

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

    def test_find_by_owner(self, project_repo, make_project):
        for i in range(3):
            project_repo.save(make_project(owner_id="user_1", name=f"p{i}"))
        project_repo.save(make_project(owner_id="user_2", name="other"))

        result = project_repo.find_by_owner_user_id("user_1")
        assert len(result) == 3
        assert all(p.owner_user_id == "user_1" for p in result)

    def test_count_by_owner(self, project_repo, make_project):
        for i in range(5):
            project_repo.save(make_project(owner_id="user_1", name=f"p{i}"))
        project_repo.save(make_project(owner_id="user_2", name="other"))
        assert project_repo.count_by_owner("user_1") == 5
        assert project_repo.count_by_owner("user_2") == 1
        assert project_repo.count_by_owner("user_3") == 0

    def test_delete_existing(self, project_repo, make_project):
        proj = make_project()
        project_repo.save(proj)
        result = project_repo.delete(proj.id)
        assert result is True
        assert project_repo.find_by_id(proj.id) is None

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

    def test_save_updates(self, project_repo, make_project):
        proj = make_project(name="v1")
        project_repo.save(proj)
        proj.name = "v2"
        project_repo.save(proj)
        found = project_repo.find_by_id(proj.id)
        assert found.name == "v2"

    def test_find_accessible_projects(self, project_repo, make_project):
        p1 = make_project(owner_id="user_1", name="owned")
        project_repo.save(p1)
        # 自己拥有的项目可访问
        result = project_repo.find_accessible_projects("user_1")
        assert len(result) >= 1
        assert any(p.id == p1.id for p in result)


# ── Classification Job Repository ──────────────────────────────────────────


@pytest.fixture
def classif_repo():
    return InMemoryClassificationJobRepository()


class TestInMemoryClassificationJobRepository:
    """分类任务仓储."""

    def test_create_and_get(self, classif_repo):
        job = ClassificationJob(
            id="job_1",
            project_id="p1",
            asset_id="a1",
            status="pending",
        )
        classif_repo.create(job)
        found = classif_repo.get("job_1")
        assert found is not None
        assert found.id == "job_1"
        assert found.status == "pending"

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

    def test_update(self, classif_repo):
        job = ClassificationJob(
            id="job_1",
            project_id="p1",
            asset_id="a1",
            status="pending",
        )
        classif_repo.create(job)
        job.status = "completed"
        classif_repo.update(job)
        found = classif_repo.get("job_1")
        assert found.status == "completed"

    def test_update_nonexistent_creates(self, classif_repo):
        """update 对不存在的也会写入（dict 赋值）"""
        job = ClassificationJob(
            id="new_job",
            project_id="p1",
            asset_id="a1",
            status="running",
        )
        classif_repo.update(job)
        found = classif_repo.get("new_job")
        assert found is not None


# ── Ingest Job Repository ──────────────────────────────────────────────────


@pytest.fixture
def ingest_repo():
    return InMemoryIngestJobRepository()


class TestInMemoryIngestJobRepository:
    """导入任务仓储."""

    def test_create_and_get(self, ingest_repo):
        job = IngestJob(
            id="ingest_1",
            project_id="p1",
            library_id="lib1",
            storage_key="key1",
            status="pending",
        )
        ingest_repo.create(job)
        found = ingest_repo.get("ingest_1")
        assert found is not None
        assert found.id == "ingest_1"
        assert found.status == "pending"

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

    def test_update(self, ingest_repo):
        job = IngestJob(
            id="ingest_1",
            project_id="p1",
            library_id="lib1",
            storage_key="key1",
            status="pending",
        )
        ingest_repo.create(job)
        job.status = "completed"
        ingest_repo.update(job)
        found = ingest_repo.get("ingest_1")
        assert found.status == "completed"

    def test_update_nonexistent_creates(self, ingest_repo):
        job = IngestJob(
            id="new_ingest",
            project_id="p1",
            library_id="lib1",
            storage_key="key1",
            status="running",
        )
        ingest_repo.update(job)
        found = ingest_repo.get("new_ingest")
        assert found is not None