From 540df908f5425ea5636eeb5934b3b269fc34ed22 Mon Sep 17 00:00:00 2001
From: xiaoxia <xiaoxia@xiaoxiajianji.com>
Date: Thu, 30 Jul 2026 00:08:36 +0800
Subject: [PATCH] =?UTF-8?q?test(wave207):=20InMemory=E5=B0=8F=E5=9E=8B?=
 =?UTF-8?q?=E4=BB=93=E5=82=A8=E5=8D=95=E6=B5=8B=E8=A1=A5=E5=85=A8=20+41?=
 =?UTF-8?q?=E6=B5=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- asset_library_repository: 13测（CRUD/按项目查询/kind过滤/计数增减）
- tag_repository: 8测（CRUD/按名称查找/分页/排序/计数）
- project_repository: 8测（保存/查找/共享访问/计数/删除）
- ingest_job_repository: 4测（创建/获取/更新状态/更新结果）
- classification_job_repository: 4测（创建/获取/更新结果/更新失败）

总计: 41个单测，pytest全绿
---
 tests/unit/test_inmemory_small_repos.py | 408 ++++++++++++++++++++++++
 1 file changed, 408 insertions(+)
 create mode 100755 tests/unit/test_inmemory_small_repos.py

diff --git a/tests/unit/test_inmemory_small_repos.py b/tests/unit/test_inmemory_small_repos.py
new file mode 100755
index 000000000..146fa4b4d
--- /dev/null
+++ b/tests/unit/test_inmemory_small_repos.py
@@ -0,0 +1,408 @@
+"""InMemory 小型仓储模块单元测试（asset_library/tag/project/ingest_job/classification_job）."""
+
+import pytest
+
+from packages.domain.entities import AssetLibrary, IngestJob, Project
+from packages.domain.tag import Tag
+from packages.domain.classification import (
+    AssetLibraryKind,
+    ClassificationJob,
+    ClassificationJobStatus,
+    IngestJobStatus,
+)
+from packages.adapters.in_memory.asset_library_repository import InMemoryAssetLibraryRepository
+from packages.adapters.in_memory.tag_repository import InMemoryTagRepository
+from packages.adapters.in_memory.project_repository import InMemoryProjectRepository
+from packages.adapters.in_memory.ingest_job_repository import InMemoryIngestJobRepository
+from packages.adapters.in_memory.classification_job_repository import InMemoryClassificationJobRepository
+
+# ==================== 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, 1024)
+
+        lib = repo.get(lib_video.id)
+        assert lib.asset_count == 1
+        assert lib.total_size == 1024
+
+        repo.increment_asset_count(lib_video.id, 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", 100)
+
+    def test_decrement_asset_count(self, repo, lib_video):
+        repo.create(lib_video)
+        repo.increment_asset_count(lib_video.id, 1024)
+        repo.increment_asset_count(lib_video.id, 512)
+
+        repo.decrement_asset_count(lib_video.id, 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, 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", 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"