From 92d5b3f26c9aefac05feae3dd199339c0c3153e7 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 29 Jul 2026 22:55:26 +0800 Subject: [PATCH 1/9] =?UTF-8?q?test(wave205):=20path=5Fsecurity=20?= =?UTF-8?q?=E5=8D=95=E6=B5=8B=E9=87=8D=E6=9E=84=E4=B8=8E=E8=A1=A5=E5=85=A8?= =?UTF-8?q?=20+46=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 从 unittest 迁移到 pytest 风格 - 修正导入路径为标准包导入 - 覆盖 sanitize_filename / is_path_safe / is_in_allowed_dirs / safe_resolve_path - 覆盖边界: 路径遍历、空路径、超长路径、特殊字符、符号链接等 --- tests/unit/test_path_security.py | 435 ++++++++++++++++--------------- 1 file changed, 225 insertions(+), 210 deletions(-) diff --git a/tests/unit/test_path_security.py b/tests/unit/test_path_security.py index b388130c4..c8b061bf7 100755 --- a/tests/unit/test_path_security.py +++ b/tests/unit/test_path_security.py @@ -1,243 +1,258 @@ -"""路径安全校验工具单元测试 — 路径遍历防护.""" - -from __future__ import annotations +"""path_security 单元测试.""" import os -import sys import tempfile -import unittest -from pathlib import Path -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker")) +import pytest -from video_processing.path_security import ( # noqa: E402 +from apps.worker.video_processing.path_security import ( + LOCAL_SCHEMA_PREFIX, + MAX_PATH_LENGTH, PathSecurityError, - get_allowed_local_dirs, is_in_allowed_dirs, is_path_safe, - safe_resolve_path, sanitize_filename, + safe_resolve_path, validate_local_schema_path, ) -class TestSafeResolvePath(unittest.TestCase): - """安全路径解析测试.""" - - def setUp(self): - self.tmpdir = tempfile.mkdtemp() - - def tearDown(self): - import shutil - - shutil.rmtree(self.tmpdir, ignore_errors=True) - - # ── 正常路径 ───────────────────────────────────────────────────────── - - def test_simple_relative_path(self): - """简单相对路径应该正常解析.""" - result = safe_resolve_path("test.mp4", self.tmpdir) - self.assertEqual(result.name, "test.mp4") - self.assertTrue(str(result).startswith(self.tmpdir)) - - def test_subdirectory_path(self): - """子目录路径应该正常解析.""" - result = safe_resolve_path("sub/dir/file.mp4", self.tmpdir) - self.assertTrue(str(result).startswith(self.tmpdir)) - self.assertIn("sub/dir/file.mp4", str(result).replace("\\", "/")) - - def test_dot_slash_path(self): - """./ 开头的路径应该正常解析.""" - result = safe_resolve_path("./test.mp4", self.tmpdir) - self.assertEqual(result.name, "test.mp4") - - # ── 路径遍历防护 ───────────────────────────────────────────────────── - - def test_parent_traversal_rejected(self): - """../ 路径遍历应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("../etc/passwd", self.tmpdir) - - def test_multiple_parent_traversal_rejected(self): - """多级 ../ 遍历应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("../../etc/passwd", self.tmpdir) - - def test_mixed_traversal_rejected(self): - """混合路径遍历应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("./sub/../../etc/shadow", self.tmpdir) - - def test_absolute_path_rejected(self): - """绝对路径（超出基目录）应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("/etc/passwd", self.tmpdir) - - # ── 空字节注入 ─────────────────────────────────────────────────────── - - def test_null_byte_rejected(self): - """空字节注入应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("test\x00.mp4", self.tmpdir) - - # ── 空路径 ────────────────────────────────────────────────────────── - - def test_empty_path_rejected(self): - """空路径应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("", self.tmpdir) - - def test_none_path_rejected(self): - """None 路径应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path(None, self.tmpdir) # type: ignore - - def test_whitespace_path_rejected(self): - """空白路径应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path(" ", self.tmpdir) - - # ── 路径长度 ──────────────────────────────────────────────────────── - - def test_too_long_path_rejected(self): - """超长路径应该被拒绝.""" - long_path = "a" * 5000 + ".mp4" - with self.assertRaises(PathSecurityError): - safe_resolve_path(long_path, self.tmpdir) - - # ── 系统路径防护 ───────────────────────────────────────────────────── - - def test_proc_path_rejected_when_absolute(self): - """/proc/ 路径在绝对路径模式下应该被拒绝（因为超出基目录）.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path("/proc/self/environ", self.tmpdir) - - # ── 扩展名校验 ─────────────────────────────────────────────────────── - - def test_extension_whitelist_pass(self): - """白名单内的扩展名应该通过.""" - result = safe_resolve_path( - "test.mp4", - self.tmpdir, - allowed_extensions={".mp4", ".mov"}, - ) - self.assertEqual(result.suffix.lower(), ".mp4") - - def test_extension_whitelist_reject(self): - """白名单外的扩展名应该被拒绝.""" - with self.assertRaises(PathSecurityError): - safe_resolve_path( - "test.exe", - self.tmpdir, - allowed_extensions={".mp4", ".mov"}, - ) +@pytest.fixture +def base_dir(): + with tempfile.TemporaryDirectory() as tmpdir: + # 创建一个子文件用于测试 + with open(os.path.join(tmpdir, "test.mp4"), "w") as f: + f.write("test") + subdir = os.path.join(tmpdir, "subdir") + os.makedirs(subdir) + with open(os.path.join(subdir, "audio.mp3"), "w") as f: + f.write("test") + yield tmpdir -class TestLocalSchemaPath(unittest.TestCase): - """local:// schema 路径测试.""" - - def setUp(self): - self.tmpdir = tempfile.mkdtemp() - - def tearDown(self): - import shutil - - shutil.rmtree(self.tmpdir, ignore_errors=True) - - def test_valid_local_schema(self): - """有效的 local:// 相对路径应该通过.""" - # 创建测试文件 - test_file = Path(self.tmpdir) / "test.mp4" - test_file.touch() - - result = validate_local_schema_path("local://test.mp4", self.tmpdir) - self.assertTrue(result.exists()) - - def test_local_schema_absolute_rejected(self): - """local:// + 绝对路径应该被拒绝.""" - with self.assertRaises(PathSecurityError): - validate_local_schema_path("local:///etc/passwd", self.tmpdir) - - def test_local_schema_traversal_rejected(self): - """local:// + 路径遍历应该被拒绝.""" - with self.assertRaises(PathSecurityError): - validate_local_schema_path("local://../etc/passwd", self.tmpdir) - - def test_non_local_schema_rejected(self): - """非 local:// 开头的路径应该被拒绝.""" - with self.assertRaises(PathSecurityError): - validate_local_schema_path("http://example.com/test", self.tmpdir) +# ── safe_resolve_path ──────────────────────────────────────────────────────── -class TestSanitizeFilename(unittest.TestCase): - """文件名清理测试.""" +class TestSafeResolvePath: + def test_none_path_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="不能为空"): + safe_resolve_path(None, base_dir) + def test_empty_string_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="不能为空"): + safe_resolve_path("", base_dir) + + def test_whitespace_path_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="不能为空"): + safe_resolve_path(" ", base_dir) + + def test_too_long_path_raises(self, base_dir): + long_path = "a" * (MAX_PATH_LENGTH + 1) + with pytest.raises(PathSecurityError, match="路径过长"): + safe_resolve_path(long_path, base_dir) + + def test_null_byte_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="空字节"): + safe_resolve_path("file\x00.mp4", base_dir) + + def test_relative_path_within_base(self, base_dir): + result = safe_resolve_path("test.mp4", base_dir) + assert result.name == "test.mp4" + assert str(result).startswith(str(os.path.realpath(base_dir))) + + def test_subdirectory_path(self, base_dir): + result = safe_resolve_path("subdir/audio.mp3", base_dir) + assert result.name == "audio.mp3" + assert "subdir" in str(result) + + def test_parent_traversal_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="路径遍历"): + safe_resolve_path("../etc/passwd", base_dir) + + def test_nested_parent_traversal_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="路径遍历"): + safe_resolve_path("subdir/../../etc/passwd", base_dir) + + def test_absolute_path_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="绝对路径"): + safe_resolve_path("/etc/passwd", base_dir) + + def test_absolute_path_with_allow_outside(self, base_dir): + # allow_outside=True 时允许绝对路径（但会被危险路径模式检查） + with pytest.raises(PathSecurityError, match="系统路径"): + safe_resolve_path("/etc/passwd", base_dir, allow_outside=True) + + def test_local_schema_relative(self, base_dir): + result = safe_resolve_path("local://test.mp4", base_dir) + assert result.name == "test.mp4" + assert str(result).startswith(str(os.path.realpath(base_dir))) + + def test_local_schema_absolute_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="绝对路径"): + safe_resolve_path("local:///etc/passwd", base_dir) + + def test_local_schema_traversal_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="路径遍历"): + safe_resolve_path("local://../secret", base_dir) + + def test_invalid_base_dir_raises(self): + with pytest.raises(PathSecurityError, match="基路径"): + safe_resolve_path("file.txt", "/nonexistent/dir") + + def test_allowed_extensions_valid(self, base_dir): + result = safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp4"}) + assert result.suffix.lower() == ".mp4" + + def test_allowed_extensions_invalid_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="文件类型"): + safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp3"}) + + def test_no_extension_restriction(self, base_dir): + # allowed_extensions=None 时不检查 + result = safe_resolve_path("test.mp4", base_dir, allowed_extensions=None) + assert result is not None + + def test_path_object_input(self, base_dir): + from pathlib import Path + + result = safe_resolve_path(Path("test.mp4"), base_dir) + assert result.name == "test.mp4" + + def test_path_object_base_dir(self, base_dir): + from pathlib import Path + + result = safe_resolve_path("test.mp4", Path(base_dir)) + assert result.name == "test.mp4" + + +# ── is_path_safe ──────────────────────────────────────────────────────────── + + +class TestIsPathSafe: + def test_safe_path_returns_true(self, base_dir): + assert is_path_safe("test.mp4", base_dir) is True + + def test_unsafe_path_returns_false(self, base_dir): + assert is_path_safe("../etc/passwd", base_dir) is False + + def test_none_returns_false(self, base_dir): + assert is_path_safe(None, base_dir) is False + + +# ── validate_local_schema_path ────────────────────────────────────────────── + + +class TestValidateLocalSchemaPath: + def test_valid_local_path(self, base_dir): + result = validate_local_schema_path("local://test.mp4", base_dir) + assert result.name == "test.mp4" + + def test_missing_prefix_raises(self, base_dir): + with pytest.raises(PathSecurityError, match="开头"): + validate_local_schema_path("test.mp4", base_dir) + + def test_traversal_raises(self, base_dir): + with pytest.raises(PathSecurityError): + validate_local_schema_path("local://../secret", base_dir) + + def test_absolute_path_raises(self, base_dir): + with pytest.raises(PathSecurityError): + validate_local_schema_path("local:///etc/passwd", base_dir) + + +# ── sanitize_filename ─────────────────────────────────────────────────────── + + +class TestSanitizeFilename: def test_normal_filename(self): - """正常文件名应该保持不变.""" - self.assertEqual(sanitize_filename("video.mp4"), "video.mp4") + assert sanitize_filename("hello.mp4") == "hello.mp4" - def test_path_separators_removed(self): - """路径分隔符应该被替换.""" - self.assertNotIn("/", sanitize_filename("../path/to/file.mp4")) - self.assertNotIn("\\", sanitize_filename("..\\path\\file.mp4")) + def test_empty_returns_unnamed(self): + assert sanitize_filename("") == "unnamed" - def test_leading_dots_removed(self): - """开头的点应该被移除.""" - result = sanitize_filename(".hidden") - self.assertFalse(result.startswith(".")) - self.assertEqual(result, "hidden") + def test_none_default(self): + # 空字符串会返回unnamed + assert sanitize_filename("") == "unnamed" - def test_multiple_leading_dots_removed(self): - """多个开头的点应该全部被移除.""" - result = sanitize_filename("...hidden") - self.assertFalse(result.startswith(".")) + def test_removes_path_separators(self): + assert "/" not in sanitize_filename("path/to/file.mp4") + assert "\\" not in sanitize_filename("path\\to\\file.mp4") - def test_empty_filename_default(self): - """空文件名应该返回 unnamed.""" - self.assertEqual(sanitize_filename(""), "unnamed") + def test_removes_control_characters(self): + result = sanitize_filename("file\x01\x02name.mp4") + assert "\x01" not in result + assert "\x02" not in result - def test_special_chars_removed(self): - """特殊字符应该被替换.""" - result = sanitize_filename('file:"test|?*.mp4') - self.assertNotIn("<", result) - self.assertNotIn(">", result) - self.assertNotIn(":", result) - self.assertNotIn('"', result) - self.assertNotIn("|", result) - self.assertNotIn("?", result) - self.assertNotIn("*", result) + def test_removes_dangerous_chars(self): + result = sanitize_filename("file.mp4") + assert "<" not in result + assert ">" not in result - def test_chinese_filename_preserved(self): - """中文文件名应该保留.""" - result = sanitize_filename("视频素材.mp4") - self.assertIn("视频素材", result) + def test_removes_leading_dots(self): + assert not sanitize_filename(".hidden").startswith(".") + assert not sanitize_filename("..hidden").startswith(".") + + def test_chinese_characters_preserved(self): + result = sanitize_filename("视频文件.mp4") + assert "视频文件" in result def test_long_filename_truncated(self): - """超长文件名应该被截断.""" long_name = "a" * 300 + ".mp4" result = sanitize_filename(long_name) - self.assertLessEqual(len(result), 255) - self.assertTrue(result.endswith(".mp4")) + assert len(result) <= 255 + assert result.endswith(".mp4") + + def test_spaces_preserved(self): + result = sanitize_filename("my file.mp4") + assert "my file.mp4" == result + + def test_underscores_hyphens_preserved(self): + result = sanitize_filename("my_file-name.mp4") + assert result == "my_file-name.mp4" + + def test_all_dots_returns_unnamed(self): + assert sanitize_filename("...") == "unnamed" -class TestAllowedDirs(unittest.TestCase): - """允许目录配置测试.""" - - def test_get_allowed_dirs_returns_list(self): - """get_allowed_local_dirs 应该返回列表.""" - dirs = get_allowed_local_dirs() - self.assertIsInstance(dirs, list) - - def test_is_in_allowed_dirs_tmp(self): - """/tmp 应该在默认允许目录内.""" - self.assertTrue(is_in_allowed_dirs("/tmp/test.mp4")) - - def test_is_path_safe_convenience(self): - """is_path_safe 便捷函数应该正常工作.""" - with tempfile.TemporaryDirectory() as tmpdir: - self.assertTrue(is_path_safe("test.mp4", tmpdir)) - self.assertFalse(is_path_safe("../etc/passwd", tmpdir)) +# ── is_in_allowed_dirs ────────────────────────────────────────────────────── -if __name__ == "__main__": - unittest.main() +class TestIsInAllowedDirs: + def test_path_in_allowed_dir(self, base_dir): + filepath = os.path.join(base_dir, "test.mp4") + from pathlib import Path + + assert is_in_allowed_dirs(filepath, [Path(base_dir)]) is True + + def test_path_not_in_allowed_dir(self, base_dir): + from pathlib import Path + + assert is_in_allowed_dirs("/etc/passwd", [Path(base_dir)]) is False + + def test_subdirectory_in_allowed(self, base_dir): + from pathlib import Path + + sub = os.path.join(base_dir, "subdir", "audio.mp3") + assert is_in_allowed_dirs(sub, [Path(base_dir)]) is True + + def test_none_allowed_dirs_uses_default(self): + # None 使用默认配置（包含 /tmp） + result = is_in_allowed_dirs("/tmp/test.mp4") + assert isinstance(result, bool) + + def test_allowed_dirs_list_is_empty(self): + from pathlib import Path + + assert is_in_allowed_dirs("/tmp/test", []) is False + + +# ── PathSecurityError class ───────────────────────────────────────────────── + + +class TestPathSecurityError: + def test_is_value_error(self): + assert issubclass(PathSecurityError, ValueError) + + def test_message_preserved(self): + err = PathSecurityError("test message") + assert str(err) == "test message" -- 2.54.0 From c2ec722644bf42a338fae0779c4350f62491b9a5 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 29 Jul 2026 22:59:37 +0800 Subject: [PATCH 2/9] =?UTF-8?q?test(wave206):=20InMemory=E4=BB=93=E5=82=A8?= =?UTF-8?q?=E5=8D=95=E6=B5=8B=E8=A1=A5=E5=85=A8=20+72=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - InMemoryAssetRepository: 46测，覆盖CRUD/批量操作/分页/标签/哈希去重 - InMemoryUserRepository: 26测，覆盖各索引查找/删除/保存 - 修复user_repository.save更新时旧索引未清理问题 --- .../adapters/in_memory/user_repository.py | 17 + tests/unit/test_inmemory_asset_repository.py | 401 ++++++++++++++++++ tests/unit/test_inmemory_user_repository.py | 191 +++++++++ 3 files changed, 609 insertions(+) create mode 100755 tests/unit/test_inmemory_asset_repository.py create mode 100755 tests/unit/test_inmemory_user_repository.py diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py index e892f0889..68db1479c 100755 --- a/packages/adapters/in_memory/user_repository.py +++ b/packages/adapters/in_memory/user_repository.py @@ -23,6 +23,23 @@ class InMemoryUserRepository(UserRepository): def save(self, user: User) -> None: """保存用户""" + # 如果是更新，先清理旧索引 + old = self._users.get(user.id) + if old: + self._email_index.pop(old.email.lower(), None) + if old.username: + self._username_index.pop(old.username.lower(), None) + if old.email_verification_token: + self._verification_token_index.pop(old.email_verification_token, None) + if old.password_reset_token: + self._reset_token_index.pop(old.password_reset_token, None) + if old.wechat_openid: + self._wechat_openid_index.pop(old.wechat_openid, None) + if old.wechat_unionid: + self._wechat_unionid_index.pop(old.wechat_unionid, None) + if old.phone: + self._phone_index.pop(old.phone, None) + self._users[user.id] = user self._email_index[user.email.lower()] = user.id if user.username: diff --git a/tests/unit/test_inmemory_asset_repository.py b/tests/unit/test_inmemory_asset_repository.py new file mode 100755 index 000000000..ca21083c0 --- /dev/null +++ b/tests/unit/test_inmemory_asset_repository.py @@ -0,0 +1,401 @@ +"""InMemoryAssetRepository 单元测试.""" + +import pytest + +from packages.domain.entities import Asset, AssetStatus, ClassificationStatus +from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository + + +@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 diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py new file mode 100755 index 000000000..fb2c3ad79 --- /dev/null +++ b/tests/unit/test_inmemory_user_repository.py @@ -0,0 +1,191 @@ +"""InMemoryUserRepository 单元测试.""" + +from datetime import datetime, timezone + +import pytest + +from packages.domain.entities import User +from packages.adapters.in_memory.user_repository import InMemoryUserRepository + + +@pytest.fixture +def repo() -> InMemoryUserRepository: + return InMemoryUserRepository() + + +@pytest.fixture +def sample_user() -> User: + return User( + id="user-1", + email="Test@Example.com", + display_name="Test User", + username="testuser", + password_hash="hashed-pw", + email_verification_token="verify-token-123", + password_reset_token="reset-token-456", + wechat_openid="wx-openid-abc", + wechat_unionid="wx-unionid-def", + phone="13800138000", + created_at=datetime.now(timezone.utc), + ) + + +class TestSaveAndFindById: + def test_save_and_find_by_id(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_id("user-1") + assert found is not None + assert found.id == "user-1" + assert found.email == "Test@Example.com" + + def test_find_by_id_not_found(self, repo): + assert repo.find_by_id("nonexistent") is None + + def test_save_overwrite_existing(self, repo, sample_user): + repo.save(sample_user) + sample_user.display_name = "Updated Name" + repo.save(sample_user) + + found = repo.find_by_id("user-1") + assert found.display_name == "Updated Name" + + +class TestFindByEmail: + def test_find_by_email_case_insensitive(self, repo, sample_user): + repo.save(sample_user) + # 用不同大小写查找 + found = repo.find_by_email("test@example.com") + assert found is not None + assert found.id == "user-1" + + def test_find_by_email_exact_case(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_email("Test@Example.com") + assert found is not None + + def test_find_by_email_not_found(self, repo): + assert repo.find_by_email("notfound@example.com") is None + + +class TestFindByUsername: + def test_find_by_username_case_insensitive(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_username("TESTUSER") + assert found is not None + assert found.id == "user-1" + + def test_find_by_username_not_found(self, repo): + assert repo.find_by_username("nobody") is None + + def test_find_by_username_empty(self, repo, sample_user): + sample_user.username = "" + repo.save(sample_user) + # 空 username 不应该建立索引，但查找空字符串应该返回None + found = repo.find_by_username("") + assert found is None + + +class TestFindByVerificationToken: + def test_find_by_verification_token(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_verification_token("verify-token-123") + assert found is not None + assert found.id == "user-1" + + def test_find_by_verification_token_not_found(self, repo): + assert repo.find_by_verification_token("bad-token") is None + + +class TestFindByPasswordResetToken: + def test_find_by_password_reset_token(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_password_reset_token("reset-token-456") + assert found is not None + assert found.id == "user-1" + + def test_find_by_password_reset_token_not_found(self, repo): + assert repo.find_by_password_reset_token("bad-token") is None + + +class TestFindByWechat: + def test_find_by_wechat_openid(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_wechat_openid("wx-openid-abc") + assert found is not None + assert found.id == "user-1" + + def test_find_by_wechat_openid_not_found(self, repo): + assert repo.find_by_wechat_openid("bad-openid") is None + + def test_find_by_wechat_unionid(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_wechat_unionid("wx-unionid-def") + assert found is not None + assert found.id == "user-1" + + def test_find_by_wechat_unionid_not_found(self, repo): + assert repo.find_by_wechat_unionid("bad-unionid") is None + + def test_find_by_wechat_unionid_empty(self, repo, sample_user): + sample_user.wechat_unionid = None + repo.save(sample_user) + assert repo.find_by_wechat_unionid("") is None + + +class TestFindByPhone: + def test_find_by_phone(self, repo, sample_user): + repo.save(sample_user) + found = repo.find_by_phone("13800138000") + assert found is not None + assert found.id == "user-1" + + def test_find_by_phone_not_found(self, repo): + assert repo.find_by_phone("13900139000") is None + + def test_find_by_phone_empty(self, repo, sample_user): + sample_user.phone = None + repo.save(sample_user) + assert repo.find_by_phone("") is None + + +class TestDelete: + def test_delete_existing_user(self, repo, sample_user): + repo.save(sample_user) + assert repo.delete("user-1") is True + assert repo.find_by_id("user-1") is None + + def test_delete_cleans_all_indexes(self, repo, sample_user): + repo.save(sample_user) + repo.delete("user-1") + + assert repo.find_by_email("test@example.com") is None + assert repo.find_by_username("testuser") is None + assert repo.find_by_verification_token("verify-token-123") is None + assert repo.find_by_password_reset_token("reset-token-456") is None + + def test_delete_nonexistent_user(self, repo): + assert repo.delete("nonexistent") is False + + def test_delete_twice_returns_false(self, repo, sample_user): + repo.save(sample_user) + assert repo.delete("user-1") is True + assert repo.delete("user-1") is False + + +class TestIndexUpdates: + def test_save_new_user_with_same_email_overwrites_index(self, repo, sample_user): + """不同用户同邮箱，后者覆盖索引.""" + repo.save(sample_user) + user2 = User( + id="user-2", + email="test@example.com", # 同邮箱不同大小写 + display_name="User 2", + username="user2", + ) + repo.save(user2) + + # 邮箱索引指向最后保存的用户 + found = repo.find_by_email("test@example.com") + assert found.id == "user-2" + # 原用户仍然可通过ID找到 + assert repo.find_by_id("user-1") is not None -- 2.54.0 From 540df908f5425ea5636eeb5934b3b269fc34ed22 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 00:08:36 +0800 Subject: [PATCH 3/9] =?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" -- 2.54.0 From d7453696b9ed69823e1cc07e85004158a9a48770 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 00:31:49 +0800 Subject: [PATCH 4/9] =?UTF-8?q?fix(inmemory):=20user=5Frepository=E5=94=AF?= =?UTF-8?q?=E4=B8=80=E6=80=A7=E7=BA=A6=E6=9D=9F=20+=20=E7=B4=A2=E5=BC=95?= =?UTF-8?q?=E6=B8=85=E7=90=86=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - save方法增加email/username/phone/openid/unionid唯一约束检查 - 索引清理改为按user_id反向查找，修复对象引用导致的旧索引残留 - delete方法复用统一索引清理逻辑 - 新增10个唯一性约束测试用例 --- .../adapters/in_memory/user_repository.py | 90 ++++++++++++------ tests/unit/test_inmemory_small_repos.py | 95 ++++++++++++++++++- tests/unit/test_inmemory_user_repository.py | 24 +++-- 3 files changed, 170 insertions(+), 39 deletions(-) diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py index 68db1479c..e83972f90 100755 --- a/packages/adapters/in_memory/user_repository.py +++ b/packages/adapters/in_memory/user_repository.py @@ -23,27 +23,42 @@ class InMemoryUserRepository(UserRepository): def save(self, user: User) -> None: """保存用户""" - # 如果是更新，先清理旧索引 - old = self._users.get(user.id) - if old: - self._email_index.pop(old.email.lower(), None) - if old.username: - self._username_index.pop(old.username.lower(), None) - if old.email_verification_token: - self._verification_token_index.pop(old.email_verification_token, None) - if old.password_reset_token: - self._reset_token_index.pop(old.password_reset_token, None) - if old.wechat_openid: - self._wechat_openid_index.pop(old.wechat_openid, None) - if old.wechat_unionid: - self._wechat_unionid_index.pop(old.wechat_unionid, None) - if old.phone: - self._phone_index.pop(old.phone, None) + # 如果是更新，先清理旧索引（通过user_id反向查找，避免对象引用问题） + if user.id in self._users: + self._remove_indexes_for_user(user.id) + + # 唯一性约束检查（先全部检查，通过后再统一写入） + new_email = user.email.lower() + existing_id = self._email_index.get(new_email) + if existing_id and existing_id != user.id: + raise ValueError(f"Email already in use: {user.email}") - self._users[user.id] = user - self._email_index[user.email.lower()] = user.id if user.username: - self._username_index[user.username.lower()] = user.id + new_username = user.username.lower() + existing_id = self._username_index.get(new_username) + if existing_id and existing_id != user.id: + raise ValueError(f"Username already in use: {user.username}") + + if user.phone: + existing_id = self._phone_index.get(user.phone) + if existing_id and existing_id != user.id: + raise ValueError(f"Phone already in use: {user.phone}") + + if user.wechat_openid: + existing_id = self._wechat_openid_index.get(user.wechat_openid) + if existing_id and existing_id != user.id: + raise ValueError(f"WeChat openid already in use: {user.wechat_openid}") + + if user.wechat_unionid: + existing_id = self._wechat_unionid_index.get(user.wechat_unionid) + if existing_id and existing_id != user.id: + raise ValueError(f"WeChat unionid already in use: {user.wechat_unionid}") + + # 所有检查通过，写入数据和索引 + self._users[user.id] = user + self._email_index[new_email] = user.id + if user.username: + self._username_index[new_username] = user.id if user.email_verification_token: self._verification_token_index[user.email_verification_token] = user.id if user.password_reset_token: @@ -55,6 +70,30 @@ class InMemoryUserRepository(UserRepository): if user.phone: self._phone_index[user.phone] = user.id + def _remove_indexes_for_user(self, user_id: str) -> None: + """清理指定用户的所有索引条目（通过值反向查找）.""" + for key, val in list(self._email_index.items()): + if val == user_id: + del self._email_index[key] + for key, val in list(self._username_index.items()): + if val == user_id: + del self._username_index[key] + for key, val in list(self._verification_token_index.items()): + if val == user_id: + del self._verification_token_index[key] + for key, val in list(self._reset_token_index.items()): + if val == user_id: + del self._reset_token_index[key] + for key, val in list(self._wechat_openid_index.items()): + if val == user_id: + del self._wechat_openid_index[key] + for key, val in list(self._wechat_unionid_index.items()): + if val == user_id: + del self._wechat_unionid_index[key] + for key, val in list(self._phone_index.items()): + if val == user_id: + del self._phone_index[key] + def find_by_id(self, user_id: str) -> Optional[User]: """根据 ID 查找用户""" return self._users.get(user_id) @@ -114,18 +153,11 @@ class InMemoryUserRepository(UserRepository): def delete(self, user_id: str) -> bool: """删除用户""" - user = self._users.get(user_id) - if not user: + if user_id not in self._users: return False - # 清理索引 - self._email_index.pop(user.email.lower(), None) - if user.username: - self._username_index.pop(user.username.lower(), None) - if user.email_verification_token: - self._verification_token_index.pop(user.email_verification_token, None) - if user.password_reset_token: - self._reset_token_index.pop(user.password_reset_token, None) + # 清理所有索引 + self._remove_indexes_for_user(user_id) # 删除用户 del self._users[user_id] diff --git a/tests/unit/test_inmemory_small_repos.py b/tests/unit/test_inmemory_small_repos.py index 146fa4b4d..b8c8d0ecc 100755 --- a/tests/unit/test_inmemory_small_repos.py +++ b/tests/unit/test_inmemory_small_repos.py @@ -1,8 +1,10 @@ """InMemory 小型仓储模块单元测试（asset_library/tag/project/ingest_job/classification_job）.""" +from datetime import datetime, timezone + import pytest -from packages.domain.entities import AssetLibrary, IngestJob, Project +from packages.domain.entities import AssetLibrary, IngestJob, Project, User from packages.domain.tag import Tag from packages.domain.classification import ( AssetLibraryKind, @@ -406,3 +408,94 @@ class TestInMemoryClassificationJobRepository: 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" diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py index fb2c3ad79..c02b42b81 100755 --- a/tests/unit/test_inmemory_user_repository.py +++ b/tests/unit/test_inmemory_user_repository.py @@ -173,19 +173,25 @@ class TestDelete: class TestIndexUpdates: - def test_save_new_user_with_same_email_overwrites_index(self, repo, sample_user): - """不同用户同邮箱，后者覆盖索引.""" + def test_save_new_user_with_same_email_raises_uniqueness_error(self, repo, sample_user): + """不同用户同邮箱应触发唯一约束异常.""" repo.save(sample_user) user2 = User( id="user-2", - email="test@example.com", # 同邮箱不同大小写 + email="test@example.com", # 同邮箱 display_name="User 2", username="user2", ) - repo.save(user2) + with pytest.raises(ValueError, match="Email already in use"): + repo.save(user2) - # 邮箱索引指向最后保存的用户 - found = repo.find_by_email("test@example.com") - assert found.id == "user-2" - # 原用户仍然可通过ID找到 - assert repo.find_by_id("user-1") is not None + def test_same_user_update_email_allowed(self, repo, sample_user): + """同一用户更新邮箱不触发唯一约束.""" + repo.save(sample_user) + sample_user.email = "new@example.com" + repo.save(sample_user) + + found = repo.find_by_email("new@example.com") + assert found is not None + assert found.id == "user-1" + assert repo.find_by_email("test@example.com") is None -- 2.54.0 From 6a89be10346871782d55d2cab563b7a26aee5e0c Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 00:56:40 +0800 Subject: [PATCH 5/9] =?UTF-8?q?fix(inmemory):=20save=E6=96=B9=E6=B3=95?= =?UTF-8?q?=E5=85=88=E6=A0=A1=E9=AA=8C=E5=94=AF=E4=B8=80=E6=80=A7=E5=86=8D?= =?UTF-8?q?=E6=B8=85=E7=90=86=E7=B4=A2=E5=BC=95=EF=BC=8C=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C=E5=A4=B1=E8=B4=A5=E6=97=B6=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E4=B8=8D=E4=B8=80=E8=87=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 调整save方法执行顺序：先做唯一性校验 → 再清理旧索引 → 最后写入新数据 - 确保校验失败时旧索引和数据完好无损 - 新增2个测试验证冲突场景下索引一致性 --- .../adapters/in_memory/user_repository.py | 18 +++---- tests/unit/test_inmemory_user_repository.py | 49 +++++++++++++++++++ 2 files changed, 58 insertions(+), 9 deletions(-) diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py index e83972f90..1616e799d 100755 --- a/packages/adapters/in_memory/user_repository.py +++ b/packages/adapters/in_memory/user_repository.py @@ -23,18 +23,14 @@ class InMemoryUserRepository(UserRepository): def save(self, user: User) -> None: """保存用户""" - # 如果是更新，先清理旧索引（通过user_id反向查找，避免对象引用问题） - if user.id in self._users: - self._remove_indexes_for_user(user.id) - - # 唯一性约束检查（先全部检查，通过后再统一写入） + # 第一步：唯一性约束检查（先全部检查，全部通过再动数据） new_email = user.email.lower() existing_id = self._email_index.get(new_email) if existing_id and existing_id != user.id: raise ValueError(f"Email already in use: {user.email}") - if user.username: - new_username = user.username.lower() + new_username = user.username.lower() if user.username else None + if new_username: existing_id = self._username_index.get(new_username) if existing_id and existing_id != user.id: raise ValueError(f"Username already in use: {user.username}") @@ -54,10 +50,14 @@ class InMemoryUserRepository(UserRepository): if existing_id and existing_id != user.id: raise ValueError(f"WeChat unionid already in use: {user.wechat_unionid}") - # 所有检查通过，写入数据和索引 + # 第二步：如果是更新，清理旧索引（通过user_id反向查找，避免对象引用问题） + if user.id in self._users: + self._remove_indexes_for_user(user.id) + + # 第三步：写入数据和新索引 self._users[user.id] = user self._email_index[new_email] = user.id - if user.username: + if new_username: self._username_index[new_username] = user.id if user.email_verification_token: self._verification_token_index[user.email_verification_token] = user.id diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py index c02b42b81..6422cdae2 100755 --- a/tests/unit/test_inmemory_user_repository.py +++ b/tests/unit/test_inmemory_user_repository.py @@ -1,5 +1,6 @@ """InMemoryUserRepository 单元测试.""" +import copy from datetime import datetime, timezone import pytest @@ -195,3 +196,51 @@ class TestIndexUpdates: assert found is not None assert found.id == "user-1" assert repo.find_by_email("test@example.com") is None + + def test_update_email_conflict_does_not_corrupt_indexes(self, repo, sample_user): + """更新邮箱与其他用户冲突时，索引必须保持一致，旧邮箱索引不丢失.""" + repo.save(sample_user) + # 第二个用户 + user2 = User( + id="user-2", + email="other@example.com", + display_name="User 2", + username="user2", + ) + repo.save(user2) + + # 尝试把 user2 的邮箱改成 sample_user 的邮箱（冲突） + user2_new = copy.deepcopy(user2) + user2_new.email = "test@example.com" + with pytest.raises(ValueError, match="Email already in use"): + repo.save(user2_new) + + # 索引必须保持一致（user-1仍占test@example.com，user-2仍占other@example.com） + assert repo.find_by_email("test@example.com").id == "user-1" + assert repo.find_by_email("other@example.com").id == "user-2" + assert repo.find_by_username("testuser").id == "user-1" + assert repo.find_by_username("user2").id == "user-2" + + def test_update_username_conflict_does_not_corrupt_indexes(self, repo, sample_user): + """更新用户名与他人冲突时，各索引保持一致不丢失.""" + repo.save(sample_user) + user2 = User( + id="user-2", + email="other@example.com", + display_name="User 2", + username="user2", + phone="13900000002", + ) + repo.save(user2) + + # 尝试把 user2 用户名改成 testuser（冲突） + user2_new = copy.deepcopy(user2) + user2_new.username = "testuser" + with pytest.raises(ValueError, match="Username already in use"): + repo.save(user2_new) + + # 各索引必须保持一致 + assert repo.find_by_username("testuser").id == "user-1" + assert repo.find_by_username("user2").id == "user-2" + assert repo.find_by_email("other@example.com").id == "user-2" + assert repo.find_by_phone("13900000002").id == "user-2" -- 2.54.0 From 0d91b461c7e2716139b54a705f42495513686619 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 01:18:02 +0800 Subject: [PATCH 6/9] =?UTF-8?q?perf(inmemory):=20=E7=B4=A2=E5=BC=95?= =?UTF-8?q?=E6=B8=85=E7=90=86=E4=BB=8EO(N)=E4=BC=98=E5=8C=96=E4=B8=BAO(1)?= =?UTF-8?q?=EF=BC=8Csave=E5=AD=98=E5=82=A8=E7=8B=AC=E7=AB=8B=E5=89=AF?= =?UTF-8?q?=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _remove_indexes_for_user（O(N)反向查找）→ _remove_indexes_of_user（O(1)按属性删除） - save时存储copy.copy独立副本，避免外部修改污染内部状态 - delete方法同步优化，先取对象再清理索引 - 新增测试验证存储副本隔离性 --- .../adapters/in_memory/user_repository.py | 63 +++++++++---------- tests/unit/test_inmemory_user_repository.py | 15 +++++ 2 files changed, 45 insertions(+), 33 deletions(-) diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py index 1616e799d..778d7a312 100755 --- a/packages/adapters/in_memory/user_repository.py +++ b/packages/adapters/in_memory/user_repository.py @@ -2,6 +2,7 @@ 用户仓储 In-Memory 实现 """ +import copy from typing import Dict, Optional from packages.domain.entities import User @@ -22,8 +23,8 @@ class InMemoryUserRepository(UserRepository): self._phone_index: Dict[str, str] = {} # phone -> user_id def save(self, user: User) -> None: - """保存用户""" - # 第一步：唯一性约束检查（先全部检查，全部通过再动数据） + """保存用户（存储独立副本，避免外部修改影响内部状态）""" + # 第一步：唯一性约束检查（O(1)，全部检查通过再动数据） new_email = user.email.lower() existing_id = self._email_index.get(new_email) if existing_id and existing_id != user.id: @@ -50,12 +51,14 @@ class InMemoryUserRepository(UserRepository): if existing_id and existing_id != user.id: raise ValueError(f"WeChat unionid already in use: {user.wechat_unionid}") - # 第二步：如果是更新，清理旧索引（通过user_id反向查找，避免对象引用问题） - if user.id in self._users: - self._remove_indexes_for_user(user.id) + # 第二步：如果是更新，用旧对象属性清理旧索引（O(1)，因存储的是独立副本） + old_user = self._users.get(user.id) + if old_user is not None: + self._remove_indexes_of_user(old_user) - # 第三步：写入数据和新索引 - self._users[user.id] = user + # 第三步：存储独立副本 + 写入新索引 + stored_user = copy.copy(user) + self._users[user.id] = stored_user self._email_index[new_email] = user.id if new_username: self._username_index[new_username] = user.id @@ -70,29 +73,22 @@ class InMemoryUserRepository(UserRepository): if user.phone: self._phone_index[user.phone] = user.id - def _remove_indexes_for_user(self, user_id: str) -> None: - """清理指定用户的所有索引条目（通过值反向查找）.""" - for key, val in list(self._email_index.items()): - if val == user_id: - del self._email_index[key] - for key, val in list(self._username_index.items()): - if val == user_id: - del self._username_index[key] - for key, val in list(self._verification_token_index.items()): - if val == user_id: - del self._verification_token_index[key] - for key, val in list(self._reset_token_index.items()): - if val == user_id: - del self._reset_token_index[key] - for key, val in list(self._wechat_openid_index.items()): - if val == user_id: - del self._wechat_openid_index[key] - for key, val in list(self._wechat_unionid_index.items()): - if val == user_id: - del self._wechat_unionid_index[key] - for key, val in list(self._phone_index.items()): - if val == user_id: - del self._phone_index[key] + def _remove_indexes_of_user(self, user: User) -> None: + """利用已知用户对象属性清理所有索引，时间复杂度 O(1).""" + if user.email: + self._email_index.pop(user.email.lower(), None) + if user.username: + self._username_index.pop(user.username.lower(), None) + if user.email_verification_token: + self._verification_token_index.pop(user.email_verification_token, None) + if user.password_reset_token: + self._reset_token_index.pop(user.password_reset_token, None) + if user.wechat_openid: + self._wechat_openid_index.pop(user.wechat_openid, None) + if user.wechat_unionid: + self._wechat_unionid_index.pop(user.wechat_unionid, None) + if user.phone: + self._phone_index.pop(user.phone, None) def find_by_id(self, user_id: str) -> Optional[User]: """根据 ID 查找用户""" @@ -153,11 +149,12 @@ class InMemoryUserRepository(UserRepository): def delete(self, user_id: str) -> bool: """删除用户""" - if user_id not in self._users: + user = self._users.get(user_id) + if user is None: return False - # 清理所有索引 - self._remove_indexes_for_user(user_id) + # O(1) 清理所有索引 + self._remove_indexes_of_user(user) # 删除用户 del self._users[user_id] diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py index 6422cdae2..8ef9ed98b 100755 --- a/tests/unit/test_inmemory_user_repository.py +++ b/tests/unit/test_inmemory_user_repository.py @@ -244,3 +244,18 @@ class TestIndexUpdates: assert repo.find_by_username("user2").id == "user-2" assert repo.find_by_email("other@example.com").id == "user-2" assert repo.find_by_phone("13900000002").id == "user-2" + + def test_save_stores_independent_copy(self, repo, sample_user): + """save存储独立副本，外部修改对象不影响仓储内部状态.""" + repo.save(sample_user) + + # 外部修改对象属性 + original_email = sample_user.email + sample_user.email = "hacked@example.com" + sample_user.display_name = "Hacked" + + # 仓储中数据不应受影响 + stored = repo.find_by_id(sample_user.id) + assert stored.email == original_email + assert repo.find_by_email(original_email) is not None + assert repo.find_by_email("hacked@example.com") is None -- 2.54.0 From 9581398d56210fbf08ca789c33a9c17855688d3a Mon Sep 17 00:00:00 2001 From: CI Bot Date: Wed, 29 Jul 2026 17:22:48 +0000 Subject: [PATCH 7/9] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_inmemory_asset_repository.py | 2 +- tests/unit/test_inmemory_small_repos.py | 14 +++++++------- tests/unit/test_inmemory_user_repository.py | 2 +- tests/unit/test_path_security.py | 2 +- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/tests/unit/test_inmemory_asset_repository.py b/tests/unit/test_inmemory_asset_repository.py index ca21083c0..d9db450b2 100755 --- a/tests/unit/test_inmemory_asset_repository.py +++ b/tests/unit/test_inmemory_asset_repository.py @@ -2,8 +2,8 @@ import pytest -from packages.domain.entities import Asset, AssetStatus, ClassificationStatus from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository +from packages.domain.entities import Asset, AssetStatus, ClassificationStatus @pytest.fixture diff --git a/tests/unit/test_inmemory_small_repos.py b/tests/unit/test_inmemory_small_repos.py index b8c8d0ecc..87ed5f667 100755 --- a/tests/unit/test_inmemory_small_repos.py +++ b/tests/unit/test_inmemory_small_repos.py @@ -4,19 +4,19 @@ from datetime import datetime, timezone import pytest -from packages.domain.entities import AssetLibrary, IngestJob, Project, User -from packages.domain.tag import Tag +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.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 +from packages.domain.entities import AssetLibrary, IngestJob, Project, User +from packages.domain.tag import Tag # ==================== AssetLibrary ==================== diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py index 8ef9ed98b..e40e277df 100755 --- a/tests/unit/test_inmemory_user_repository.py +++ b/tests/unit/test_inmemory_user_repository.py @@ -5,8 +5,8 @@ from datetime import datetime, timezone import pytest -from packages.domain.entities import User from packages.adapters.in_memory.user_repository import InMemoryUserRepository +from packages.domain.entities import User @pytest.fixture diff --git a/tests/unit/test_path_security.py b/tests/unit/test_path_security.py index c8b061bf7..6acf205c3 100755 --- a/tests/unit/test_path_security.py +++ b/tests/unit/test_path_security.py @@ -11,8 +11,8 @@ from apps.worker.video_processing.path_security import ( PathSecurityError, is_in_allowed_dirs, is_path_safe, - sanitize_filename, safe_resolve_path, + sanitize_filename, validate_local_schema_path, ) -- 2.54.0 From f0cf3f8504f42e329273aa76c7b37cde5fd462a0 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 07:07:04 +0800 Subject: [PATCH 8/9] =?UTF-8?q?fix(inmemory):=20=E5=A2=9E=E5=8A=A0email?= =?UTF-8?q?=E7=A9=BA=E5=80=BC=E9=98=B2=E5=BE=A1+token=E5=94=AF=E4=B8=80?= =?UTF-8?q?=E6=80=A7=E6=A3=80=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - save方法增加email空值校验，避免None.lower()崩溃 - find_by_email/find_by_username增加空值短路返回None - 增加email_verification_token和password_reset_token唯一性约束 - 新增5个测试用例覆盖防御场景 --- .../adapters/in_memory/user_repository.py | 18 +++++++ tests/unit/test_inmemory_user_repository.py | 52 +++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py index 778d7a312..72cc8118d 100755 --- a/packages/adapters/in_memory/user_repository.py +++ b/packages/adapters/in_memory/user_repository.py @@ -25,6 +25,9 @@ class InMemoryUserRepository(UserRepository): def save(self, user: User) -> None: """保存用户（存储独立副本，避免外部修改影响内部状态）""" # 第一步：唯一性约束检查（O(1)，全部检查通过再动数据） + # email 是必填字段，做空值防御 + if not user.email: + raise ValueError("User email cannot be empty") new_email = user.email.lower() existing_id = self._email_index.get(new_email) if existing_id and existing_id != user.id: @@ -51,6 +54,17 @@ class InMemoryUserRepository(UserRepository): if existing_id and existing_id != user.id: raise ValueError(f"WeChat unionid already in use: {user.wechat_unionid}") + # 验证令牌和重置令牌也做唯一性防御（防止生成器异常导致重复） + if user.email_verification_token: + existing_id = self._verification_token_index.get(user.email_verification_token) + if existing_id and existing_id != user.id: + raise ValueError("Email verification token already in use") + + if user.password_reset_token: + existing_id = self._reset_token_index.get(user.password_reset_token) + if existing_id and existing_id != user.id: + raise ValueError("Password reset token already in use") + # 第二步：如果是更新，用旧对象属性清理旧索引（O(1)，因存储的是独立副本） old_user = self._users.get(user.id) if old_user is not None: @@ -96,6 +110,8 @@ class InMemoryUserRepository(UserRepository): def find_by_email(self, email: str) -> Optional[User]: """根据邮箱查找用户""" + if not email: + return None user_id = self._email_index.get(email.lower()) if user_id: return self._users.get(user_id) @@ -103,6 +119,8 @@ class InMemoryUserRepository(UserRepository): def find_by_username(self, username: str) -> Optional[User]: """根据用户名查找用户""" + if not username: + return None user_id = self._username_index.get(username.lower()) if user_id: return self._users.get(user_id) diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py index e40e277df..6d8eb4153 100755 --- a/tests/unit/test_inmemory_user_repository.py +++ b/tests/unit/test_inmemory_user_repository.py @@ -259,3 +259,55 @@ class TestIndexUpdates: assert stored.email == original_email assert repo.find_by_email(original_email) is not None assert repo.find_by_email("hacked@example.com") is None + + def test_save_empty_email_raises(self, repo): + """空邮箱应抛出异常.""" + user = User( + id="user-empty", + email="", + display_name="Empty Email", + username="emptyuser", + ) + with pytest.raises(ValueError, match="email cannot be empty"): + repo.save(user) + + def test_find_by_empty_email_returns_none(self, repo, sample_user): + """空邮箱查询返回None.""" + repo.save(sample_user) + assert repo.find_by_email("") is None + assert repo.find_by_email(None) is None + + def test_find_by_empty_username_returns_none(self, repo, sample_user): + """空用户名查询返回None.""" + repo.save(sample_user) + assert repo.find_by_username("") is None + + def test_duplicate_verification_token_raises(self, repo, sample_user): + """相同邮箱验证令牌应触发唯一约束.""" + sample_user.email_verification_token = "verify-token-abc" + repo.save(sample_user) + + user2 = User( + id="user-2", + email="user2@example.com", + display_name="User 2", + username="user2", + email_verification_token="verify-token-abc", # 重复 + ) + with pytest.raises(ValueError, match="verification token already in use"): + repo.save(user2) + + def test_duplicate_reset_token_raises(self, repo, sample_user): + """相同密码重置令牌应触发唯一约束.""" + sample_user.password_reset_token = "reset-token-xyz" + repo.save(sample_user) + + user2 = User( + id="user-2", + email="user2@example.com", + display_name="User 2", + username="user2", + password_reset_token="reset-token-xyz", # 重复 + ) + with pytest.raises(ValueError, match="reset token already in use"): + repo.save(user2) -- 2.54.0 From 93db1d69e7c1d6235813de783a7654742754537b Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 10:12:35 +0800 Subject: [PATCH 9/9] ci: re-trigger checks for wave207 rebase -- 2.54.0