test(unit): P3-1 核心模块单元测试覆盖率提升 - 新增12个模块400+测试 #661

Merged
auto-approve-bot merged 3 commits from test/unit-test-coverage-improvement-p3 into develop 2026-07-21 00:21:38 +08:00
12 changed files with 4053 additions and 256 deletions
+333 -256
View File
@@ -1,274 +1,351 @@
"""查重域模型单元测试。
覆盖:
- DuplicationRecord.create() 工厂方法及验证
- DuplicationRecord 状态转换(mark_processing / mark_completed / mark_failed
- DuplicationRecord.can_retry() / reset_for_retry()
- DuplicateSegment.create() 工厂方法及验证
"""
from __future__ import annotations
Duplication 查重记录领域模型单元测试
"""
import pytest
from packages.domain.duplication import DuplicateSegment, DuplicationRecord
class TestDuplicationRecordCreate:
"""DuplicationRecord.create() 工厂方法测试"""
class TestDuplicateSegmentCreate:
"""DuplicateSegment.create 测试"""
def test_create_success(self):
record = DuplicationRecord.create(
user_id="user-1",
filename="test.mp4",
file_size=1024,
storage_key="oss/key/test.mp4",
duration_seconds=30.0,
seg = DuplicateSegment.create(
source_start=10.0,
source_end=20.0,
matched_video_id="vid_123",
matched_video_name="测试视频",
matched_start=5.0,
matched_end=15.0,
similarity=85.5,
)
assert record.user_id == "user-1"
assert seg.id is not None
assert len(seg.id) == 32
assert seg.source_start == 10.0
assert seg.source_end == 20.0
assert seg.matched_video_id == "vid_123"
assert seg.matched_video_name == "测试视频"
assert seg.matched_start == 5.0
assert seg.matched_end == 15.0
assert seg.similarity == 85.5
def test_invalid_source_negative_start(self):
with pytest.raises(ValueError, match="invalid source segment range"):
DuplicateSegment.create(
source_start=-1.0,
source_end=10.0,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=50,
)
def test_invalid_source_end_before_start(self):
with pytest.raises(ValueError, match="invalid source segment range"):
DuplicateSegment.create(
source_start=20.0,
source_end=10.0,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=50,
)
def test_invalid_source_end_equals_start(self):
with pytest.raises(ValueError, match="invalid source segment range"):
DuplicateSegment.create(
source_start=10.0,
source_end=10.0,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=50,
)
def test_invalid_matched_negative_start(self):
with pytest.raises(ValueError, match="invalid matched segment range"):
DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=-5,
matched_end=10,
similarity=50,
)
def test_invalid_matched_end_before_start(self):
with pytest.raises(ValueError, match="invalid matched segment range"):
DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=15,
matched_end=10,
similarity=50,
)
def test_invalid_similarity_negative(self):
with pytest.raises(ValueError, match="similarity must be between 0 and 100"):
DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=-1,
)
def test_invalid_similarity_over_100(self):
with pytest.raises(ValueError, match="similarity must be between 0 and 100"):
DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=101,
)
def test_similarity_boundary_zero(self):
seg = DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=0,
)
assert seg.similarity == 0
def test_similarity_boundary_100(self):
seg = DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=100,
)
assert seg.similarity == 100
class TestDuplicationRecordCreate:
"""DuplicationRecord.create 测试"""
def test_create_minimal(self):
record = DuplicationRecord.create(
user_id="user123",
filename="test.mp4",
file_size=1024000,
storage_key="oss://bucket/test.mp4",
)
assert record.id is not None
assert len(record.id) == 32
assert record.user_id == "user123"
assert record.filename == "test.mp4"
assert record.file_size == 1024
assert record.storage_key == "oss/key/test.mp4"
assert record.duration_seconds == 30.0
assert record.file_size == 1024000
assert record.storage_key == "oss://bucket/test.mp4"
assert record.status == "pending"
assert record.duplicate_rate is None
assert record.duplicate_count == 0
assert record.segments == []
assert record.duration_seconds == 0.0
assert record.created_at is not None
assert record.updated_at is not None
def test_create_with_duration(self):
record = DuplicationRecord.create(
user_id="u1",
filename="video.mp4",
file_size=5000,
storage_key="key",
duration_seconds=120.5,
)
assert record.duration_seconds == 120.5
def test_create_strips_whitespace(self):
record = DuplicationRecord.create(
user_id=" user456 ",
filename=" my video.mp4 ",
file_size=100,
storage_key="key",
)
assert record.user_id == "user456"
assert record.filename == "my video.mp4"
def test_empty_user_id_raises(self):
with pytest.raises(ValueError, match="user_id cannot be empty"):
DuplicationRecord.create(
user_id=" ",
filename="test.mp4",
file_size=100,
storage_key="key",
)
def test_empty_filename_raises(self):
with pytest.raises(ValueError, match="filename cannot be empty"):
DuplicationRecord.create(
user_id="u1",
filename=" ",
file_size=100,
storage_key="key",
)
def test_zero_file_size_raises(self):
with pytest.raises(ValueError, match="file_size must be positive"):
DuplicationRecord.create(
user_id="u1",
filename="test.mp4",
file_size=0,
storage_key="key",
)
def test_negative_file_size_raises(self):
with pytest.raises(ValueError, match="file_size must be positive"):
DuplicationRecord.create(
user_id="u1",
filename="test.mp4",
file_size=-100,
storage_key="key",
)
class TestDuplicationRecordLifecycle:
"""生命周期状态转换测试"""
def test_mark_processing(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
old_updated = record.updated_at
record.mark_processing()
assert record.status == "processing"
assert record.updated_at >= old_updated
def test_mark_completed(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
segments = [
DuplicateSegment.create(
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=90,
)
]
record.mark_completed(
duplicate_rate=25.5,
duplicate_count=1,
segments=segments,
)
assert record.status == "completed"
assert record.duplicate_rate == 25.5
assert record.duplicate_count == 1
assert len(record.segments) == 1
assert record.error_message == ""
def test_mark_completed_zero_rate(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_completed(duplicate_rate=0.0, duplicate_count=0, segments=[])
assert record.status == "completed"
assert record.duplicate_rate == 0.0
assert record.duplicate_count == 0
assert record.segments == []
def test_mark_completed_100_rate(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_completed(duplicate_rate=100.0, duplicate_count=5, segments=[])
assert record.duplicate_rate == 100.0
def test_mark_completed_invalid_rate_negative(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
record.mark_completed(duplicate_rate=-1, duplicate_count=0, segments=[])
def test_mark_completed_invalid_rate_over_100(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
record.mark_completed(duplicate_rate=101, duplicate_count=0, segments=[])
def test_mark_failed(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
record.mark_failed("网络超时")
assert record.status == "failed"
assert record.error_message == "网络超时"
assert record.duplicate_rate is None
def test_mark_failed_from_pending(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_failed("文件损坏")
assert record.status == "failed"
assert record.error_message == "文件损坏"
class TestDuplicationRecordRetry:
"""重试逻辑测试"""
def test_can_retry_failed(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_failed("error")
assert record.can_retry() is True
def test_cannot_retry_pending(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
assert record.can_retry() is False
def test_cannot_retry_processing(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
assert record.can_retry() is False
def test_cannot_retry_completed(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_completed(duplicate_rate=10, duplicate_count=1, segments=[])
assert record.can_retry() is False
def test_reset_for_retry(self):
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
segments = [
DuplicateSegment.create(
source_start=0,
source_end=5,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=5,
similarity=80,
)
]
record.mark_completed(duplicate_rate=30, duplicate_count=1, segments=segments)
record.status = "failed"
record.error_message = "something wrong"
record.video_fingerprint = {"hash": "abc"}
record.reset_for_retry()
assert record.status == "pending"
assert record.duplicate_rate is None
assert record.duplicate_count == 0
assert record.error_message == ""
assert record.segments == []
assert record.video_fingerprint is None
assert record.id # 自动生成 ID
assert record.updated_at is not None
def test_create_with_default_duration(self):
record = DuplicationRecord.create(
user_id="user-1",
filename="test.mp4",
file_size=1024,
storage_key="oss/key",
)
assert record.duration_seconds == 0.0
def test_create_empty_user_id_raises(self):
with pytest.raises(ValueError, match="user_id"):
DuplicationRecord.create(
user_id="",
filename="test.mp4",
file_size=1024,
storage_key="oss/key",
)
def test_create_whitespace_user_id_raises(self):
with pytest.raises(ValueError, match="user_id"):
DuplicationRecord.create(
user_id=" ",
filename="test.mp4",
file_size=1024,
storage_key="oss/key",
)
def test_create_empty_filename_raises(self):
with pytest.raises(ValueError, match="filename"):
DuplicationRecord.create(
user_id="user-1",
filename="",
file_size=1024,
storage_key="oss/key",
)
def test_create_zero_file_size_raises(self):
with pytest.raises(ValueError, match="file_size"):
DuplicationRecord.create(
user_id="user-1",
filename="test.mp4",
file_size=0,
storage_key="oss/key",
)
def test_create_negative_file_size_raises(self):
with pytest.raises(ValueError, match="file_size"):
DuplicationRecord.create(
user_id="user-1",
filename="test.mp4",
file_size=-100,
storage_key="oss/key",
)
class TestDuplicationRecordStateTransitions:
"""状态转换测试。"""
@pytest.fixture
def record(self):
return DuplicationRecord.create(
user_id="user-1",
filename="test.mp4",
file_size=1024,
storage_key="oss/key",
)
def test_mark_processing(self, record):
record.mark_processing()
assert record.status == "processing"
def test_mark_completed_success(self, record):
record.mark_processing()
segments = [
DuplicateSegment.create(
source_start=0.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="existing.mp4",
matched_start=0.0,
matched_end=5.0,
similarity=92.5,
)
]
record.mark_completed(duplicate_rate=15.0, duplicate_count=1, segments=segments)
assert record.status == "completed"
assert record.duplicate_rate == 15.0
assert record.duplicate_count == 1
assert len(record.segments) == 1
def test_mark_completed_invalid_rate_raises(self, record):
record.mark_processing()
with pytest.raises(ValueError, match="duplicate_rate"):
record.mark_completed(duplicate_rate=101.0, duplicate_count=0, segments=[])
def test_mark_completed_negative_rate_raises(self, record):
record.mark_processing()
with pytest.raises(ValueError, match="duplicate_rate"):
record.mark_completed(duplicate_rate=-1.0, duplicate_count=0, segments=[])
def test_mark_failed(self, record):
record.mark_processing()
record.mark_failed("处理超时")
assert record.status == "failed"
assert record.error_message == "处理超时"
class TestDuplicationRecordRetry:
"""can_retry() 和 reset_for_retry() 测试。"""
@pytest.fixture
def record(self):
return DuplicationRecord.create(
user_id="user-1",
filename="test.mp4",
file_size=1024,
storage_key="oss/key",
)
def test_mark_failed_sets_status_and_error(self, record):
record.mark_processing()
record.mark_failed("处理失败")
assert record.status == "failed"
assert record.error_message == "处理失败"
def test_mark_failed_updates_timestamp(self, record):
old_updated = record.updated_at
record.mark_processing()
record.mark_failed("错误")
assert record.updated_at >= old_updated
def test_failed_record_preserves_result_fields(self, record):
"""mark_failed 不改变 duplicate_rate 等结果字段(由 use case 层重置)。"""
record.mark_processing()
record.mark_completed(duplicate_rate=10.0, duplicate_count=1, segments=[])
record.mark_failed("重试失败")
assert record.status == "failed"
assert record.error_message == "重试失败"
assert record.duplicate_rate == 10.0
class TestDuplicateSegmentCreate:
"""DuplicateSegment.create() 工厂方法测试。"""
def test_create_success(self):
seg = DuplicateSegment.create(
source_start=1.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="existing.mp4",
matched_start=2.0,
matched_end=6.0,
similarity=85.5,
)
assert seg.source_start == 1.0
assert seg.source_end == 5.0
assert seg.matched_video_id == "vid-1"
assert seg.matched_video_name == "existing.mp4"
assert seg.matched_start == 2.0
assert seg.matched_end == 6.0
assert seg.similarity == 85.5
assert seg.id # 自动生成 ID
def test_create_negative_source_start_raises(self):
with pytest.raises(ValueError, match="invalid source segment range"):
DuplicateSegment.create(
source_start=-1.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="v.mp4",
matched_start=0.0,
matched_end=5.0,
similarity=80.0,
)
def test_create_source_end_le_start_raises(self):
with pytest.raises(ValueError, match="invalid source segment range"):
DuplicateSegment.create(
source_start=5.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="v.mp4",
matched_start=0.0,
matched_end=5.0,
similarity=80.0,
)
def test_create_negative_matched_start_raises(self):
with pytest.raises(ValueError, match="invalid matched segment range"):
DuplicateSegment.create(
source_start=0.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="v.mp4",
matched_start=-1.0,
matched_end=5.0,
similarity=80.0,
)
def test_create_matched_end_le_start_raises(self):
with pytest.raises(ValueError, match="invalid matched segment range"):
DuplicateSegment.create(
source_start=0.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="v.mp4",
matched_start=2.0,
matched_end=1.0,
similarity=80.0,
)
def test_create_similarity_out_of_range_raises(self):
with pytest.raises(ValueError, match="similarity"):
DuplicateSegment.create(
source_start=0.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="v.mp4",
matched_start=0.0,
matched_end=5.0,
similarity=101.0,
)
def test_create_negative_similarity_raises(self):
with pytest.raises(ValueError, match="similarity"):
DuplicateSegment.create(
source_start=0.0,
source_end=5.0,
matched_video_id="vid-1",
matched_video_name="v.mp4",
matched_start=0.0,
matched_end=5.0,
similarity=-1.0,
)
def test_reset_for_retry_from_pending(self):
"""即使从 pending 也能重置(调用方负责判断 can_retry)"""
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.reset_for_retry()
assert record.status == "pending"
assert record.duplicate_count == 0
+154
View File
@@ -0,0 +1,154 @@
"""
EditTemplate 剪辑模板领域模型单元测试
"""
import pytest
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.editing_mode import EditingMode
class TestEditTemplateStatus:
"""EditTemplateStatus 枚举测试"""
def test_status_values(self):
assert EditTemplateStatus.ACTIVE == "active"
assert EditTemplateStatus.INACTIVE == "inactive"
class TestEditTemplateCreate:
"""create 工厂方法测试"""
def test_create_minimal(self):
template = EditTemplate.create(name="测试模板")
assert template.id is not None
assert len(template.id) == 32
assert template.name == "测试模板"
assert template.description == ""
assert template.template_type == "default"
assert template.editing_mode == EditingMode.ONE_TAKE.value
assert template.config == {}
assert template.preview_url == ""
assert template.sort_weight == 0
assert template.status == EditTemplateStatus.ACTIVE
assert template.version == 1
assert template.created_at is not None
assert template.updated_at is not None
def test_create_with_all_fields(self):
template = EditTemplate.create(
name="口播模板",
description="适合口播类视频",
template_type="talking_head",
editing_mode=EditingMode.VOICE_OVER.value,
config={"aspect_ratio": "9:16", "duration": 60},
preview_url="https://example.com/preview.mp4",
sort_weight=100,
status=EditTemplateStatus.INACTIVE,
version=3,
)
assert template.name == "口播模板"
assert template.description == "适合口播类视频"
assert template.template_type == "talking_head"
assert template.editing_mode == EditingMode.VOICE_OVER.value
assert template.config == {"aspect_ratio": "9:16", "duration": 60}
assert template.preview_url == "https://example.com/preview.mp4"
assert template.sort_weight == 100
assert template.status == EditTemplateStatus.INACTIVE
assert template.version == 3
def test_create_strips_whitespace(self):
template = EditTemplate.create(
name=" 我的模板 ",
description=" 描述 ",
template_type=" ",
editing_mode=" ",
preview_url=" url ",
)
assert template.name == "我的模板"
assert template.description == "描述"
assert template.template_type == "default" # 空字符串回退到default
assert template.editing_mode == EditingMode.ONE_TAKE.value # 空回退
assert template.preview_url == "url"
def test_create_empty_name_raises(self):
with pytest.raises(ValueError, match="模板名称不能为空"):
EditTemplate.create(name=" ")
def test_create_invalid_editing_mode_raises(self):
with pytest.raises(ValueError, match="无效的 editing_mode"):
EditTemplate.create(name="测试", editing_mode="invalid_mode")
def test_create_config_none(self):
template = EditTemplate.create(name="test", config=None)
assert template.config == {}
# 两个模板不共享同一个 dict
t2 = EditTemplate.create(name="test2")
assert template.config is not t2.config
def test_create_all_editing_modes(self):
"""所有合法 editing_mode 都能创建"""
for mode in EditingMode:
template = EditTemplate.create(name=f"test_{mode.value}", editing_mode=mode.value)
assert template.editing_mode == mode.value
class TestEditTemplateActivateDeactivate:
"""激活/停用测试"""
def test_activate_from_inactive(self):
template = EditTemplate.create(name="test", status=EditTemplateStatus.INACTIVE)
old_updated = template.updated_at
template.activate()
assert template.status == EditTemplateStatus.ACTIVE
assert template.updated_at >= old_updated
def test_activate_from_active(self):
template = EditTemplate.create(name="test")
template.activate()
assert template.status == EditTemplateStatus.ACTIVE
def test_deactivate_from_active(self):
template = EditTemplate.create(name="test")
template.deactivate()
assert template.status == EditTemplateStatus.INACTIVE
def test_deactivate_from_inactive(self):
template = EditTemplate.create(name="test", status=EditTemplateStatus.INACTIVE)
template.deactivate()
assert template.status == EditTemplateStatus.INACTIVE
class TestEditTemplateIsActive:
"""is_active 属性测试"""
def test_is_active_true(self):
template = EditTemplate.create(name="test")
assert template.is_active is True
def test_is_active_false(self):
template = EditTemplate.create(name="test", status=EditTemplateStatus.INACTIVE)
assert template.is_active is False
class TestEditTemplateBumpVersion:
"""版本号递增测试"""
def test_bump_version_increments(self):
template = EditTemplate.create(name="test", version=1)
old_updated = template.updated_at
template.bump_version()
assert template.version == 2
assert template.updated_at >= old_updated
def test_bump_version_multiple(self):
template = EditTemplate.create(name="test", version=5)
template.bump_version()
template.bump_version()
assert template.version == 7
def test_bump_version_updates_timestamp(self):
template = EditTemplate.create(name="test")
old_updated = template.updated_at
template.bump_version()
assert template.updated_at > old_updated or template.updated_at == old_updated
+347
View File
@@ -0,0 +1,347 @@
"""
Feature Flags 基础设施层单元测试
"""
import pytest
from packages.infrastructure.feature_flags import (
FeatureFlag,
FeatureFlags,
FeatureScope,
feature_flags,
)
class TestFeatureFlag:
"""FeatureFlag 单个开关测试"""
def test_default_values(self):
"""测试默认值"""
flag = FeatureFlag(name="test_flag")
assert flag.name == "test_flag"
assert flag.description == ""
assert flag.global_enabled is True
assert flag.plan_overrides == {}
assert flag.user_overrides == {}
def test_is_enabled_global_true(self):
"""测试全局启用"""
flag = FeatureFlag(name="test", global_enabled=True)
assert flag.is_enabled() is True
def test_is_enabled_global_false(self):
"""测试全局禁用"""
flag = FeatureFlag(name="test", global_enabled=False)
assert flag.is_enabled() is False
def test_is_enabled_plan_override(self):
"""测试套餐级别覆盖"""
flag = FeatureFlag(
name="test",
global_enabled=True,
plan_overrides={"free": False, "premium": True},
)
# free 套餐被覆盖为 False
assert flag.is_enabled(user_plan="free") is False
# premium 套餐覆盖为 True
assert flag.is_enabled(user_plan="premium") is True
# 没有覆盖的套餐用全局值
assert flag.is_enabled(user_plan="basic") is True
def test_is_enabled_user_override_priority(self):
"""测试用户白名单优先级最高"""
flag = FeatureFlag(
name="test",
global_enabled=False,
plan_overrides={"premium": True},
user_overrides={"user-1": True, "user-2": False},
)
# 用户白名单 True → 全局禁用也能启用
assert flag.is_enabled(user_plan="free", user_id="user-1") is True
# 用户白名单 False → premium 套餐也禁用
assert flag.is_enabled(user_plan="premium", user_id="user-2") is False
# 没有用户白名单 → 走套餐级别
assert flag.is_enabled(user_plan="premium", user_id="user-3") is True
def test_is_enabled_no_params(self):
"""测试不传任何参数时使用全局值"""
flag = FeatureFlag(name="test", global_enabled=True)
assert flag.is_enabled() is True
def test_is_enabled_empty_strings_treated_as_none(self):
"""测试空字符串 user_id/user_plan 不触发覆盖"""
flag = FeatureFlag(
name="test",
global_enabled=True,
plan_overrides={"free": False},
user_overrides={"": True}, # 空字符串key
)
# 空字符串 user_id 被当作 falsy,不走用户白名单分支
assert flag.is_enabled(user_id="", user_plan="") is True
def test_plan_override_does_not_affect_other_plans(self):
"""测试套餐覆盖不影响其他套餐"""
flag = FeatureFlag(
name="test",
global_enabled=True,
plan_overrides={"free": False},
)
assert flag.is_enabled(user_plan="free") is False
assert flag.is_enabled(user_plan="basic") is True
assert flag.is_enabled(user_plan="premium") is True
def test_user_override_can_enable_for_disabled_plan(self):
"""测试用户白名单可以为被禁用的套餐用户单独启用"""
flag = FeatureFlag(
name="test",
global_enabled=False,
plan_overrides={"premium": True},
user_overrides={"special-user": True},
)
# free 套餐用户 + 白名单 → 启用
assert flag.is_enabled(user_plan="free", user_id="special-user") is True
def test_user_override_can_disable_for_enabled_plan(self):
"""测试用户白名单可以为启用套餐的用户单独禁用"""
flag = FeatureFlag(
name="test",
global_enabled=True,
user_overrides={"bad-user": False},
)
assert flag.is_enabled(user_id="bad-user") is False
class TestFeatureScope:
"""FeatureScope 常量测试"""
def test_scope_constants(self):
"""测试所有常量存在"""
assert FeatureScope.AI_VOICE_GENERATION == "ai_voice_generation"
assert FeatureScope.DEDUPLICATION_REPORT == "deduplication_report"
assert FeatureScope.BATCH_EXPORT == "batch_export"
assert FeatureScope.MULTI_PLATFORM_OUTPUT == "multi_platform_output"
assert FeatureScope.RECIPE_REUSE == "recipe_reuse"
class TestFeatureFlags:
"""FeatureFlags 管理器测试"""
@pytest.fixture
def flags(self):
"""创建新的 FeatureFlags 实例(不影响全局单例)"""
return FeatureFlags()
# ===== 初始化 =====
def test_default_flags_exist(self, flags):
"""测试默认 flags 已注册"""
all_flags = flags.list_flags()
assert FeatureScope.AI_VOICE_GENERATION in all_flags
assert FeatureScope.DEDUPLICATION_REPORT in all_flags
assert FeatureScope.BATCH_EXPORT in all_flags
assert FeatureScope.MULTI_PLATFORM_OUTPUT in all_flags
assert FeatureScope.RECIPE_REUSE in all_flags
def test_default_ai_voice_generation(self, flags):
"""测试 AI 配音功能默认配置"""
# free 套餐不可用
assert flags.is_enabled("ai_voice_generation", user_plan="free") is False
# basic 套餐可用
assert flags.is_enabled("ai_voice_generation", user_plan="basic") is True
# premium 套餐可用
assert flags.is_enabled("ai_voice_generation", user_plan="premium") is True
def test_default_deduplication_report(self, flags):
"""测试去重报告默认配置(仅 premium)"""
assert flags.is_enabled("deduplication_report", user_plan="free") is False
assert flags.is_enabled("deduplication_report", user_plan="basic") is False
assert flags.is_enabled("deduplication_report", user_plan="premium") is True
def test_default_multi_platform_output(self, flags):
"""测试多平台输出默认配置(仅 premium)"""
assert flags.is_enabled("multi_platform_output", user_plan="free") is False
assert flags.is_enabled("multi_platform_output", user_plan="basic") is False
assert flags.is_enabled("multi_platform_output", user_plan="premium") is True
def test_default_batch_export(self, flags):
"""测试批量导出默认配置"""
assert flags.is_enabled("batch_export", user_plan="free") is False
assert flags.is_enabled("batch_export", user_plan="basic") is True
assert flags.is_enabled("batch_export", user_plan="premium") is True
def test_default_recipe_reuse(self, flags):
"""测试配方复用默认配置"""
assert flags.is_enabled("recipe_reuse", user_plan="free") is False
assert flags.is_enabled("recipe_reuse", user_plan="basic") is True
assert flags.is_enabled("recipe_reuse", user_plan="premium") is True
# ===== 注册新 flag =====
def test_register_new_flag(self, flags):
"""测试注册新的 feature flag"""
new_flag = FeatureFlag(name="new_feature", description="新功能", global_enabled=False)
flags.register(new_flag)
assert flags.get("new_feature") is not None
assert flags.get("new_feature").description == "新功能"
assert flags.is_enabled("new_feature") is False
def test_register_overwrites_existing(self, flags):
"""测试注册同名 flag 会覆盖"""
flag1 = FeatureFlag(name="test", global_enabled=True, description="v1")
flags.register(flag1)
assert flags.get("test").description == "v1"
flag2 = FeatureFlag(name="test", global_enabled=False, description="v2")
flags.register(flag2)
assert flags.get("test").description == "v2"
assert flags.is_enabled("test") is False
# ===== get 方法 =====
def test_get_existing_flag(self, flags):
"""测试获取存在的 flag"""
flag = flags.get("ai_voice_generation")
assert flag is not None
assert flag.name == "ai_voice_generation"
def test_get_nonexistent_flag(self, flags):
"""测试获取不存在的 flag 返回 None"""
assert flags.get("nonexistent") is None
# ===== is_enabled 方法 =====
def test_is_enabled_nonexistent_flag_returns_false(self, flags):
"""测试不存在的 flag 返回 False"""
assert flags.is_enabled("nonexistent_flag") is False
def test_is_enabled_without_plan_or_user(self, flags):
"""测试不传套餐和用户ID"""
assert flags.is_enabled("ai_voice_generation") is True
# ===== set_global =====
def test_set_global_enable(self, flags):
"""测试设置全局启用"""
flags.set_global("ai_voice_generation", enabled=False)
assert flags.is_enabled("ai_voice_generation", user_plan="premium") is False
def test_set_global_disable(self, flags):
"""测试设置全局禁用"""
flags.set_global("deduplication_report", enabled=False)
assert flags.is_enabled("deduplication_report", user_plan="premium") is False
def test_set_global_nonexistent_raises(self, flags):
"""测试设置不存在的 flag 抛出异常"""
with pytest.raises(KeyError, match="not found"):
flags.set_global("nonexistent", enabled=True)
# ===== set_plan_override =====
def test_set_plan_override(self, flags):
"""测试设置套餐覆盖"""
# 先确认 basic 套餐默认是去重报告禁用
assert flags.is_enabled("deduplication_report", user_plan="basic") is False
flags.set_plan_override("deduplication_report", "basic", True)
assert flags.is_enabled("deduplication_report", user_plan="basic") is True
def test_set_plan_override_nonexistent_raises(self, flags):
"""测试设置不存在 flag 的套餐覆盖抛出异常"""
with pytest.raises(KeyError, match="not found"):
flags.set_plan_override("nonexistent", "free", True)
# ===== set_user_override =====
def test_set_user_override_enable(self, flags):
"""测试设置用户白名单启用"""
assert flags.is_enabled("deduplication_report", user_plan="free", user_id="user-1") is False
flags.set_user_override("deduplication_report", "user-1", True)
assert flags.is_enabled("deduplication_report", user_plan="free", user_id="user-1") is True
def test_set_user_override_disable(self, flags):
"""测试设置用户白名单禁用"""
assert flags.is_enabled("batch_export", user_plan="premium", user_id="user-2") is True
flags.set_user_override("batch_export", "user-2", False)
assert flags.is_enabled("batch_export", user_plan="premium", user_id="user-2") is False
def test_set_user_override_nonexistent_raises(self, flags):
"""测试设置不存在 flag 的用户覆盖抛出异常"""
with pytest.raises(KeyError, match="not found"):
flags.set_user_override("nonexistent", "user-1", True)
# ===== list_flags =====
def test_list_flags_returns_copy(self, flags):
"""测试 list_flags 返回副本"""
all_flags = flags.list_flags()
all_flags["fake"] = FeatureFlag(name="fake")
# 原注册表不应被修改
assert "fake" not in flags.list_flags()
def test_list_flags_count(self, flags):
"""测试默认 flag 数量"""
all_flags = flags.list_flags()
assert len(all_flags) == 5 # 5 个默认 flag
# ===== get_enabled_for_plan =====
def test_get_enabled_for_free_plan(self, flags):
"""测试 free 套餐启用的功能"""
enabled = flags.get_enabled_for_plan("free")
# free 套餐应该只有 0 个默认启用的功能?不对,让我看看...
# 所有5个默认功能 free 套餐都是 False 吗?
# AI_VOICE_GENERATION: free=False
# DEDUPLICATION_REPORT: free=False, basic=False
# BATCH_EXPORT: free=False
# MULTI_PLATFORM_OUTPUT: free=False, basic=False
# RECIPE_REUSE: free=False
# 所以 free 套餐一个都没有?
assert len(enabled) == 0
def test_get_enabled_for_premium_plan(self, flags):
"""测试 premium 套餐启用的功能"""
enabled = flags.get_enabled_for_plan("premium")
# premium 套餐所有功能都应该启用
assert len(enabled) == 5
assert "ai_voice_generation" in enabled
assert "deduplication_report" in enabled
assert "batch_export" in enabled
assert "multi_platform_output" in enabled
assert "recipe_reuse" in enabled
def test_get_enabled_for_basic_plan(self, flags):
"""测试 basic 套餐启用的功能"""
enabled = flags.get_enabled_for_plan("basic")
# basic: ai_voice=True, dedup=False, batch=True, multi=False, recipe=True
assert "ai_voice_generation" in enabled
assert "deduplication_report" not in enabled
assert "batch_export" in enabled
assert "multi_platform_output" not in enabled
assert "recipe_reuse" in enabled
assert len(enabled) == 3
class TestGlobalSingleton:
"""全局单例测试"""
def test_global_singleton_exists(self):
"""测试全局单例存在"""
assert feature_flags is not None
assert isinstance(feature_flags, FeatureFlags)
def test_global_singleton_has_defaults(self):
"""测试全局单例有默认配置"""
assert feature_flags.get("ai_voice_generation") is not None
assert feature_flags.get("deduplication_report") is not None
def test_global_singleton_independent_from_new_instance(self):
"""测试全局单例与新实例相互独立"""
new_flags = FeatureFlags()
new_flags.set_global("ai_voice_generation", False)
# 全局单例不应受影响
assert feature_flags.is_enabled("ai_voice_generation") is True
+512
View File
@@ -0,0 +1,512 @@
"""
GenerationTask 领域模型单元测试
"""
import json
import time
from datetime import datetime, timezone
import pytest
from packages.domain.generation_task import (
TERMINAL_STATUSES,
GenerationTask,
GenerationTaskStatus,
)
class TestGenerationTaskStatus:
"""GenerationTaskStatus 枚举测试"""
def test_status_values(self):
assert GenerationTaskStatus.PENDING == "pending"
assert GenerationTaskStatus.RUNNING == "running"
assert GenerationTaskStatus.COMPLETED == "completed"
assert GenerationTaskStatus.FAILED == "failed"
assert GenerationTaskStatus.CANCELLED == "cancelled"
def test_terminal_statuses(self):
assert GenerationTaskStatus.COMPLETED in TERMINAL_STATUSES
assert GenerationTaskStatus.FAILED in TERMINAL_STATUSES
assert GenerationTaskStatus.CANCELLED in TERMINAL_STATUSES
assert GenerationTaskStatus.PENDING not in TERMINAL_STATUSES
assert GenerationTaskStatus.RUNNING not in TERMINAL_STATUSES
class TestGenerationTaskCreate:
"""GenerationTask 创建测试"""
def test_create_basic(self):
"""测试基本创建"""
task = GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
)
assert task.id is not None
assert task.project_id == "proj-123"
assert task.asset_library_id == "lib-456"
assert task.status == GenerationTaskStatus.PENDING
assert task.progress == 0.0
assert task.result_count == 0
assert task.retry_count == 0
assert task.asset_ids == []
assert task.title_ids == []
assert task.voice_ids == []
assert task.logs == "[]"
assert task.created_at is not None
assert task.started_at is None
assert task.completed_at is None
def test_create_with_all_params(self):
"""测试创建带所有参数"""
task = GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
strategy_id="strat-789",
voice_library_id="voice-lib-001",
template_id="tmpl-001",
asset_ids=["asset-1", "asset-2"],
title_ids=["title-1"],
voice_ids=["voice-1"],
created_by_user_id="user-001",
source_edit_plan_id="plan-001",
asset_select_mode="smart",
batch_id="batch-001",
video_title="测试视频",
auto_retry_enabled=True,
auto_retry_max=3,
)
assert task.strategy_id == "strat-789"
assert task.voice_library_id == "voice-lib-001"
assert task.template_id == "tmpl-001"
assert task.asset_ids == ["asset-1", "asset-2"]
assert task.title_ids == ["title-1"]
assert task.voice_ids == ["voice-1"]
assert task.created_by_user_id == "user-001"
assert task.source_edit_plan_id == "plan-001"
assert task.asset_select_mode == "smart"
assert task.batch_id == "batch-001"
assert task.video_title == "测试视频"
assert task.auto_retry_enabled is True
assert task.auto_retry_max == 3
def test_create_with_template_id_only(self):
"""测试只提供 template_id 不提供 project_id(应通过校验)"""
task = GenerationTask.create(
project_id="",
asset_library_id="lib-456",
template_id="tmpl-001",
)
assert task.project_id == ""
assert task.template_id == "tmpl-001"
def test_create_with_asset_ids_only(self):
"""测试只提供 asset_ids 不提供 asset_library_id(应通过校验)"""
task = GenerationTask.create(
project_id="proj-123",
asset_library_id="",
asset_ids=["asset-1", "asset-2"],
)
assert task.asset_library_id == ""
assert task.asset_ids == ["asset-1", "asset-2"]
def test_create_missing_project_and_template(self):
"""测试 project_id 和 template_id 都为空"""
with pytest.raises(ValueError, match="至少需要提供一个"):
GenerationTask.create(
project_id="",
asset_library_id="lib-456",
)
def test_create_missing_library_and_assets(self):
"""测试 asset_library_id 和素材列表都为空"""
with pytest.raises(ValueError, match="至少需要提供一个"):
GenerationTask.create(
project_id="proj-123",
asset_library_id="",
)
def test_create_strips_strings(self):
"""测试字符串字段被 strip"""
task = GenerationTask.create(
project_id=" proj-123 ",
asset_library_id=" lib-456 ",
strategy_id=" strat-789 ",
template_id=" tmpl-001 ",
video_title=" 测试视频 ",
)
assert task.project_id == "proj-123"
assert task.asset_library_id == "lib-456"
assert task.strategy_id == "strat-789"
assert task.template_id == "tmpl-001"
assert task.video_title == "测试视频"
def test_create_default_empty_lists(self):
"""测试 None 列表默认化为空列表"""
task = GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
asset_ids=None,
title_ids=None,
voice_ids=None,
)
assert task.asset_ids == []
assert task.title_ids == []
assert task.voice_ids == []
def test_create_copies_list(self):
"""测试列表被复制(不共享引用)"""
assets = ["a", "b"]
task = GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
asset_ids=assets,
)
assets.append("c")
assert task.asset_ids == ["a", "b"]
class TestGenerationTaskStatusProperties:
"""状态属性测试"""
@pytest.fixture
def new_task(self):
return GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
)
def test_is_terminal_pending(self, new_task):
assert not new_task.is_terminal
def test_is_terminal_running(self, new_task):
new_task.mark_processing()
assert not new_task.is_terminal
def test_is_terminal_completed(self, new_task):
new_task.mark_processing()
new_task.mark_completed()
assert new_task.is_terminal
def test_is_terminal_failed(self, new_task):
new_task.mark_processing()
new_task.mark_failed("error")
assert new_task.is_terminal
def test_is_terminal_cancelled(self, new_task):
new_task.mark_cancelled()
assert new_task.is_terminal
def test_is_completed(self, new_task):
assert not new_task.is_completed
new_task.mark_processing()
assert not new_task.is_completed
new_task.mark_completed()
assert new_task.is_completed
def test_is_failed(self, new_task):
assert not new_task.is_failed
new_task.mark_processing()
new_task.mark_failed("error")
assert new_task.is_failed
def test_is_running(self, new_task):
assert not new_task.is_running
new_task.mark_processing()
assert new_task.is_running
class TestGenerationTaskTransitions:
"""状态转换测试"""
@pytest.fixture
def new_task(self):
return GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
)
# ===== Pending → Running =====
def test_pending_to_running(self, new_task):
"""测试 pending → running"""
new_task.mark_processing()
assert new_task.status == GenerationTaskStatus.RUNNING
assert new_task.started_at is not None
assert new_task.error_message == ""
def test_pending_to_running_clears_error(self, new_task):
"""测试 processing 清除错误信息"""
# pending 可以直接 failed(跟 job 不同)
new_task.mark_failed("some error")
# 然后重试回 pending
new_task.mark_pending_from_failed()
new_task.mark_processing()
assert new_task.error_message == ""
# ===== Pending → Failed =====
def test_pending_to_failed(self, new_task):
"""测试 pending → failed(可以直接失败)"""
new_task.mark_failed("task failed before start")
assert new_task.status == GenerationTaskStatus.FAILED
assert new_task.error_message == "task failed before start"
assert new_task.completed_at is not None
def test_pending_to_failed_with_error_info(self, new_task):
"""测试 pending → failed 带 error_info"""
error_info = {"error_type": "ValidationError", "stage": "init"}
new_task.mark_failed("validation failed", error_info=error_info)
assert new_task.error_info == error_info
def test_pending_to_failed_default_error_info(self, new_task):
"""测试 pending → failed 默认 error_info"""
new_task.mark_failed("some error")
assert new_task.error_info["error_type"] == "UnknownError"
assert new_task.error_info["message"] == "some error"
assert "failed_at" in new_task.error_info
# ===== Pending → Cancelled =====
def test_pending_to_cancelled(self, new_task):
"""测试 pending → cancelled"""
new_task.mark_cancelled()
assert new_task.status == GenerationTaskStatus.CANCELLED
assert new_task.completed_at is not None
# ===== Running → Completed =====
def test_running_to_completed(self, new_task):
"""测试 running → completed"""
new_task.mark_processing()
new_task.mark_completed(result_count=3)
assert new_task.status == GenerationTaskStatus.COMPLETED
assert new_task.progress == 100.0
assert new_task.result_count == 3
assert new_task.error_message == ""
assert new_task.completed_at is not None
def test_running_to_completed_default_count(self, new_task):
"""测试 running → completed 默认 result_count=1"""
new_task.mark_processing()
new_task.mark_completed()
assert new_task.result_count == 1
# ===== Running → Failed =====
def test_running_to_failed(self, new_task):
"""测试 running → failed"""
new_task.mark_processing()
new_task.mark_failed("render timeout")
assert new_task.status == GenerationTaskStatus.FAILED
assert new_task.error_message == "render timeout"
assert new_task.completed_at is not None
# ===== Running → Cancelled =====
def test_running_to_cancelled(self, new_task):
"""测试 running → cancelled"""
new_task.mark_processing()
new_task.mark_cancelled()
assert new_task.status == GenerationTaskStatus.CANCELLED
assert new_task.completed_at is not None
# ===== Failed → Pending (重试) =====
def test_failed_to_pending_retry(self, new_task):
"""测试 failed → pending(重试)"""
new_task.mark_processing()
new_task.mark_failed("error", error_info={"error_type": "TimeoutError"})
assert new_task.retry_count == 0
new_task.mark_pending_from_failed()
assert new_task.status == GenerationTaskStatus.PENDING
assert new_task.retry_count == 1
assert new_task.error_message == ""
assert new_task.error_info == {}
assert new_task.started_at is None
assert new_task.completed_at is None
assert new_task.progress == 0.0
assert new_task.result_count == 0
def test_failed_to_pending_multiple_retries(self, new_task):
"""测试多次重试"""
for i in range(3):
new_task.mark_processing()
new_task.mark_failed(f"error {i}")
new_task.mark_pending_from_failed()
assert new_task.retry_count == 3
assert new_task.status == GenerationTaskStatus.PENDING
def test_mark_pending_from_failed_wrong_status(self, new_task):
"""测试非 failed 状态调用 mark_pending_from_failed 报错"""
with pytest.raises(ValueError, match="只有 failed 状态"):
new_task.mark_pending_from_failed() # pending 状态
# ===== 非法状态转换 =====
def test_invalid_completed_to_running(self, new_task):
"""测试 completed → running 非法"""
new_task.mark_processing()
new_task.mark_completed()
with pytest.raises(ValueError, match="非法状态转换"):
new_task.mark_processing()
def test_invalid_cancelled_to_running(self, new_task):
"""测试 cancelled → running 非法"""
new_task.mark_cancelled()
with pytest.raises(ValueError, match="非法状态转换"):
new_task.mark_processing()
def test_invalid_completed_to_failed(self, new_task):
"""测试 completed → failed 非法"""
new_task.mark_processing()
new_task.mark_completed()
with pytest.raises(ValueError, match="非法状态转换"):
new_task.mark_failed("error")
def test_invalid_status_string(self, new_task):
"""测试无效状态字符串"""
with pytest.raises(ValueError, match="无效状态"):
new_task.transition_to("invalid_status")
class TestGenerationTaskLogs:
"""日志系统测试"""
@pytest.fixture
def new_task(self):
return GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
)
def test_initial_logs_empty(self, new_task):
"""测试初始日志为空"""
assert new_task.get_logs() == []
assert new_task.logs == "[]"
def test_append_log(self, new_task):
"""测试追加日志"""
new_task.append_log(stage="下载素材", message="开始下载")
logs = new_task.get_logs()
assert len(logs) == 1
assert logs[0]["stage"] == "下载素材"
assert logs[0]["message"] == "开始下载"
assert logs[0]["level"] == "INFO"
assert "ts" in logs[0]
def test_append_log_with_custom_level(self, new_task):
"""测试带自定义日志级别"""
new_task.append_log(stage="渲染", message="渲染失败", level="ERROR")
logs = new_task.get_logs()
assert logs[0]["level"] == "ERROR"
def test_append_log_with_extra_fields(self, new_task):
"""测试带额外字段的日志"""
new_task.append_log(
stage="下载",
message="下载完成",
asset_id="asset-001",
duration=30.5,
)
logs = new_task.get_logs()
assert logs[0]["asset_id"] == "asset-001"
assert logs[0]["duration"] == 30.5
def test_multiple_logs(self, new_task):
"""测试多条日志"""
new_task.append_log(stage="步骤1", message="开始")
new_task.append_log(stage="步骤2", message="进行中")
new_task.append_log(stage="步骤3", message="完成")
logs = new_task.get_logs()
assert len(logs) == 3
assert logs[0]["stage"] == "步骤1"
assert logs[2]["stage"] == "步骤3"
def test_logs_corrupted_json(self, new_task):
"""测试 logs 字段损坏时仍能正常工作"""
new_task.logs = "not valid json {{{"
logs = new_task.get_logs()
assert logs == []
# 追加新日志应该能正常工作
new_task.append_log(stage="test", message="after corruption")
logs = new_task.get_logs()
assert len(logs) == 1
assert logs[0]["message"] == "after corruption"
def test_logs_max_limit(self, new_task):
"""测试日志数量上限"""
for i in range(250):
new_task.append_log(stage="loop", message=f"log {i}")
logs = new_task.get_logs()
assert len(logs) == 200 # _MAX_LOGS
# 应该保留最近的 200 条
assert logs[0]["message"] == "log 50"
assert logs[-1]["message"] == "log 249"
def test_logs_empty_string(self, new_task):
"""测试 logs 为空字符串时返回空列表"""
new_task.logs = ""
assert new_task.get_logs() == []
def test_append_log_preserves_existing(self, new_task):
"""测试追加日志保留已有日志"""
new_task.append_log(stage="first", message="first message")
new_task.append_log(stage="second", message="second message")
logs = new_task.get_logs()
assert len(logs) == 2
assert logs[0]["message"] == "first message"
assert logs[1]["message"] == "second message"
class TestGenerationTaskTimestamps:
"""时间戳测试"""
@pytest.fixture
def new_task(self):
return GenerationTask.create(
project_id="proj-123",
asset_library_id="lib-456",
)
def test_created_at_set(self, new_task):
assert new_task.created_at is not None
assert isinstance(new_task.created_at, datetime)
assert new_task.created_at.tzinfo is not None
def test_mark_processing_sets_started_at(self, new_task):
assert new_task.started_at is None
before = datetime.now(timezone.utc)
time.sleep(0.01)
new_task.mark_processing()
time.sleep(0.01)
after = datetime.now(timezone.utc)
assert before < new_task.started_at < after
def test_mark_completed_sets_completed_at(self, new_task):
new_task.mark_processing()
assert new_task.completed_at is None
new_task.mark_completed()
assert new_task.completed_at is not None
def test_mark_failed_sets_completed_at(self, new_task):
new_task.mark_processing()
assert new_task.completed_at is None
new_task.mark_failed("error")
assert new_task.completed_at is not None
def test_retry_clears_timestamps(self, new_task):
new_task.mark_processing()
new_task.mark_failed("error")
new_task.mark_pending_from_failed()
assert new_task.started_at is None
assert new_task.completed_at is None
+542
View File
@@ -0,0 +1,542 @@
"""
Job 领域模型单元测试
"""
import time
from datetime import datetime, timezone
from unittest.mock import patch
import pytest
from packages.domain.job import (
TERMINAL_STATUSES,
Job,
JobStatus,
JobType,
)
class TestJobType:
"""JobType 枚举测试"""
def test_job_type_values(self):
"""测试所有 JobType 值"""
assert JobType.VIDEO_COMPOSE == "video_compose"
assert JobType.RENDER_EDIT_PLAN == "render_edit_plan"
assert JobType.ASSET_INGEST == "asset_ingest"
assert JobType.CLASSIFICATION == "classification"
assert JobType.VOICE_EXTRACTION == "voice_extraction"
assert JobType.GENERATION == "generation"
def test_job_type_is_string(self):
"""测试 StrEnum 行为"""
assert isinstance(JobType.VIDEO_COMPOSE, str)
assert JobType.VIDEO_COMPOSE == "video_compose"
class TestJobStatus:
"""JobStatus 枚举测试"""
def test_job_status_values(self):
"""测试所有 JobStatus 值"""
assert JobStatus.PENDING == "pending"
assert JobStatus.RUNNING == "running"
assert JobStatus.SUCCESS == "success"
assert JobStatus.FAILED == "failed"
assert JobStatus.CANCELLED == "cancelled"
def test_terminal_statuses(self):
"""测试终态集合"""
assert JobStatus.SUCCESS in TERMINAL_STATUSES
assert JobStatus.FAILED in TERMINAL_STATUSES
assert JobStatus.CANCELLED in TERMINAL_STATUSES
assert JobStatus.PENDING not in TERMINAL_STATUSES
assert JobStatus.RUNNING not in TERMINAL_STATUSES
class TestJobCreate:
"""Job 创建测试"""
def test_create_basic_job(self):
"""测试创建基本任务"""
job = Job.create(
project_id="proj-123",
job_type=JobType.VIDEO_COMPOSE,
)
assert job.id is not None
assert len(job.id) > 0
assert job.project_id == "proj-123"
assert job.job_type == JobType.VIDEO_COMPOSE
assert job.status == JobStatus.PENDING
assert job.progress == 0.0
assert job.payload == {}
assert job.result == {}
assert job.error_message == ""
assert job.retry_count == 0
assert job.max_retries == 3
assert job.created_at is not None
assert job.updated_at is not None
assert job.started_at is None
assert job.completed_at is None
def test_create_with_all_params(self):
"""测试创建带所有参数的任务"""
job = Job.create(
project_id="proj-456",
job_type=JobType.GENERATION,
payload={"key": "value"},
source_id="src-789",
created_by_user_id="user-001",
max_retries=5,
)
assert job.project_id == "proj-456"
assert job.job_type == JobType.GENERATION
assert job.payload == {"key": "value"}
assert job.source_id == "src-789"
assert job.created_by_user_id == "user-001"
assert job.max_retries == 5
def test_create_with_string_job_type(self):
"""测试用字符串创建任务"""
job = Job.create(
project_id="proj-123",
job_type="video_compose",
)
assert job.job_type == JobType.VIDEO_COMPOSE
def test_create_with_invalid_job_type(self):
"""测试无效任务类型"""
with pytest.raises(ValueError, match="不支持的任务类型"):
Job.create(
project_id="proj-123",
job_type="invalid_type",
)
def test_create_empty_project_id(self):
"""测试空 project_id"""
with pytest.raises(ValueError, match="project_id 不能为空"):
Job.create(
project_id="",
job_type=JobType.VIDEO_COMPOSE,
)
def test_create_whitespace_project_id(self):
"""测试空白 project_id 被 strip 后为空"""
with pytest.raises(ValueError, match="project_id 不能为空"):
Job.create(
project_id=" ",
job_type=JobType.VIDEO_COMPOSE,
)
def test_create_strips_strings(self):
"""测试字符串字段被 strip"""
job = Job.create(
project_id=" proj-123 ",
job_type=JobType.VIDEO_COMPOSE,
source_id=" src-456 ",
created_by_user_id=" user-789 ",
)
assert job.project_id == "proj-123"
assert job.source_id == "src-456"
assert job.created_by_user_id == "user-789"
def test_create_default_payload(self):
"""测试 None payload 默认化为空 dict"""
job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE, payload=None)
assert job.payload == {}
def test_create_generates_unique_ids(self):
"""测试每次创建生成不同的 ID"""
job1 = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
job2 = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
assert job1.id != job2.id
def test_create_sets_timestamps(self):
"""测试创建时设置时间戳"""
before = datetime.now(timezone.utc)
time.sleep(0.01)
job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
time.sleep(0.01)
after = datetime.now(timezone.utc)
assert before < job.created_at < after
assert before < job.updated_at < after
class TestJobStateTransitions:
"""Job 状态转换测试"""
@pytest.fixture
def new_job(self):
return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
# ===== Pending → Running =====
def test_pending_to_running(self, new_job):
"""测试 pending → running"""
assert new_job.status == JobStatus.PENDING
new_job.mark_running()
assert new_job.status == JobStatus.RUNNING
assert new_job.started_at is not None
assert new_job.completed_at is None
assert not new_job.is_terminal
def test_pending_to_running_with_stage(self, new_job):
"""测试 pending → running 带阶段描述"""
new_job.mark_running(stage="初始化")
assert new_job.current_stage == "初始化"
# ===== Pending → Success =====
def test_pending_to_success(self, new_job):
"""测试 pending → success(直接成功)"""
new_job.mark_success()
assert new_job.status == JobStatus.SUCCESS
assert new_job.progress == 100.0
assert new_job.current_stage == "完成"
assert new_job.completed_at is not None
assert new_job.is_terminal
def test_pending_to_success_with_result(self, new_job):
"""测试 pending → success 带结果"""
result = {"output_url": "http://example.com/video.mp4"}
new_job.mark_success(result=result)
assert new_job.result == result
# ===== Pending → Cancelled =====
def test_pending_to_cancelled(self, new_job):
"""测试 pending → cancelled"""
new_job.mark_cancelled()
assert new_job.status == JobStatus.CANCELLED
assert new_job.current_stage == "已取消"
assert new_job.is_terminal
# ===== Running → Success =====
def test_running_to_success(self, new_job):
"""测试 running → success"""
new_job.mark_running()
new_job.mark_success()
assert new_job.status == JobStatus.SUCCESS
assert new_job.completed_at is not None
assert new_job.progress == 100.0
assert new_job.is_terminal
def test_running_to_success_preserves_started_at(self, new_job):
"""测试 running → success 保留 started_at"""
new_job.mark_running()
started_at = new_job.started_at
new_job.mark_success()
assert new_job.started_at == started_at
# ===== Running → Failed =====
def test_running_to_failed(self, new_job):
"""测试 running → failed"""
new_job.mark_running()
new_job.mark_failed("Something went wrong")
assert new_job.status == JobStatus.FAILED
assert new_job.error_message == "Something went wrong"
assert new_job.current_stage == "失败"
assert new_job.completed_at is not None
assert new_job.is_terminal
# ===== Running → Cancelled =====
def test_running_to_cancelled(self, new_job):
"""测试 running → cancelled"""
new_job.mark_running()
new_job.mark_cancelled()
assert new_job.status == JobStatus.CANCELLED
assert new_job.is_terminal
# ===== Failed → Pending (Retry) =====
def test_failed_to_pending_retry(self, new_job):
"""测试 failed → pending(重试)"""
new_job.mark_running()
new_job.mark_failed("error")
assert new_job.retry_count == 0
new_job.prepare_retry()
assert new_job.status == JobStatus.PENDING
assert new_job.retry_count == 1
assert new_job.progress == 0.0
assert new_job.error_message == ""
assert new_job.started_at is None
assert new_job.completed_at is None
assert new_job.celery_task_id == ""
assert "第 1 次重试" in new_job.current_stage
def test_retry_up_to_max_retries(self, new_job):
"""测试最多重试 max_retries 次"""
new_job.max_retries = 2
new_job.mark_running()
# 第一次失败重试
new_job.mark_failed("error 1")
assert new_job.is_retryable # 失败后可重试
new_job.prepare_retry()
assert new_job.retry_count == 1
# 第二次失败重试
new_job.mark_running()
new_job.mark_failed("error 2")
assert new_job.is_retryable # retry_count=1 < max_retries=2
new_job.prepare_retry()
assert new_job.retry_count == 2
# 第三次失败后不可重试(retry_count == max_retries
new_job.mark_running()
new_job.mark_failed("error 3")
assert not new_job.is_retryable # 达到上限
with pytest.raises(ValueError, match="任务不可重试"):
new_job.prepare_retry()
def test_retry_not_from_failed(self, new_job):
"""测试非 failed 状态不可重试"""
with pytest.raises(ValueError, match="任务不可重试"):
new_job.prepare_retry() # pending 状态
# ===== 非法状态转换 =====
def test_invalid_transition_success_to_running(self, new_job):
"""测试 success → running 非法"""
new_job.mark_success()
with pytest.raises(ValueError, match="非法状态转换"):
new_job.mark_running()
def test_invalid_transition_cancelled_to_running(self, new_job):
"""测试 cancelled → running 非法"""
new_job.mark_cancelled()
with pytest.raises(ValueError, match="非法状态转换"):
new_job.mark_running()
def test_invalid_transition_pending_to_failed(self, new_job):
"""测试 pending → failed 非法(必须经过 running"""
with pytest.raises(ValueError, match="非法状态转换"):
new_job.mark_failed("test error")
def test_invalid_status_string(self, new_job):
"""测试无效状态字符串"""
with pytest.raises(ValueError, match="无效状态"):
new_job.transition_to("invalid_status")
class TestJobProperties:
"""Job 属性测试"""
@pytest.fixture
def new_job(self):
return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
def test_is_terminal_pending(self, new_job):
"""测试 pending 不是终态"""
assert not new_job.is_terminal
def test_is_terminal_running(self, new_job):
"""测试 running 不是终态"""
new_job.mark_running()
assert not new_job.is_terminal
def test_is_terminal_success(self, new_job):
"""测试 success 是终态"""
new_job.mark_success()
assert new_job.is_terminal
def test_is_terminal_failed(self, new_job):
"""测试 failed 是终态"""
new_job.mark_running()
new_job.mark_failed("error")
assert new_job.is_terminal
def test_is_terminal_cancelled(self, new_job):
"""测试 cancelled 是终态"""
new_job.mark_cancelled()
assert new_job.is_terminal
def test_is_retryable_failed_under_limit(self, new_job):
"""测试失败且未达上限时可重试"""
new_job.mark_running()
new_job.mark_failed("error")
assert new_job.is_retryable
def test_is_retryable_failed_at_limit(self, new_job):
"""测试失败且达上限时不可重试"""
new_job.max_retries = 0
new_job.mark_running()
new_job.mark_failed("error")
assert not new_job.is_retryable
def test_is_retryable_not_failed(self, new_job):
"""测试非失败状态不可重试"""
assert not new_job.is_retryable # pending
new_job.mark_running()
assert not new_job.is_retryable # running
new_job.mark_success()
assert not new_job.is_retryable # success
class TestJobProgress:
"""Job 进度更新测试"""
@pytest.fixture
def running_job(self):
job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
job.mark_running()
return job
def test_update_progress_normal(self, running_job):
"""测试正常更新进度"""
running_job.update_progress(50.0, stage="处理中")
assert running_job.progress == 50.0
assert running_job.current_stage == "处理中"
def test_update_progress_zero(self, running_job):
"""测试更新进度为 0"""
running_job.update_progress(0.0)
assert running_job.progress == 0.0
def test_update_progress_hundred(self, running_job):
"""测试更新进度为 100"""
running_job.update_progress(100.0)
assert running_job.progress == 100.0
def test_update_progress_negative(self, running_job):
"""测试负进度报错"""
with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
running_job.update_progress(-1.0)
def test_update_progress_over_hundred(self, running_job):
"""测试超过 100 的进度报错"""
with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
running_job.update_progress(101.0)
def test_update_progress_without_stage(self, running_job):
"""测试更新进度但不改变阶段"""
running_job.current_stage = "初始阶段"
running_job.update_progress(30.0)
assert running_job.progress == 30.0
assert running_job.current_stage == "初始阶段" # 保留原值
def test_update_progress_updates_updated_at(self, running_job):
"""测试更新进度会更新 updated_at"""
old_updated = running_job.updated_at
time.sleep(0.01)
running_job.update_progress(50.0)
assert running_job.updated_at > old_updated
class TestJobToDict:
"""Job 序列化测试"""
def test_to_dict_pending_job(self):
"""测试 pending 状态的 Job 序列化为字典"""
job = Job.create(
project_id="proj-123",
job_type=JobType.VIDEO_COMPOSE,
payload={"input": "data"},
source_id="src-456",
)
d = job.to_dict()
assert d["id"] == job.id
assert d["project_id"] == "proj-123"
assert d["job_type"] == "video_compose"
assert d["status"] == "pending"
assert d["progress"] == 0.0
assert d["payload"] == {"input": "data"}
assert d["result"] == {}
assert d["error_message"] == ""
assert d["retry_count"] == 0
assert d["max_retries"] == 3
assert d["source_id"] == "src-456"
assert d["is_retryable"] is False
assert d["started_at"] is None
assert d["completed_at"] is None
assert d["created_at"] is not None
assert d["updated_at"] is not None
def test_to_dict_completed_job(self):
"""测试完成状态的 Job 序列化为字典"""
job = Job.create(project_id="proj-123", job_type=JobType.GENERATION)
job.mark_running()
job.mark_success(result={"output": "result"})
d = job.to_dict()
assert d["status"] == "success"
assert d["progress"] == 100.0
assert d["result"] == {"output": "result"}
assert d["started_at"] is not None
assert d["completed_at"] is not None
def test_to_dict_failed_job(self):
"""测试失败状态的 Job 序列化为字典"""
job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
job.mark_running()
job.mark_failed("timeout error")
d = job.to_dict()
assert d["status"] == "failed"
assert d["error_message"] == "timeout error"
assert d["is_retryable"] is True
class TestTransitionTimestamps:
"""状态转换时间戳测试"""
@pytest.fixture
def new_job(self):
return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
def test_mark_running_sets_started_at(self, new_job):
"""测试 mark_running 设置 started_at"""
assert new_job.started_at is None
new_job.mark_running()
assert new_job.started_at is not None
assert isinstance(new_job.started_at, datetime)
assert new_job.started_at.tzinfo is not None
def test_mark_running_twice_preserves_started_at(self, new_job):
"""测试再次 mark_running 不覆盖 started_at"""
# 先手动转换到 running
new_job.transition_to(JobStatus.RUNNING)
first_started = new_job.started_at
# 不能直接再调 mark_running(会报错),但可以验证 started_at 不被重复设置
# transition_to 已经处理了 started_at is None 的逻辑
assert first_started == new_job.started_at
def test_mark_success_sets_completed_at(self, new_job):
"""测试 mark_success 设置 completed_at"""
new_job.mark_running()
assert new_job.completed_at is None
new_job.mark_success()
assert new_job.completed_at is not None
def test_mark_failed_sets_completed_at(self, new_job):
"""测试 mark_failed 设置 completed_at"""
new_job.mark_running()
assert new_job.completed_at is None
new_job.mark_failed("error")
assert new_job.completed_at is not None
def test_transition_updates_updated_at(self, new_job):
"""测试每次状态转换都更新 updated_at"""
old_updated = new_job.updated_at
time.sleep(0.01)
new_job.mark_running()
assert new_job.updated_at > old_updated
+117
View File
@@ -243,3 +243,120 @@ class TestJWTService:
access = service.create_access_token(user_id="user-123")
refresh = service.create_refresh_token(user_id="user-123", session_id="sess-abc")
assert access != refresh
class TestJWTHandler:
"""JWT Handler 委托层测试"""
def test_create_access_token(self):
"""测试创建 access token"""
from packages.application.auth.jwt_handler import JWTHandler
handler = JWTHandler(secret_key="test-secret-key")
token = handler.create_access_token(user_id="user-123", role="admin")
assert token is not None
assert len(token) > 20
# 验证token内容
payload = jwt.decode(token, "test-secret-key", algorithms=["HS256"])
assert payload["sub"] == "user-123"
assert payload["role"] == "admin"
assert payload["type"] == "access"
def test_create_access_token_with_additional_claims(self):
"""测试带额外声明创建 token"""
from packages.application.auth.jwt_handler import JWTHandler
handler = JWTHandler(secret_key="test-secret-key")
token = handler.create_access_token(
user_id="user-456",
additional_claims={"custom_field": "custom_value"},
)
payload = jwt.decode(token, "test-secret-key", algorithms=["HS256"])
assert payload["sub"] == "user-456"
assert payload["custom_field"] == "custom_value"
def test_verify_access_token(self):
"""测试验证 access token"""
from packages.application.auth.jwt_handler import JWTHandler
handler = JWTHandler(secret_key="test-secret-key")
token = handler.create_access_token(user_id="user-123", role="user")
payload = handler.verify_access_token(token)
assert payload["sub"] == "user-123"
assert payload["role"] == "user"
assert payload["type"] == "access"
def test_verify_access_token_expired(self):
"""测试验证过期的 access token"""
from packages.application.auth.jwt_handler import JWTHandler
handler = JWTHandler(secret_key="test-secret-key", access_token_expire_minutes=0)
token = handler.create_access_token(user_id="user-123")
time.sleep(1) # 确保过期
with pytest.raises(ExpiredSignatureError):
handler.verify_access_token(token)
def test_verify_token(self):
"""测试验证任意类型 token"""
from packages.application.auth.jwt_handler import JWTHandler
handler = JWTHandler(secret_key="test-secret-key")
token = handler.create_access_token(user_id="user-123")
payload = handler.verify_token(token)
assert payload["sub"] == "user-123"
def test_verify_invalid_token(self):
"""测试验证无效 token"""
from packages.application.auth.jwt_handler import JWTHandler
handler = JWTHandler(secret_key="test-secret-key")
with pytest.raises(InvalidTokenError):
handler.verify_token("invalid.token.here")
def test_configure_and_get_default_handler(self):
"""测试配置和获取全局默认 handler"""
from packages.application.auth import jwt_handler as handler_module
from packages.application.auth.jwt_handler import (
configure_jwt_handler,
get_jwt_handler,
)
# 重置全局状态
handler_module._default_handler = None
# 配置
handler = configure_jwt_handler(
secret_key="global-secret",
algorithm="HS256",
access_token_expire_minutes=60,
)
assert handler is not None
# 获取
same_handler = get_jwt_handler()
assert same_handler is handler
# 验证能正常工作
token = same_handler.create_access_token(user_id="global-user")
payload = jwt.decode(token, "global-secret", algorithms=["HS256"])
assert payload["sub"] == "global-user"
# 重置全局状态,避免影响其他测试
handler_module._default_handler = None
def test_get_jwt_handler_not_configured(self):
"""测试未配置时获取 handler 抛出异常"""
from packages.application.auth import jwt_handler as handler_module
from packages.application.auth.jwt_handler import get_jwt_handler
# 确保未配置
handler_module._default_handler = None
with pytest.raises(RuntimeError, match="JWT handler not configured"):
get_jwt_handler()
+97
View File
@@ -170,3 +170,100 @@ class TestPasswordValidator:
valid, error = validator.validate("longenoughpassword")
assert valid is True
assert error is None
class TestPasswordHandler:
"""Password Handler 委托层测试"""
def test_hash_and_verify_password(self):
"""测试哈希和验证密码"""
from packages.application.auth.password_handler import PasswordHandler
handler = PasswordHandler(rounds=4)
hashed = handler.hash_password("MySecurePass123")
assert hashed != "MySecurePass123"
assert len(hashed) > 20
assert handler.verify_password("MySecurePass123", hashed) is True
assert handler.verify_password("WrongPassword", hashed) is False
def test_hash_empty_password_raises(self):
"""测试空密码抛出异常"""
from packages.application.auth.password_handler import PasswordHandler
handler = PasswordHandler(rounds=4)
with pytest.raises(ValueError):
handler.hash_password("")
def test_needs_rehash(self):
"""测试检测需要重新哈希"""
from packages.application.auth.password_handler import PasswordHandler
handler = PasswordHandler(rounds=4)
hashed = handler.hash_password("TestPass123")
# 相同 rounds 不需要重新哈希
assert handler.needs_rehash(hashed) is False
# 用更高 rounds 的 handler 检查,应该需要重新哈希
# 注意:bcrypt 的 rounds 体现在 hash 中,这里用不同 rounds 测试
high_rounds_handler = PasswordHandler(rounds=5)
# 低 rounds 的 hash 在高 rounds 配置下应该需要 rehash
assert high_rounds_handler.needs_rehash(hashed) is True
def test_validate_strength(self):
"""测试密码强度验证"""
from packages.application.auth.password_handler import PasswordHandler
handler = PasswordHandler(rounds=4)
# 弱密码
valid, error = handler.validate_strength("weak")
assert valid is False
assert error is not None
# 强密码
valid, error = handler.validate_strength("StrongPass123")
assert valid is True
assert error is None
def test_configure_and_get_default_handler(self):
"""测试配置和获取全局默认 handler"""
from packages.application.auth import password_handler as handler_module
from packages.application.auth.password_handler import (
configure_password_handler,
get_password_handler,
)
# 重置全局状态
handler_module._default_handler = None
# 配置
handler = configure_password_handler(rounds=4)
assert handler is not None
# 获取
same_handler = get_password_handler()
assert same_handler is handler
# 验证能正常工作
hashed = same_handler.hash_password("TestPass123")
assert same_handler.verify_password("TestPass123", hashed) is True
# 重置全局状态,避免影响其他测试
handler_module._default_handler = None
def test_get_password_handler_auto_creates_default(self):
"""测试未配置时获取 handler 会自动创建默认实例"""
from packages.application.auth import password_handler as handler_module
from packages.application.auth.password_handler import get_password_handler
# 重置全局状态
handler_module._default_handler = None
# 自动创建默认实例
handler = get_password_handler()
assert handler is not None
# 重置
handler_module._default_handler = None
+551
View File
@@ -0,0 +1,551 @@
"""
Quota 配额系统单元测试
"""
import math
import pytest
from packages.domain.quota import (
QuotaChecker,
QuotaCheckResult,
QuotaDimension,
QuotaRegistry,
QuotaTier,
QuotaWarningLevel,
get_warning_level,
quota_checker,
quota_registry,
)
class TestQuotaDimension:
"""配额维度枚举测试"""
def test_builtin_dimensions_exist(self):
"""测试内置维度存在"""
assert QuotaDimension.STORAGE_GB == "storage_gb"
assert QuotaDimension.VIDEOS_PER_MONTH == "videos_per_month"
assert QuotaDimension.MAX_CONCURRENT == "max_concurrent"
assert QuotaDimension.MAX_TEMPLATES == "max_templates"
assert QuotaDimension.MAX_TITLES == "max_titles"
assert QuotaDimension.MAX_VOICEOVERS == "max_voiceovers"
assert QuotaDimension.AI_VOICE_ENABLED == "ai_voice_enabled"
def test_extended_dimensions_exist(self):
"""测试扩展维度存在"""
assert QuotaDimension.AI_VOICE_CREDITS == "ai_voice_credits"
assert QuotaDimension.BATCH_EXPORT_ENABLED == "batch_export_enabled"
assert QuotaDimension.MULTI_PLATFORM_ENABLED == "multi_platform_enabled"
assert QuotaDimension.DEDUP_REPORT_ENABLED == "dedup_report_enabled"
def test_dimension_is_string(self):
"""测试枚举值是字符串"""
assert isinstance(QuotaDimension.STORAGE_GB, str)
assert QuotaDimension.STORAGE_GB == "storage_gb"
class TestQuotaTier:
"""配额等级测试"""
def test_get_limit_defined(self):
"""测试获取已定义的配额限制"""
tier = QuotaTier(name="test", limits={"storage_gb": 10, "videos_per_month": 50})
assert tier.get_limit("storage_gb") == 10
assert tier.get_limit("videos_per_month") == 50
def test_get_limit_undefined_returns_zero(self):
"""测试未定义维度返回 0"""
tier = QuotaTier(name="test", limits={"storage_gb": 10})
assert tier.get_limit("unknown_dim") == 0
def test_is_unlimited_with_inf(self):
"""测试不限量判断(inf"""
tier = QuotaTier(name="test", limits={"templates": float("inf")})
assert tier.is_unlimited("templates") is True
def test_is_unlimited_with_finite(self):
"""测试有限量判断"""
tier = QuotaTier(name="test", limits={"storage_gb": 10})
assert tier.is_unlimited("storage_gb") is False
def test_is_unlimited_undefined(self):
"""测试未定义维度默认不限量(因为默认值是 inf)"""
tier = QuotaTier(name="test", limits={})
# is_unlimited 使用 limits.get(dim, float("inf")) == float("inf")
# 未定义时默认是 inf,所以返回 True
assert tier.is_unlimited("undefined") is True
def test_default_limits_empty(self):
"""测试默认 limits 为空 dict"""
tier = QuotaTier(name="test")
assert tier.limits == {}
class TestQuotaTiers:
"""预定义配额等级测试"""
def test_free_tier_limits(self):
"""测试 free 套餐限制"""
from packages.domain.quota import QUOTA_TIERS
free = QUOTA_TIERS["free"]
assert free.name == "free"
assert free.get_limit("storage_gb") == 2
assert free.get_limit("videos_per_month") == 5
assert free.get_limit("max_concurrent") == 3
assert free.get_limit("max_templates") == 3
assert free.get_limit("max_titles") == 50
assert free.get_limit("max_voiceovers") == 10
assert free.get_limit("ai_voice_enabled") == 0
assert free.get_limit("ai_voice_credits") == 0
def test_basic_tier_limits(self):
"""测试 basic 套餐限制"""
from packages.domain.quota import QUOTA_TIERS
basic = QUOTA_TIERS["basic"]
assert basic.name == "basic"
assert basic.get_limit("storage_gb") == 20
assert basic.get_limit("videos_per_month") == 30
assert basic.get_limit("max_concurrent") == 10
assert basic.get_limit("max_templates") == 15
assert basic.get_limit("max_titles") == 500
assert basic.get_limit("max_voiceovers") == 100
assert basic.get_limit("ai_voice_enabled") == 1
assert basic.get_limit("ai_voice_credits") == 100
assert basic.get_limit("batch_export_enabled") == 1
def test_premium_tier_limits(self):
"""测试 premium 套餐限制"""
from packages.domain.quota import QUOTA_TIERS
premium = QUOTA_TIERS["premium"]
assert premium.name == "premium"
assert premium.get_limit("storage_gb") == 100
assert premium.get_limit("videos_per_month") == 100
assert premium.get_limit("max_concurrent") == 20
assert premium.is_unlimited("max_templates") is True
assert premium.get_limit("max_titles") == 500
assert premium.get_limit("max_voiceovers") == 100
assert premium.get_limit("ai_voice_enabled") == 1
assert premium.get_limit("ai_voice_credits") == 500
assert premium.get_limit("batch_export_enabled") == 1
assert premium.get_limit("multi_platform_enabled") == 1
assert premium.get_limit("dedup_report_enabled") == 1
def test_three_tiers_exist(self):
"""测试三个套餐等级都存在"""
from packages.domain.quota import QUOTA_TIERS
assert "free" in QUOTA_TIERS
assert "basic" in QUOTA_TIERS
assert "premium" in QUOTA_TIERS
class TestQuotaWarningLevel:
"""告警级别测试"""
def test_warning_level_values(self):
"""测试告警级别常量值"""
assert QuotaWarningLevel.NORMAL == "normal"
assert QuotaWarningLevel.WARNING == "warning"
assert QuotaWarningLevel.CRITICAL == "critical"
assert QuotaWarningLevel.EXCEEDED == "exceeded"
class TestQuotaCheckResult:
"""配额检查结果测试"""
def test_usage_percent_normal(self):
"""测试正常使用率计算"""
result = QuotaCheckResult(
allowed=True,
dimension="storage_gb",
limit=100,
used=50,
remaining=50,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 50.0
def test_usage_percent_over_limit(self):
"""测试超出限制时 capped at 100%"""
result = QuotaCheckResult(
allowed=False,
dimension="storage_gb",
limit=100,
used=150,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_zero_limit_with_usage(self):
"""测试限制为 0 但有使用量时返回 100%"""
result = QuotaCheckResult(
allowed=False,
dimension="ai_voice",
limit=0,
used=5,
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
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_exactly_100(self):
"""测试刚好 100% 使用"""
result = QuotaCheckResult(
allowed=False,
dimension="storage_gb",
limit=100,
used=100,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
class TestQuotaRegistry:
"""配额注册表测试"""
def test_initial_builtin_dimensions(self):
"""测试初始化后内置维度已注册"""
registry = QuotaRegistry()
dims = registry.list_dimensions()
assert "storage_gb" in dims
assert "videos_per_month" in dims
assert "max_concurrent" in dims
assert "max_templates" in dims
assert "max_titles" in dims
assert "max_voiceovers" in dims
assert "ai_voice_enabled" in dims
def test_register_new_dimension(self):
"""测试注册新维度"""
registry = QuotaRegistry()
registry.register_dimension("custom_dim", "自定义维度")
dims = registry.list_dimensions()
assert "custom_dim" in dims
assert dims["custom_dim"] == "自定义维度"
def test_register_dimension_with_default_limits(self):
"""测试注册带默认限制的新维度"""
registry = QuotaRegistry()
registry.register_dimension(
"custom_feature",
"自定义功能",
default_limits={"free": 0, "basic": 1, "premium": 5},
)
assert registry.get_limit("free", "custom_feature") == 0
assert registry.get_limit("basic", "custom_feature") == 1
assert registry.get_limit("premium", "custom_feature") == 5
def test_register_dimension_without_default_limits(self):
"""测试注册不带默认限制的新维度(所有套餐默认 0)"""
registry = QuotaRegistry()
registry.register_dimension("new_feature", "新功能")
assert registry.get_limit("free", "new_feature") == 0
assert registry.get_limit("basic", "new_feature") == 0
assert registry.get_limit("premium", "new_feature") == 0
def test_register_dimension_idempotent(self):
"""测试重复注册是幂等的"""
registry = QuotaRegistry()
registry.register_dimension("test_dim", "测试维度", default_limits={"free": 10})
# 第二次注册不应该改变任何东西
registry.register_dimension("test_dim", "另一个描述", default_limits={"free": 999})
dims = registry.list_dimensions()
assert dims["test_dim"] == "测试维度" # 保留第一次的描述
assert registry.get_limit("free", "test_dim") == 10 # 保留第一次的限制
def test_register_unknown_plan_ignored(self):
"""测试未知套餐的默认限制被忽略"""
registry = QuotaRegistry()
registry.register_dimension(
"test_dim",
"测试",
default_limits={"free": 1, "enterprise": 100},
)
assert registry.get_limit("free", "test_dim") == 1
# enterprise 套餐不存在,不影响
assert "enterprise" not in registry.list_tiers()
def test_get_tier_existing(self):
"""测试获取存在的套餐"""
registry = QuotaRegistry()
tier = registry.get_tier("free")
assert tier is not None
assert tier.name == "free"
def test_get_tier_nonexistent(self):
"""测试获取不存在的套餐返回 None"""
registry = QuotaRegistry()
assert registry.get_tier("nonexistent") is None
def test_get_limit_nonexistent_plan(self):
"""测试不存在套餐的限制返回 0"""
registry = QuotaRegistry()
assert registry.get_limit("enterprise", "storage_gb") == 0
def test_list_tiers(self):
"""测试列出所有套餐"""
registry = QuotaRegistry()
tiers = registry.list_tiers()
assert "free" in tiers
assert "basic" in tiers
assert "premium" in tiers
assert len(tiers) == 3
def test_list_dimensions_returns_copy(self):
"""测试 list_dimensions 返回副本(修改不影响内部)"""
registry = QuotaRegistry()
dims = registry.list_dimensions()
dims["fake_dim"] = "fake"
# 原始注册表不应被修改
assert "fake_dim" not in registry.list_dimensions()
class TestQuotaChecker:
"""配额检查器测试"""
@pytest.fixture
def checker(self):
return QuotaChecker()
# ===== 基础检查 =====
def test_check_free_storage_under_limit(self, checker):
"""测试 free 套餐存储未超限"""
result = checker.check("free", "storage_gb", 1.0)
assert result.allowed is True
assert result.limit == 2
assert result.used == 1.0
assert result.remaining == 1.0
assert result.warning_level == QuotaWarningLevel.NORMAL
assert result.dimension == "storage_gb"
def test_check_free_storage_over_limit(self, checker):
"""测试 free 套餐存储超限"""
result = checker.check("free", "storage_gb", 3.0)
assert result.allowed is False
assert result.remaining == 0
assert result.warning_level == QuotaWarningLevel.EXCEEDED
def test_check_free_storage_exactly_at_limit(self, checker):
"""测试刚好达到限制(不允许)"""
result = checker.check("free", "storage_gb", 2.0)
# used < limit → 2 < 2 → False
assert result.allowed is False
assert result.warning_level == QuotaWarningLevel.EXCEEDED
# ===== 告警级别 =====
def test_warning_level_normal(self, checker):
"""测试正常级别(< 80%"""
result = checker.check("free", "storage_gb", 1.0) # 50%
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_warning_level_warning(self, checker):
"""测试警告级别(80% ~ 95%"""
result = checker.check("free", "storage_gb", 1.7) # 85%
assert result.warning_level == QuotaWarningLevel.WARNING
def test_warning_level_critical(self, checker):
"""测试严重级别(95% ~ 100%"""
result = checker.check("free", "storage_gb", 1.95) # 97.5%
assert result.warning_level == QuotaWarningLevel.CRITICAL
def test_warning_level_exceeded(self, checker):
"""测试超限级别(>= 100%"""
result = checker.check("free", "storage_gb", 2.0) # 100%
assert result.warning_level == QuotaWarningLevel.EXCEEDED
# ===== 不限量 =====
def test_check_unlimited_templates_premium(self, checker):
"""测试 premium 套餐模板不限量"""
result = checker.check("premium", "max_templates", 9999)
assert result.allowed is True
assert result.limit == float("inf")
assert result.remaining == float("inf")
assert result.warning_level == QuotaWarningLevel.NORMAL
# ===== 0 限制 =====
def test_check_zero_limit_with_usage(self, checker):
"""测试限制为 0 但有使用量"""
result = checker.check("free", "ai_voice_enabled", 1)
assert result.allowed is False
assert result.warning_level == QuotaWarningLevel.EXCEEDED
def test_check_zero_limit_no_usage(self, checker):
"""测试限制为 0 且无使用量"""
result = checker.check("free", "ai_voice_enabled", 0)
# used < limit → 0 < 0 → False? 让我们看看...
# 实际上 0 < 0 是 False,所以 allowed = False
# 但 warning_level: limit <= 0 and used == 0 → NORMAL
# 等一下,看看代码逻辑:
# if limit <= 0: return EXCEEDED if used > 0 else NORMAL
assert result.warning_level == QuotaWarningLevel.NORMAL
# ===== 多维度检查 =====
def test_check_multiple(self, checker):
"""测试批量检查多个维度"""
usage = {
"storage_gb": 1.0,
"videos_per_month": 3,
"max_concurrent": 2,
}
results = checker.check_multiple("free", usage)
assert len(results) == 3
dims = {r.dimension: r for r in results}
assert dims["storage_gb"].allowed is True
assert dims["videos_per_month"].allowed is True
assert dims["max_concurrent"].allowed is True
def test_check_multiple_some_exceeded(self, checker):
"""测试批量检查中有超限的"""
usage = {
"storage_gb": 5.0, # 超限
"videos_per_month": 3, # 正常
}
results = checker.check_multiple("free", usage)
dims = {r.dimension: r for r in results}
assert dims["storage_gb"].allowed is False
assert dims["videos_per_month"].allowed is True
# ===== 自定义 registry =====
def test_check_with_custom_registry(self):
"""测试使用自定义 registry"""
registry = QuotaRegistry()
registry.register_dimension(
"custom_feature",
"自定义",
default_limits={"free": 5, "basic": 20},
)
checker = QuotaChecker(registry)
result = checker.check("free", "custom_feature", 3)
assert result.allowed is True
assert result.limit == 5
result = checker.check("basic", "custom_feature", 25)
assert result.allowed is False
def test_check_unknown_plan(self, checker):
"""测试未知套餐(限制为 0"""
result = checker.check("enterprise", "storage_gb", 1)
assert result.allowed is False
assert result.limit == 0
class TestGetWarningLevel:
"""便捷函数 get_warning_level 测试"""
def test_normal(self):
assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL
def test_warning(self):
assert get_warning_level(85, 100) == QuotaWarningLevel.WARNING
def test_critical(self):
assert get_warning_level(96, 100) == QuotaWarningLevel.CRITICAL
def test_exceeded(self):
assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED
def test_zero_limit_with_usage(self):
assert get_warning_level(5, 0) == QuotaWarningLevel.EXCEEDED
def test_zero_limit_no_usage(self):
assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL
def test_unlimited(self):
assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL
def test_boundary_79_percent(self):
"""测试 79% 仍是 normal"""
assert get_warning_level(79, 100) == QuotaWarningLevel.NORMAL
def test_boundary_80_percent(self):
"""测试 80% 是 warning"""
assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING
def test_boundary_94_percent(self):
"""测试 94% 仍是 warning"""
assert get_warning_level(94, 100) == QuotaWarningLevel.WARNING
def test_boundary_95_percent(self):
"""测试 95% 是 critical"""
assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL
def test_boundary_99_percent(self):
"""测试 99% 仍是 critical"""
assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL
def test_zero_usage(self):
"""测试 0 使用量"""
assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL
class TestGlobalSingletons:
"""全局单例测试"""
def test_quota_registry_exists(self):
"""测试全局 quota_registry 存在"""
assert quota_registry is not None
assert isinstance(quota_registry, QuotaRegistry)
assert "free" in quota_registry.list_tiers()
def test_quota_checker_exists(self):
"""测试全局 quota_checker 存在"""
assert quota_checker is not None
assert isinstance(quota_checker, QuotaChecker)
def test_global_checker_uses_global_registry(self):
"""测试全局 checker 使用全局 registry"""
result = quota_checker.check("free", "storage_gb", 1.0)
assert result.limit == 2
+503
View File
@@ -0,0 +1,503 @@
"""
Subtitle 字幕领域模型单元测试
"""
import pytest
from packages.domain.subtitle import (
SubtitleSegment,
SubtitleTimeline,
SubtitleWord,
)
class TestSubtitleWord:
"""SubtitleWord 测试"""
def test_duration_positive(self):
word = SubtitleWord(text="你好", start=1.0, end=2.5)
assert word.duration == pytest.approx(1.5)
def test_duration_zero(self):
word = SubtitleWord(text="a", start=5.0, end=5.0)
assert word.duration == 0.0
def test_duration_negative_returns_zero(self):
"""测试结束时间小于开始时间时返回 0"""
word = SubtitleWord(text="a", start=3.0, end=1.0)
assert word.duration == 0.0
class TestSubtitleSegment:
"""SubtitleSegment 测试"""
def test_duration(self):
seg = SubtitleSegment(text="你好世界", start=0.0, end=3.0)
assert seg.duration == pytest.approx(3.0)
def test_duration_zero(self):
seg = SubtitleSegment(text="test", start=5.0, end=5.0)
assert seg.duration == 0.0
def test_duration_negative_returns_zero(self):
seg = SubtitleSegment(text="test", start=5.0, end=2.0)
assert seg.duration == 0.0
def test_char_count(self):
seg = SubtitleSegment(text="你好世界", start=0, end=1)
assert seg.char_count == 4
def test_char_count_empty(self):
seg = SubtitleSegment(text="", start=0, end=1)
assert seg.char_count == 0
def test_default_words_empty(self):
seg = SubtitleSegment(text="test", start=0, end=1)
assert seg.words == []
def test_with_words(self):
words = [
SubtitleWord(text="你好", start=0.0, end=1.0),
SubtitleWord(text="世界", start=1.0, end=2.0),
]
seg = SubtitleSegment(text="你好世界", start=0.0, end=2.0, words=words)
assert len(seg.words) == 2
assert seg.words[0].text == "你好"
assert seg.words[1].text == "世界"
class TestSubtitleTimelineBasics:
"""SubtitleTimeline 基础属性测试"""
def test_empty_timeline(self):
tl = SubtitleTimeline()
assert tl.segment_count == 0
assert tl.total_chars == 0
assert tl.language == "zh"
assert tl.total_duration == 0.0
def test_segment_count(self):
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="a", start=0, end=1),
SubtitleSegment(text="b", start=1, end=2),
SubtitleSegment(text="c", start=2, end=3),
]
)
assert tl.segment_count == 3
def test_total_chars(self):
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=0, end=1),
SubtitleSegment(text="世界", start=1, end=2),
SubtitleSegment(text="abcde", start=2, end=3),
]
)
assert tl.total_chars == 9
def test_custom_language(self):
tl = SubtitleTimeline(language="en")
assert tl.language == "en"
def test_custom_total_duration(self):
tl = SubtitleTimeline(total_duration=60.0)
assert tl.total_duration == 60.0
class TestMergeShortSegments:
"""merge_short_segments 测试"""
def test_single_segment_no_merge(self):
"""单个片段不需要合并"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="a", start=0, end=1),
]
)
result = tl.merge_short_segments(min_chars=8)
assert result.segment_count == 1
assert result.segments[0].text == "a"
def test_empty_timeline(self):
"""空时间轴"""
tl = SubtitleTimeline()
result = tl.merge_short_segments(min_chars=8)
assert result.segment_count == 0
def test_all_short_segments_merge_into_one(self):
"""所有短片段合并成一个"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="", start=0, end=0.5),
SubtitleSegment(text="", start=0.5, end=1.0),
SubtitleSegment(text="", start=1.0, end=1.5),
SubtitleSegment(text="", start=1.5, end=2.0),
]
)
result = tl.merge_short_segments(min_chars=8)
assert result.segment_count == 1
assert result.segments[0].text == "你好世界"
assert result.segments[0].start == 0
assert result.segments[0].end == 2.0
def test_merge_short_segments_preserves_timing(self):
"""合并后时间轴正确"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=1.0, end=2.0),
SubtitleSegment(text="世界", start=2.0, end=3.5),
]
)
result = tl.merge_short_segments(min_chars=10)
assert result.segment_count == 1
assert result.segments[0].start == 1.0
assert result.segments[0].end == 3.5
def test_merge_short_segments_with_words(self):
"""合并后词级信息保留"""
w1 = SubtitleWord(text="你好", start=0.0, end=1.0)
w2 = SubtitleWord(text="世界", start=1.0, end=2.0)
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=0.0, end=1.0, words=[w1]),
SubtitleSegment(text="世界", start=1.0, end=2.0, words=[w2]),
]
)
result = tl.merge_short_segments(min_chars=10)
assert len(result.segments[0].words) == 2
assert result.segments[0].words[0].text == "你好"
assert result.segments[0].words[1].text == "世界"
def test_multiple_merged_groups(self):
"""多个合并组 — 短段会和后续段累积到够数才提交"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="一二三四五六七八", start=0, end=2), # 8字,够数,提交
SubtitleSegment(text="", start=2, end=2.5), # 1字,入buffer
SubtitleSegment(text="", start=2.5, end=3), # 1字,入buffer(共2字)
SubtitleSegment(text="一二三四五六七八九十", start=3, end=5), # 10字,入buffer后共12字,够数提交
]
)
result = tl.merge_short_segments(min_chars=8)
# 第1段:"一二三四五六七八"(8字直接提交)
# 第2段:"九十" + "一二三四五六七八九十" 累积到12字一起提交
assert result.segment_count == 2
assert result.segments[0].text == "一二三四五六七八"
assert result.segments[1].text == "九十一二三四五六七八九十"
def test_remaining_short_merged_with_last(self):
"""剩余短片段合并到最后一段"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="一二三四五六七八", start=0, end=2), # 8字
SubtitleSegment(text="一二三", start=2, end=3), # 3字,不够
]
)
result = tl.merge_short_segments(min_chars=8)
# 最后的3字会合并到上一段(因为 < min_chars
assert result.segment_count == 1
assert result.segments[0].text == "一二三四五六七八一二三"
def test_custom_min_chars(self):
"""自定义最小字数 — 累积到够数就提交,剩余短的合并到最后"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="一二", start=0, end=1),
SubtitleSegment(text="三四", start=1, end=2),
SubtitleSegment(text="五六", start=2, end=3),
]
)
# min_chars=3
# "一二"(2字) → 不够
# +"三四"(共4字) → 够了,提交"一二三四"buffer清空
# "五六"(2字) → 循环结束,剩余<min_chars且merged非空 → 合并到最后一段
# 结果:1段 "一二三四五六"
result = tl.merge_short_segments(min_chars=3)
assert result.segment_count == 1
assert result.segments[0].text == "一二三四五六"
def test_preserves_language_and_duration(self):
"""合并后保留语言和总时长"""
tl = SubtitleTimeline(
segments=[SubtitleSegment(text="a", start=0, end=1)],
language="en",
total_duration=60.0,
)
result = tl.merge_short_segments(min_chars=8)
assert result.language == "en"
assert result.total_duration == 60.0
def test_does_not_modify_original(self):
"""不修改原时间轴"""
segments = [
SubtitleSegment(text="a", start=0, end=1),
SubtitleSegment(text="b", start=1, end=2),
]
tl = SubtitleTimeline(segments=segments)
result = tl.merge_short_segments(min_chars=5)
# 原时间轴不变
assert tl.segment_count == 2
assert result is not tl
class TestSplitLongSegments:
"""split_long_segments 测试"""
def test_short_segments_no_split(self):
"""短片段不需要拆分"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=0, end=1),
]
)
result = tl.split_long_segments(max_chars=20)
assert result.segment_count == 1
assert result.segments[0].text == "你好"
def test_single_long_segment_split_by_punctuation(self):
"""长片段按标点拆分"""
text = "你好世界。今天天气真好,我们出去玩吧!"
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=10)
# 应该被拆成多段
assert result.segment_count > 1
# 每段都不超过 max_chars(除了硬切的情况)
for seg in result.segments:
assert seg.char_count <= len(text) # 至少比原文短
def test_split_preserves_total_text(self):
"""拆分后总文本不变"""
text = "你好世界。今天天气真好,我们出去玩吧!明天再见。"
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=8)
merged_text = "".join(s.text for s in result.segments)
assert merged_text == text
def test_split_time_proportional(self):
"""拆分后时间按字数比例分配"""
text = "一二三四五六七八九十。" # 11字
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=5)
# 总时长不变
assert result.segments[0].start == 0.0
assert result.segments[-1].end == pytest.approx(10.0)
# 各段首尾相接
for i in range(len(result.segments) - 1):
assert result.segments[i].end == pytest.approx(result.segments[i + 1].start)
def test_split_with_words(self):
"""拆分时词级信息正确分配"""
words = [
SubtitleWord(text="你好", start=0.0, end=1.0),
SubtitleWord(text="世界", start=1.0, end=2.0),
SubtitleWord(text="你好吗", start=2.0, end=3.5),
]
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好世界。你好吗?", start=0.0, end=3.5, words=words),
]
)
result = tl.split_long_segments(max_chars=4)
# 第一段应该有前几个词
assert len(result.segments) >= 2
total_words = sum(len(s.words) for s in result.segments)
assert total_words == 3 # 词的总数不变
def test_multiple_mixed_segments(self):
"""混合长短片段"""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="", start=0, end=1), # 短
SubtitleSegment(text="一二三四五六七八九十一二三四五六七八九十", start=1, end=5), # 长
SubtitleSegment(text="也短", start=5, end=6), # 短
]
)
result = tl.split_long_segments(max_chars=10)
assert result.segment_count >= 3 # 至少3段(中间被拆成多段)
# 第一段还是原来的短的
assert result.segments[0].text == ""
# 最后一段还是原来的短的
assert result.segments[-1].text == "也短"
def test_no_punctuation_hard_split(self):
"""没有标点时硬切"""
text = "一二三四五六七八九十一二三四五六七八九十一二三四五"
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=10)
assert result.segment_count >= 3
for seg in result.segments:
# 硬切的每段应该 <= max_chars
assert seg.char_count <= 10
def test_preserves_language_and_duration(self):
"""拆分后保留语言和总时长"""
tl = SubtitleTimeline(
segments=[SubtitleSegment(text="a", start=0, end=1)],
language="ja",
total_duration=30.0,
)
result = tl.split_long_segments(max_chars=20)
assert result.language == "ja"
assert result.total_duration == 30.0
def test_does_not_modify_original(self):
"""不修改原时间轴"""
original_text = "一二三四五六七八九十一二三四五六七八九十"
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=original_text, start=0, end=5),
]
)
result = tl.split_long_segments(max_chars=8)
assert tl.segment_count == 1
assert tl.segments[0].text == original_text
assert result is not tl
class TestSplitTextByPunctuation:
"""_split_text_by_punctuation 静态方法测试"""
def test_short_text_no_split(self):
result = SubtitleTimeline._split_text_by_punctuation("你好世界", 10)
assert result == ["你好世界"]
def test_split_at_sentence_end(self):
"""在句末标点处断开"""
result = SubtitleTimeline._split_text_by_punctuation("你好。世界。", 5)
assert len(result) == 2
assert result[0] == "你好。"
assert result[1] == "世界。"
def test_split_at_comma(self):
"""在逗号处断开(超过最大长度时)"""
text = "一二三四五六七八,二二三四五六七八。"
result = SubtitleTimeline._split_text_by_punctuation(text, 10)
assert len(result) >= 2
def test_no_punctuation_hard_split(self):
"""没有标点时硬切"""
result = SubtitleTimeline._split_text_by_punctuation("一二三四五六七八九十", 5)
assert len(result) == 2
assert result[0] == "一二三四五"
assert result[1] == "六七八九十"
def test_empty_text(self):
# 空字符串循环不执行,current为空不append,返回空列表
result = SubtitleTimeline._split_text_by_punctuation("", 10)
assert result == []
def test_mixed_punctuation(self):
"""混合标点"""
text = "你好!吃饭了吗?是的,我吃过了。"
result = SubtitleTimeline._split_text_by_punctuation(text, 6)
# 验证所有段加起来等于原文
assert "".join(result) == text
def test_sentence_end_with_min_length(self):
"""句末标点断句的「半长门槛」只在未超max_chars时生效;
超过max_chars回溯找标点时,即使首段很短也会断开。"""
# "你好。" 3字 < max_chars//2(5),未超max_chars时不会主动断开
# 但加上后面的"世界很大很美好"后超过10字,回溯找标点找到"。",强制断开
text = "你好。世界很大很美好。"
result = SubtitleTimeline._split_text_by_punctuation(text, 10)
# 超过max_chars时回溯断开,首段可能很短
assert len(result) == 2
assert result[0] == "你好。"
assert result[1] == "世界很大很美好。"
# 总文本不变
assert "".join(result) == text
def test_exclamation_and_question_marks(self):
"""感叹号和问号也算句末标点"""
text = "你好吗!我很好!你呢?"
result = SubtitleTimeline._split_text_by_punctuation(text, 4)
assert len(result) >= 3
class TestMergeSegments:
"""_merge_segments 静态方法测试"""
def test_merge_two_segments(self):
result = SubtitleTimeline._merge_segments(
[
SubtitleSegment(text="你好", start=0.0, end=1.0),
SubtitleSegment(text="世界", start=1.0, end=2.0),
]
)
assert result.text == "你好世界"
assert result.start == 0.0
assert result.end == 2.0
def test_merge_empty_list(self):
result = SubtitleTimeline._merge_segments([])
assert result.text == ""
assert result.start == 0
assert result.end == 0
def test_merge_single_segment(self):
seg = SubtitleSegment(text="test", start=1.0, end=2.0)
result = SubtitleTimeline._merge_segments([seg])
assert result.text == "test"
assert result.start == 1.0
assert result.end == 2.0
def test_merge_preserves_words(self):
w1 = SubtitleWord(text="你好", start=0.0, end=1.0)
w2 = SubtitleWord(text="世界", start=1.0, end=2.0)
result = SubtitleTimeline._merge_segments(
[
SubtitleSegment(text="你好", start=0.0, end=1.0, words=[w1]),
SubtitleSegment(text="世界", start=1.0, end=2.0, words=[w2]),
]
)
assert len(result.words) == 2
assert result.words[0].text == "你好"
assert result.words[1].text == "世界"
def test_merge_non_contiguous_segments(self):
"""合并非连续片段(有间隙)"""
result = SubtitleTimeline._merge_segments(
[
SubtitleSegment(text="a", start=0.0, end=1.0),
SubtitleSegment(text="b", start=3.0, end=4.0),
]
)
assert result.start == 0.0
assert result.end == 4.0
assert result.text == "ab"
class TestMergeAndSplitRoundtrip:
"""合并和拆分的组合测试"""
def test_split_then_merge_approximate(self):
"""拆分后再合并,总字数和总时长基本一致"""
original_text = "你好世界。今天天气真好,我们出去玩吧!明天见。"
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=original_text, start=0.0, end=10.0),
]
)
split = tl.split_long_segments(max_chars=5)
merged = split.merge_short_segments(min_chars=50) # 足够大的min_chars让它们都合并
assert merged.segment_count == 1
assert merged.segments[0].text == original_text
assert merged.segments[0].start == 0.0
assert merged.segments[0].end == pytest.approx(10.0)
+202
View File
@@ -0,0 +1,202 @@
"""
TTS 配音配置模型单元测试
"""
import pytest
from packages.domain.tts_config import TtsConfig
class TestTtsConfigDefaults:
"""默认值测试"""
def test_default_values(self):
config = TtsConfig()
assert config.enabled is False
assert config.voice_id == ""
assert config.speed == 1.0
assert config.pitch == 0.0
assert config.volume == 0.8
assert config.text == ""
assert config.align_mode == "full"
assert config.overlap_mode == "replace"
class TestTtsConfigParse:
"""parse 方法测试"""
def test_parse_none(self):
config = TtsConfig.parse(None)
assert config.enabled is False
assert config.speed == 1.0
def test_parse_empty_dict(self):
config = TtsConfig.parse({})
assert config.enabled is False
def test_parse_not_dict(self):
config = TtsConfig.parse("not a dict")
assert config.enabled is False
def test_parse_disabled_returns_minimal(self):
"""disabled 时直接返回 enabled=False,忽略其他字段"""
config = TtsConfig.parse(
{
"enabled": False,
"voice_id": "v123",
"speed": 1.5,
}
)
assert config.enabled is False
assert config.voice_id == "" # 不保留
def test_parse_enabled_true(self):
config = TtsConfig.parse(
{
"enabled": True,
"voice_id": "voice_001",
"speed": 1.2,
"pitch": 2.5,
"volume": 0.5,
"text": "你好世界",
"align_mode": "subtitle",
"overlap_mode": "mix",
}
)
assert config.enabled is True
assert config.voice_id == "voice_001"
assert config.speed == 1.2
assert config.pitch == 2.5
assert config.volume == 0.5
assert config.text == "你好世界"
assert config.align_mode == "subtitle"
assert config.overlap_mode == "mix"
def test_parse_enabled_not_bool_false(self):
"""enabled 不是 bool 时视为 False"""
config = TtsConfig.parse({"enabled": "true"})
assert config.enabled is False
def test_parse_enabled_not_bool_zero(self):
config = TtsConfig.parse({"enabled": 0})
assert config.enabled is False
def test_parse_voice_id_not_string(self):
config = TtsConfig.parse({"enabled": True, "voice_id": 123})
assert config.voice_id == ""
def test_parse_speed_not_number(self):
config = TtsConfig.parse({"enabled": True, "speed": "fast"})
assert config.speed == 1.0
def test_parse_pitch_not_number(self):
config = TtsConfig.parse({"enabled": True, "pitch": "high"})
assert config.pitch == 0.0
def test_parse_volume_not_number(self):
config = TtsConfig.parse({"enabled": True, "volume": "loud"})
assert config.volume == 0.8
def test_parse_text_not_string(self):
config = TtsConfig.parse({"enabled": True, "text": 12345})
assert config.text == ""
def test_parse_align_mode_invalid(self):
config = TtsConfig.parse({"enabled": True, "align_mode": "invalid"})
assert config.align_mode == "full"
def test_parse_align_mode_subtitle(self):
config = TtsConfig.parse({"enabled": True, "align_mode": "subtitle"})
assert config.align_mode == "subtitle"
def test_parse_align_mode_full(self):
config = TtsConfig.parse({"enabled": True, "align_mode": "full"})
assert config.align_mode == "full"
def test_parse_overlap_mode_invalid(self):
config = TtsConfig.parse({"enabled": True, "overlap_mode": "invalid"})
assert config.overlap_mode == "replace"
def test_parse_overlap_mode_replace(self):
config = TtsConfig.parse({"enabled": True, "overlap_mode": "replace"})
assert config.overlap_mode == "replace"
def test_parse_overlap_mode_mix(self):
config = TtsConfig.parse({"enabled": True, "overlap_mode": "mix"})
assert config.overlap_mode == "mix"
def test_parse_integer_speed(self):
"""int 类型的 speed 应该被转成 float"""
config = TtsConfig.parse({"enabled": True, "speed": 2})
assert config.speed == 2.0
assert isinstance(config.speed, float)
def test_parse_integer_pitch(self):
config = TtsConfig.parse({"enabled": True, "pitch": -5})
assert config.pitch == -5.0
assert isinstance(config.pitch, float)
def test_parse_integer_volume(self):
config = TtsConfig.parse({"enabled": True, "volume": 1})
assert config.volume == 1.0
assert isinstance(config.volume, float)
class TestTtsConfigClamp:
"""边界钳制测试"""
def test_speed_too_low(self):
config = TtsConfig.parse({"enabled": True, "speed": 0.1})
assert config.speed == 0.5
def test_speed_too_high(self):
config = TtsConfig.parse({"enabled": True, "speed": 3.0})
assert config.speed == 2.0
def test_speed_lower_boundary(self):
config = TtsConfig.parse({"enabled": True, "speed": 0.5})
assert config.speed == 0.5
def test_speed_upper_boundary(self):
config = TtsConfig.parse({"enabled": True, "speed": 2.0})
assert config.speed == 2.0
def test_pitch_too_low(self):
config = TtsConfig.parse({"enabled": True, "pitch": -20})
assert config.pitch == -12
def test_pitch_too_high(self):
config = TtsConfig.parse({"enabled": True, "pitch": 20})
assert config.pitch == 12
def test_pitch_lower_boundary(self):
config = TtsConfig.parse({"enabled": True, "pitch": -12})
assert config.pitch == -12
def test_pitch_upper_boundary(self):
config = TtsConfig.parse({"enabled": True, "pitch": 12})
assert config.pitch == 12
def test_volume_negative(self):
config = TtsConfig.parse({"enabled": True, "volume": -0.5})
assert config.volume == 0.0
def test_volume_over_one(self):
config = TtsConfig.parse({"enabled": True, "volume": 1.5})
assert config.volume == 1.0
def test_volume_zero(self):
config = TtsConfig.parse({"enabled": True, "volume": 0.0})
assert config.volume == 0.0
def test_volume_one(self):
config = TtsConfig.parse({"enabled": True, "volume": 1.0})
assert config.volume == 1.0
def test_clamp_via_direct_construction(self):
"""直接构造也应该钳制(通过 _clamp 方法)"""
config = TtsConfig(enabled=True, speed=5.0, pitch=100, volume=-1)
config._clamp()
assert config.speed == 2.0
assert config.pitch == 12
assert config.volume == 0.0
@@ -0,0 +1,389 @@
"""
VoiceCloneProfile 音色克隆档案领域模型单元测试
"""
import pytest
from packages.domain.voice_clone_profile import (
TERMINAL_STATUSES,
VoiceCloneProfile,
VoiceCloneStatus,
)
class TestVoiceCloneStatus:
"""VoiceCloneStatus 枚举测试"""
def test_status_values(self):
assert VoiceCloneStatus.PENDING == "pending"
assert VoiceCloneStatus.PROCESSING == "processing"
assert VoiceCloneStatus.READY == "ready"
assert VoiceCloneStatus.FAILED == "failed"
assert VoiceCloneStatus.DISABLED == "disabled"
def test_terminal_statuses(self):
assert VoiceCloneStatus.READY in TERMINAL_STATUSES
assert VoiceCloneStatus.FAILED in TERMINAL_STATUSES
assert VoiceCloneStatus.DISABLED in TERMINAL_STATUSES
assert VoiceCloneStatus.PENDING not in TERMINAL_STATUSES
assert VoiceCloneStatus.PROCESSING not in TERMINAL_STATUSES
class TestVoiceCloneProfileCreate:
"""VoiceCloneProfile.create 工厂方法测试"""
def test_create_minimal(self):
profile = VoiceCloneProfile.create(user_id="user123", name="我的音色")
assert profile.id is not None
assert len(profile.id) == 32 # uuid4 hex
assert profile.user_id == "user123"
assert profile.name == "我的音色"
assert profile.status == VoiceCloneStatus.PENDING
assert profile.retry_count == 0
assert profile.max_retries == 3
assert profile.gender == "unknown"
assert profile.language == "zh-CN"
assert profile.created_at is not None
assert profile.updated_at is not None
def test_create_with_all_fields(self):
profile = VoiceCloneProfile.create(
user_id="user456",
name="测试音色",
description="这是一个测试音色",
source_audio_url="https://example.com/audio.wav",
voice_model="cosyvoice-v2",
language="en-US",
gender="MALE",
max_retries=5,
metadata={"source": "upload"},
)
assert profile.user_id == "user456"
assert profile.name == "测试音色"
assert profile.description == "这是一个测试音色"
assert profile.source_audio_url == "https://example.com/audio.wav"
assert profile.voice_model == "cosyvoice-v2"
assert profile.language == "en-US"
assert profile.gender == "male" # 转小写
assert profile.max_retries == 5
assert profile.metadata == {"source": "upload"}
def test_create_strips_whitespace(self):
profile = VoiceCloneProfile.create(
user_id=" user789 ",
name=" 我的音色 ",
description=" 描述 ",
)
assert profile.user_id == "user789"
assert profile.name == "我的音色"
assert profile.description == "描述"
def test_create_empty_user_id_raises(self):
with pytest.raises(ValueError, match="user_id 不能为空"):
VoiceCloneProfile.create(user_id=" ", name="测试")
def test_create_empty_name_raises(self):
with pytest.raises(ValueError, match="name 不能为空"):
VoiceCloneProfile.create(user_id="user1", name=" ")
def test_create_name_too_long_raises(self):
long_name = "a" * 101
with pytest.raises(ValueError, match="name 长度不能超过 100 字符"):
VoiceCloneProfile.create(user_id="user1", name=long_name)
def test_create_name_exactly_100_chars_ok(self):
name = "a" * 100
profile = VoiceCloneProfile.create(user_id="user1", name=name)
assert profile.name == name
def test_create_default_metadata_is_dict(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
assert profile.metadata == {}
# 不应该共享同一个 dict
p2 = VoiceCloneProfile.create(user_id="u2", name="test2")
assert profile.metadata is not p2.metadata
class TestVoiceCloneProfileProperties:
"""属性测试"""
def test_is_terminal_ready(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
assert profile.is_terminal is True
def test_is_terminal_failed(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.FAILED
assert profile.is_terminal is True
def test_is_terminal_disabled(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.DISABLED
assert profile.is_terminal is True
def test_is_terminal_pending(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
assert profile.is_terminal is False
def test_is_terminal_processing(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
assert profile.is_terminal is False
def test_is_retryable_failed_within_limit(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
profile.status = VoiceCloneStatus.FAILED
profile.retry_count = 1
assert profile.is_retryable is True
def test_is_retryable_failed_at_limit(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
profile.status = VoiceCloneStatus.FAILED
profile.retry_count = 3
assert profile.is_retryable is False
def test_is_retryable_pending(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
assert profile.is_retryable is False
def test_is_ready_with_voice_id(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
profile.voice_id = "voice_123"
assert profile.is_ready is True
def test_is_ready_without_voice_id(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
profile.voice_id = ""
assert profile.is_ready is False
def test_is_ready_wrong_status(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.voice_id = "voice_123"
assert profile.is_ready is False # pending status
class TestStateTransitions:
"""状态转换测试"""
def test_pending_to_processing(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
old_updated = profile.updated_at
profile.transition_to(VoiceCloneStatus.PROCESSING)
assert profile.status == VoiceCloneStatus.PROCESSING
assert profile.updated_at >= old_updated
def test_pending_to_failed(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.transition_to(VoiceCloneStatus.FAILED)
assert profile.status == VoiceCloneStatus.FAILED
def test_pending_to_disabled(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.transition_to(VoiceCloneStatus.DISABLED)
assert profile.status == VoiceCloneStatus.DISABLED
def test_processing_to_ready(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.transition_to(VoiceCloneStatus.READY)
assert profile.status == VoiceCloneStatus.READY
def test_processing_to_failed(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.transition_to(VoiceCloneStatus.FAILED)
assert profile.status == VoiceCloneStatus.FAILED
def test_processing_to_disabled(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.transition_to(VoiceCloneStatus.DISABLED)
assert profile.status == VoiceCloneStatus.DISABLED
def test_failed_to_pending_retry(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.FAILED
profile.transition_to(VoiceCloneStatus.PENDING)
assert profile.status == VoiceCloneStatus.PENDING
def test_ready_to_disabled(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
profile.transition_to(VoiceCloneStatus.DISABLED)
assert profile.status == VoiceCloneStatus.DISABLED
def test_transition_with_string(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.transition_to("processing")
assert profile.status == VoiceCloneStatus.PROCESSING
def test_transition_invalid_string_raises(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
with pytest.raises(ValueError, match="无效状态"):
profile.transition_to("invalid_status")
def test_invalid_transition_pending_to_ready(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
with pytest.raises(ValueError, match="非法状态转换"):
profile.transition_to(VoiceCloneStatus.READY)
def test_invalid_transition_ready_to_processing(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
with pytest.raises(ValueError, match="非法状态转换"):
profile.transition_to(VoiceCloneStatus.PROCESSING)
def test_invalid_transition_failed_to_ready(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.FAILED
with pytest.raises(ValueError, match="非法状态转换"):
profile.transition_to(VoiceCloneStatus.READY)
def test_invalid_transition_disabled_to_pending(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.DISABLED
with pytest.raises(ValueError, match="非法状态转换"):
profile.transition_to(VoiceCloneStatus.PENDING)
class TestMarkMethods:
"""标记方法测试"""
def test_mark_processing(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.error_message = "some error"
profile.mark_processing()
assert profile.status == VoiceCloneStatus.PROCESSING
assert profile.error_message == ""
def test_mark_ready(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.error_message = "old error"
profile.mark_ready("voice_abc123")
assert profile.status == VoiceCloneStatus.READY
assert profile.voice_id == "voice_abc123"
assert profile.error_message == ""
def test_mark_ready_empty_voice_id_raises(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
with pytest.raises(ValueError, match="voice_id 不能为空"):
profile.mark_ready(" ")
def test_mark_ready_strips_whitespace(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.mark_ready(" voice_123 ")
assert profile.voice_id == "voice_123"
def test_mark_failed(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.mark_failed("音频质量太差")
assert profile.status == VoiceCloneStatus.FAILED
assert profile.error_message == "音频质量太差"
def test_mark_disabled_from_pending(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.mark_disabled()
assert profile.status == VoiceCloneStatus.DISABLED
def test_mark_disabled_from_ready(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
profile.voice_id = "v1"
profile.mark_disabled()
assert profile.status == VoiceCloneStatus.DISABLED
assert profile.voice_id == "v1" # 禁用不清除voice_id
class TestPrepareRetry:
"""重试准备测试"""
def test_prepare_retry_success(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
profile.status = VoiceCloneStatus.FAILED
profile.retry_count = 1
profile.error_message = "timeout"
profile.voice_id = "old_voice"
profile.prepare_retry()
assert profile.status == VoiceCloneStatus.PENDING
assert profile.retry_count == 2
assert profile.error_message == ""
assert profile.voice_id == ""
def test_prepare_retry_first_time(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
profile.status = VoiceCloneStatus.FAILED
profile.prepare_retry()
assert profile.status == VoiceCloneStatus.PENDING
assert profile.retry_count == 1
def test_prepare_retry_exceeds_max_raises(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
profile.status = VoiceCloneStatus.FAILED
profile.retry_count = 3
with pytest.raises(ValueError, match="不可重试"):
profile.prepare_retry()
def test_prepare_retry_from_pending_raises(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
with pytest.raises(ValueError, match="不可重试"):
profile.prepare_retry()
def test_prepare_retry_from_ready_raises(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.READY
with pytest.raises(ValueError, match="不可重试"):
profile.prepare_retry()
class TestToDict:
"""序列化测试"""
def test_to_dict_basic(self):
profile = VoiceCloneProfile.create(
user_id="user1",
name="测试音色",
description="desc",
max_retries=2,
)
d = profile.to_dict()
assert d["id"] == profile.id
assert d["user_id"] == "user1"
assert d["name"] == "测试音色"
assert d["description"] == "desc"
assert d["status"] == "pending"
assert d["retry_count"] == 0
assert d["max_retries"] == 2
assert d["is_retryable"] is False
assert d["is_ready"] is False
assert isinstance(d["created_at"], str)
assert isinstance(d["updated_at"], str)
def test_to_dict_ready_state(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.status = VoiceCloneStatus.PROCESSING
profile.mark_ready("voice_123")
d = profile.to_dict()
assert d["status"] == "ready"
assert d["voice_id"] == "voice_123"
assert d["is_ready"] is True
assert d["is_retryable"] is False
def test_to_dict_failed_state(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test")
profile.mark_failed("some error")
d = profile.to_dict()
assert d["status"] == "failed"
assert d["error_message"] == "some error"
assert d["is_retryable"] is True # retry_count=0, max_retries=3
def test_to_dict_includes_metadata(self):
profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata={"key": "value", "num": 42})
d = profile.to_dict()
assert d["metadata"] == {"key": "value", "num": 42}
+306
View File
@@ -0,0 +1,306 @@
"""
微信同步登录/注册 Use Case 测试
"""
from datetime import datetime, timezone
from unittest.mock import Mock, patch
import pytest
from packages.application.auth.wechat_sync_use_case import (
WechatSyncRequest,
WechatSyncUseCase,
)
from packages.domain.entities import User
class TestWechatSyncRequest:
"""微信同步请求对象测试"""
def test_request_with_basic_fields(self):
"""测试基本字段初始化"""
request = WechatSyncRequest(openid="openid123")
assert request.openid == "openid123"
assert request.unionid == ""
assert request.nickname == "微信用户"
assert request.avatar_url == ""
assert request.source == "miniapp"
def test_request_with_all_fields(self):
"""测试完整字段初始化"""
request = WechatSyncRequest(
openid=" openid123 ",
unionid=" unionid456 ",
nickname="测试用户",
avatar_url="http://example.com/avatar.jpg",
source="h5",
)
assert request.openid == "openid123" # stripped
assert request.unionid == "unionid456" # stripped
assert request.nickname == "测试用户"
assert request.avatar_url == "http://example.com/avatar.jpg"
assert request.source == "h5"
def test_request_empty_unionid_stays_empty(self):
"""测试空 unionid 处理"""
request = WechatSyncRequest(openid="openid123", unionid="")
assert request.unionid == ""
def test_request_none_nickname_defaults(self):
"""测试空昵称使用默认值"""
request = WechatSyncRequest(openid="openid123", nickname="")
assert request.nickname == "微信用户"
class TestWechatSyncUseCase:
"""微信同步登录/注册用例测试"""
@pytest.fixture
def mock_user_repo(self):
"""Mock 用户仓储"""
repo = Mock()
repo.find_by_wechat_openid = Mock(return_value=None)
repo.find_by_wechat_unionid = Mock(return_value=None)
repo.find_by_username = Mock(return_value=None)
repo.find_by_email = Mock(return_value=None)
repo.save = Mock()
repo.get = Mock(return_value=None)
return repo
@pytest.fixture
def mock_session_store(self):
"""Mock Session 存储"""
store = Mock()
store.save_session = Mock(return_value=True)
store.get_refresh_token = Mock(return_value=None)
store.get_session_by_refresh_token = Mock(return_value=None)
store.delete_session = Mock(return_value=True)
return store
@pytest.fixture
def test_user(self):
"""测试用户"""
return User(
id="user-123",
email="test@example.com",
username="testuser",
display_name="测试用户",
password_hash="hashed_password",
wechat_openid="openid123",
wechat_unionid="unionid456",
)
@pytest.fixture
def use_case(self, mock_user_repo, mock_session_store):
"""创建微信同步用例"""
return WechatSyncUseCase(
user_repository=mock_user_repo,
session_store=mock_session_store,
jwt_secret_key="test-secret-key-for-unit-tests",
)
# ===== 登录场景:openid 找到用户 =====
def test_login_by_openid_success(self, use_case, mock_user_repo, mock_session_store, test_user):
"""测试通过 openid 登录成功"""
mock_user_repo.find_by_wechat_openid.return_value = test_user
request = WechatSyncRequest(openid="openid123")
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.user_id == "user-123"
assert response.nickname == "测试用户"
assert response.is_new_user is False
assert response.access_token != ""
assert response.refresh_token != ""
assert response.expires_in > 0
# 验证 session 已保存
mock_session_store.save_session.assert_called_once()
save_kwargs = mock_session_store.save_session.call_args.kwargs
assert save_kwargs["user_id"] == "user-123"
assert "wechat_miniapp" in save_kwargs["device_info"]
# 验证更新了最后登录信息
mock_user_repo.save.assert_called_once()
saved_user = mock_user_repo.save.call_args[0][0]
assert saved_user.last_login_at is not None
assert saved_user.last_login_ip == "bff_gateway"
# 验证 to_dict 包含兼容字段
data = response.to_dict()
assert data["access_token"] == response.access_token
assert data["token"] == response.access_token # 兼容字段
assert data["user"]["id"] == "user-123"
assert data["user_info"]["id"] == "user-123"
# ===== 登录场景:openid 没找到,通过 unionid 找到 =====
def test_login_by_unionid_binds_openid(self, use_case, mock_user_repo, mock_session_store, test_user):
"""测试通过 unionid 找到用户并绑定当前 openid"""
# openid 没找到
mock_user_repo.find_by_wechat_openid.return_value = None
# unionid 找到了(但 openid 字段为空)
test_user.wechat_openid = None
mock_user_repo.find_by_wechat_unionid.return_value = test_user
request = WechatSyncRequest(
openid="new_openid_789",
unionid="unionid456",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.is_new_user is False
assert response.user_id == "user-123"
# 验证绑定了新的 openid(save 被调用了两次:一次绑定 openid,一次更新登录信息)
assert mock_user_repo.save.call_count == 2
# 第一次 save 应该是绑定 openid
first_save_user = mock_user_repo.save.call_args_list[0][0][0]
assert first_save_user.wechat_openid == "new_openid_789"
def test_login_by_unionid_no_binding_needed(self, use_case, mock_user_repo, mock_session_store, test_user):
"""测试通过 unionid 找到用户且 openid 已存在时(不需要额外绑定)"""
mock_user_repo.find_by_wechat_openid.return_value = None
mock_user_repo.find_by_wechat_unionid.return_value = test_user
request = WechatSyncRequest(
openid="openid123", # 跟用户已有的一样
unionid="unionid456",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.is_new_user is False
# 还是会 save(绑定)+ save(更新登录信息)= 2次
assert mock_user_repo.save.call_count == 2
# ===== 注册场景:openid 和 unionid 都没找到,创建新用户 =====
def test_register_new_user(self, use_case, mock_user_repo, mock_session_store):
"""测试创建新微信用户"""
mock_user_repo.find_by_wechat_openid.return_value = None
mock_user_repo.find_by_wechat_unionid.return_value = None
mock_user_repo.find_by_username.return_value = None
request = WechatSyncRequest(
openid="new_openid",
unionid="new_unionid",
nickname="新用户",
avatar_url="http://example.com/avatar.jpg",
source="miniapp",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.is_new_user is True
assert response.nickname == "新用户"
assert response.access_token != ""
assert response.refresh_token != ""
# 验证用户被创建并保存
assert mock_user_repo.save.call_count >= 1
# 找到 save 的用户(可能有多次save,找第一次即创建用户的那次)
created_user = None
for call in mock_user_repo.save.call_args_list:
user = call[0][0]
if user.wechat_openid == "new_openid":
created_user = user
break
assert created_user is not None
assert created_user.wechat_openid == "new_openid"
assert created_user.wechat_unionid == "new_unionid"
assert created_user.email_verified is True
assert created_user.username.startswith("wx_")
assert "@wechat.local" in created_user.email
def test_register_new_user_without_unionid(self, use_case, mock_user_repo, mock_session_store):
"""测试创建无 unionid 的新用户"""
mock_user_repo.find_by_wechat_openid.return_value = None
mock_user_repo.find_by_username.return_value = None
request = WechatSyncRequest(openid="openid_no_union")
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.is_new_user is True
created_user = mock_user_repo.save.call_args_list[0][0][0]
assert created_user.wechat_unionid is None
def test_register_username_conflict_adds_suffix(self, use_case, mock_user_repo, mock_session_store):
"""测试用户名冲突时自动加后缀"""
mock_user_repo.find_by_wechat_openid.return_value = None
mock_user_repo.find_by_wechat_unionid.return_value = None
# 第一次 find_by_username 返回存在(冲突),第二次返回 None(生成了带后缀的新名)
mock_user_repo.find_by_username.side_effect = [Mock(), None]
request = WechatSyncRequest(openid="conflict_openid")
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.is_new_user is True
# find_by_username 应该被调用了两次
assert mock_user_repo.find_by_username.call_count == 2
# 第二个用户名应该带后缀 _1
second_call_username = mock_user_repo.find_by_username.call_args_list[1][0][0]
assert "_1" in second_call_username
def test_register_default_nickname_when_empty(self, use_case, mock_user_repo, mock_session_store):
"""测试新用户空昵称时使用默认值"""
mock_user_repo.find_by_wechat_openid.return_value = None
mock_user_repo.find_by_username.return_value = None
request = WechatSyncRequest(openid="openid123", nickname="")
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.nickname == "微信用户"
# ===== 错误场景 =====
def test_missing_openid(self, use_case):
"""测试缺少 openid"""
request = WechatSyncRequest(openid="")
response, error = use_case.execute(request)
assert response is None
assert error == "openid is required"
def test_exception_handling(self, use_case, mock_user_repo):
"""测试异常处理"""
mock_user_repo.find_by_wechat_openid.side_effect = Exception("DB error")
request = WechatSyncRequest(openid="openid123")
response, error = use_case.execute(request)
assert response is None
assert "Internal error" in error
assert "DB error" in error
# ===== Session 保存验证 =====
def test_session_saved_with_correct_params(self, use_case, mock_user_repo, mock_session_store, test_user):
"""测试 session 保存参数正确"""
mock_user_repo.find_by_wechat_openid.return_value = test_user
request = WechatSyncRequest(openid="openid123", source="h5")
use_case.execute(request)
mock_session_store.save_session.assert_called_once()
kwargs = mock_session_store.save_session.call_args.kwargs
assert kwargs["user_id"] == "user-123"
assert kwargs["refresh_token"] != ""
assert "wechat_h5" in kwargs["device_info"]
assert kwargs["ip_address"] == "bff_gateway"
assert kwargs["expires_in_seconds"] == 30 * 24 * 3600 # 30天