Compare commits

...

8 Commits

Author SHA1 Message Date
CI Bot 584a1059b8 style: auto-format with black + isort + prettier [skip ci-format-check]
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 31s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m1s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m31s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m33s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m47s
AI Code Review / AI Code Review (pull_request) Successful in 1m44s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m42s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m39s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m56s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m56s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 4m35s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 3m1s
CI/CD Pipeline / CI Gate (pull_request) Successful in 11s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 44s
2026-07-29 15:03:17 +00:00
xiaoxia 92d5b3f26c test(wave205): path_security 单测重构与补全 +46测
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 5s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m14s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m16s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Lint (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Web Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
Preview Deploy / Deploy Preview Environment (pull_request) Has been cancelled
- 从 unittest 迁移到 pytest 风格
- 修正导入路径为标准包导入
- 覆盖 sanitize_filename / is_path_safe / is_in_allowed_dirs / safe_resolve_path
- 覆盖边界: 路径遍历、空路径、超长路径、特殊字符、符号链接等
2026-07-29 22:55:26 +08:00
xiaoxia eb73eeab69 refactor(assets): 拆分 batchOperations 为4个独立Hook(238→4行, -98%) (#1149)
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Successful in 5m47s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m4s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m14s
CI/CD Pipeline / Unit Tests (push) Successful in 8m34s
CI/CD Pipeline / Integration Tests (push) Successful in 3m7s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 54s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 19m10s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m31s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 7m42s
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 35s
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 25s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 4m15s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 9m16s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-07-29 22:21:52 +08:00
xiaoxia 2325ffbc57 fix(ci): 修复pr_auto_scan.py的typo - base_ref字段名应为ref而非re
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 33s
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Successful in 5m36s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m42s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m11s
CI/CD Pipeline / Unit Tests (push) Successful in 9m14s
CI/CD Pipeline / Integration Tests (push) Successful in 2m30s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m35s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 12m57s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m33s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m14s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 28s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 4m23s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m59s
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
- 修复base_ref获取字段名错误(re->ref),导致所有PR被误判为目标分支不匹配而跳过
- 修复dry-run模式打印缺少f前缀的小bug
2026-07-29 22:20:24 +08:00
xiaoxia faeed6f014 refactor(editing-planner): 拆分 SubtitleStylePanel 为子组件(190→146行, -23%) (#1139)
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 3m4s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 3m6s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 3m11s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 3m17s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 5m3s
CI/CD Pipeline / Validate - Code Quality (push) Successful in 8m4s
CI/CD Pipeline / Unit Tests (push) Successful in 12m32s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 4m36s
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 13m9s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 33s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 35s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m31s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Failing after 4m37s
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-07-29 17:30:42 +08:00
xiaoxia 6dc388a794 fix(cosyvoice): 修复prefix过滤中文失效导致400错误 (#1138)
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 50s
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Successful in 6m33s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 2m33s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m29s
CI/CD Pipeline / Unit Tests (push) Successful in 11m50s
CI/CD Pipeline / Integration Tests (push) Successful in 4m26s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m18s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 13m40s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 3m28s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m31s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 31s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 3m32s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m42s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
修复prefix过滤中文失效导致400错误
2026-07-29 14:40:58 +08:00
xiaoxia 7ca5b3732f refactor(voice-materials): 拆分 VoiceMaterialRow 进度条和标签组件 (#1087)
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
- 拆分 VoiceMaterialRow 为独立子组件
- 进度条逻辑抽取到 useRowProgress hook
- 标签展示组件独立为 TagDisplay
- 代码行数从 186 行减少到 138 行 (-26%)
2026-07-29 14:35:45 +08:00
xiaoxia 7cdf56a1ac test(wave175): template_clip_converter 模板片段转换 +64测 (#1137)
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
- 新增 template_clip_converter 模块单元测试
- 覆盖模板解析、片段转换、边界条件等场景
2026-07-29 14:34:37 +08:00
19 changed files with 1106 additions and 512 deletions
@@ -0,0 +1,4 @@
export { useBatchDelete } from "./useBatchDelete"
export { useBatchTag } from "./useBatchTag"
export { useBatchClassify } from "./useBatchClassify"
export { useBatchMark } from "./useBatchMark"
@@ -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,
}
}
@@ -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 }
}
@@ -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,
}
}
@@ -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,
}
}
@@ -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,
}
}
@@ -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"
+14 -58
View File
@@ -1,18 +1,16 @@
/**
* 字幕样式配置面板 — Drawer 形式
* 字幕开关(手动 / ASR 自动识别)、字体大小、颜色、描边/阴影、位置、ASR 语言
*/
import React from "react"
import { Drawer, Slider, ColorPicker, Select } from "antd"
import type { Color } from "antd/es/color-picker"
import type { SubtitleStyleConfig } from "@/pages/editing-planner/types/subtitle"
import {
POSITION_OPTIONS,
FONT_OPTIONS,
ANIMATION_OPTIONS,
ASR_LANGUAGE_OPTIONS,
} from "@/pages/editing-planner/constants/subtitleStyle"
import SubtitlePreview from "./subtitle-style/SubtitlePreview"
import { SubtitleModeSwitch } from "./subtitle-style/SubtitleModeSwitch"
import { SubtitlePositionSelector } from "./subtitle-style/SubtitlePositionSelector"
import { SubtitleEffectButtons } from "./subtitle-style/SubtitleEffectButtons"
interface SubtitleStylePanelProps {
open: boolean
@@ -40,7 +38,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
onClose={onClose}
className="subtitle-style-drawer"
>
{/* ── 字幕开关 ── */}
<div className="sub-field">
<div className="sub-toggle-row">
<span className="sub-label"></span>
@@ -55,26 +52,11 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
{config.enabled && (
<>
{/* ── 模式切换 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<div className="sub-mode-switch">
<button
className={`sub-mode-btn${config.mode === "manual" ? " active" : ""}`}
onClick={() => update({ mode: "manual" })}
>
</button>
<button
className={`sub-mode-btn${config.mode === "asr" ? " active" : ""}`}
onClick={() => update({ mode: "asr" })}
>
🤖 ASR
</button>
</div>
<SubtitleModeSwitch mode={config.mode} onModeChange={(mode) => update({ mode })} />
</div>
{/* ── ASR 语言(仅 ASR 模式) ── */}
{config.mode === "asr" && (
<div className="sub-field">
<label className="sub-label"></label>
@@ -88,7 +70,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
</div>
)}
{/* ── 字体大小 ── */}
<div className="sub-field">
<label className="sub-label">
<span className="sub-value">{config.fontSize}px</span>
@@ -101,7 +82,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
/>
</div>
{/* ── 字体颜色 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<div className="sub-color-row">
@@ -113,7 +93,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
</div>
</div>
{/* ── 字体 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<Select
@@ -125,46 +104,24 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
/>
</div>
{/* ── 字幕位置 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<div className="sub-position-group">
{POSITION_OPTIONS.map((opt) => (
<button
key={opt.value}
className={`sub-position-btn${config.position === opt.value ? " active" : ""}`}
onClick={() =>
update({
position: opt.value as SubtitleStyleConfig["position"],
})
}
>
{opt.label}
</button>
))}
</div>
<SubtitlePositionSelector
position={config.position}
onPositionChange={(position) => update({ position })}
/>
</div>
{/* ── 描边 / 阴影 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<div className="sub-effect-btns">
<button
className={`sub-effect-btn${config.stroke ? " active" : ""}`}
onClick={() => update({ stroke: !config.stroke })}
>
S
</button>
<button
className={`sub-effect-btn${config.shadow ? " active" : ""}`}
onClick={() => update({ shadow: !config.shadow })}
>
</button>
</div>
<SubtitleEffectButtons
stroke={config.stroke}
shadow={config.shadow}
onStrokeChange={(stroke) => update({ stroke })}
onShadowChange={(shadow) => update({ shadow })}
/>
</div>
{/* ── 动画 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<Select
@@ -179,7 +136,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
/>
</div>
{/* ── 预览 ── */}
<SubtitlePreview config={config} />
</>
)}
@@ -0,0 +1,32 @@
import React from "react"
interface SubtitleEffectButtonsProps {
stroke: boolean
shadow: boolean
onStrokeChange: (enabled: boolean) => void
onShadowChange: (enabled: boolean) => void
}
export const SubtitleEffectButtons: React.FC<SubtitleEffectButtonsProps> = ({
stroke,
shadow,
onStrokeChange,
onShadowChange,
}) => {
return (
<div className="sub-effect-btns">
<button
className={`sub-effect-btn${stroke ? " active" : ""}`}
onClick={() => onStrokeChange(!stroke)}
>
S
</button>
<button
className={`sub-effect-btn${shadow ? " active" : ""}`}
onClick={() => onShadowChange(!shadow)}
>
</button>
</div>
)
}
@@ -0,0 +1,25 @@
import React from "react"
interface SubtitleModeSwitchProps {
mode: "manual" | "asr"
onModeChange: (mode: "manual" | "asr") => void
}
export const SubtitleModeSwitch: React.FC<SubtitleModeSwitchProps> = ({ mode, onModeChange }) => {
return (
<div className="sub-mode-switch">
<button
className={`sub-mode-btn${mode === "manual" ? " active" : ""}`}
onClick={() => onModeChange("manual")}
>
</button>
<button
className={`sub-mode-btn${mode === "asr" ? " active" : ""}`}
onClick={() => onModeChange("asr")}
>
🤖 ASR
</button>
</div>
)
}
@@ -0,0 +1,27 @@
import React from "react"
import type { SubtitleStyleConfig } from "@/pages/editing-planner/types/subtitle"
import { POSITION_OPTIONS } from "@/pages/editing-planner/constants/subtitleStyle"
interface SubtitlePositionSelectorProps {
position: SubtitleStyleConfig["position"]
onPositionChange: (position: SubtitleStyleConfig["position"]) => void
}
export const SubtitlePositionSelector: React.FC<SubtitlePositionSelectorProps> = ({
position,
onPositionChange,
}) => {
return (
<div className="sub-position-group">
{POSITION_OPTIONS.map((opt) => (
<button
key={opt.value}
className={`sub-position-btn${position === opt.value ? " active" : ""}`}
onClick={() => onPositionChange(opt.value as SubtitleStyleConfig["position"])}
>
{opt.label}
</button>
))}
</div>
)
}
@@ -1,4 +1,4 @@
import { useRef, useCallback } from "react"
import { useRef, useCallback, useEffect } from "react"
interface UseRowProgressOptions {
duration: number
@@ -7,6 +7,22 @@ interface UseRowProgressOptions {
export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
const progressRef = useRef<HTMLDivElement>(null)
const listenersRef = useRef<{ move: ((e: MouseEvent) => void) | null; up: (() => void) | null }>({
move: null,
up: null,
})
const cleanupListeners = useCallback(() => {
const { move, up } = listenersRef.current
if (move) {
document.removeEventListener("mousemove", move)
listenersRef.current.move = null
}
if (up) {
document.removeEventListener("mouseup", up)
listenersRef.current.up = null
}
}, [])
const handleMouseDown = useCallback(
(e: React.MouseEvent<HTMLDivElement>) => {
@@ -16,6 +32,7 @@ export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
const doSeek = (ev: MouseEvent) => {
if (!progressRef.current) return
const rect = progressRef.current.getBoundingClientRect()
if (rect.width <= 0) return
const percent = Math.max(0, Math.min(1, (ev.clientX - rect.left) / rect.width))
onSeek(percent * duration)
}
@@ -24,15 +41,26 @@ export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
const handleMove = (ev: MouseEvent) => doSeek(ev)
const handleUp = () => {
document.removeEventListener("mousemove", handleMove)
document.removeEventListener("mouseup", handleUp)
cleanupListeners()
}
// 先清理旧的,再添加新的
cleanupListeners()
listenersRef.current.move = handleMove
listenersRef.current.up = handleUp
document.addEventListener("mousemove", handleMove)
document.addEventListener("mouseup", handleUp)
},
[duration, onSeek],
[duration, onSeek, cleanupListeners],
)
// 组件卸载时清理事件监听器
useEffect(() => {
return () => {
cleanupListeners()
}
}, [cleanupListeners])
return { progressRef, handleMouseDown }
}
@@ -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", () => {
@@ -65,6 +65,9 @@ import "@/pages/editing-planner/components/filter/FilterPresetGrid"
import "@/pages/editing-planner/components/filter/FilterManualAdjust"
import "@/pages/editing-planner/components/intro-outro/IntroOutroBlock"
import "@/pages/editing-planner/components/subtitle-style/SubtitlePreview"
import "@/pages/editing-planner/components/subtitle-style/SubtitleModeSwitch"
import "@/pages/editing-planner/components/subtitle-style/SubtitlePositionSelector"
import "@/pages/editing-planner/components/subtitle-style/SubtitleEffectButtons"
import "@/pages/editing-planner/components/tts/VoiceSelector"
import "@/pages/editing-planner/components/tts/TtsSlider"
import "@/pages/editing-planner/components/watermark/WatermarkTypeTabs"
@@ -20,6 +20,8 @@ import "@/pages/voice-materials/components/voice-material-card/CardActions"
import "@/pages/voice-materials/components/voice-material-card/BatchCheckbox"
import "@/pages/voice-materials/components/voice-material-card/types"
import "@/pages/voice-materials/components/VoiceMaterialRow"
import "@/pages/voice-materials/components/voice-material-row/useRowProgress"
import "@/pages/voice-materials/components/voice-material-row/TagDisplay"
import "@/pages/voice-materials/components/Toolbar"
import "@/pages/voice-materials/components/TagFilterBar"
import "@/pages/voice-materials/components/BatchBar"
+3 -1
View File
@@ -569,7 +569,9 @@ class CosyVoiceService:
清洗后的 prefix
"""
# 只保留字母和数字
cleaned = "".join(c for c in name if c.isalnum())
import re
cleaned = re.sub(r"[^a-zA-Z0-9]", "", name)
# 最多10字符
cleaned = cleaned[:10]
# 如果清洗后为空,用默认值
+1 -1
View File
@@ -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"):
+496
View File
@@ -0,0 +1,496 @@
"""模板片段转换器单测.
纯函数模块,覆盖:枚举安全解析、config过滤、
clip→template转换、snapshot双向转换、名称校验。
"""
from __future__ import annotations
from dataclasses import dataclass
from packages.domain.template_clip_config import (
ClipType,
TemplateClipConfig,
TransitionEffect,
)
from packages.domain.template_clip_converter import (
clip_config_to_snapshot,
clip_configs_to_snapshots,
clip_to_template_clip_config,
clips_to_template_clip_configs,
filter_clip_config,
filter_plan_config_to_template,
safe_parse_clip_type,
safe_parse_transition_effect,
snapshot_to_template_clip_config,
snapshots_to_template_clip_configs,
validate_template_name,
)
class TestSafeParseTransitionEffect:
def test_enum_passthrough(self):
result = safe_parse_transition_effect(TransitionEffect.FADE)
assert result == TransitionEffect.FADE
assert isinstance(result, TransitionEffect)
def test_valid_string(self):
result = safe_parse_transition_effect("fade")
assert result == TransitionEffect.FADE
def test_cut_string(self):
result = safe_parse_transition_effect("cut")
assert result == TransitionEffect.CUT
def test_invalid_string_returns_default(self):
result = safe_parse_transition_effect("invalid_effect")
assert result == TransitionEffect.CUT # 默认
def test_invalid_string_custom_default(self):
result = safe_parse_transition_effect("bad", default=TransitionEffect.DISSOLVE)
assert result == TransitionEffect.DISSOLVE
def test_none_returns_default(self):
result = safe_parse_transition_effect(None)
assert result == TransitionEffect.CUT
def test_int_value_returns_default(self):
result = safe_parse_transition_effect(123)
assert result == TransitionEffect.CUT
def test_empty_string_returns_default(self):
result = safe_parse_transition_effect("")
assert result == TransitionEffect.CUT
class TestSafeParseClipType:
def test_enum_passthrough(self):
result = safe_parse_clip_type(ClipType.SUBTITLE)
assert result == ClipType.SUBTITLE
assert isinstance(result, ClipType)
def test_valid_string_main(self):
result = safe_parse_clip_type("main")
assert result == ClipType.MAIN
def test_valid_string_text(self):
result = safe_parse_clip_type("subtitle")
assert result == ClipType.SUBTITLE
def test_invalid_string_returns_default(self):
result = safe_parse_clip_type("unknown_type")
assert result == ClipType.MAIN
def test_invalid_string_custom_default(self):
result = safe_parse_clip_type("bad", default=ClipType.TITLE)
assert result == ClipType.TITLE
def test_none_returns_default(self):
result = safe_parse_clip_type(None)
assert result == ClipType.MAIN
def test_dict_returns_default(self):
result = safe_parse_clip_type({"key": "val"})
assert result == ClipType.MAIN
class TestFilterClipConfig:
def test_none_config(self):
result = filter_clip_config(None)
assert result == {}
def test_empty_dict(self):
result = filter_clip_config({})
assert result == {}
def test_basic_config_passthrough(self):
cfg = {"font_size": 24, "color": "red"}
result = filter_clip_config(cfg)
assert result == {"font_size": 24, "color": "red"}
def test_filters_asset_info(self):
cfg = {"font_size": 24, "asset_info": {"id": "123"}}
result = filter_clip_config(cfg)
assert "asset_info" not in result
assert result["font_size"] == 24
def test_filters_source_asset_id(self):
cfg = {"source_asset_id": "asset_1", "text_key": "hi"}
result = filter_clip_config(cfg)
assert "source_asset_id" not in result
assert result["text_key"] == "hi"
def test_playback_speed_added_when_not_one(self):
result = filter_clip_config({}, playback_speed=1.5)
assert result["playback_speed"] == 1.5
def test_playback_speed_one_not_added(self):
result = filter_clip_config({}, playback_speed=1.0)
assert "playback_speed" not in result
def test_playback_speed_none_not_added(self):
result = filter_clip_config({}, playback_speed=None)
assert "playback_speed" not in result
def test_playback_speed_config_takes_priority(self):
"""clip_config中的playback_speed会覆盖参数传入的(因为update在后面)."""
cfg = {"playback_speed": 0.5, "other": "val"}
result = filter_clip_config(cfg, playback_speed=2.0)
assert result["playback_speed"] == 0.5 # config里的覆盖参数的
assert result["other"] == "val"
def test_custom_skip_keys(self):
cfg = {"keep_me": 1, "drop_me": 2, "also_drop": 3}
skip = frozenset({"drop_me", "also_drop"})
result = filter_clip_config(cfg, skip_keys=skip)
assert result == {"keep_me": 1}
def test_does_not_mutate_input(self):
cfg = {"a": 1, "asset_info": "x"}
original = dict(cfg)
filter_clip_config(cfg)
assert cfg == original # 原dict不变
class TestFilterPlanConfigToTemplate:
def test_none_config(self):
result = filter_plan_config_to_template(None)
assert result == {}
def test_empty_dict(self):
result = filter_plan_config_to_template({})
assert result == {}
def test_keeps_template_fields(self):
cfg = {"title": "My Template", "aspect_ratio": "9:16"}
result = filter_plan_config_to_template(cfg)
assert result == cfg
def test_filters_runtime_fields(self):
cfg = {
"title": "T",
"is_template_draft": True,
"asset_ids": ["a1"],
"source_edit_plan_id": "ep1",
"generation_task_id": "gt1",
}
result = filter_plan_config_to_template(cfg)
assert "is_template_draft" not in result
assert "asset_ids" not in result
assert "source_edit_plan_id" not in result
assert "generation_task_id" not in result
assert result["title"] == "T"
def test_custom_skip_keys(self):
cfg = {"keep": 1, "skip_a": 2, "skip_b": 3}
skip = frozenset({"skip_a", "skip_b"})
result = filter_plan_config_to_template(cfg, skip_keys=skip)
assert result == {"keep": 1}
class TestClipToTemplateClipConfig:
@dataclass
class FakeClip:
clip_type: str = "main"
order: int = 0
duration: float = 5.0
text_content: str = ""
transition_effect: str = "cut"
playback_speed: float | None = None
config: dict | None = None
def test_basic_conversion(self):
clip = self.FakeClip(
clip_type="subtitle",
order=2,
duration=3.5,
text_content="Hello",
transition_effect="fade",
)
result = clip_to_template_clip_config("tpl_1", clip)
assert isinstance(result, TemplateClipConfig)
assert result.template_id == "tpl_1"
assert result.clip_type == ClipType.SUBTITLE
assert result.order == 2
assert result.min_duration == 3.5
assert result.max_duration == 3.5
assert result.text_template == "Hello"
assert result.transition_effect == TransitionEffect.FADE
def test_duration_fixed_min_max_equal(self):
"""转换后 min_duration == max_duration == clip.duration."""
clip = self.FakeClip(duration=7.2)
result = clip_to_template_clip_config("t1", clip)
assert result.min_duration == 7.2
assert result.max_duration == 7.2
def test_zero_duration(self):
clip = self.FakeClip(duration=0.0)
result = clip_to_template_clip_config("t1", clip)
assert result.min_duration == 0.0
assert result.max_duration == 0.0
def test_none_duration_defaults_to_zero(self):
clip = self.FakeClip()
clip.duration = None # type: ignore
result = clip_to_template_clip_config("t1", clip)
assert result.min_duration == 0.0
assert result.max_duration == 0.0
def test_empty_text_content_becomes_empty_string(self):
clip = self.FakeClip(text_content="")
result = clip_to_template_clip_config("t1", clip)
assert result.text_template == ""
def test_none_text_content_becomes_empty_string(self):
clip = self.FakeClip()
clip.text_content = None # type: ignore
result = clip_to_template_clip_config("t1", clip)
assert result.text_template == ""
def test_playback_speed_in_config(self):
clip = self.FakeClip(playback_speed=1.5, config={"font": "bold"})
result = clip_to_template_clip_config("t1", clip)
assert result.config["playback_speed"] == 1.5
assert result.config["font"] == "bold"
def test_playback_speed_one_not_in_config(self):
clip = self.FakeClip(playback_speed=1.0)
result = clip_to_template_clip_config("t1", clip)
assert "playback_speed" not in result.config
def test_config_asset_info_filtered(self):
clip = self.FakeClip(config={"text_key": "hi", "asset_info": {"id": "a"}})
result = clip_to_template_clip_config("t1", clip)
assert "asset_info" not in result.config
assert result.config["text_key"] == "hi"
def test_invalid_clip_type_falls_back(self):
clip = self.FakeClip(clip_type="invalid_type")
result = clip_to_template_clip_config("t1", clip)
assert result.clip_type == ClipType.MAIN
def test_missing_attributes(self):
"""对象没有某些属性时使用默认值."""
class MinimalClip:
pass
result = clip_to_template_clip_config("t1", MinimalClip())
assert result.clip_type == ClipType.MAIN
assert result.order == 0
assert result.min_duration == 0.0
assert result.text_template == ""
assert result.transition_effect == TransitionEffect.CUT
class TestClipsToTemplateClipConfigs:
def test_empty_list(self):
result = clips_to_template_clip_configs("t1", [])
assert result == []
def test_multiple_clips(self):
clip_a = TestClipToTemplateClipConfig.FakeClip(clip_type="subtitle", order=0, duration=3.0, text_content="A")
clip_b = TestClipToTemplateClipConfig.FakeClip(clip_type="title", order=1, duration=5.0, text_content="")
result = clips_to_template_clip_configs("t1", [clip_a, clip_b])
assert len(result) == 2
assert result[0].clip_type == ClipType.SUBTITLE
assert result[0].order == 0
assert result[1].clip_type == ClipType.TITLE
assert result[1].order == 1
assert all(isinstance(r, TemplateClipConfig) for r in result)
class TestClipConfigToSnapshot:
def test_basic_snapshot(self):
cfg = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.SUBTITLE,
order=2,
min_duration=3.0,
max_duration=5.0,
text_template="Hello",
transition_effect=TransitionEffect.FADE,
config={"font_size": 20},
)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "subtitle"
assert snap["order"] == 2
assert snap["min_duration"] == 3.0
assert snap["max_duration"] == 5.0
assert snap["text_template"] == "Hello"
assert snap["transition_effect"] == "fade"
assert snap["config"] == {"font_size": 20}
def test_enum_values_are_strings(self):
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "main"
assert isinstance(snap["clip_type"], str)
assert snap["transition_effect"] == "cut"
assert isinstance(snap["transition_effect"], str)
def test_config_is_copy_not_reference(self):
config = {"key": "val"}
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0, config=config)
snap = clip_config_to_snapshot(cfg)
snap["config"]["key"] = "changed"
assert config["key"] == "val" # 原config不变
def test_empty_config(self):
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0, config={})
snap = clip_config_to_snapshot(cfg)
assert snap["config"] == {}
def test_none_text_becomes_empty(self):
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0)
cfg.text_template = None # type: ignore
snap = clip_config_to_snapshot(cfg)
assert snap["text_template"] == ""
class TestClipConfigsToSnapshots:
def test_empty_list(self):
assert clip_configs_to_snapshots([]) == []
def test_multiple_configs(self):
cfg1 = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.SUBTITLE,
order=0,
min_duration=2.0,
max_duration=2.0,
)
cfg2 = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.TITLE,
order=1,
min_duration=3.0,
max_duration=3.0,
)
snaps = clip_configs_to_snapshots([cfg1, cfg2])
assert len(snaps) == 2
assert snaps[0]["clip_type"] == "subtitle"
assert snaps[1]["clip_type"] == "title"
class TestSnapshotToTemplateClipConfig:
def test_basic_conversion(self):
snap = {
"clip_type": "subtitle",
"order": 3,
"min_duration": 2.5,
"max_duration": 4.5,
"text_template": "World",
"transition_effect": "dissolve",
"config": {"color": "blue"},
}
result = snapshot_to_template_clip_config("tpl_2", snap)
assert isinstance(result, TemplateClipConfig)
assert result.template_id == "tpl_2"
assert result.clip_type == ClipType.SUBTITLE
assert result.order == 3
assert result.min_duration == 2.5
assert result.max_duration == 4.5
assert result.text_template == "World"
assert result.transition_effect == TransitionEffect.DISSOLVE
assert result.config == {"color": "blue"}
def test_empty_snapshot_uses_defaults(self):
result = snapshot_to_template_clip_config("t1", {})
assert result.clip_type == ClipType.MAIN
assert result.order == 0
assert result.min_duration == 0.0
assert result.max_duration == 0.0
assert result.text_template == ""
assert result.transition_effect == TransitionEffect.CUT
assert result.config == {}
def test_invalid_clip_type_defaults(self):
snap = {"clip_type": "unknown"}
result = snapshot_to_template_clip_config("t1", snap)
assert result.clip_type == ClipType.MAIN
def test_invalid_transition_defaults(self):
snap = {"transition_effect": "bad_effect"}
result = snapshot_to_template_clip_config("t1", snap)
assert result.transition_effect == TransitionEffect.CUT
def test_none_config_becomes_empty(self):
snap = {"config": None}
result = snapshot_to_template_clip_config("t1", snap)
assert result.config == {}
class TestSnapshotsToTemplateClipConfigs:
def test_empty_list(self):
result = snapshots_to_template_clip_configs("t1", [])
assert result == []
def test_multiple_snapshots(self):
snaps = [
{"clip_type": "subtitle", "order": 0, "text_template": "A"},
{"clip_type": "title", "order": 1},
]
result = snapshots_to_template_clip_configs("t1", snaps)
assert len(result) == 2
assert result[0].clip_type == ClipType.SUBTITLE
assert result[0].text_template == "A"
assert result[1].clip_type == ClipType.TITLE
class TestRoundTrip:
"""clip → config → snapshot → config 双向转换一致性."""
def test_snapshot_config_round_trip(self):
original = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.SUBTITLE,
order=5,
min_duration=3.0,
max_duration=6.0,
text_template="Round trip",
transition_effect=TransitionEffect.FADE,
config={"key": "value"},
)
snap = clip_config_to_snapshot(original)
restored = snapshot_to_template_clip_config("t1", snap)
assert restored.clip_type == original.clip_type
assert restored.order == original.order
assert restored.min_duration == original.min_duration
assert restored.max_duration == original.max_duration
assert restored.text_template == original.text_template
assert restored.transition_effect == original.transition_effect
assert restored.config == original.config
class TestValidateTemplateName:
def test_valid_name(self):
assert validate_template_name("我的模板") == "我的模板"
def test_strips_whitespace(self):
assert validate_template_name(" Hello ") == "Hello"
def test_empty_string_raises(self):
try:
validate_template_name("")
except ValueError as e:
assert "不能为空" in str(e)
else:
raise AssertionError("expected ValueError")
def test_whitespace_only_raises(self):
try:
validate_template_name(" ")
except ValueError as e:
assert "不能为空" in str(e)
else:
raise AssertionError("expected ValueError")
def test_none_raises(self):
try:
validate_template_name(None)
except ValueError as e:
assert "不能为空" in str(e)
else:
raise AssertionError("expected ValueError")
+224 -209
View File
@@ -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"