test: 第80波单测 - module_registry+subtitle+voice_library (+74) #888

Merged
xiaoxia merged 1 commits from test/wave80-module-registry-subtitle-more into develop 2026-07-26 01:31:53 +08:00
3 changed files with 475 additions and 74 deletions
+294
View File
@@ -382,3 +382,297 @@ class TestModuleStatus:
assert ModuleStatus.ACTIVE.value == "active"
assert ModuleStatus.DISABLED.value == "disabled"
assert ModuleStatus.ERROR.value == "error"
# ── Module 更多状态转换测试 ──────────────────────────────────────────────────
class TestModuleStateTransitions:
"""Module 状态转换补充测试."""
def test_activate_twice_idempotent(self):
"""多次激活不报错."""
mod = Module(name="m1")
mod.activate()
mod.activate()
assert mod.status == ModuleStatus.ACTIVE
def test_disable_twice_idempotent(self):
"""多次禁用不报错."""
mod = Module(name="m1", status=ModuleStatus.ACTIVE)
mod.disable()
mod.disable()
assert mod.status == ModuleStatus.DISABLED
def test_disable_from_registered(self):
"""从registered状态禁用."""
mod = Module(name="m1")
mod.disable()
assert mod.status == ModuleStatus.DISABLED
def test_activate_after_disable(self):
"""禁用后重新激活."""
mod = Module(name="m1")
mod.activate()
mod.disable()
mod.activate()
assert mod.status == ModuleStatus.ACTIVE
def test_error_state_cannot_be_activated(self):
"""error状态不能被激活."""
mod = Module(name="m1", status=ModuleStatus.ERROR)
mod.activate()
assert mod.status == ModuleStatus.ERROR
def test_error_state_can_be_disabled(self):
"""error状态可以被禁用(disable 不检查状态)."""
mod = Module(name="m1", status=ModuleStatus.ERROR)
mod.disable()
assert mod.status == ModuleStatus.DISABLED
# ── ModuleRegistry 注册补充测试 ──────────────────────────────────────────────
class TestModuleRegistryRegisterMore:
"""模块注册补充场景."""
def test_register_multiple_modules(self):
"""注册多个模块."""
registry = ModuleRegistry()
for i in range(5):
registry.register(Module(name=f"mod_{i}"))
assert len(registry.list_modules()) == 5
def test_register_order_independent_deps(self):
"""先注册依赖方,后注册被依赖方,依赖方不会自动激活."""
registry = ModuleRegistry()
registry.register(Module(name="plugin", dependencies=["core"]))
assert registry.get("plugin").status == ModuleStatus.REGISTERED
# 注册core后,plugin仍然是REGISTERED(不会自动检查)
registry.register(Module(name="core"))
assert registry.get("core").status == ModuleStatus.ACTIVE
def test_register_with_multiple_deps_all_satisfied(self):
"""所有依赖都满足时自动激活."""
registry = ModuleRegistry()
registry.register(Module(name="dep_a"))
registry.register(Module(name="dep_b"))
registry.register(Module(name="plugin", dependencies=["dep_a", "dep_b"]))
assert registry.get("plugin").status == ModuleStatus.ACTIVE
def test_register_with_multiple_deps_partial_missing(self):
"""部分依赖缺失时不激活."""
registry = ModuleRegistry()
registry.register(Module(name="dep_a"))
registry.register(Module(name="plugin", dependencies=["dep_a", "dep_b"]))
assert registry.get("plugin").status == ModuleStatus.REGISTERED
def test_register_self_dependency_handled(self):
"""自依赖不会导致死循环(依赖检查时找不到自己)."""
registry = ModuleRegistry()
registry.register(Module(name="self_dep", dependencies=["self_dep"]))
# 注册时自己还没加入 _modules,检查依赖时找不到,保持REGISTERED
assert registry.get("self_dep").status == ModuleStatus.REGISTERED
# ── ModuleRegistry 能力查询补充 ─────────────────────────────────────────────
class TestModuleRegistryCapabilitiesMore:
"""能力查询补充测试."""
def test_multiple_modules_same_capability_returns_first(self):
"""多个模块提供同一能力,get_capability返回第一个."""
registry = ModuleRegistry()
cap1 = ModuleCapability(name="render", description="渲染器A")
cap2 = ModuleCapability(name="render", description="渲染器B")
registry.register(Module(name="mod_a", capabilities=[cap1]))
registry.register(Module(name="mod_b", capabilities=[cap2]))
result = registry.get_capability("render")
assert result is not None
assert result.name == "render"
def test_has_capability_case_sensitive(self):
"""能力名大小写敏感."""
registry = ModuleRegistry()
registry.register(Module(name="m1", capabilities=[ModuleCapability(name="Render")]))
assert registry.has_capability("Render") is True
assert registry.has_capability("render") is False
def test_get_quota_rules_nonexistent_capability(self):
"""不存在的能力返回空配额列表."""
registry = ModuleRegistry()
rules = registry.get_quota_rules("nonexistent")
assert rules == []
def test_get_active_capabilities_empty_registry(self):
"""空注册中心返回空字典."""
registry = ModuleRegistry()
result = registry.get_active_capabilities()
assert result == {}
def test_get_active_capabilities_skips_inactive(self):
"""非激活模块的能力不计入."""
registry = ModuleRegistry()
registry.register(Module(name="active_mod", capabilities=[ModuleCapability(name="cap1")]))
disabled = Module(
name="disabled_mod",
status=ModuleStatus.DISABLED,
capabilities=[ModuleCapability(name="cap2")],
)
registry._modules["disabled_mod"] = disabled
result = registry.get_active_capabilities()
assert "active_mod" in result
assert "disabled_mod" not in result
def test_get_active_capabilities_skips_no_cap_modules(self):
"""无能力的模块不出现在结果中."""
registry = ModuleRegistry()
registry.register(Module(name="no_cap_mod"))
registry.register(Module(name="has_cap_mod", capabilities=[ModuleCapability(name="cap1")]))
result = registry.get_active_capabilities()
assert "no_cap_mod" not in result
assert "has_cap_mod" in result
def test_module_with_multiple_capabilities(self):
"""单个模块有多个能力."""
registry = ModuleRegistry()
caps = [
ModuleCapability(name="cap_a"),
ModuleCapability(name="cap_b"),
ModuleCapability(name="cap_c"),
]
registry.register(Module(name="multi_mod", capabilities=caps))
assert registry.has_capability("cap_a")
assert registry.has_capability("cap_b")
assert registry.has_capability("cap_c")
assert len(registry.get_active_capabilities()["multi_mod"]) == 3
# ── ModuleRegistry 依赖检查补充 ────────────────────────────────────────────
class TestModuleRegistryDependenciesMore:
"""依赖检查补充测试."""
def test_multiple_dependencies_all_active(self):
"""多个依赖都激活."""
registry = ModuleRegistry()
registry.register(Module(name="dep1"))
registry.register(Module(name="dep2"))
registry.register(Module(name="dep3"))
registry.register(Module(name="plugin", dependencies=["dep1", "dep2", "dep3"]))
assert registry.check_dependencies("plugin") is True
def test_multiple_dependencies_one_inactive(self):
"""多个依赖中有一个未激活."""
registry = ModuleRegistry()
registry.register(Module(name="dep1"))
dep2 = Module(name="dep2", status=ModuleStatus.DISABLED)
registry._modules["dep2"] = dep2
registry.register(Module(name="plugin", dependencies=["dep1", "dep2"]))
# 注册时dep2不是ACTIVEplugin不会自动激活
assert registry.check_dependencies("plugin") is False
def test_chain_dependencies(self):
"""链式依赖 A→B→C."""
registry = ModuleRegistry()
registry.register(Module(name="c"))
registry.register(Module(name="b", dependencies=["c"]))
registry.register(Module(name="a", dependencies=["b"]))
# a 依赖 bACTIVE),b 依赖 cACTIVE
# check_dependencies 只检查直接依赖,b 是 ACTIVE 的
assert registry.check_dependencies("a") is True
assert registry.get("a").status == ModuleStatus.ACTIVE
def test_no_dependencies_always_satisfied(self):
"""无依赖的模块总是满足依赖检查."""
registry = ModuleRegistry()
registry.register(Module(name="standalone"))
assert registry.check_dependencies("standalone") is True
# ── ModuleRegistry list_modules 补充 ──────────────────────────────────────
class TestModuleRegistryListMore:
"""list_modules 补充测试."""
def test_list_modules_empty(self):
"""空注册中心."""
registry = ModuleRegistry()
assert registry.list_modules() == []
def test_list_modules_registered_status(self):
"""按registered状态过滤."""
registry = ModuleRegistry()
registry.register(Module(name="active_mod")) # auto ACTIVE
pending = Module(name="pending_mod")
registry._modules["pending_mod"] = pending # REGISTERED
registered = registry.list_modules(status=ModuleStatus.REGISTERED)
assert len(registered) == 1
assert registered[0].name == "pending_mod"
def test_list_modules_error_status(self):
"""按error状态过滤."""
registry = ModuleRegistry()
error_mod = Module(name="err", status=ModuleStatus.ERROR)
registry._modules["err"] = error_mod
errors = registry.list_modules(status=ModuleStatus.ERROR)
assert len(errors) == 1
assert errors[0].name == "err"
# ── QuotaRule 补充测试 ──────────────────────────────────────────────────────
class TestQuotaRuleMore:
"""QuotaRule 补充测试."""
def test_zero_per_operation(self):
"""零消耗配额规则."""
rule = QuotaRule(dimension="free_ops", per_operation=0.0)
assert rule.per_operation == 0.0
def test_fractional_per_operation(self):
"""小数消耗配额规则."""
rule = QuotaRule(dimension="storage", per_operation=0.001)
assert rule.per_operation == 0.001
def test_large_per_operation(self):
"""大数值消耗."""
rule = QuotaRule(dimension="tokens", per_operation=10000.0)
assert rule.per_operation == 10000.0
# ── ModuleCapability 补充测试 ──────────────────────────────────────────────
class TestModuleCapabilityMore:
"""ModuleCapability 补充测试."""
def test_empty_metadata(self):
"""默认metadata为空字典."""
cap = ModuleCapability(name="test")
assert cap.metadata == {}
def test_metadata_preserved(self):
"""元数据完整保存."""
meta = {"model": "v1", "speed": 1.5, "enabled": True}
cap = ModuleCapability(name="test", metadata=meta)
assert cap.metadata["model"] == "v1"
assert cap.metadata["speed"] == 1.5
assert cap.metadata["enabled"] is True
def test_multiple_quota_rules(self):
"""多个配额规则."""
rules = [
QuotaRule("dim1", 1.0),
QuotaRule("dim2", 2.0),
QuotaRule("dim3", 3.0),
]
cap = ModuleCapability(name="test", quota_rules=rules)
assert len(cap.quota_rules) == 3
assert cap.quota_rules[0].dimension == "dim1"
assert cap.quota_rules[2].per_operation == 3.0
+125
View File
@@ -415,6 +415,48 @@ class TestMergeShortSegmentsEdgeCases:
result = timeline.merge_short_segments(min_chars=8)
assert result.segment_count == 1
assert result.segments[0].text == "一二三四五六七八九"
def test_min_chars_one_no_merge(self):
"""min_chars=1 时每个都够,不合并."""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="", start=0.0, end=0.5),
SubtitleSegment(text="", start=0.5, end=1.0),
SubtitleSegment(text="", start=1.0, end=1.5),
],
)
result = tl.merge_short_segments(min_chars=1)
assert result.segment_count == 3
def test_min_chars_very_large_all_merged(self):
"""min_chars 极大,全部合并成一段."""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="第一句", start=0.0, end=1.0),
SubtitleSegment(text="第二句", start=1.0, end=2.0),
SubtitleSegment(text="第三句", start=2.0, end=3.0),
],
)
result = tl.merge_short_segments(min_chars=100)
assert result.segment_count == 1
assert result.total_chars == tl.total_chars
def test_merge_preserves_word_level_info(self):
"""合并后词级信息完整保留,顺序正确."""
w1 = [SubtitleWord(text="", start=0.0, end=0.3)]
w2 = [SubtitleWord(text="", start=0.3, end=0.6)]
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="", start=0.0, end=0.3, words=w1),
SubtitleSegment(text="", start=0.3, end=0.6, words=w2),
],
)
result = tl.merge_short_segments(min_chars=4)
assert result.segment_count == 1
assert len(result.segments[0].words) == 2
assert result.segments[0].words[0].text == ""
assert result.segments[0].words[1].text == ""
class TestSplitLongSegmentsEdgeCases:
@@ -473,6 +515,49 @@ class TestSplitLongSegmentsEdgeCases:
timeline = SubtitleTimeline(segments=[], total_duration=0.0)
result = timeline.split_long_segments(max_chars=10)
assert result.segment_count == 0
def test_exactly_max_chars_no_split(self):
"""恰好等于 max_chars 不拆分."""
text = "一二三四五六七八九十" # 10字
seg = SubtitleSegment(text=text, start=0.0, end=5.0)
tl = SubtitleTimeline(segments=[seg])
result = tl.split_long_segments(max_chars=10)
assert result.segment_count == 1
def test_one_char_over_triggers_split(self):
"""超过1个字符就触发拆分."""
text = "一二三四五六七八九十1" # 11字
seg = SubtitleSegment(text=text, start=0.0, end=5.0)
tl = SubtitleTimeline(segments=[seg])
result = tl.split_long_segments(max_chars=10)
assert result.segment_count >= 2
def test_mixed_short_and_long_segments(self):
"""长短片段混合,只拆分超长的."""
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="", start=0.0, end=0.5),
SubtitleSegment(
text="这是一段很长很长需要拆分的字幕内容",
start=0.5,
end=3.0,
),
SubtitleSegment(text="", start=3.0, end=3.5),
],
)
result = tl.split_long_segments(max_chars=10)
assert result.segment_count > 3
assert result.segments[0].text == ""
assert result.segments[-1].text == ""
def test_total_chars_preserved_after_split(self):
"""拆分后总字数保持不变."""
text = "一二三四五六七八九十一二三四五六七八九十一二三四五"
seg = SubtitleSegment(text=text, start=0.0, end=10.0)
tl = SubtitleTimeline(segments=[seg])
result = tl.split_long_segments(max_chars=10)
assert result.total_chars == len(text)
class TestSplitTextByPunctuationDeep:
@@ -516,6 +601,46 @@ class TestSplitTextByPunctuationDeep:
assert result[-1].endswith("")
class TestSplitTextByPunctuationEdgeCases:
"""_split_text_by_punctuation 边界场景补充."""
def test_colon_semicolon_splits(self):
"""冒号分号也能触发拆分."""
text = "第一段:第二段;第三段"
result = SubtitleTimeline._split_text_by_punctuation(text, 5)
assert len(result) >= 2
def test_punctuation_at_start(self):
"""标点在开头不崩溃,字符完整保留."""
text = ",你好世界"
result = SubtitleTimeline._split_text_by_punctuation(text, 10)
assert "".join(result) == text
def test_consecutive_punctuation(self):
"""连续标点符号,字符完整保留."""
text = "你好!!!测试。。。"
result = SubtitleTimeline._split_text_by_punctuation(text, 3)
assert "".join(result) == text
def test_single_character_text(self):
"""单字符文本不拆分."""
result = SubtitleTimeline._split_text_by_punctuation("", 10)
assert len(result) == 1
assert result[0] == ""
def test_only_punctuation(self):
"""纯标点符号文本不崩溃."""
result = SubtitleTimeline._split_text_by_punctuation("。。。", 10)
assert isinstance(result, list)
assert "".join(result) == "。。。"
def test_mixed_fullwidth_halfwidth_punctuation(self):
"""全角半角标点混合."""
text = "你好,世界!测试?完成"
result = SubtitleTimeline._split_text_by_punctuation(text, 5)
assert "".join(result) == text
assert len(result) >= 2
class TestSubtitleTimelineProperties:
"""SubtitleTimeline 属性计算深度测试."""
+56 -74
View File
@@ -95,91 +95,73 @@ class TestVoiceLibraryItem:
assert item.updated_at is not None
class TestVoiceLibraryItemExtended:
"""VoiceLibraryItem 深度补充测试"""
# ── VoiceLibraryItem 更多边界测试 ──────────────────────────────────────
def test_zero_duration(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", duration=0)
class TestVoiceLibraryItemMore:
"""VoiceLibraryItem 补充测试."""
def test_status_pending(self):
"""pending状态."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", status="pending")
assert item.status == "pending"
def test_status_processing(self):
"""processing状态."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", status="processing")
assert item.status == "processing"
def test_status_failed(self):
"""failed状态."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", status="failed")
assert item.status == "failed"
def test_zero_duration_and_size(self):
"""零时长零大小."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", duration=0, file_size=0)
assert item.duration == 0
def test_negative_duration(self):
"""负数 duration 领域层不校验"""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", duration=-1.5)
assert item.duration == -1.5
def test_large_duration(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", duration=9999.99)
assert item.duration == 9999.99
def test_zero_file_size(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", file_size=0)
assert item.file_size == 0
def test_large_file_size(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", file_size=10**9)
assert item.file_size == 10**9
def test_large_duration_and_size(self):
"""大时长大文件."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", duration=3600.0, file_size=1024 * 1024 * 100)
assert item.duration == 3600.0
assert item.file_size == 104857600
def test_empty_audio_url(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", audio_url="")
assert item.audio_url == ""
def test_tags_independence(self):
item1 = VoiceLibraryItem(id="v1", user_id="u1", name="n1")
item2 = VoiceLibraryItem(id="v2", user_id="u1", name="n2")
item1.tags.append("新标签")
assert "新标签" not in item2.tags
assert len(item2.tags) == 0
def test_metadata_independence(self):
item1 = VoiceLibraryItem(id="v1", user_id="u1", name="n1")
item2 = VoiceLibraryItem(id="v2", user_id="u1", name="n2")
item1.metadata_["key"] = "val"
assert "key" not in item2.metadata_
def test_empty_tags_list(self):
"""空标签列表."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", tags=[])
assert item.tags == []
def test_many_tags(self):
tags = [f"tag_{i}" for i in range(30)]
"""多个标签."""
tags = ["温柔", "女声", "情感", "治愈", "朗读", "故事"]
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", tags=tags)
assert len(item.tags) == 30
assert item.tags[0] == "tag_0"
assert len(item.tags) == 6
assert "治愈" in item.tags
def test_empty_text(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", text="")
assert item.text == ""
def test_empty_metadata(self):
"""空元数据."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n")
assert item.metadata_ == {}
def test_long_text(self):
long_text = "这是一段很长的配音文本" * 100
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", text=long_text)
assert item.text == long_text
assert len(item.text) == 1100
def test_metadata_with_nested_values(self):
"""嵌套元数据."""
meta = {
"settings": {"speed": 1.0, "pitch": 0.5},
"source": "upload",
"version": 2,
}
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", metadata_=meta)
assert item.metadata_["settings"]["speed"] == 1.0
assert item.metadata_["source"] == "upload"
def test_empty_voice_id(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", voice_id="")
assert item.voice_id == ""
def test_empty_project_id(self):
"""project_id 默认是 None 不是空字符串"""
def test_project_id_none_by_default(self):
"""默认project_id为None."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n")
assert item.project_id is None
def test_project_id_with_string(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", project_id="")
# 传空字符串的话就是空字符串
assert item.project_id == ""
def test_status_empty_string(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", status="")
assert item.status == ""
def test_unicode_name(self):
item = VoiceLibraryItem(id="v1", user_id="u1", name="🎙️ 专业配音 · 龙小淳")
assert "🎙️" in item.name
assert "龙小淳" in item.name
def test_special_characters_in_name(self):
special = "配!@#$%^&*()音"
item = VoiceLibraryItem(id="v1", user_id="u1", name=special)
assert item.name == special
def test_empty_user_id(self):
item = VoiceLibraryItem(id="v1", user_id="", name="n")
assert item.user_id == ""
def test_project_id_set(self):
"""设置project_id."""
item = VoiceLibraryItem(id="v1", user_id="u1", name="n", project_id="proj-abc-123")
assert item.project_id == "proj-abc-123"