diff --git a/tests/unit/test_module_registry.py b/tests/unit/test_module_registry.py index 41ccced6e..e473c2a28 100755 --- a/tests/unit/test_module_registry.py +++ b/tests/unit/test_module_registry.py @@ -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不是ACTIVE,plugin不会自动激活 + 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 依赖 b(ACTIVE),b 依赖 c(ACTIVE) + # 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 diff --git a/tests/unit/test_subtitle_domain.py b/tests/unit/test_subtitle_domain.py index 734c3c038..74dfe6c69 100755 --- a/tests/unit/test_subtitle_domain.py +++ b/tests/unit/test_subtitle_domain.py @@ -270,3 +270,210 @@ class TestMergeSegments: assert result.text == "第一第二" assert result.start == 0.0 assert result.end == 2.0 + + +# ── merge_short_segments 更多边界 ────────────────────────────────────── + + +class TestMergeShortSegmentsMore: + """merge_short_segments 补充边界测试.""" + + def test_all_short_segments_merged_into_one(self): + """所有片段都很短,合并成一段.""" + segs = [ + SubtitleSegment(text="你", start=0.0, end=0.1), + SubtitleSegment(text="好", start=0.1, end=0.2), + SubtitleSegment(text="世", start=0.2, end=0.3), + SubtitleSegment(text="界", start=0.3, end=0.4), + ] + tl = SubtitleTimeline(segments=segs) + result = tl.merge_short_segments(min_chars=8) + assert result.segment_count == 1 + assert result.segments[0].text == "你好世界" + assert result.segments[0].start == 0.0 + assert result.segments[0].end == 0.4 + + def test_min_chars_very_small(self): + """min_chars 很小(1),几乎不合并.""" + segs = [ + SubtitleSegment(text="一", start=0.0, end=0.5), + SubtitleSegment(text="二", start=0.5, end=1.0), + SubtitleSegment(text="三", start=1.0, end=1.5), + ] + tl = SubtitleTimeline(segments=segs) + result = tl.merge_short_segments(min_chars=1) + # 每个1字 >= min_chars(1),所以都独立成段 + assert result.segment_count == 3 + + def test_min_chars_very_large(self): + """min_chars 很大,全部合并.""" + segs = [ + SubtitleSegment(text="第一句很长的内容", start=0.0, end=1.0), + SubtitleSegment(text="第二句也不短", start=1.0, end=2.0), + SubtitleSegment(text="第三句还行", start=2.0, end=3.0), + ] + tl = SubtitleTimeline(segments=segs) + result = tl.merge_short_segments(min_chars=100) + assert result.segment_count == 1 + assert result.total_chars == tl.total_chars + + def test_merge_preserves_total_duration(self): + """合并后总时长不变.""" + segs = [ + SubtitleSegment(text="短1", start=0.0, end=0.5), + SubtitleSegment(text="短2", start=0.5, end=1.0), + SubtitleSegment(text="很长的一段内容", start=1.0, end=3.0), + ] + tl = SubtitleTimeline(segments=segs, total_duration=3.0) + result = tl.merge_short_segments(min_chars=4) + assert result.total_duration == 3.0 + + 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)] + segs = [ + SubtitleSegment(text="你", start=0.0, end=0.3, words=w1), + SubtitleSegment(text="好", start=0.3, end=0.6, words=w2), + ] + tl = SubtitleTimeline(segments=segs) + 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 == "好" + + def test_trailing_short_merged_to_last(self): + """尾部极短段合并到最后一段.""" + segs = [ + SubtitleSegment(text="一二三四五六七八九十", start=0.0, end=1.0), + SubtitleSegment(text="尾", start=1.0, end=1.1), + ] + tl = SubtitleTimeline(segments=segs) + result = tl.merge_short_segments(min_chars=5) + # 第一段10字 >= 5 → 先入merged + # "尾" = 1字 < 5 → 进入buffer + # 循环结束,buffer非空,检查是否合并到最后一段 + # merged[-1] = 10字 + 1字 = 11字 <= max_chars(这里没限制) + assert result.segment_count == 1 + assert result.segments[0].text == "一二三四五六七八九十尾" + + +# ── split_long_segments 更多边界 ────────────────────────────────────── + + +class TestSplitLongSegmentsMore: + """split_long_segments 补充边界测试.""" + + 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_over_max_chars_splits(self): + """超过1个字符就拆分.""" + text = "一二三四五六七八九十一" # 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 + assert result.total_chars == 11 + + def test_mixed_short_and_long_segments(self): + """长短片段混合,只拆分长的.""" + segs = [ + SubtitleSegment(text="短", start=0.0, end=0.5), + SubtitleSegment(text="很长很长很长的一段文字内容超过限制", start=0.5, end=3.0), + SubtitleSegment(text="也短", start=3.0, end=3.5), + ] + tl = SubtitleTimeline(segments=segs) + result = tl.split_long_segments(max_chars=10) + # 第1段和第3段保持,第2段被拆分 + 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=8) + assert result.total_chars == len(text) + + def test_total_duration_preserved(self): + """拆分后最后一段的结束时间等于原始结束时间.""" + text = "一二三四。五六七八。九十一二。" + seg = SubtitleSegment(text=text, start=1.0, end=5.0) + tl = SubtitleTimeline(segments=[seg]) + result = tl.split_long_segments(max_chars=5) + assert abs(result.segments[-1].end - 5.0) < 0.01 + assert abs(result.segments[0].start - 1.0) < 0.01 + + +# ── _split_text_by_punctuation 更多边界 ─────────────────────────────── + + +class TestSplitTextByPunctuationMore: + """_split_text_by_punctuation 补充边界测试.""" + + def test_comma_splits_when_over_max(self): + """逗号在超长时触发拆分.""" + text = "一二三四五六七八,八七六五四三二一" + result = SubtitleTimeline._split_text_by_punctuation(text, 10) + assert len(result) >= 2 + assert "," in result[0] + + def test_colon_semicolon_splits(self): + """冒号分号也能触发拆分.""" + text = "首先:第一点内容;第二点内容更多一些。" + result = SubtitleTimeline._split_text_by_punctuation(text, 8) + assert len(result) >= 2 + + def test_punctuation_at_start(self): + """标点在开头位置.""" + text = "。这是正文内容开始了" + result = SubtitleTimeline._split_text_by_punctuation(text, 10) + # 开头的句号在 max_chars//2 之前,不会触发句末标点拆分 + # 但整个文本不长,可能不拆 + assert len(result) >= 1 + + def test_consecutive_punctuation(self): + """连续多个标点符号.""" + text = "你好!!!世界???" + result = SubtitleTimeline._split_text_by_punctuation(text, 5) + assert len(result) >= 1 + + def test_english_sentence_end(self): + """英文句末标点也触发拆分.""" + text = "Hello world. This is a test sentence with enough words." + result = SubtitleTimeline._split_text_by_punctuation(text, 15) + assert len(result) >= 2 + + def test_no_punctuation_hard_cut_evenly(self): + """没有标点时均匀硬切.""" + text = "一二三四五六七八九十一二三四五六七八九十一二三四五" # 25字 + result = SubtitleTimeline._split_text_by_punctuation(text, 10) + assert len(result) == 3 + assert len(result[0]) == 10 + assert len(result[1]) == 10 + assert len(result[2]) == 5 + + def test_single_character_text(self): + """单字符文本.""" + result = SubtitleTimeline._split_text_by_punctuation("好", 10) + assert result == ["好"] + + def test_only_punctuation(self): + """只有标点符号.""" + result = SubtitleTimeline._split_text_by_punctuation("。!?", 10) + assert len(result) == 1 + + def test_mixed_fullwidth_halfwidth_punctuation(self): + """全角半角标点混合.""" + text = "你好。再见!谢谢?不用谢。" + result = SubtitleTimeline._split_text_by_punctuation(text, 5) + assert len(result) >= 2 diff --git a/tests/unit/test_voice_library_domain.py b/tests/unit/test_voice_library_domain.py index 9ebe1b6e3..972a77582 100755 --- a/tests/unit/test_voice_library_domain.py +++ b/tests/unit/test_voice_library_domain.py @@ -93,3 +93,75 @@ class TestVoiceLibraryItem: item = VoiceLibraryItem(id="v1", user_id="u1", name="n") assert item.created_at is not None assert item.updated_at is not None + + +# ── VoiceLibraryItem 更多边界测试 ────────────────────────────────────── + + +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 + assert item.file_size == 0 + + 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_tags_list(self): + """空标签列表.""" + item = VoiceLibraryItem(id="v1", user_id="u1", name="n", tags=[]) + assert item.tags == [] + + def test_many_tags(self): + """多个标签.""" + tags = ["温柔", "女声", "情感", "治愈", "朗读", "故事"] + item = VoiceLibraryItem(id="v1", user_id="u1", name="n", tags=tags) + assert len(item.tags) == 6 + assert "治愈" in item.tags + + def test_empty_metadata(self): + """空元数据.""" + item = VoiceLibraryItem(id="v1", user_id="u1", name="n") + assert item.metadata_ == {} + + 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_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_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"