style: auto-format with black + isort + prettier [skip ci-format-check]

- InMemoryAssetRepository: 46测，覆盖CRUD/批量操作/分页/标签/哈希去重
- InMemoryUserRepository: 26测，覆盖各索引查找/删除/保存
- 修复user_repository.save更新时旧索引未清理问题

apps/web/src/pages/assets/hooks/asset-operations/batch-operations/index.ts

@@ -0,0 +1,4 @@
+export { useBatchDelete } from "./useBatchDelete"
+export { useBatchTag } from "./useBatchTag"
+export { useBatchClassify } from "./useBatchClassify"
+export { useBatchMark } from "./useBatchMark"

apps/web/src/pages/assets/hooks/asset-operations/batch-operations/useBatchClassify.ts

@@ -0,0 +1,58 @@
+import { useState, useCallback } from "react"
+import { message } from "antd"
+import { batchClassifyAssets, type BatchOperationResult } from "@/api/assets"
+
+interface UseBatchClassifyOptions {
+  selectedIds: Set<string>
+  queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
+  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
+}
+
+export const useBatchClassify = ({
+  selectedIds,
+  queryClient,
+  showResult,
+}: UseBatchClassifyOptions) => {
+  const [classifyModalOpen, setClassifyModalOpen] = useState(false)
+  const [batchCategory, setBatchCategory] = useState("")
+  const [batchLoading, setBatchLoading] = useState(false)
+
+  const handleBatchClassify = useCallback(async () => {
+    if (!batchCategory) {
+      message.warning("请选择分类")
+      return
+    }
+    const ids = Array.from(selectedIds)
+    setBatchLoading(true)
+    try {
+      const result = await batchClassifyAssets({
+        asset_ids: ids,
+        category: batchCategory,
+      })
+      queryClient.invalidateQueries({ queryKey: ["assets"] })
+      showResult(result, "批量改分类")
+      setClassifyModalOpen(false)
+      setBatchCategory("")
+      if (result.failure_count === 0) {
+        message.success(`成功将 ${result.success_count} 个素材改为「${batchCategory}」`)
+      } else {
+        message.warning(
+          `改分类完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
+        )
+      }
+    } catch {
+      message.error("批量改分类失败，请重试")
+    } finally {
+      setBatchLoading(false)
+    }
+  }, [batchCategory, selectedIds, queryClient, showResult])
+
+  return {
+    classifyModalOpen,
+    setClassifyModalOpen,
+    batchCategory,
+    setBatchCategory,
+    batchLoading,
+    handleBatchClassify,
+  }
+}

apps/web/src/pages/assets/hooks/asset-operations/batch-operations/useBatchDelete.ts

@@ -0,0 +1,40 @@
+import { useState, useCallback } from "react"
+import { message } from "antd"
+import { batchDeleteAssets, type BatchOperationResult } from "@/api/assets"
+
+interface UseBatchDeleteOptions {
+  selectedIds: Set<string>
+  invalidateAssets: () => void
+  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
+}
+
+export const useBatchDelete = ({
+  selectedIds,
+  invalidateAssets,
+  showResult,
+}: UseBatchDeleteOptions) => {
+  const [batchLoading, setBatchLoading] = useState(false)
+
+  const handleBatchDelete = useCallback(async () => {
+    const ids = Array.from(selectedIds)
+    setBatchLoading(true)
+    try {
+      const result = await batchDeleteAssets(ids)
+      invalidateAssets()
+      showResult(result, "批量删除")
+      if (result.failure_count === 0) {
+        message.success(`成功删除 ${result.success_count} 个素材`)
+      } else {
+        message.warning(
+          `删除完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
+        )
+      }
+    } catch {
+      message.error("批量删除失败，请重试")
+    } finally {
+      setBatchLoading(false)
+    }
+  }, [selectedIds, invalidateAssets, showResult])
+
+  return { batchLoading, handleBatchDelete }
+}

apps/web/src/pages/assets/hooks/asset-operations/batch-operations/useBatchMark.ts

@@ -0,0 +1,53 @@
+import { useState, useCallback } from "react"
+import { message } from "antd"
+import { batchMarkAssets, type BatchOperationResult } from "@/api/assets"
+import type { SmartViewType } from "../../../components/BatchMarkModal"
+import { SMART_VIEW_LABELS } from "../constants"
+
+interface UseBatchMarkOptions {
+  selectedIds: Set<string>
+  queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
+  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
+}
+
+export const useBatchMark = ({ selectedIds, queryClient, showResult }: UseBatchMarkOptions) => {
+  const [markModalOpen, setMarkModalOpen] = useState(false)
+  const [batchSmartView, setBatchSmartView] = useState<SmartViewType>("recommended")
+  const [batchLoading, setBatchLoading] = useState(false)
+
+  const handleBatchMark = useCallback(async () => {
+    const ids = Array.from(selectedIds)
+    setBatchLoading(true)
+    try {
+      const result = await batchMarkAssets({
+        asset_ids: ids,
+        smart_view: batchSmartView,
+      })
+      queryClient.invalidateQueries({ queryKey: ["assets"] })
+      showResult(result, "批量智能标记")
+      setMarkModalOpen(false)
+      if (result.failure_count === 0) {
+        message.success(
+          `成功将 ${result.success_count} 个素材标记为「${SMART_VIEW_LABELS[batchSmartView]}」`,
+        )
+      } else {
+        message.warning(
+          `智能标记完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
+        )
+      }
+    } catch {
+      message.error("批量智能标记失败，请重试")
+    } finally {
+      setBatchLoading(false)
+    }
+  }, [batchSmartView, selectedIds, queryClient, showResult])
+
+  return {
+    markModalOpen,
+    setMarkModalOpen,
+    batchSmartView,
+    setBatchSmartView,
+    batchLoading,
+    handleBatchMark,
+  }
+}

apps/web/src/pages/assets/hooks/asset-operations/batch-operations/useBatchTag.ts

@@ -0,0 +1,86 @@
+import { useState, useCallback } from "react"
+import { message } from "antd"
+import { batchTagAssets, type BatchOperationResult } from "@/api/assets"
+
+interface UseBatchTagOptions {
+  selectedIds: Set<string>
+  queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
+  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
+}
+
+export const useBatchTag = ({ selectedIds, queryClient, showResult }: UseBatchTagOptions) => {
+  const [tagModalOpen, setTagModalOpen] = useState(false)
+  const [batchTagInput, setBatchTagInput] = useState("")
+  const [batchTags, setBatchTags] = useState<string[]>([])
+  const [tagMode, setTagMode] = useState<"add" | "replace">("add")
+  const [batchLoading, setBatchLoading] = useState(false)
+
+  const handleBatchTag = useCallback(async () => {
+    if (batchTags.length === 0) {
+      message.warning("请至少输入一个标签")
+      return
+    }
+    const ids = Array.from(selectedIds)
+    setBatchLoading(true)
+    try {
+      const result = await batchTagAssets({
+        asset_ids: ids,
+        tags: batchTags,
+        mode: tagMode,
+      })
+      queryClient.invalidateQueries({ queryKey: ["assets"] })
+      showResult(result, "批量打标签")
+      setTagModalOpen(false)
+      setBatchTags([])
+      setBatchTagInput("")
+      setTagMode("add")
+      if (result.failure_count === 0) {
+        message.success(`成功为 ${result.success_count} 个素材打标签`)
+      } else {
+        message.warning(
+          `打标签完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
+        )
+      }
+    } catch {
+      message.error("批量打标签失败，请重试")
+    } finally {
+      setBatchLoading(false)
+    }
+  }, [batchTags, selectedIds, tagMode, queryClient, showResult])
+
+  const handleTagInputKeyDown = useCallback(
+    (e: React.KeyboardEvent) => {
+      if (e.key === "Enter" && batchTagInput.trim()) {
+        e.preventDefault()
+        const tag = batchTagInput.trim()
+        if (!batchTags.includes(tag)) {
+          setBatchTags([...batchTags, tag])
+        }
+        setBatchTagInput("")
+      }
+    },
+    [batchTagInput, batchTags],
+  )
+
+  const removeBatchTag = useCallback(
+    (tag: string) => {
+      setBatchTags(batchTags.filter((t) => t !== tag))
+    },
+    [batchTags],
+  )
+
+  return {
+    tagModalOpen,
+    setTagModalOpen,
+    batchTagInput,
+    setBatchTagInput,
+    batchTags,
+    setBatchTags,
+    tagMode,
+    setTagMode,
+    batchLoading,
+    handleBatchTag,
+    handleTagInputKeyDown,
+    removeBatchTag,
+  }
+}

apps/web/src/pages/assets/hooks/asset-operations/batchOperations.ts

@@ -1,238 +0,0 @@
-import { useState, useCallback } from "react"
-import { message } from "antd"
-import {
-  batchDeleteAssets,
-  batchTagAssets,
-  batchClassifyAssets,
-  batchMarkAssets,
-  type BatchOperationResult,
-} from "@/api/assets"
-import type { SmartViewType } from "../../components/BatchMarkModal"
-import { SMART_VIEW_LABELS } from "./constants"
-
-/* ── 批量删除 ── */
-interface UseBatchDeleteOptions {
-  selectedIds: Set<string>
-  invalidateAssets: () => void
-  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
-}
-
-export const useBatchDelete = ({
-  selectedIds,
-  invalidateAssets,
-  showResult,
-}: UseBatchDeleteOptions) => {
-  const [batchLoading, setBatchLoading] = useState(false)
-
-  const handleBatchDelete = useCallback(async () => {
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchDeleteAssets(ids)
-      invalidateAssets()
-      showResult(result, "批量删除")
-      if (result.failure_count === 0) {
-        message.success(`成功删除 ${result.success_count} 个素材`)
-      } else {
-        message.warning(
-          `删除完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
-        )
-      }
-    } catch {
-      message.error("批量删除失败，请重试")
-    } finally {
-      setBatchLoading(false)
-    }
-  }, [selectedIds, invalidateAssets, showResult])
-
-  return { batchLoading, handleBatchDelete }
-}
-
-/* ── 批量打标签 ── */
-interface UseBatchTagOptions {
-  selectedIds: Set<string>
-  queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
-  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
-}
-
-export const useBatchTag = ({ selectedIds, queryClient, showResult }: UseBatchTagOptions) => {
-  const [tagModalOpen, setTagModalOpen] = useState(false)
-  const [batchTagInput, setBatchTagInput] = useState("")
-  const [batchTags, setBatchTags] = useState<string[]>([])
-  const [tagMode, setTagMode] = useState<"add" | "replace">("add")
-  const [batchLoading, setBatchLoading] = useState(false)
-
-  const handleBatchTag = useCallback(async () => {
-    if (batchTags.length === 0) {
-      message.warning("请至少输入一个标签")
-      return
-    }
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchTagAssets({
-        asset_ids: ids,
-        tags: batchTags,
-        mode: tagMode,
-      })
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      showResult(result, "批量打标签")
-      setTagModalOpen(false)
-      setBatchTags([])
-      setBatchTagInput("")
-      setTagMode("add")
-      if (result.failure_count === 0) {
-        message.success(`成功为 ${result.success_count} 个素材打标签`)
-      } else {
-        message.warning(
-          `打标签完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
-        )
-      }
-    } catch {
-      message.error("批量打标签失败，请重试")
-    } finally {
-      setBatchLoading(false)
-    }
-  }, [batchTags, selectedIds, tagMode, queryClient, showResult])
-
-  const handleTagInputKeyDown = useCallback(
-    (e: React.KeyboardEvent) => {
-      if (e.key === "Enter" && batchTagInput.trim()) {
-        e.preventDefault()
-        const tag = batchTagInput.trim()
-        if (!batchTags.includes(tag)) {
-          setBatchTags([...batchTags, tag])
-        }
-        setBatchTagInput("")
-      }
-    },
-    [batchTagInput, batchTags],
-  )
-
-  const removeBatchTag = useCallback(
-    (tag: string) => {
-      setBatchTags(batchTags.filter((t) => t !== tag))
-    },
-    [batchTags],
-  )
-
-  return {
-    tagModalOpen,
-    setTagModalOpen,
-    batchTagInput,
-    setBatchTagInput,
-    batchTags,
-    setBatchTags,
-    tagMode,
-    setTagMode,
-    batchLoading,
-    handleBatchTag,
-    handleTagInputKeyDown,
-    removeBatchTag,
-  }
-}
-
-/* ── 批量改分类 ── */
-interface UseBatchClassifyOptions {
-  selectedIds: Set<string>
-  queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
-  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
-}
-
-export const useBatchClassify = ({
-  selectedIds,
-  queryClient,
-  showResult,
-}: UseBatchClassifyOptions) => {
-  const [classifyModalOpen, setClassifyModalOpen] = useState(false)
-  const [batchCategory, setBatchCategory] = useState("")
-  const [batchLoading, setBatchLoading] = useState(false)
-
-  const handleBatchClassify = useCallback(async () => {
-    if (!batchCategory) {
-      message.warning("请选择分类")
-      return
-    }
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchClassifyAssets({
-        asset_ids: ids,
-        category: batchCategory,
-      })
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      showResult(result, "批量改分类")
-      setClassifyModalOpen(false)
-      setBatchCategory("")
-      if (result.failure_count === 0) {
-        message.success(`成功将 ${result.success_count} 个素材改为「${batchCategory}」`)
-      } else {
-        message.warning(
-          `改分类完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
-        )
-      }
-    } catch {
-      message.error("批量改分类失败，请重试")
-    } finally {
-      setBatchLoading(false)
-    }
-  }, [batchCategory, selectedIds, queryClient, showResult])
-
-  return {
-    classifyModalOpen,
-    setClassifyModalOpen,
-    batchCategory,
-    setBatchCategory,
-    batchLoading,
-    handleBatchClassify,
-  }
-}
-
-/* ── 批量智能标记 ── */
-interface UseBatchMarkOptions {
-  selectedIds: Set<string>
-  queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
-  showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
-}
-
-export const useBatchMark = ({ selectedIds, queryClient, showResult }: UseBatchMarkOptions) => {
-  const [markModalOpen, setMarkModalOpen] = useState(false)
-  const [batchSmartView, setBatchSmartView] = useState<SmartViewType>("recommended")
-  const [batchLoading, setBatchLoading] = useState(false)
-
-  const handleBatchMark = useCallback(async () => {
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchMarkAssets({
-        asset_ids: ids,
-        smart_view: batchSmartView,
-      })
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      showResult(result, "批量智能标记")
-      setMarkModalOpen(false)
-      if (result.failure_count === 0) {
-        message.success(
-          `成功将 ${result.success_count} 个素材标记为「${SMART_VIEW_LABELS[batchSmartView]}」`,
-        )
-      } else {
-        message.warning(
-          `智能标记完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
-        )
-      }
-    } catch {
-      message.error("批量智能标记失败，请重试")
-    } finally {
-      setBatchLoading(false)
-    }
-  }, [batchSmartView, selectedIds, queryClient, showResult])
-
-  return {
-    markModalOpen,
-    setMarkModalOpen,
-    batchSmartView,
-    setBatchSmartView,
-    batchLoading,
-    handleBatchMark,
-  }
-}

apps/web/src/pages/assets/hooks/useAssetOperations.ts

@@ -12,7 +12,7 @@ import {
   useBatchTag,
   useBatchClassify,
   useBatchMark,
-} from "./asset-operations/batchOperations"
+} from "./asset-operations/batch-operations"
 import type { BatchOperationResult } from "@/api/assets"
 import type { SmartViewType } from "../components/BatchMarkModal"
 

apps/web/src/test/pages/assets/smoke.test.tsx

@@ -36,6 +36,11 @@ import "@/pages/assets/hooks/useLibraryManagement"
 import "@/pages/assets/hooks/useAssetUpload"
 import "@/pages/assets/hooks/useAssetSelection"
 import "@/pages/assets/hooks/useAssetOperations"
+import "@/pages/assets/hooks/asset-operations/batch-operations"
+import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchDelete"
+import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchTag"
+import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchClassify"
+import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchMark"
 
 describe("AssetLibrary module smoke test", () => {
   it("should load all asset modules", () => {

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:

scripts/ci/pr_auto_scan.py

@@ -301,7 +301,7 @@ def main():
         pr_num = pr["number"]
         pr_title = pr["title"]
         head_sha = pr["head"]["sha"]
-        base_ref = pr.get("base", {}).get("re", "")
+        base_ref = pr.get("base", {}).get("ref", "")
 
         # 跳过draft
         if pr.get("draft"):

tests/unit/domain/test_verification_code.py

@@ -1,280 +0,0 @@
-"""VerificationCode 单元测试."""
-
-from __future__ import annotations
-
-from datetime import datetime, timedelta, timezone
-from unittest.mock import patch
-
-import pytest
-from domain.verification_code import VerificationCode
-
-
-class TestVerificationCodeCreate:
-    """create() 工厂方法测试."""
-
-    def test_create_basic(self):
-        vc = VerificationCode.create("test@example.com", "email_login")
-        assert vc.id is not None
-        assert len(vc.id) == 32
-        assert vc.recipient == "test@example.com"
-        assert vc.code_type == "email_login"
-        assert len(vc.code) == 6
-        assert vc.code.isdigit()
-        assert vc.used_at is None
-        assert vc.attempts == 0
-        assert vc.created_at is not None
-        assert vc.expires_at > vc.created_at
-
-    def test_create_recipient_stripped(self):
-        vc = VerificationCode.create("  test@example.com  ", "email_login")
-        assert vc.recipient == "test@example.com"
-
-    def test_create_custom_code(self):
-        vc = VerificationCode.create("test@example.com", "email_login", custom_code="123456")
-        assert vc.code == "123456"
-
-    def test_create_custom_ttl(self):
-        fixed_now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
-        with patch("domain.verification_code.datetime") as mock_dt:
-            mock_dt.now.return_value = fixed_now
-            mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
-            vc = VerificationCode.create("test@example.com", "email_login", ttl_seconds=60)
-        assert vc.expires_at == fixed_now + timedelta(seconds=60)
-
-    def test_create_default_ttl_300(self):
-        fixed_now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
-        with patch("domain.verification_code.datetime") as mock_dt:
-            mock_dt.now.return_value = fixed_now
-            mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
-            vc = VerificationCode.create("test@example.com", "email_login")
-        assert vc.expires_at == fixed_now + timedelta(seconds=300)
-
-    def test_create_unique_ids(self):
-        vc1 = VerificationCode.create("a@b.com", "email_login")
-        vc2 = VerificationCode.create("a@b.com", "email_login")
-        assert vc1.id != vc2.id
-
-    def test_create_unique_codes(self):
-        codes = set()
-        for _ in range(20):
-            vc = VerificationCode.create("a@b.com", "email_login")
-            codes.add(vc.code)
-        # 20个随机6位码几乎肯定不都一样
-        assert len(codes) > 1
-
-    def test_create_phone_recipient(self):
-        vc = VerificationCode.create("13800138000", "phone_login")
-        assert vc.recipient == "13800138000"
-        assert vc.code_type == "phone_login"
-
-    def test_create_all_code_types(self):
-        for ct in ["email_bind", "phone_bind", "email_login", "phone_login", "reset_password"]:
-            vc = VerificationCode.create("test@example.com", ct)
-            assert vc.code_type == ct
-
-
-class TestVerificationCodeIsExpired:
-    """is_expired 属性测试."""
-
-    def test_not_expired_future(self):
-        future = datetime.now(timezone.utc) + timedelta(hours=1)
-        vc = VerificationCode(
-            id="1",
-            recipient="a@b.com",
-            code="123456",
-            code_type="email_login",
-            expires_at=future,
-        )
-        assert vc.is_expired is False
-
-    def test_expired_past(self):
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        vc = VerificationCode(
-            id="1",
-            recipient="a@b.com",
-            code="123456",
-            code_type="email_login",
-            expires_at=past,
-        )
-        assert vc.is_expired is True
-
-    def test_expired_boundary_exact(self):
-        # 用mock固定时间，expires_at等于当前时间不算过期
-        fixed_now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
-        with patch("domain.verification_code.datetime") as mock_dt:
-            mock_dt.now.return_value = fixed_now
-            mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
-            vc = VerificationCode(
-                id="1",
-                recipient="a@b.com",
-                code="123456",
-                code_type="email_login",
-                expires_at=fixed_now,
-            )
-            assert vc.is_expired is False
-
-
-class TestVerificationCodeIsUsed:
-    """is_used 属性测试."""
-
-    def test_not_used_default(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        assert vc.is_used is False
-
-    def test_is_used_after_mark(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        vc.mark_used()
-        assert vc.is_used is True
-
-
-class TestVerificationCodeIsValid:
-    """is_valid 属性测试."""
-
-    def test_valid_fresh(self):
-        future = datetime.now(timezone.utc) + timedelta(hours=1)
-        vc = VerificationCode(
-            id="1",
-            recipient="a@b.com",
-            code="123456",
-            code_type="email_login",
-            expires_at=future,
-        )
-        assert vc.is_valid is True
-
-    def test_invalid_expired(self):
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        vc = VerificationCode(
-            id="1",
-            recipient="a@b.com",
-            code="123456",
-            code_type="email_login",
-            expires_at=past,
-        )
-        assert vc.is_valid is False
-
-    def test_invalid_used(self):
-        future = datetime.now(timezone.utc) + timedelta(hours=1)
-        vc = VerificationCode(
-            id="1",
-            recipient="a@b.com",
-            code="123456",
-            code_type="email_login",
-            expires_at=future,
-        )
-        vc.mark_used()
-        assert vc.is_valid is False
-
-    def test_invalid_expired_and_used(self):
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        vc = VerificationCode(
-            id="1",
-            recipient="a@b.com",
-            code="123456",
-            code_type="email_login",
-            expires_at=past,
-        )
-        vc.mark_used()
-        assert vc.is_valid is False
-
-
-class TestVerificationCodeMarkUsed:
-    """mark_used 方法测试."""
-
-    def test_mark_used_sets_timestamp(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        assert vc.used_at is None
-        before = datetime.now(timezone.utc)
-        vc.mark_used()
-        after = datetime.now(timezone.utc)
-        assert vc.used_at is not None
-        assert before <= vc.used_at <= after
-
-    def test_mark_used_twice_overwrites(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        vc.mark_used()
-        first = vc.used_at
-        # 时间足够短，一般不会不同，但确保可以重复调用
-        vc.mark_used()
-        assert vc.used_at is not None
-
-
-class TestVerificationCodeIncrementAttempts:
-    """increment_attempts 方法测试."""
-
-    def test_default_zero(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        assert vc.attempts == 0
-
-    def test_increment_once(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        vc.increment_attempts()
-        assert vc.attempts == 1
-
-    def test_increment_multiple(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        for _i in range(5):
-            vc.increment_attempts()
-        assert vc.attempts == 5
-
-
-class TestVerificationCodeBasics:
-    """基础构造和 slots 测试."""
-
-    def test_direct_construction(self):
-        now = datetime.now(timezone.utc)
-        vc = VerificationCode(
-            id="abc123",
-            recipient="test@test.com",
-            code="000000",
-            code_type="email_bind",
-            expires_at=now + timedelta(minutes=5),
-            used_at=None,
-            attempts=0,
-            created_at=now,
-        )
-        assert vc.id == "abc123"
-        assert vc.recipient == "test@test.com"
-        assert vc.code == "000000"
-
-    def test_slots_no_extra_attrs(self):
-        vc = VerificationCode.create("a@b.com", "email_login")
-        with pytest.raises((AttributeError, TypeError)):
-            vc.new_field = "value"  # type: ignore[attr-defined]
-
-    def test_equality_same_id(self):
-        now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
-        vc1 = VerificationCode(
-            id="same",
-            recipient="a@b.com",
-            code="111",
-            code_type="email_login",
-            expires_at=now,
-            created_at=now,
-        )
-        vc2 = VerificationCode(
-            id="same",
-            recipient="a@b.com",
-            code="111",
-            code_type="email_login",
-            expires_at=now,
-            created_at=now,
-        )
-        assert vc1 == vc2
-
-    def test_equality_different_id(self):
-        now = datetime.now(timezone.utc)
-        vc1 = VerificationCode(
-            id="id1",
-            recipient="a@b.com",
-            code="111",
-            code_type="email_login",
-            expires_at=now,
-        )
-        vc2 = VerificationCode(
-            id="id2",
-            recipient="a@b.com",
-            code="111",
-            code_type="email_login",
-            expires_at=now,
-        )
-        assert vc1 != vc2

tests/unit/test_inmemory_asset_repository.py

@@ -0,0 +1,401 @@
+"""InMemoryAssetRepository 单元测试."""
+
+import pytest
+
+from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository
+from packages.domain.entities import Asset, AssetStatus, ClassificationStatus
+
+
+@pytest.fixture
+def repo() -> InMemoryAssetRepository:
+    return InMemoryAssetRepository()
+
+
+@pytest.fixture
+def sample_asset() -> Asset:
+    return Asset.create(
+        project_id="proj-1",
+        library_id="lib-1",
+        name="test.mp4",
+        storage_key="storage/key1",
+        mime_type="video/mp4",
+        file_size=1024,
+        file_hash="hash-abc",
+    )
+
+
+@pytest.fixture
+def asset2() -> Asset:
+    return Asset.create(
+        project_id="proj-1",
+        library_id="lib-1",
+        name="test2.jpg",
+        storage_key="storage/key2",
+        mime_type="image/jpeg",
+        file_size=512,
+        file_hash="hash-def",
+    )
+
+
+@pytest.fixture
+def asset_other_project() -> Asset:
+    return Asset.create(
+        project_id="proj-2",
+        library_id="lib-2",
+        name="other.mp3",
+        storage_key="storage/key3",
+        mime_type="audio/mpeg",
+        file_size=256,
+        file_hash="hash-ghi",
+    )
+
+
+class TestCreateAndGet:
+    def test_create_returns_asset(self, repo, sample_asset):
+        result = repo.create(sample_asset)
+        assert result.id == sample_asset.id
+        assert result.name == "test.mp4"
+
+    def test_get_existing_asset(self, repo, sample_asset):
+        repo.create(sample_asset)
+        result = repo.get(sample_asset.id)
+        assert result is not None
+        assert result.id == sample_asset.id
+
+    def test_get_nonexistent_returns_none(self, repo):
+        assert repo.get("nonexistent") is None
+
+    def test_find_by_id_same_as_get(self, repo, sample_asset):
+        repo.create(sample_asset)
+        assert repo.find_by_id(sample_asset.id).id == repo.get(sample_asset.id).id
+
+
+class TestListByProject:
+    def test_list_by_project_filters_correctly(self, repo, sample_asset, asset2, asset_other_project):
+        repo.create(sample_asset)
+        repo.create(asset2)
+        repo.create(asset_other_project)
+
+        proj1 = repo.list_by_project("proj-1")
+        assert len(proj1) == 2
+        assert all(a.project_id == "proj-1" for a in proj1)
+
+        proj2 = repo.list_by_project("proj-2")
+        assert len(proj2) == 1
+        assert proj2[0].id == asset_other_project.id
+
+    def test_list_by_project_empty(self, repo):
+        assert repo.list_by_project("nonexistent") == []
+
+
+class TestListByLibrary:
+    def test_list_by_library_filters_correctly(self, repo, sample_asset, asset2, asset_other_project):
+        repo.create(sample_asset)
+        repo.create(asset2)
+        repo.create(asset_other_project)
+
+        lib1 = repo.list_by_library("lib-1")
+        assert len(lib1) == 2
+
+        lib2 = repo.list_by_library("lib-2")
+        assert len(lib2) == 1
+        assert lib2[0].id == asset_other_project.id
+
+    def test_find_by_library_is_alias(self, repo, sample_asset):
+        repo.create(sample_asset)
+        assert repo.find_by_library("lib-1") == repo.list_by_library("lib-1")
+
+    def test_list_by_library_empty(self, repo):
+        assert repo.list_by_library("nonexistent") == []
+
+
+class TestFindByLibraryAndFileType:
+    def test_filter_by_video(self, repo, sample_asset, asset2, asset_other_project):
+        repo.create(sample_asset)
+        repo.create(asset2)
+        repo.create(asset_other_project)
+
+        videos = repo.find_by_library_and_file_type("lib-1", "video")
+        assert len(videos) == 1
+        assert videos[0].mime_type.startswith("video/")
+
+    def test_filter_by_image(self, repo, sample_asset, asset2):
+        repo.create(sample_asset)
+        repo.create(asset2)
+
+        images = repo.find_by_library_and_file_type("lib-1", "image")
+        assert len(images) == 1
+        assert images[0].mime_type.startswith("image/")
+
+    def test_filter_by_audio(self, repo, sample_asset, asset_other_project):
+        repo.create(sample_asset)
+        repo.create(asset_other_project)
+
+        audio = repo.find_by_library_and_file_type("lib-2", "audio")
+        assert len(audio) == 1
+
+    def test_empty_result(self, repo, sample_asset):
+        repo.create(sample_asset)
+        assert repo.find_by_library_and_file_type("lib-1", "audio") == []
+
+
+class TestUpdate:
+    def test_update_existing_asset(self, repo, sample_asset):
+        repo.create(sample_asset)
+        sample_asset.name = "updated.mp4"
+        sample_asset.file_size = 2048
+
+        result = repo.update(sample_asset)
+        assert result.name == "updated.mp4"
+        assert result.file_size == 2048
+
+        fetched = repo.get(sample_asset.id)
+        assert fetched.name == "updated.mp4"
+
+    def test_update_nonexistent_creates(self, repo, sample_asset):
+        """update 直接覆盖，不存在则相当于 create."""
+        result = repo.update(sample_asset)
+        assert result.id == sample_asset.id
+        assert repo.get(sample_asset.id) is not None
+
+
+class TestDelete:
+    def test_delete_existing(self, repo, sample_asset):
+        repo.create(sample_asset)
+        assert repo.delete(sample_asset.id) is True
+        assert repo.get(sample_asset.id) is None
+
+    def test_delete_nonexistent(self, repo):
+        assert repo.delete("nonexistent") is False
+
+
+class TestBatchDelete:
+    def test_batch_delete_soft_delete(self, repo, sample_asset, asset2):
+        repo.create(sample_asset)
+        repo.create(asset2)
+
+        count = repo.batch_delete([sample_asset.id, asset2.id])
+        assert count == 2
+
+        a1 = repo.get(sample_asset.id)
+        a2 = repo.get(asset2.id)
+        assert a1.status == AssetStatus.DELETED
+        assert a2.status == AssetStatus.DELETED
+        assert a1.updated_at is not None
+        assert a2.updated_at is not None
+
+    def test_batch_delete_skip_already_deleted(self, repo, sample_asset):
+        repo.create(sample_asset)
+        sample_asset.status = AssetStatus.DELETED
+        repo.update(sample_asset)
+
+        count = repo.batch_delete([sample_asset.id])
+        assert count == 0
+
+    def test_batch_delete_nonexistent(self, repo):
+        count = repo.batch_delete(["nonexistent-1", "nonexistent-2"])
+        assert count == 0
+
+    def test_batch_delete_partial(self, repo, sample_asset):
+        repo.create(sample_asset)
+        count = repo.batch_delete([sample_asset.id, "nonexistent"])
+        assert count == 1
+
+
+class TestBatchUpdateMetadata:
+    def test_batch_update_metadata_merge(self, repo, sample_asset, asset2):
+        sample_asset.metadata = {"key1": "val1"}
+        repo.create(sample_asset)
+        repo.create(asset2)
+
+        count = repo.batch_update_metadata(
+            [sample_asset.id, asset2.id],
+            {"key2": "val2"},
+        )
+        assert count == 2
+
+        a1 = repo.get(sample_asset.id)
+        a2 = repo.get(asset2.id)
+        assert a1.metadata == {"key1": "val1", "key2": "val2"}
+        assert a2.metadata == {"key2": "val2"}
+
+    def test_batch_update_metadata_overwrite_existing_key(self, repo, sample_asset):
+        sample_asset.metadata = {"key1": "old"}
+        repo.create(sample_asset)
+
+        count = repo.batch_update_metadata([sample_asset.id], {"key1": "new"})
+        assert count == 1
+        assert repo.get(sample_asset.id).metadata["key1"] == "new"
+
+    def test_batch_update_metadata_nonexistent(self, repo):
+        count = repo.batch_update_metadata(["nonexistent"], {"key": "val"})
+        assert count == 0
+
+
+class TestBatchAddTags:
+    def test_batch_add_tags_new_tags(self, repo, sample_asset, asset2):
+        repo.create(sample_asset)
+        repo.create(asset2)
+
+        count = repo.batch_add_tags([sample_asset.id, asset2.id], ["tag1", "tag2"])
+        assert count == 2
+
+        a1 = repo.get(sample_asset.id)
+        a2 = repo.get(asset2.id)
+        assert set(a1.tag_ids) == {"tag1", "tag2"}
+        assert set(a2.tag_ids) == {"tag1", "tag2"}
+
+    def test_batch_add_tags_dedup(self, repo, sample_asset):
+        sample_asset.tag_ids = ["tag1"]
+        repo.create(sample_asset)
+
+        count = repo.batch_add_tags([sample_asset.id], ["tag1", "tag2"])
+        assert count == 1  # tag1已存在，但tag2新增，所以有变化
+
+        tags = repo.get(sample_asset.id).tag_ids
+        assert tags.count("tag1") == 1
+        assert "tag2" in tags
+
+    def test_batch_add_tags_no_change_when_all_exist(self, repo, sample_asset):
+        sample_asset.tag_ids = ["tag1", "tag2"]
+        repo.create(sample_asset)
+
+        count = repo.batch_add_tags([sample_asset.id], ["tag1", "tag2"])
+        assert count == 0  # 没有变化
+
+    def test_batch_add_tags_nonexistent_assets(self, repo):
+        count = repo.batch_add_tags(["nonexistent"], ["tag1"])
+        assert count == 0
+
+
+class TestBatchReplaceTags:
+    def test_batch_replace_tags_override(self, repo, sample_asset):
+        sample_asset.tag_ids = ["old1", "old2"]
+        repo.create(sample_asset)
+
+        count = repo.batch_replace_tags([sample_asset.id], ["new1", "new2", "new3"])
+        assert count == 1
+
+        tags = repo.get(sample_asset.id).tag_ids
+        assert tags == ["new1", "new2", "new3"]
+
+    def test_batch_replace_tags_empty(self, repo, sample_asset):
+        sample_asset.tag_ids = ["tag1"]
+        repo.create(sample_asset)
+
+        count = repo.batch_replace_tags([sample_asset.id], [])
+        assert count == 1
+        assert repo.get(sample_asset.id).tag_ids == []
+
+    def test_batch_replace_tags_nonexistent(self, repo):
+        count = repo.batch_replace_tags(["nonexistent"], ["tag1"])
+        assert count == 0
+
+
+class TestFindByProjectPagination:
+    @pytest.fixture
+    def five_assets(self, repo):
+        assets = []
+        for i in range(5):
+            a = Asset.create(
+                project_id="proj-paged",
+                library_id="lib-paged",
+                name=f"asset-{i}.mp4",
+                storage_key=f"key-{i}",
+                mime_type="video/mp4",
+            )
+            repo.create(a)
+            assets.append(a)
+        return assets
+
+    def test_find_by_project_default_pagination(self, repo, five_assets):
+        result = repo.find_by_project("proj-paged")
+        assert len(result) == 5
+
+    def test_find_by_project_skip(self, repo, five_assets):
+        result = repo.find_by_project("proj-paged", skip=2)
+        assert len(result) == 3
+
+    def test_find_by_project_limit(self, repo, five_assets):
+        result = repo.find_by_project("proj-paged", limit=2)
+        assert len(result) == 2
+
+    def test_find_by_project_skip_and_limit(self, repo, five_assets):
+        result = repo.find_by_project("proj-paged", skip=1, limit=2)
+        assert len(result) == 2
+
+    def test_find_by_project_skip_past_end(self, repo, five_assets):
+        result = repo.find_by_project("proj-paged", skip=10)
+        assert result == []
+
+    def test_find_by_project_empty(self, repo):
+        assert repo.find_by_project("nonexistent") == []
+
+
+class TestFindByTagIds:
+    def test_find_by_tag_ids_match_all(self, repo, sample_asset, asset2):
+        sample_asset.tag_ids = ["tag1", "tag2", "tag3"]
+        asset2.tag_ids = ["tag1", "tag2"]
+        repo.create(sample_asset)
+        repo.create(asset2)
+
+        result = repo.find_by_tag_ids(["tag1", "tag2"])
+        assert len(result) == 2
+
+    def test_find_by_tag_ids_subset_match(self, repo, sample_asset, asset2):
+        sample_asset.tag_ids = ["tag1", "tag2"]
+        asset2.tag_ids = ["tag1"]
+        repo.create(sample_asset)
+        repo.create(asset2)
+
+        result = repo.find_by_tag_ids(["tag1", "tag2"])
+        assert len(result) == 1
+        assert result[0].id == sample_asset.id
+
+    def test_find_by_tag_ids_empty_tag_list(self, repo, sample_asset):
+        sample_asset.tag_ids = ["tag1"]
+        repo.create(sample_asset)
+        assert repo.find_by_tag_ids([]) == []
+
+    def test_find_by_tag_ids_no_match(self, repo, sample_asset):
+        sample_asset.tag_ids = ["tag1"]
+        repo.create(sample_asset)
+        assert repo.find_by_tag_ids(["tag999"]) == []
+
+    def test_find_by_tag_ids_pagination(self, repo):
+        for i in range(5):
+            a = Asset.create(
+                project_id="p1",
+                library_id="l1",
+                name=f"a{i}.mp4",
+                storage_key=f"k{i}",
+                mime_type="video/mp4",
+            )
+            a.tag_ids = ["shared-tag"]
+            repo.create(a)
+
+        result = repo.find_by_tag_ids(["shared-tag"], skip=1, limit=2)
+        assert len(result) == 2
+
+
+class TestFindByLibraryAndFileHash:
+    def test_find_by_hash_match(self, repo, sample_asset):
+        repo.create(sample_asset)
+        result = repo.find_by_library_and_file_hash("lib-1", "hash-abc")
+        assert result is not None
+        assert result.id == sample_asset.id
+
+    def test_find_by_hash_wrong_library(self, repo, sample_asset):
+        repo.create(sample_asset)
+        result = repo.find_by_library_and_file_hash("lib-2", "hash-abc")
+        assert result is None
+
+    def test_find_by_hash_wrong_hash(self, repo, sample_asset):
+        repo.create(sample_asset)
+        result = repo.find_by_library_and_file_hash("lib-1", "hash-wrong")
+        assert result is None
+
+    def test_find_by_hash_empty_hash(self, repo, sample_asset):
+        repo.create(sample_asset)
+        result = repo.find_by_library_and_file_hash("lib-1", "")
+        assert result is None

tests/unit/test_inmemory_user_repository.py

@@ -0,0 +1,191 @@
+"""InMemoryUserRepository 单元测试."""
+
+from datetime import datetime, timezone
+
+import pytest
+
+from packages.adapters.in_memory.user_repository import InMemoryUserRepository
+from packages.domain.entities import User
+
+
+@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

tests/unit/test_path_security.py

@@ -1,18 +1,14 @@
-"""路径安全校验工具单元测试 — 路径遍历防护."""
-
-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,
@@ -21,223 +17,242 @@ from video_processing.path_security import (  # noqa: E402
 )
 
 
-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<name>:"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<name>.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"