"""TTS 相关领域模块单测 — text_splitter + tts_config + tts_job.""" from __future__ import annotations import pytest from packages.application.tts_job.text_splitter import split_text from packages.domain.tts_config import TtsConfig from packages.domain.tts_job import TERMINAL_STATUSES, TTSJob, TTSJobStatus # ═══════════════════════════════════════════════════════════════════════════════ # text_splitter 文本分段 # ═══════════════════════════════════════════════════════════════════════════════ class TestSplitTextBasic: """基础分段功能.""" def test_empty_text_returns_empty_list(self): assert split_text("", max_chars=500) == [] def test_whitespace_only_returns_empty_list(self): assert split_text(" \n\n ", max_chars=500) == [] def test_short_text_returns_single_segment(self): text = "你好世界" result = split_text(text, max_chars=500) assert len(result) == 1 assert result[0] == text def test_text_equal_to_max_chars_single_segment(self): text = "a" * 500 result = split_text(text, max_chars=500) assert len(result) == 1 assert len(result[0]) == 500 def test_none_max_chars_uses_default(self): """默认 max_chars=500.""" text = "你好" result = split_text(text) # 使用默认值 assert len(result) == 1 class TestSplitTextSentenceBoundary: """按句子边界分段.""" def test_splits_at_period(self): text = "第一句。第二句。第三句。" result = split_text(text, max_chars=10) # 每句都比较短，会在句子边界处合并 assert len(result) >= 2 assert "".join(result) == text.strip() def test_splits_at_question_mark(self): text = "你是谁？我是AI。你好吗？很好。" result = split_text(text, max_chars=15) assert len(result) >= 2 assert "".join(result) == text.strip() def test_splits_at_exclamation_mark(self): text = "太棒了！真厉害！好厉害！" result = split_text(text, max_chars=10) assert len(result) >= 2 assert "".join(result) == text.strip() def test_splits_at_newline(self): text = "第一段\n第二段\n第三段" result = split_text(text, max_chars=10) assert len(result) >= 2 def test_splits_at_semicolon(self): text = "第一部分；第二部分；第三部分。" result = split_text(text, max_chars=15) assert len(result) >= 1 assert "".join(result) == text.strip() class TestSplitTextLongSentence: """长句子（超过 max_chars）强制切段.""" def test_very_long_sentence_hard_cut(self): """单个超长句子会被强制切段.""" text = "我" * 600 # 没有标点 result = split_text(text, max_chars=500) assert len(result) >= 2 total = sum(len(seg) for seg in result) assert total == len(text) def test_each_segment_leq_max_chars(self): """每个分段都不超过 max_chars.""" text = "测试句子。" * 100 result = split_text(text, max_chars=100) for seg in result: assert len(seg) <= 100 def test_no_empty_segments(self): """不产生空分段.""" text = "测试。" * 50 result = split_text(text, max_chars=50) for seg in result: assert len(seg) > 0 class TestSplitTextMergeShortSegments: """合并过短的分段.""" def test_short_segments_get_merged(self): """< 50 字符的段会被合并（如果不超限）.""" # 多个短句子，应该会被合并 text = "你好。我是。他是。她是。它是。" result = split_text(text, max_chars=50) # 每段6字符左右，应该被合并成一段 assert len(result) < 5 def test_last_short_segment_merged_to_previous(self): """最后一段如果很短，会合并到前一段.""" text = "a" * 48 + "。" + "b" * 48 + "。" + "cc" result = split_text(text, max_chars=100) # 最后的 "cc" 很短，应该被合并 assert result[-1] != "cc" class TestSplitTextEdgeCases: """边界情况.""" def test_single_character(self): result = split_text("一", max_chars=500) assert result == ["一"] def test_only_punctuation(self): text = "。。。" result = split_text(text, max_chars=500) assert len(result) == 1 def test_mixed_chinese_english(self): text = "Hello世界。Hello世界。" * 20 result = split_text(text, max_chars=50) assert len(result) >= 2 assert "".join(result) == text.strip() def test_max_chars_one(self): """极端情况：max_chars=1.""" text = "abc" result = split_text(text, max_chars=1) assert len(result) == 3 assert result == ["a", "b", "c"] # ═══════════════════════════════════════════════════════════════════════════════ # TtsConfig 配置解析 + 边界钳制 # ═══════════════════════════════════════════════════════════════════════════════ class TestTtsConfigDefaults: """默认值.""" def test_default_config_disabled(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" def test_parse_none_returns_default(self): config = TtsConfig.parse(None) assert config.enabled is False def test_parse_empty_dict_returns_default(self): config = TtsConfig.parse({}) assert config.enabled is False def test_parse_non_dict_returns_default(self): config = TtsConfig.parse("not a dict") assert config.enabled is False class TestTtsConfigParseEnabled: """enabled 字段解析.""" def test_parse_enabled_true(self): config = TtsConfig.parse({"enabled": True}) assert config.enabled is True def test_parse_enabled_false(self): config = TtsConfig.parse({"enabled": False}) assert config.enabled is False def test_parse_enabled_invalid_type(self): """enabled 不是 bool 时回退到 False.""" config = TtsConfig.parse({"enabled": "true"}) assert config.enabled is False def test_disabled_ignores_other_fields(self): """enabled=False 时其他字段都用默认值.""" config = TtsConfig.parse( { "enabled": False, "voice_id": "test", "speed": 2.0, } ) assert config.enabled is False assert config.voice_id == "" assert config.speed == 1.0 class TestTtsConfigParseFields: """各字段解析.""" def test_parse_voice_id(self): config = TtsConfig.parse({"enabled": True, "voice_id": "voice_001"}) assert config.voice_id == "voice_001" def test_parse_voice_id_invalid_type(self): config = TtsConfig.parse({"enabled": True, "voice_id": 123}) assert config.voice_id == "" def test_parse_speed(self): config = TtsConfig.parse({"enabled": True, "speed": 1.5}) assert config.speed == 1.5 def test_parse_speed_int(self): config = TtsConfig.parse({"enabled": True, "speed": 2}) assert config.speed == 2.0 def test_parse_speed_invalid_type(self): config = TtsConfig.parse({"enabled": True, "speed": "fast"}) assert config.speed == 1.0 def test_parse_pitch(self): config = TtsConfig.parse({"enabled": True, "pitch": 5}) assert config.pitch == 5.0 def test_parse_pitch_invalid_type(self): config = TtsConfig.parse({"enabled": True, "pitch": "high"}) assert config.pitch == 0.0 def test_parse_volume(self): config = TtsConfig.parse({"enabled": True, "volume": 0.5}) assert config.volume == 0.5 def test_parse_volume_invalid_type(self): config = TtsConfig.parse({"enabled": True, "volume": "loud"}) assert config.volume == 0.8 def test_parse_text(self): config = TtsConfig.parse({"enabled": True, "text": "你好世界"}) assert config.text == "你好世界" def test_parse_text_invalid_type(self): config = TtsConfig.parse({"enabled": True, "text": 12345}) assert config.text == "" def test_parse_align_mode_valid(self): config = TtsConfig.parse({"enabled": True, "align_mode": "subtitle"}) assert config.align_mode == "subtitle" def test_parse_align_mode_invalid(self): config = TtsConfig.parse({"enabled": True, "align_mode": "invalid"}) assert config.align_mode == "full" def test_parse_overlap_mode_mix(self): config = TtsConfig.parse({"enabled": True, "overlap_mode": "mix"}) assert config.overlap_mode == "mix" def test_parse_overlap_mode_invalid(self): config = TtsConfig.parse({"enabled": True, "overlap_mode": "invalid"}) assert config.overlap_mode == "replace" class TestTtsConfigClamp: """边界钳制.""" def test_speed_below_minimum_clamped(self): config = TtsConfig.parse({"enabled": True, "speed": 0.1}) assert config.speed == 0.5 def test_speed_above_maximum_clamped(self): config = TtsConfig.parse({"enabled": True, "speed": 3.0}) assert config.speed == 2.0 def test_speed_at_minimum_ok(self): config = TtsConfig.parse({"enabled": True, "speed": 0.5}) assert config.speed == 0.5 def test_speed_at_maximum_ok(self): config = TtsConfig.parse({"enabled": True, "speed": 2.0}) assert config.speed == 2.0 def test_pitch_below_minimum_clamped(self): config = TtsConfig.parse({"enabled": True, "pitch": -20}) assert config.pitch == -12 def test_pitch_above_maximum_clamped(self): config = TtsConfig.parse({"enabled": True, "pitch": 20}) assert config.pitch == 12 def test_pitch_at_minimum_ok(self): config = TtsConfig.parse({"enabled": True, "pitch": -12}) assert config.pitch == -12 def test_pitch_at_maximum_ok(self): config = TtsConfig.parse({"enabled": True, "pitch": 12}) assert config.pitch == 12 def test_volume_below_minimum_clamped(self): config = TtsConfig.parse({"enabled": True, "volume": -0.5}) assert config.volume == 0.0 def test_volume_above_maximum_clamped(self): config = TtsConfig.parse({"enabled": True, "volume": 2.0}) assert config.volume == 1.0 def test_volume_at_minimum_ok(self): config = TtsConfig.parse({"enabled": True, "volume": 0.0}) assert config.volume == 0.0 def test_volume_at_maximum_ok(self): config = TtsConfig.parse({"enabled": True, "volume": 1.0}) assert config.volume == 1.0 # ═══════════════════════════════════════════════════════════════════════════════ # TTSJob 领域模型 — 状态机 # ═══════════════════════════════════════════════════════════════════════════════ class TestTTSJobCreate: """创建任务.""" def test_create_basic(self): job = TTSJob.create(user_id="user1", input_text="你好") assert job.id is not None assert len(job.id) > 0 assert job.user_id == "user1" assert job.input_text == "你好" assert job.status == TTSJobStatus.PENDING assert job.retry_count == 0 assert job.max_retries == 3 assert job.started_at is None assert job.completed_at is None def test_create_with_voice_id(self): job = TTSJob.create(user_id="user1", input_text="你好", voice_id="voice_001") assert job.voice_id == "voice_001" def test_create_with_project_id(self): job = TTSJob.create(user_id="user1", input_text="你好", project_id="proj_001") assert job.project_id == "proj_001" def test_create_with_voice_clone_profile_id(self): job = TTSJob.create( user_id="user1", input_text="你好", voice_clone_profile_id="clone_001", ) assert job.voice_clone_profile_id == "clone_001" def test_create_with_custom_max_retries(self): job = TTSJob.create(user_id="user1", input_text="你好", max_retries=5) assert job.max_retries == 5 def test_create_with_format(self): job = TTSJob.create(user_id="user1", input_text="你好", format="wav") assert job.format == "wav" def test_create_with_sample_rate(self): job = TTSJob.create(user_id="user1", input_text="你好", sample_rate=44100) assert job.sample_rate == 44100 class TestTTSJobStatusProperties: """状态查询属性.""" def test_pending_not_terminal(self): job = TTSJob.create(user_id="u1", input_text="hi") assert job.is_terminal is False def test_processing_not_terminal(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() assert job.is_terminal is False def test_completed_is_terminal(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_completed(output_audio_url="url") assert job.is_terminal is True assert job.is_completed is True def test_failed_is_terminal(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_failed("error") assert job.is_terminal is True assert job.status == TTSJobStatus.FAILED def test_cancelled_is_terminal(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_cancelled() assert job.is_terminal is True def test_terminal_statuses_contains_all_three(self): assert TTSJobStatus.COMPLETED in TERMINAL_STATUSES assert TTSJobStatus.FAILED in TERMINAL_STATUSES assert TTSJobStatus.CANCELLED in TERMINAL_STATUSES class TestTTSJobTransitions: """状态转换.""" def test_pending_to_processing(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() assert job.status == TTSJobStatus.PROCESSING assert job.started_at is not None assert job.error_message == "" def test_pending_to_failed(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_failed("网络错误") assert job.status == TTSJobStatus.FAILED assert job.error_message == "网络错误" def test_pending_to_cancelled(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_cancelled() assert job.status == TTSJobStatus.CANCELLED def test_processing_to_completed(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_completed(output_audio_url="https://example.com/audio.mp3") assert job.status == TTSJobStatus.COMPLETED assert job.output_audio_url == "https://example.com/audio.mp3" assert job.completed_at is not None def test_processing_to_failed(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_failed("API超时") assert job.status == TTSJobStatus.FAILED assert job.error_message == "API超时" def test_processing_to_cancelled(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_cancelled() assert job.status == TTSJobStatus.CANCELLED def test_failed_to_pending_retry(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_failed("error") job.prepare_retry() assert job.status == TTSJobStatus.PENDING assert job.retry_count == 1 assert job.error_message == "" assert job.started_at is None def test_completed_cannot_transition_back(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_completed(output_audio_url="url") with pytest.raises(ValueError, match="非法状态转换"): job.mark_failed("test") def test_cancelled_cannot_retry(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_cancelled() with pytest.raises(ValueError, match="不可重试"): job.prepare_retry() def test_mark_failed_sets_error_message(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_failed("服务端错误") assert job.error_message == "服务端错误" def test_failed_status_after_mark_failed(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_failed("API超时") assert job.status == TTSJobStatus.FAILED assert job.error_message == "API超时" class TestTTSJobRetry: """重试逻辑.""" def test_retry_increments_retry_count(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_failed("error1") job.prepare_retry() assert job.retry_count == 1 job.mark_processing() job.mark_failed("error2") job.prepare_retry() assert job.retry_count == 2 def test_retry_clears_error_message(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_failed("error") job.prepare_retry() assert job.error_message == "" def test_retry_resets_timestamps(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_failed("error") job.prepare_retry() assert job.started_at is None def test_can_retry_while_below_max_retries(self): job = TTSJob.create(user_id="u1", input_text="hi", max_retries=2) # 第一次失败+重试 job.mark_failed("e1") job.prepare_retry() assert job.retry_count == 1 # 第二次失败+重试 job.mark_processing() job.mark_failed("e2") job.prepare_retry() assert job.retry_count == 2 def test_cannot_retry_when_exceeded_max_retries(self): job = TTSJob.create(user_id="u1", input_text="hi", max_retries=1) job.mark_failed("e1") job.prepare_retry() assert job.retry_count == 1 # 再次失败就不能重试了（已经用完1次重试） job.mark_processing() job.mark_failed("e2") with pytest.raises(ValueError, match="不可重试"): job.prepare_retry() def test_is_retryable_true_when_failed_and_under_limit(self): job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3) job.mark_failed("error") assert job.is_retryable is True def test_is_retryable_false_when_not_failed(self): job = TTSJob.create(user_id="u1", input_text="hi") assert job.is_retryable is False def test_prepare_retry_fails_when_not_failed(self): job = TTSJob.create(user_id="u1", input_text="hi") with pytest.raises(ValueError, match="不可重试"): job.prepare_retry() class TestTTSJobCompleted: """完成时的字段.""" def test_mark_completed_sets_output_url(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_completed( output_audio_url="https://cdn.example.com/audio.mp3", output_audio_key="audio/xxx.mp3", duration=10.5, file_size=102400, ) assert job.output_audio_url == "https://cdn.example.com/audio.mp3" assert job.output_audio_key == "audio/xxx.mp3" assert job.duration == 10.5 assert job.file_size == 102400 assert job.completed_at is not None def test_mark_completed_default_values(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_completed(output_audio_url="url") assert job.duration == 0.0 assert job.file_size == 0 class TestTTSJobMetadata: """元数据.""" def test_create_with_metadata(self): meta = {"source": "api", "priority": "high"} job = TTSJob.create(user_id="u1", input_text="hi", metadata=meta) assert job.metadata["source"] == "api" assert job.metadata["priority"] == "high" def test_default_metadata_empty_dict(self): job = TTSJob.create(user_id="u1", input_text="hi") assert job.metadata == {} class TestTTSJobCreateValidation: """创建时的参数校验.""" def test_empty_user_id_raises(self): with pytest.raises(ValueError, match="user_id"): TTSJob.create(user_id="", input_text="hi") def test_whitespace_user_id_raises(self): with pytest.raises(ValueError, match="user_id"): TTSJob.create(user_id=" ", input_text="hi") def test_empty_input_text_raises(self): with pytest.raises(ValueError, match="input_text"): TTSJob.create(user_id="u1", input_text="") def test_input_text_too_long_raises(self): long_text = "a" * 10001 with pytest.raises(ValueError, match="10000"): TTSJob.create(user_id="u1", input_text=long_text) def test_invalid_format_raises(self): with pytest.raises(ValueError, match="不支持的输出格式"): TTSJob.create(user_id="u1", input_text="hi", format="flac") def test_valid_format_wav(self): job = TTSJob.create(user_id="u1", input_text="hi", format="wav") assert job.format == "wav" def test_valid_format_pcm(self): job = TTSJob.create(user_id="u1", input_text="hi", format="pcm") assert job.format == "pcm" def test_input_text_stripped(self): job = TTSJob.create(user_id="u1", input_text=" 你好 ") assert job.input_text == "你好" class TestTTSJobToDict: """序列化.""" def test_to_dict_contains_key_fields(self): job = TTSJob.create(user_id="u1", input_text="hi", voice_id="v1") d = job.to_dict() assert d["id"] == job.id assert d["user_id"] == "u1" assert d["input_text"] == "hi" assert d["voice_id"] == "v1" assert d["status"] == "pending" assert d["retry_count"] == 0 assert d["is_retryable"] is False def test_to_dict_completed_status(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() job.mark_completed(output_audio_url="https://example.com/a.mp3", duration=10.5, file_size=1024) d = job.to_dict() assert d["status"] == "completed" assert d["output_audio_url"] == "https://example.com/a.mp3" assert d["duration"] == 10.5 assert d["file_size"] == 1024 assert d["is_completed"] is True assert d["started_at"] is not None assert d["completed_at"] is not None def test_to_dict_failed_status(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_failed("error msg") d = job.to_dict() assert d["status"] == "failed" assert d["error_message"] == "error msg" assert d["is_retryable"] is True class TestTTSJobCompletedValidation: """完成时的校验.""" def test_mark_completed_empty_url_raises(self): job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() with pytest.raises(ValueError, match="output_audio_url"): job.mark_completed(output_audio_url="") def test_is_completed_requires_url(self): """is_completed 属性需要 output_audio_url.""" job = TTSJob.create(user_id="u1", input_text="hi") job.mark_processing() # 直接设置状态为 completed 但不给 URL（模拟异常情况） # 正常流程 mark_completed 会校验 URL，所以这里不会出现 # 但确认属性逻辑：没有 URL 时 is_completed 为 False job.output_audio_url = "" # 直接绕过状态机 from packages.domain.tts_job import _VALID_TRANSITIONS # noqa job.status = TTSJobStatus.COMPLETED assert job.is_completed is False