Compare commits

..

2 Commits

Author SHA1 Message Date
CI Bot 75901bed6f style: auto-format with black + isort + prettier [skip ci-format-check] 2026-07-28 09:16:45 +00:00
xiaoxia 73ffafdca6 test(wave164): filter_presets滤镜预设 +53测
PR Automation / Auto Approve on CI Green (pull_request) Waiting to run
PR Automation / Auto Merge on CI Green + Approved (pull_request) Waiting to run
Preview Deploy / Deploy Preview Environment (pull_request) Waiting to run
AI Code Review / AI Code Review (pull_request) CI runner不可用,手动设置
Preview Cleanup / Cleanup Preview Environment (pull_request) Waiting to run
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 19s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m38s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 34s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 2m17s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m29s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m48s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m49s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m44s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 6m50s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 3m50s
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Failing after 6s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1198h2m43s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1198h2m47s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1198h2m49s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1198h6m47s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1198h10m10s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1198h6m55s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1198h10m14s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 1198h10m16s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1198h10m57s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1198h11m1s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1198h35m42s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1198h43m10s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1198h43m57s
- FilterPreset数据类:最小创建/默认值/完整参数/frozen不可变/tags独立
- 预设库:非空/ID唯一/必填字段/5个分类存在/黑白饱和度为0/原图无效果/总数
- get_filter_preset:存在ID/不存在/空串/大小写敏感/返回类型
- list_filter_presets:无筛选/5个分类筛选/无效分类/关键词搜索(名称/标签/描述)
  + 组合筛选/空关键词/返回类型
- build_ffmpeg_filter:全强度/无效预设/0强度/负强度/超100截断
  + 半强度/原图空返回/格式校验/参数分隔/各参数存在/全预设遍历
2026-07-28 16:06:26 +08:00
9 changed files with 350 additions and 1498 deletions
@@ -6,7 +6,6 @@
from __future__ import annotations
# ── 单轨时间计算 ──────────────────────────────────────────────────────────────
@@ -13,7 +13,6 @@
from __future__ import annotations
from packages.domain.speed_config import (
DEFAULT_SPEED,
SpeedConfig,
-181
View File
@@ -1,181 +0,0 @@
"""classification 单测.
domain 层素材分类模块纯逻辑,0 外部依赖。
覆盖:4个枚举 + ClassificationJob 工厂/校验。
"""
from __future__ import annotations
from packages.domain.classification import (
AssetClassification,
AssetLibraryKind,
ClassificationJob,
ClassificationJobStatus,
IngestJobStatus,
)
class TestAssetLibraryKind:
"""AssetLibraryKind 枚举测试."""
def test_two_values(self):
"""视频和配音两类."""
assert len(AssetLibraryKind) == 2
def test_video(self):
assert AssetLibraryKind.VIDEO == "video"
def test_voice(self):
assert AssetLibraryKind.VOICE == "voice"
def test_str_compatible(self):
"""StrEnum 字符串兼容."""
assert AssetLibraryKind.VIDEO == "video"
class TestIngestJobStatus:
"""IngestJobStatus 枚举测试."""
def test_four_statuses(self):
assert len(IngestJobStatus) == 4
def test_pending(self):
assert IngestJobStatus.PENDING == "pending"
def test_processing(self):
assert IngestJobStatus.PROCESSING == "processing"
def test_completed(self):
assert IngestJobStatus.COMPLETED == "completed"
def test_failed(self):
assert IngestJobStatus.FAILED == "failed"
class TestClassificationJobStatus:
"""ClassificationJobStatus 枚举测试."""
def test_four_statuses(self):
assert len(ClassificationJobStatus) == 4
def test_pending(self):
assert ClassificationJobStatus.PENDING == "pending"
def test_processing(self):
assert ClassificationJobStatus.PROCESSING == "processing"
def test_completed(self):
assert ClassificationJobStatus.COMPLETED == "completed"
def test_failed(self):
assert ClassificationJobStatus.FAILED == "failed"
def test_same_values_as_ingest(self):
"""两种任务状态值相同."""
assert set(ClassificationJobStatus) == set(IngestJobStatus)
class TestAssetClassification:
"""AssetClassification 枚举测试."""
def test_nine_categories(self):
"""9个分类."""
assert len(AssetClassification) == 9
def test_scenic(self):
assert AssetClassification.SCENIC == "scenic"
def test_product(self):
assert AssetClassification.PRODUCT == "product"
def test_person(self):
assert AssetClassification.PERSON == "person"
def test_animal(self):
assert AssetClassification.ANIMAL == "animal"
def test_food(self):
assert AssetClassification.FOOD == "food"
def test_tech(self):
assert AssetClassification.TECH == "tech"
def test_sport(self):
assert AssetClassification.SPORT == "sport"
def test_music(self):
assert AssetClassification.MUSIC == "music"
def test_other(self):
assert AssetClassification.OTHER == "other"
def test_all_values_unique(self):
"""所有分类值唯一."""
values = [c.value for c in AssetClassification]
assert len(values) == len(set(values))
class TestClassificationJobCreate:
"""ClassificationJob.create 测试."""
def test_create_valid(self):
"""正常创建."""
job = ClassificationJob.create(project_id="proj1", asset_id="asset1")
assert job.project_id == "proj1"
assert job.asset_id == "asset1"
assert job.status == ClassificationJobStatus.PENDING
assert job.classification == ""
assert job.confidence == 0.0
assert job.error_message == ""
assert isinstance(job.id, str)
assert len(job.id) > 0
def test_create_strips(self):
"""project_id 和 asset_id 会 strip."""
job = ClassificationJob.create(project_id=" proj1 ", asset_id=" asset1 ")
assert job.project_id == "proj1"
assert job.asset_id == "asset1"
def test_create_empty_project_id(self):
"""空 project_id 无效."""
try:
ClassificationJob.create(project_id="", asset_id="a1")
assert False
except ValueError as e:
assert "project_id" in str(e)
def test_create_whitespace_project_id(self):
"""纯空白 project_id 无效."""
try:
ClassificationJob.create(project_id=" ", asset_id="a1")
assert False
except ValueError as e:
assert "project_id" in str(e)
def test_create_empty_asset_id(self):
"""空 asset_id 无效."""
try:
ClassificationJob.create(project_id="p1", asset_id="")
assert False
except ValueError as e:
assert "asset_id" in str(e)
def test_create_whitespace_asset_id(self):
"""纯空白 asset_id 无效."""
try:
ClassificationJob.create(project_id="p1", asset_id=" ")
assert False
except ValueError as e:
assert "asset_id" in str(e)
def test_create_unique_id(self):
"""不同 job id 不同."""
j1 = ClassificationJob.create("p", "a")
j2 = ClassificationJob.create("p", "a")
assert j1.id != j2.id
def test_create_has_timestamps(self):
"""有创建和更新时间."""
job = ClassificationJob.create("p", "a")
assert job.created_at is not None
assert job.updated_at is not None
-376
View File
@@ -1,376 +0,0 @@
"""duplication 单测.
domain 层查重记录纯逻辑模块,0 外部依赖。
覆盖:DuplicateSegment 工厂/校验、DuplicationRecord 创建/状态流转/重试。
"""
from __future__ import annotations
from packages.domain.duplication import DuplicateSegment, DuplicationRecord
class TestDuplicateSegmentCreate:
"""DuplicateSegment.create 工厂方法测试."""
def test_create_valid(self):
"""正常创建."""
seg = DuplicateSegment.create(
source_start=1.0,
source_end=5.0,
matched_video_id="vid123",
matched_video_name="测试视频",
matched_start=10.0,
matched_end=14.0,
similarity=85.5,
)
assert seg.source_start == 1.0
assert seg.source_end == 5.0
assert seg.matched_video_id == "vid123"
assert seg.matched_video_name == "测试视频"
assert seg.matched_start == 10.0
assert seg.matched_end == 14.0
assert seg.similarity == 85.5
assert isinstance(seg.id, str)
assert len(seg.id) > 0
def test_create_generates_unique_id(self):
"""每次创建生成不同的 id."""
seg1 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
seg2 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
assert seg1.id != seg2.id
def test_create_negative_source_start(self):
"""source_start 为负抛出 ValueError."""
try:
DuplicateSegment.create(-1, 5, "v", "n", 0, 1, 50.0)
assert False, "应该抛出 ValueError"
except ValueError as e:
assert "source" in str(e).lower()
def test_create_source_end_equals_start(self):
"""source_end 等于 source_start 无效."""
try:
DuplicateSegment.create(5, 5, "v", "n", 0, 1, 50.0)
assert False
except ValueError as e:
assert "source" in str(e).lower()
def test_create_source_end_less_than_start(self):
"""source_end 小于 source_start 无效."""
try:
DuplicateSegment.create(5, 3, "v", "n", 0, 1, 50.0)
assert False
except ValueError as e:
assert "source" in str(e).lower()
def test_create_negative_matched_start(self):
"""matched_start 为负无效."""
try:
DuplicateSegment.create(0, 5, "v", "n", -1, 1, 50.0)
assert False
except ValueError as e:
assert "matched" in str(e).lower()
def test_create_matched_end_invalid(self):
"""matched_end <= matched_start 无效."""
try:
DuplicateSegment.create(0, 5, "v", "n", 5, 5, 50.0)
assert False
except ValueError as e:
assert "matched" in str(e).lower()
def test_create_similarity_zero(self):
"""similarity = 0 是合法的."""
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 0.0)
assert seg.similarity == 0.0
def test_create_similarity_100(self):
"""similarity = 100 是合法的."""
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 100.0)
assert seg.similarity == 100.0
def test_create_similarity_negative(self):
"""similarity < 0 无效."""
try:
DuplicateSegment.create(0, 1, "v", "n", 0, 1, -1.0)
assert False
except ValueError as e:
assert "similarity" in str(e).lower()
def test_create_similarity_over_100(self):
"""similarity > 100 无效."""
try:
DuplicateSegment.create(0, 1, "v", "n", 0, 1, 101.0)
assert False
except ValueError as e:
assert "similarity" in str(e).lower()
class TestDuplicationRecordCreate:
"""DuplicationRecord.create 工厂方法测试."""
def test_create_minimal(self):
"""最简创建."""
rec = DuplicationRecord.create(
user_id="user1",
filename="test.mp4",
file_size=1024,
storage_key="oss://bucket/test.mp4",
)
assert rec.user_id == "user1"
assert rec.filename == "test.mp4"
assert rec.file_size == 1024
assert rec.storage_key == "oss://bucket/test.mp4"
assert rec.duration_seconds == 0.0
assert rec.status == "pending"
assert rec.duplicate_rate is None
assert rec.duplicate_count == 0
assert rec.segments == []
assert rec.error_message == ""
assert isinstance(rec.id, str)
assert len(rec.id) > 0
def test_create_with_duration(self):
"""带时长创建."""
rec = DuplicationRecord.create(
user_id="user1",
filename="test.mp4",
file_size=1024,
storage_key="oss://key",
duration_seconds=120.5,
)
assert rec.duration_seconds == 120.5
def test_create_strips_whitespace(self):
"""user_id 和 filename 会 strip."""
rec = DuplicationRecord.create(
user_id=" user1 ",
filename=" test.mp4 ",
file_size=1024,
storage_key="oss://key",
)
assert rec.user_id == "user1"
assert rec.filename == "test.mp4"
def test_create_empty_user_id(self):
"""空 user_id 无效."""
try:
DuplicationRecord.create("", "test.mp4", 1024, "oss://key")
assert False
except ValueError as e:
assert "user_id" in str(e)
def test_create_whitespace_user_id(self):
"""纯空白 user_id 无效."""
try:
DuplicationRecord.create(" ", "test.mp4", 1024, "oss://key")
assert False
except ValueError as e:
assert "user_id" in str(e)
def test_create_empty_filename(self):
"""空 filename 无效."""
try:
DuplicationRecord.create("user1", "", 1024, "oss://key")
assert False
except ValueError as e:
assert "filename" in str(e)
def test_create_whitespace_filename(self):
"""纯空白 filename 无效."""
try:
DuplicationRecord.create("user1", " ", 1024, "oss://key")
assert False
except ValueError as e:
assert "filename" in str(e)
def test_create_zero_file_size(self):
"""file_size = 0 无效."""
try:
DuplicationRecord.create("user1", "test.mp4", 0, "oss://key")
assert False
except ValueError as e:
assert "file_size" in str(e)
def test_create_negative_file_size(self):
"""file_size 为负无效."""
try:
DuplicationRecord.create("user1", "test.mp4", -1, "oss://key")
assert False
except ValueError as e:
assert "file_size" in str(e)
def test_create_unique_id(self):
"""不同记录 id 不同."""
r1 = DuplicationRecord.create("u", "f", 1, "k")
r2 = DuplicationRecord.create("u", "f", 1, "k")
assert r1.id != r2.id
def test_create_has_timestamps(self):
"""有创建和更新时间."""
rec = DuplicationRecord.create("u", "f", 1, "k")
assert rec.created_at is not None
assert rec.updated_at is not None
# 两者应该很接近(都是 now
delta = (rec.updated_at - rec.created_at).total_seconds()
assert abs(delta) < 1.0
class TestDuplicationRecordStatusFlow:
"""状态流转测试."""
def _make_record(self):
return DuplicationRecord.create("user1", "test.mp4", 1024, "oss://key")
def test_initial_status_pending(self):
"""初始状态 pending."""
rec = self._make_record()
assert rec.status == "pending"
def test_mark_processing(self):
"""标记为处理中."""
rec = self._make_record()
old_updated = rec.updated_at
rec.mark_processing()
assert rec.status == "processing"
assert rec.updated_at >= old_updated
def test_mark_completed(self):
"""标记为完成."""
rec = self._make_record()
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 80.0)
rec.mark_completed(duplicate_rate=45.5, duplicate_count=3, segments=[seg])
assert rec.status == "completed"
assert rec.duplicate_rate == 45.5
assert rec.duplicate_count == 3
assert len(rec.segments) == 1
assert rec.segments[0].similarity == 80.0
def test_mark_completed_zero_rate(self):
"""重复率为 0 合法."""
rec = self._make_record()
rec.mark_completed(0.0, 0, [])
assert rec.status == "completed"
assert rec.duplicate_rate == 0.0
assert rec.duplicate_count == 0
assert rec.segments == []
def test_mark_completed_full_rate(self):
"""重复率 100 合法."""
rec = self._make_record()
rec.mark_completed(100.0, 1, [])
assert rec.duplicate_rate == 100.0
def test_mark_completed_negative_rate(self):
"""重复率为负无效."""
rec = self._make_record()
try:
rec.mark_completed(-1, 0, [])
assert False
except ValueError as e:
assert "duplicate_rate" in str(e)
def test_mark_completed_over_100(self):
"""重复率超过 100 无效."""
rec = self._make_record()
try:
rec.mark_completed(101, 0, [])
assert False
except ValueError as e:
assert "duplicate_rate" in str(e)
def test_mark_failed(self):
"""标记为失败."""
rec = self._make_record()
rec.mark_failed("网络超时")
assert rec.status == "failed"
assert rec.error_message == "网络超时"
def test_mark_failed_empty_message(self):
"""失败信息可以为空字符串."""
rec = self._make_record()
rec.mark_failed("")
assert rec.status == "failed"
assert rec.error_message == ""
def test_can_retry_failed(self):
"""failed 状态可以重试."""
rec = self._make_record()
rec.mark_failed("error")
assert rec.can_retry() is True
def test_cannot_retry_pending(self):
"""pending 状态不可重试."""
rec = self._make_record()
assert rec.can_retry() is False
def test_cannot_retry_processing(self):
"""processing 状态不可重试."""
rec = self._make_record()
rec.mark_processing()
assert rec.can_retry() is False
def test_cannot_retry_completed(self):
"""completed 状态不可重试."""
rec = self._make_record()
rec.mark_completed(50, 1, [])
assert rec.can_retry() is False
def test_reset_for_retry(self):
"""重置回 pending."""
rec = self._make_record()
rec.mark_failed("error")
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
rec.segments = [seg]
rec.video_fingerprint = {"hash": "abc"}
rec.duplicate_rate = 50.0
rec.duplicate_count = 5
rec.reset_for_retry()
assert rec.status == "pending"
assert rec.duplicate_rate is None
assert rec.duplicate_count == 0
assert rec.error_message == ""
assert rec.segments == []
assert rec.video_fingerprint is None
def test_reset_updates_timestamp(self):
"""重置更新 updated_at."""
rec = self._make_record()
rec.mark_failed("error")
old_updated = rec.updated_at
rec.reset_for_retry()
assert rec.updated_at >= old_updated
class TestDuplicationRecordSegments:
"""segments 列表相关测试."""
def _make_record(self):
return DuplicationRecord.create("user1", "test.mp4", 1024, "oss://key")
def test_segments_default_empty(self):
"""初始 segments 为空列表."""
rec = self._make_record()
assert rec.segments == []
def test_segments_independent_list(self):
"""不同记录的 segments 是独立列表."""
r1 = self._make_record()
r2 = self._make_record()
r1.segments.append("fake")
assert len(r2.segments) == 0
def test_completed_with_multiple_segments(self):
"""完成时带多个片段."""
rec = self._make_record()
segs = [
DuplicateSegment.create(0, 1, "v1", "n1", 0, 1, 90.0),
DuplicateSegment.create(2, 3, "v2", "n2", 5, 6, 70.0),
DuplicateSegment.create(4, 5, "v3", "n3", 10, 11, 85.0),
]
rec.mark_completed(60.0, 3, segs)
assert len(rec.segments) == 3
assert rec.segments[0].similarity == 90.0
assert rec.segments[1].matched_video_id == "v2"
assert rec.segments[2].matched_video_name == "n3"
-362
View File
@@ -1,362 +0,0 @@
"""edit_plan_clip 单测.
domain 层剪辑计划片段纯逻辑模块,0 外部依赖。
覆盖:枚举常量、create工厂/校验、素材分配、状态流转、属性计算。
"""
from __future__ import annotations
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
class TestEditPlanClipStatus:
"""EditPlanClipStatus 枚举测试."""
def test_four_statuses(self):
"""四种状态."""
assert len(EditPlanClipStatus) == 4
def test_pending(self):
"""pending 状态."""
assert EditPlanClipStatus.PENDING == "pending"
def test_ready(self):
"""ready 状态."""
assert EditPlanClipStatus.READY == "ready"
def test_rendered(self):
"""rendered 状态."""
assert EditPlanClipStatus.RENDERED == "rendered"
def test_failed(self):
"""failed 状态."""
assert EditPlanClipStatus.FAILED == "failed"
def test_is_string(self):
"""枚举值是字符串."""
for status in EditPlanClipStatus:
assert isinstance(status.value, str)
assert len(status.value) > 0
def test_str_compatible(self):
"""StrEnum 可与字符串比较."""
assert EditPlanClipStatus.PENDING == "pending"
assert EditPlanClipStatus.READY + "" == "ready"
class TestEditPlanClipCreate:
"""EditPlanClip.create 工厂方法测试."""
def test_create_minimal(self):
"""最简创建."""
clip = EditPlanClip.create(plan_id="plan1", clip_type="video", order=0)
assert clip.plan_id == "plan1"
assert clip.clip_type == "video"
assert clip.order == 0
assert clip.status == EditPlanClipStatus.PENDING
assert clip.template_clip_config_id == ""
assert clip.asset_id == ""
assert clip.text_content == ""
assert clip.start_time == 0.0
assert clip.duration == 0.0
assert clip.transition_effect == "cut"
assert clip.config == {}
assert isinstance(clip.id, str)
assert len(clip.id) > 0
def test_create_with_all_fields(self):
"""带全部字段创建."""
clip = EditPlanClip.create(
plan_id="plan1",
clip_type="video",
order=2,
template_clip_config_id="tpl1",
asset_id="asset1",
text_content=" 你好世界 ",
start_time=10.5,
duration=5.0,
transition_effect="fade",
config={"key": "value"},
)
assert clip.order == 2
assert clip.template_clip_config_id == "tpl1"
assert clip.asset_id == "asset1"
assert clip.text_content == "你好世界" # strip了
assert clip.start_time == 10.5
assert clip.duration == 5.0
assert clip.transition_effect == "fade"
assert clip.config == {"key": "value"}
def test_create_strips_ids(self):
"""plan_id 和 clip_type 会 strip."""
clip = EditPlanClip.create(plan_id=" plan1 ", clip_type=" video ", order=0)
assert clip.plan_id == "plan1"
assert clip.clip_type == "video"
def test_create_empty_plan_id(self):
"""空 plan_id 无效."""
try:
EditPlanClip.create(plan_id="", clip_type="video", order=0)
assert False
except ValueError as e:
assert "plan_id" in str(e)
def test_create_whitespace_plan_id(self):
"""纯空白 plan_id 无效."""
try:
EditPlanClip.create(plan_id=" ", clip_type="video", order=0)
assert False
except ValueError as e:
assert "plan_id" in str(e)
def test_create_empty_clip_type(self):
"""空 clip_type 无效."""
try:
EditPlanClip.create(plan_id="p1", clip_type="", order=0)
assert False
except ValueError as e:
assert "clip_type" in str(e)
def test_create_whitespace_clip_type(self):
"""纯空白 clip_type 无效."""
try:
EditPlanClip.create(plan_id="p1", clip_type=" ", order=0)
assert False
except ValueError as e:
assert "clip_type" in str(e)
def test_create_negative_start_time(self):
"""start_time 为负无效."""
try:
EditPlanClip.create(plan_id="p1", clip_type="v", order=0, start_time=-1.0)
assert False
except ValueError as e:
assert "start_time" in str(e)
def test_create_negative_duration(self):
"""duration 为负无效."""
try:
EditPlanClip.create(plan_id="p1", clip_type="v", order=0, duration=-1.0)
assert False
except ValueError as e:
assert "duration" in str(e)
def test_create_zero_duration_valid(self):
"""duration 为 0 合法."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, duration=0.0)
assert clip.duration == 0.0
def test_create_empty_transition_defaults_to_cut(self):
"""空 transition_effect 默认 cut."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, transition_effect="")
assert clip.transition_effect == "cut"
def test_create_whitespace_transition_defaults_to_cut(self):
"""空白 transition_effect 默认 cut."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, transition_effect=" ")
assert clip.transition_effect == "cut"
def test_create_config_none_defaults_empty_dict(self):
"""config=None 默认为空 dict."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, config=None)
assert clip.config == {}
def test_create_unique_id(self):
"""不同 clip id 不同."""
c1 = EditPlanClip.create("p1", "v", 0)
c2 = EditPlanClip.create("p1", "v", 0)
assert c1.id != c2.id
def test_create_has_timestamps(self):
"""有创建和更新时间."""
clip = EditPlanClip.create("p1", "v", 0)
assert clip.created_at is not None
assert clip.updated_at is not None
def test_create_negative_order_valid(self):
"""order 可以为负(表示排序位置)."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=-1)
assert clip.order == -1
class TestEditPlanClipAssignAsset:
"""素材分配测试."""
def _make_clip(self):
return EditPlanClip.create(plan_id="p1", clip_type="video", order=0)
def test_assign_asset(self):
"""正常分配素材."""
clip = self._make_clip()
old_updated = clip.updated_at
clip.assign_asset("asset123")
assert clip.asset_id == "asset123"
assert clip.updated_at >= old_updated
def test_assign_asset_strips(self):
"""asset_id 会 strip."""
clip = self._make_clip()
clip.assign_asset(" asset123 ")
assert clip.asset_id == "asset123"
def test_assign_asset_empty(self):
"""空 asset_id 无效."""
clip = self._make_clip()
try:
clip.assign_asset("")
assert False
except ValueError as e:
assert "asset_id" in str(e)
def test_assign_asset_whitespace(self):
"""纯空白 asset_id 无效."""
clip = self._make_clip()
try:
clip.assign_asset(" ")
assert False
except ValueError as e:
assert "asset_id" in str(e)
def test_has_asset_false_initially(self):
"""初始无素材."""
clip = self._make_clip()
assert clip.has_asset is False
def test_has_asset_true_after_assign(self):
"""分配后有素材."""
clip = self._make_clip()
clip.assign_asset("a1")
assert clip.has_asset is True
class TestEditPlanClipStatusFlow:
"""状态流转测试."""
def _make_clip(self):
return EditPlanClip.create(plan_id="p1", clip_type="video", order=0)
def test_initial_status_pending(self):
"""初始状态 pending."""
clip = self._make_clip()
assert clip.status == EditPlanClipStatus.PENDING
def test_pending_to_ready(self):
"""pending -> ready."""
clip = self._make_clip()
clip.mark_ready()
assert clip.status == EditPlanClipStatus.READY
def test_ready_to_rendered(self):
"""ready -> rendered."""
clip = self._make_clip()
clip.mark_ready()
clip.mark_rendered()
assert clip.status == EditPlanClipStatus.RENDERED
def test_ready_to_failed(self):
"""ready -> failed."""
clip = self._make_clip()
clip.mark_ready()
clip.mark_failed()
assert clip.status == EditPlanClipStatus.FAILED
def test_cannot_ready_from_rendered(self):
"""rendered 状态不能再 mark_ready."""
clip = self._make_clip()
clip.mark_ready()
clip.mark_rendered()
try:
clip.mark_ready()
assert False
except ValueError as e:
assert "pending" in str(e).lower()
def test_cannot_ready_from_failed(self):
"""failed 状态不能 mark_ready."""
clip = self._make_clip()
clip.mark_ready()
clip.mark_failed()
try:
clip.mark_ready()
assert False
except ValueError as e:
assert "pending" in str(e).lower()
def test_cannot_render_from_pending(self):
"""pending 不能直接 mark_rendered."""
clip = self._make_clip()
try:
clip.mark_rendered()
assert False
except ValueError as e:
assert "ready" in str(e).lower()
def test_cannot_failed_from_pending(self):
"""pending 不能直接 mark_failed."""
clip = self._make_clip()
try:
clip.mark_failed()
assert False
except ValueError as e:
assert "ready" in str(e).lower()
def test_status_change_updates_timestamp(self):
"""状态变更更新 updated_at."""
clip = self._make_clip()
old_updated = clip.updated_at
clip.mark_ready()
assert clip.updated_at >= old_updated
class TestEditPlanClipProperties:
"""属性计算测试."""
def test_end_time(self):
"""end_time = start_time + duration."""
clip = EditPlanClip.create(
plan_id="p1",
clip_type="v",
order=0,
start_time=10.0,
duration=5.5,
)
assert clip.end_time == 15.5
def test_end_time_zero_duration(self):
"""零时长 end_time = start_time."""
clip = EditPlanClip.create(
plan_id="p1",
clip_type="v",
order=0,
start_time=10.0,
duration=0.0,
)
assert clip.end_time == 10.0
def test_end_time_zero_start(self):
"""零起点 end_time = duration."""
clip = EditPlanClip.create(
plan_id="p1",
clip_type="v",
order=0,
start_time=0.0,
duration=7.0,
)
assert clip.end_time == 7.0
def test_has_asset_empty_string(self):
"""空字符串无素材."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, asset_id="")
assert clip.has_asset is False
def test_has_asset_with_value(self):
"""有值则有素材."""
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, asset_id="a1")
assert clip.has_asset is True
def test_config_independent_between_clips(self):
"""不同 clip 的 config 独立."""
c1 = EditPlanClip.create(plan_id="p1", clip_type="v", order=0)
c2 = EditPlanClip.create(plan_id="p1", clip_type="v", order=1)
c1.config["key"] = "value"
assert "key" not in c2.config
+350
View File
@@ -0,0 +1,350 @@
"""filter_presets 单元测试 - wave164
覆盖:
- FilterPreset 数据类(frozen/默认值/字段)
- FILTER_PRESET_LIBRARY 预设库(数量/分类/ID唯一性)
- get_filter_preset 按ID获取
- list_filter_presets 筛选列表(分类/关键词)
- build_ffmpeg_filter 滤镜生成(强度/参数插值/边界/空值)
"""
import dataclasses
import pytest
from packages.domain.filter_presets import (
FILTER_PRESET_LIBRARY,
FilterPreset,
build_ffmpeg_filter,
get_filter_preset,
list_filter_presets,
)
# ============================================================
# FilterPreset 数据类
# ============================================================
class TestFilterPresetDataclass:
def test_minimal_creation(self):
p = FilterPreset(id="test", name="Test", category="basic")
assert p.id == "test"
assert p.name == "Test"
assert p.category == "basic"
def test_default_values(self):
p = FilterPreset(id="t", name="T", category="basic")
assert p.description == ""
assert p.tags == []
assert p.brightness == 0.0
assert p.contrast == 1.0
assert p.saturation == 1.0
assert p.gamma == 1.0
assert p.gamma_r == 1.0
assert p.gamma_g == 1.0
assert p.gamma_b == 1.0
assert p.hue == 0.0
assert p.lut_url == ""
def test_full_creation(self):
p = FilterPreset(
id="full",
name="Full",
category="cinematic",
description="test desc",
tags=["tag1", "tag2"],
brightness=0.5,
contrast=1.5,
saturation=2.0,
gamma=1.2,
gamma_r=1.3,
gamma_g=0.9,
gamma_b=0.8,
hue=30.0,
lut_url="http://lut.png",
)
assert p.description == "test desc"
assert p.tags == ["tag1", "tag2"]
assert p.brightness == 0.5
assert p.contrast == 1.5
assert p.gamma == 1.2
assert p.hue == 30.0
assert p.lut_url == "http://lut.png"
def test_frozen_immutable(self):
p = FilterPreset(id="t", name="T", category="basic")
with pytest.raises((AttributeError, dataclasses.FrozenInstanceError)):
p.brightness = 0.5 # frozen=True,不能修改
def test_tags_default_new_list(self):
# 每个实例有独立的列表
p1 = FilterPreset(id="t1", name="T1", category="basic")
p2 = FilterPreset(id="t2", name="T2", category="basic")
assert p1.tags is not p2.tags
# ============================================================
# FILTER_PRESET_LIBRARY 预设库
# ============================================================
class TestFilterPresetLibrary:
def test_library_not_empty(self):
assert len(FILTER_PRESET_LIBRARY) > 0
def test_all_ids_unique(self):
ids = [p.id for p in FILTER_PRESET_LIBRARY]
assert len(ids) == len(set(ids))
def test_all_have_required_fields(self):
for p in FILTER_PRESET_LIBRARY:
assert p.id, f"预设缺少id: {p}"
assert p.name, f"预设缺少name: {p.id}"
assert p.category, f"预设缺少category: {p.id}"
def test_categories_are_valid(self):
valid_categories = {"basic", "cinematic", "vintage", "bw", "style"}
for p in FILTER_PRESET_LIBRARY:
assert p.category in valid_categories, f"无效分类: {p.id} -> {p.category}"
def test_basic_category_exists(self):
basics = [p for p in FILTER_PRESET_LIBRARY if p.category == "basic"]
assert len(basics) >= 3
def test_cinematic_category_exists(self):
cinematic = [p for p in FILTER_PRESET_LIBRARY if p.category == "cinematic"]
assert len(cinematic) >= 1
def test_vintage_category_exists(self):
vintage = [p for p in FILTER_PRESET_LIBRARY if p.category == "vintage"]
assert len(vintage) >= 1
def test_bw_category_exists(self):
bw = [p for p in FILTER_PRESET_LIBRARY if p.category == "bw"]
assert len(bw) >= 1
# 所有黑白预设饱和度为0
for p in bw:
assert p.saturation == 0.0
def test_style_category_exists(self):
style = [p for p in FILTER_PRESET_LIBRARY if p.category == "style"]
assert len(style) >= 1
def test_none_filter_has_no_effect(self):
none_preset = get_filter_preset("filter_none")
assert none_preset is not None
assert none_preset.brightness == 0.0
assert none_preset.contrast == 1.0
assert none_preset.saturation == 1.0
assert none_preset.gamma == 1.0
def test_total_count(self):
# 至少有15个预设
assert len(FILTER_PRESET_LIBRARY) >= 15
# ============================================================
# get_filter_preset
# ============================================================
class TestGetFilterPreset:
def test_existing_id(self):
p = get_filter_preset("filter_brighten")
assert p is not None
assert p.id == "filter_brighten"
assert p.name == "明亮"
def test_nonexistent_id(self):
assert get_filter_preset("nonexistent") is None
def test_empty_id(self):
assert get_filter_preset("") is None
def test_case_sensitive(self):
# 大小写敏感
assert get_filter_preset("Filter_Brighten") is None
def test_returns_preset_object(self):
p = get_filter_preset("filter_cinematic")
assert isinstance(p, FilterPreset)
# ============================================================
# list_filter_presets
# ============================================================
class TestListFilterPresets:
def test_no_filters_returns_all(self):
result = list_filter_presets()
assert len(result) == len(FILTER_PRESET_LIBRARY)
def test_filter_by_category_basic(self):
result = list_filter_presets(category="basic")
assert len(result) > 0
for p in result:
assert p.category == "basic"
def test_filter_by_category_cinematic(self):
result = list_filter_presets(category="cinematic")
for p in result:
assert p.category == "cinematic"
def test_filter_by_category_bw(self):
result = list_filter_presets(category="bw")
for p in result:
assert p.category == "bw"
def test_filter_by_category_vintage(self):
result = list_filter_presets(category="vintage")
for p in result:
assert p.category == "vintage"
def test_filter_by_category_style(self):
result = list_filter_presets(category="style")
for p in result:
assert p.category == "style"
def test_invalid_category_returns_empty(self):
result = list_filter_presets(category="nonexistent")
assert result == []
def test_keyword_search_name(self):
result = list_filter_presets(keyword="电影")
assert len(result) >= 1
names = [p.name for p in result]
assert any("电影" in n for n in names)
def test_keyword_search_tags(self):
result = list_filter_presets(keyword="复古")
assert len(result) >= 1
# 至少有一个的标签或名称包含复古
found = any(any("复古" in t for t in p.tags) or "复古" in p.name for p in result)
assert found
def test_keyword_search_description(self):
result = list_filter_presets(keyword="青橙")
assert len(result) >= 1
def test_keyword_case_insensitive(self):
# 中文不区分大小写,用英文标签/名称试
result = list_filter_presets(keyword="FILTER")
# 搜索 filter 应该能匹配到很多(id里有但搜索name/desc/tags
# 用确定的中文关键词更可靠
assert isinstance(result, list)
def test_keyword_nonexistent_returns_empty(self):
result = list_filter_presets(keyword="zzz不存在的关键词zzz")
assert result == []
def test_category_and_keyword_combined(self):
result = list_filter_presets(category="basic", keyword="")
for p in result:
assert p.category == "basic"
assert "" in p.name or "" in p.description or any("" in t for t in p.tags)
def test_keyword_empty_returns_all(self):
result = list_filter_presets(keyword="")
assert len(result) == len(FILTER_PRESET_LIBRARY)
def test_returns_list_of_presets(self):
result = list_filter_presets()
for p in result:
assert isinstance(p, FilterPreset)
# ============================================================
# build_ffmpeg_filter
# ============================================================
class TestBuildFfmpegFilter:
def test_valid_preset_full_intensity(self):
result = build_ffmpeg_filter("filter_brighten", 100)
assert result.startswith("eq=")
assert "brightness=" in result
assert "contrast=" in result
def test_invalid_preset_returns_empty(self):
assert build_ffmpeg_filter("nonexistent", 100) == ""
def test_zero_intensity_returns_empty(self):
assert build_ffmpeg_filter("filter_brighten", 0) == ""
def test_negative_intensity_returns_empty(self):
assert build_ffmpeg_filter("filter_brighten", -10) == ""
def test_intensity_over_100_clamped(self):
# >100 按100算
r100 = build_ffmpeg_filter("filter_saturate", 100)
r200 = build_ffmpeg_filter("filter_saturate", 200)
assert r100 == r200
def test_half_intensity_half_effect(self):
full = build_ffmpeg_filter("filter_brighten", 100)
half = build_ffmpeg_filter("filter_brighten", 50)
# 半强度的brightness应该小一些
assert full != half
# 半强度仍然有效果
assert half.startswith("eq=")
def test_none_filter_returns_empty(self):
# filter_none所有参数都是默认值,应该返回空
result = build_ffmpeg_filter("filter_none", 100)
assert result == ""
def test_format_has_eq_prefix(self):
result = build_ffmpeg_filter("filter_saturate", 100)
assert result.startswith("eq=")
def test_parameters_separated_by_colon(self):
result = build_ffmpeg_filter("filter_cinematic", 100)
parts = result[len("eq=") :].split(":")
assert len(parts) >= 3 # 至少有3个参数
def test_brightness_value_format(self):
result = build_ffmpeg_filter("filter_brighten", 100)
# brightness参数应该存在且是数值
assert "brightness=0." in result or "brightness=-0." in result
def test_saturation_value_format(self):
result = build_ffmpeg_filter("filter_saturate", 100)
assert "saturation=" in result
def test_contrast_value_format(self):
result = build_ffmpeg_filter("filter_contrast", 100)
assert "contrast=" in result
def test_gamma_parameters_present(self):
result = build_ffmpeg_filter("filter_warm", 100)
# 暖色预设应该有gamma_r/gamma_b变化
assert "gamma_r=" in result or "gamma_b=" in result
def test_very_low_intensity_near_zero(self):
# 强度1%,效果接近0,但因为有brightness等可能仍然在阈值以上
result = build_ffmpeg_filter("filter_brighten", 1)
# 至少应该有一个参数(brightness * 0.01 = 0.0012 > 0.001
assert isinstance(result, str)
def test_all_presets_generate_valid_output(self):
# 每个预设在100强度下都能生成合法结果
for preset in FILTER_PRESET_LIBRARY:
result = build_ffmpeg_filter(preset.id, 100)
assert isinstance(result, str)
if result: # filter_none 可能为空
assert result.startswith("eq=")
def test_intensity_50_between_0_and_100(self):
# 验证50%强度确实在0和100之间插值
r0 = build_ffmpeg_filter("filter_saturate", 0)
r50 = build_ffmpeg_filter("filter_saturate", 50)
r100 = build_ffmpeg_filter("filter_saturate", 100)
assert r0 == ""
assert r50 != r100
assert r50 != ""
def test_hue_not_included_in_eq(self):
# hue 不在 eq 滤镜参数中(当前实现没有hue)
result = build_ffmpeg_filter("filter_cinematic", 100)
assert "hue=" not in result
-575
View File
@@ -1,575 +0,0 @@
"""quota 单测.
domain 层配额系统纯逻辑模块,0 外部依赖。
覆盖:枚举常量、QuotaTier、QUOTA_TIERS常量、QuotaWarningLevel、
QuotaCheckResult、QuotaRegistry注册/查询、QuotaChecker检查/告警级别。
"""
from __future__ import annotations
import math
from packages.domain.quota import (
QUOTA_TIERS,
QuotaCheckResult,
QuotaChecker,
QuotaDimension,
QuotaRegistry,
QuotaTier,
QuotaWarningLevel,
get_warning_level,
quota_checker,
quota_registry,
)
class TestQuotaDimension:
"""QuotaDimension 枚举测试."""
def test_member_count(self):
"""内置维度数量."""
assert len(QuotaDimension) == 11
def test_storage_gb(self):
"""存储维度值."""
assert QuotaDimension.STORAGE_GB.value == "storage_gb"
def test_videos_per_month(self):
"""每月视频数维度."""
assert QuotaDimension.VIDEOS_PER_MONTH.value == "videos_per_month"
def test_max_concurrent(self):
"""并发数维度."""
assert QuotaDimension.MAX_CONCURRENT.value == "max_concurrent"
def test_max_templates(self):
"""模板数维度."""
assert QuotaDimension.MAX_TEMPLATES.value == "max_templates"
def test_max_titles(self):
"""标题库维度."""
assert QuotaDimension.MAX_TITLES.value == "max_titles"
def test_max_voiceovers(self):
"""配音库维度."""
assert QuotaDimension.MAX_VOICEOVERS.value == "max_voiceovers"
def test_ai_voice_enabled(self):
"""AI配音开关维度."""
assert QuotaDimension.AI_VOICE_ENABLED.value == "ai_voice_enabled"
def test_ai_voice_credits(self):
"""AI配音积分维度."""
assert QuotaDimension.AI_VOICE_CREDITS.value == "ai_voice_credits"
def test_all_values_are_strings(self):
"""所有枚举值都是字符串."""
for dim in QuotaDimension:
assert isinstance(dim.value, str)
assert len(dim.value) > 0
class TestQuotaTier:
"""QuotaTier 数据类测试."""
def test_create_empty(self):
"""创建空配额等级."""
tier = QuotaTier(name="test")
assert tier.name == "test"
assert tier.limits == {}
def test_create_with_limits(self):
"""创建带限制的配额等级."""
tier = QuotaTier(name="pro", limits={"storage": 100, "videos": 50})
assert tier.name == "pro"
assert tier.get_limit("storage") == 100
assert tier.get_limit("videos") == 50
def test_get_limit_undefined_returns_zero(self):
"""未定义的维度返回 0."""
tier = QuotaTier(name="test")
assert tier.get_limit("nonexistent") == 0
def test_is_unlimited_inf(self):
"""inf 视为不限量."""
tier = QuotaTier(name="test", limits={"templates": float("inf")})
assert tier.is_unlimited("templates") is True
def test_is_unlimited_finite(self):
"""有限值不是不限量."""
tier = QuotaTier(name="test", limits={"storage": 100})
assert tier.is_unlimited("storage") is False
def test_is_unlimited_undefined(self):
"""未定义的维度默认不限量."""
tier = QuotaTier(name="test")
# 未定义的 key 取默认值 inf,因此 is_unlimited 应该返回 True
assert tier.is_unlimited("unknown") is True
class TestQuotaTiers:
"""QUOTA_TIERS 常量测试."""
def test_three_tiers_exist(self):
"""三个套餐等级都存在."""
assert "free" in QUOTA_TIERS
assert "basic" in QUOTA_TIERS
assert "premium" in QUOTA_TIERS
def test_free_storage(self):
"""免费版 2GB 存储."""
assert QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB) == 2
def test_free_videos_per_month(self):
"""免费版 5个视频/月."""
assert QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 5
def test_free_ai_voice_disabled(self):
"""免费版无AI配音."""
assert QUOTA_TIERS["free"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 0
def test_basic_storage(self):
"""基础版 20GB 存储."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB) == 20
def test_basic_videos(self):
"""基础版 30视频/月."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 30
def test_basic_ai_voice_enabled(self):
"""基础版有AI配音."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 1
def test_basic_ai_voice_credits(self):
"""基础版 100 AI配音积分."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_CREDITS) == 100
def test_premium_storage(self):
"""高级版 100GB 存储."""
assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB) == 100
def test_premium_videos(self):
"""高级版 100视频/月."""
assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 100
def test_premium_templates_unlimited(self):
"""高级版模板不限量."""
assert QUOTA_TIERS["premium"].is_unlimited(QuotaDimension.MAX_TEMPLATES)
def test_premium_multi_platform_enabled(self):
"""高级版多平台发布."""
assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.MULTI_PLATFORM_ENABLED) == 1
def test_storage_monotonic(self):
"""存储量随套餐升级单调递增."""
free = QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB)
basic = QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB)
premium = QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB)
assert free < basic < premium
def test_videos_monotonic(self):
"""视频数随套餐升级单调递增."""
free = QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
basic = QUOTA_TIERS["basic"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
premium = QUOTA_TIERS["premium"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
assert free < basic < premium
class TestQuotaWarningLevel:
"""告警级别常量测试."""
def test_levels_defined(self):
"""四个级别都有定义."""
assert QuotaWarningLevel.NORMAL == "normal"
assert QuotaWarningLevel.WARNING == "warning"
assert QuotaWarningLevel.CRITICAL == "critical"
assert QuotaWarningLevel.EXCEEDED == "exceeded"
def test_four_distinct_levels(self):
"""四个级别各不相同."""
levels = {
QuotaWarningLevel.NORMAL,
QuotaWarningLevel.WARNING,
QuotaWarningLevel.CRITICAL,
QuotaWarningLevel.EXCEEDED,
}
assert len(levels) == 4
class TestQuotaCheckResult:
"""QuotaCheckResult 测试."""
def test_usage_percent_normal(self):
"""正常使用百分比."""
result = QuotaCheckResult(
allowed=True,
dimension="storage",
limit=100,
used=30,
remaining=70,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 30.0
def test_usage_percent_zero_used(self):
"""使用量为 0."""
result = QuotaCheckResult(
allowed=True,
dimension="storage",
limit=100,
used=0,
remaining=100,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 0.0
def test_usage_percent_exactly_100(self):
"""刚好用完."""
result = QuotaCheckResult(
allowed=False,
dimension="storage",
limit=100,
used=100,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_over_limit_capped(self):
"""超出限制时封顶 100%."""
result = QuotaCheckResult(
allowed=False,
dimension="storage",
limit=100,
used=150,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_unlimited(self):
"""不限量时使用率为 0."""
result = QuotaCheckResult(
allowed=True,
dimension="templates",
limit=float("inf"),
used=1000,
remaining=float("inf"),
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 0.0
def test_usage_percent_zero_limit_with_usage(self):
"""限制为 0 但有使用量,返回 100%."""
result = QuotaCheckResult(
allowed=False,
dimension="ai_voice",
limit=0,
used=1,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_zero_limit_no_usage(self):
"""限制为 0 且无使用量,返回 0%."""
result = QuotaCheckResult(
allowed=True,
dimension="ai_voice",
limit=0,
used=0,
remaining=0,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 0.0
class TestGetWarningLevel:
"""get_warning_level 便捷函数测试."""
def test_zero_usage(self):
"""0% 使用 - normal."""
assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL
def test_below_80_percent(self):
"""低于80% - normal."""
assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL
assert get_warning_level(79, 100) == QuotaWarningLevel.NORMAL
def test_at_80_percent(self):
"""刚好80% - warning."""
assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING
def test_between_80_and_95(self):
"""80%-95%之间 - warning."""
assert get_warning_level(90, 100) == QuotaWarningLevel.WARNING
def test_at_95_percent(self):
"""刚好95% - critical."""
assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL
def test_between_95_and_100(self):
"""95%-100%之间 - critical."""
assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL
def test_at_100_percent(self):
"""刚好100% - exceeded."""
assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
def test_over_100_percent(self):
"""超过100% - exceeded."""
assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED
def test_zero_limit_with_usage(self):
"""限制为0但有使用 - exceeded."""
assert get_warning_level(1, 0) == QuotaWarningLevel.EXCEEDED
def test_zero_limit_no_usage(self):
"""限制为0且无使用 - normal."""
assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL
def test_unlimited(self):
"""不限量 - 始终 normal."""
assert get_warning_level(0, float("inf")) == QuotaWarningLevel.NORMAL
assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL
def test_negative_usage(self):
"""负使用量 - normal."""
assert get_warning_level(-10, 100) == QuotaWarningLevel.NORMAL
class TestQuotaRegistry:
"""QuotaRegistry 测试."""
def test_initial_dimensions(self):
"""初始化后内置维度都在."""
reg = QuotaRegistry()
dims = reg.list_dimensions()
for dim in QuotaDimension:
assert dim.value in dims
def test_initial_dimensions_count(self):
"""初始维度数量等于枚举数量."""
reg = QuotaRegistry()
assert len(reg.list_dimensions()) == len(QuotaDimension)
def test_list_tiers(self):
"""三个套餐等级."""
reg = QuotaRegistry()
tiers = reg.list_tiers()
assert "free" in tiers
assert "basic" in tiers
assert "premium" in tiers
assert len(tiers) == 3
def test_get_tier_existing(self):
"""获取已有的套餐."""
reg = QuotaRegistry()
tier = reg.get_tier("free")
assert tier is not None
assert tier.name == "free"
def test_get_tier_nonexistent(self):
"""获取不存在的套餐返回 None."""
reg = QuotaRegistry()
assert reg.get_tier("enterprise") is None
def test_get_limit_existing(self):
"""获取已有限制."""
reg = QuotaRegistry()
assert reg.get_limit("free", QuotaDimension.STORAGE_GB) == 2
def test_get_limit_nonexistent_plan(self):
"""不存在的套餐返回 0."""
reg = QuotaRegistry()
assert reg.get_limit("enterprise", QuotaDimension.STORAGE_GB) == 0
def test_get_limit_unknown_dimension(self):
"""未知维度返回 0."""
reg = QuotaRegistry()
assert reg.get_limit("free", "unknown_dim") == 0
def test_register_dimension_new(self):
"""注册新维度."""
reg = QuotaRegistry()
count_before = len(reg.list_dimensions())
reg.register_dimension("custom_dim", "自定义维度")
dims = reg.list_dimensions()
assert "custom_dim" in dims
assert len(dims) == count_before + 1
def test_register_dimension_with_default_limits(self):
"""注册带默认限制的维度."""
reg = QuotaRegistry()
reg.register_dimension(
"api_calls",
"API调用次数",
default_limits={"free": 100, "basic": 1000, "premium": 10000},
)
assert reg.get_limit("free", "api_calls") == 100
assert reg.get_limit("basic", "api_calls") == 1000
assert reg.get_limit("premium", "api_calls") == 10000
def test_register_dimension_default_zero(self):
"""不带默认限制的维度,各套餐默认为 0."""
reg = QuotaRegistry()
reg.register_dimension("beta_feature", "测试功能")
assert reg.get_limit("free", "beta_feature") == 0
assert reg.get_limit("basic", "beta_feature") == 0
assert reg.get_limit("premium", "beta_feature") == 0
def test_register_dimension_idempotent(self):
"""重复注册幂等."""
reg = QuotaRegistry()
reg.register_dimension("dup", "重复测试", default_limits={"free": 10})
count_before = len(reg.list_dimensions())
# 第二次注册不同的限制,应该不生效
reg.register_dimension("dup", "重复测试2", default_limits={"free": 999})
count_after = len(reg.list_dimensions())
assert count_before == count_after
assert reg.get_limit("free", "dup") == 10 # 仍然是第一次的值
def test_list_dimensions_returns_copy(self):
"""list_dimensions 返回副本,修改不影响内部."""
reg = QuotaRegistry()
dims = reg.list_dimensions()
dims["hacked"] = "hack"
assert "hacked" not in reg.list_dimensions()
def test_register_partial_default_limits(self):
"""只给部分套餐设置默认限制."""
reg = QuotaRegistry()
reg.register_dimension(
"partial",
"部分套餐",
default_limits={"premium": 100},
)
assert reg.get_limit("free", "partial") == 0
assert reg.get_limit("basic", "partial") == 0
assert reg.get_limit("premium", "partial") == 100
class TestQuotaChecker:
"""QuotaChecker 测试."""
def test_check_within_limit(self):
"""在限制内,allowed=True."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 1)
assert result.allowed is True
assert result.dimension == QuotaDimension.STORAGE_GB
assert result.limit == 2
assert result.used == 1
assert result.remaining == 1
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_exceeded(self):
"""超出限制,allowed=False."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 3)
assert result.allowed is False
assert result.remaining == 0
assert result.warning_level == QuotaWarningLevel.EXCEEDED
def test_check_exactly_at_limit(self):
"""刚好等于限制视为超出(used < limit 才允许)."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 2)
assert result.allowed is False
def test_check_unlimited(self):
"""不限量维度."""
checker = QuotaChecker()
result = checker.check("premium", QuotaDimension.MAX_TEMPLATES, 999)
assert result.allowed is True
assert math.isinf(result.remaining)
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_warning_level_boundaries(self):
"""各告警级别的边界值."""
checker = QuotaChecker()
# 79% - normal
assert checker.check("free", QuotaDimension.STORAGE_GB, 1.58).warning_level == QuotaWarningLevel.NORMAL
# 80% - warning
assert checker.check("free", QuotaDimension.STORAGE_GB, 1.6).warning_level == QuotaWarningLevel.WARNING
# 95% - critical
assert checker.check("free", QuotaDimension.STORAGE_GB, 1.9).warning_level == QuotaWarningLevel.CRITICAL
# 100% - exceeded
assert checker.check("free", QuotaDimension.STORAGE_GB, 2).warning_level == QuotaWarningLevel.EXCEEDED
def test_check_zero_usage(self):
"""0使用量."""
checker = QuotaChecker()
result = checker.check("basic", QuotaDimension.VIDEOS_PER_MONTH, 0)
assert result.allowed is True
assert result.remaining == 30
assert result.usage_percent == 0.0
def test_check_unknown_plan(self):
"""未知套餐,限制为0."""
checker = QuotaChecker()
result = checker.check("enterprise", QuotaDimension.STORAGE_GB, 0)
assert result.limit == 0
# used=0, limit=0 → 0 < 0 is False → allowed=False
assert result.allowed is False
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_unknown_dimension(self):
"""未知维度,限制为0."""
checker = QuotaChecker()
result = checker.check("free", "unknown", 0)
assert result.limit == 0
def test_check_multiple(self):
"""批量检查多个维度."""
checker = QuotaChecker()
usage = {
QuotaDimension.STORAGE_GB: 1,
QuotaDimension.VIDEOS_PER_MONTH: 10,
}
results = checker.check_multiple("free", usage)
assert len(results) == 2
dims = {r.dimension for r in results}
assert QuotaDimension.STORAGE_GB in dims
assert QuotaDimension.VIDEOS_PER_MONTH in dims
def test_check_multiple_empty(self):
"""空字典返回空列表."""
checker = QuotaChecker()
results = checker.check_multiple("free", {})
assert results == []
def test_remaining_never_negative(self):
"""剩余量不为负."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 100)
assert result.remaining >= 0
def test_custom_registry(self):
"""使用自定义 registry."""
reg = QuotaRegistry()
reg.register_dimension("custom", "自定义", default_limits={"free": 42})
checker = QuotaChecker(reg)
result = checker.check("free", "custom", 10)
assert result.limit == 42
assert result.allowed is True
class TestGlobalSingletons:
"""全局单例测试."""
def test_quota_registry_exists(self):
"""全局 registry 单例存在."""
assert quota_registry is not None
assert isinstance(quota_registry, QuotaRegistry)
def test_quota_checker_exists(self):
"""全局 checker 单例存在."""
assert quota_checker is not None
assert isinstance(quota_checker, QuotaChecker)
def test_global_checker_works(self):
"""全局 checker 能正常工作."""
result = quota_checker.check("free", QuotaDimension.STORAGE_GB, 1)
assert result.allowed is True
assert result.limit == 2
@@ -6,7 +6,6 @@ domain 层纯逻辑模块,0 FFmpeg 依赖,快速轻量。
from __future__ import annotations
import dataclasses
from unittest.mock import MagicMock
import pytest
-1
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
import dataclasses
import pytest
from packages.domain.transition_presets import (