diff --git a/tests/unit/test_tts_job_domain.py b/tests/unit/test_tts_job_domain.py index c76a09115..de26cb11b 100755 --- a/tests/unit/test_tts_job_domain.py +++ b/tests/unit/test_tts_job_domain.py @@ -1,242 +1,280 @@ -"""tts_job 领域模型单元测试.""" +"""TTSJob 领域层单元测试 - tts_job.py""" import pytest -from domain.tts_job import TERMINAL_STATUSES, TTSJob, TTSJobStatus + +from packages.domain.tts_job import ( + TERMINAL_STATUSES, + TTSJob, + TTSJobStatus, +) class TestTTSJobStatus: - """TTSJobStatus 枚举测试.""" + """TTSJobStatus 枚举测试""" - def test_values(self): + def test_all_statuses_have_values(self): + for s in TTSJobStatus: + assert isinstance(s.value, str) + assert s.value + + def test_str_enum_behavior(self): assert TTSJobStatus.PENDING == "pending" - assert TTSJobStatus.PROCESSING == "processing" - assert TTSJobStatus.COMPLETED == "completed" - assert TTSJobStatus.FAILED == "failed" - assert TTSJobStatus.CANCELLED == "cancelled" + assert isinstance(TTSJobStatus.PENDING, str) def test_terminal_statuses(self): assert TTSJobStatus.COMPLETED in TERMINAL_STATUSES assert TTSJobStatus.FAILED in TERMINAL_STATUSES assert TTSJobStatus.CANCELLED in TERMINAL_STATUSES + + def test_non_terminal_statuses(self): assert TTSJobStatus.PENDING not in TERMINAL_STATUSES assert TTSJobStatus.PROCESSING not in TERMINAL_STATUSES class TestTTSJobCreate: - """TTSJob.create 工厂方法测试.""" + """TTSJob.create 工厂方法测试""" - def test_create_with_required_fields(self): - job = TTSJob.create(user_id="user_001", input_text="你好世界") + def test_create_basic(self): + job = TTSJob.create(user_id="user-1", input_text="你好世界") assert job.id assert len(job.id) == 32 - assert job.user_id == "user_001" + assert job.user_id == "user-1" assert job.input_text == "你好世界" assert job.status == TTSJobStatus.PENDING assert job.voice_id == "" - assert job.sample_rate == 22050 + assert job.voice_model == "" assert job.format == "mp3" + assert job.sample_rate == 22050 assert job.retry_count == 0 assert job.max_retries == 3 assert job.metadata == {} - assert job.created_at is not None - assert job.updated_at is not None - - def test_create_with_all_fields(self): - job = TTSJob.create( - user_id="user_002", - input_text="测试文本", - voice_id="voice_001", - voice_model="cosyvoice", - project_id="proj_001", - voice_clone_profile_id="clone_001", - sample_rate=16000, - format="wav", - max_retries=5, - metadata={"key": "value"}, - ) - assert job.voice_id == "voice_001" - assert job.voice_model == "cosyvoice" - assert job.project_id == "proj_001" - assert job.voice_clone_profile_id == "clone_001" - assert job.sample_rate == 16000 - assert job.format == "wav" - assert job.max_retries == 5 - assert job.metadata == {"key": "value"} - - def test_create_strips_strings(self): - job = TTSJob.create( - user_id=" user_003 ", - input_text=" 测试文本 ", - voice_id=" voice_001 ", - voice_model=" cosyvoice ", - project_id=" proj_001 ", - voice_clone_profile_id=" clone_001 ", - format="wav", - ) - assert job.user_id == "user_003" - assert job.input_text == "测试文本" - assert job.voice_id == "voice_001" - assert job.voice_model == "cosyvoice" - assert job.project_id == "proj_001" - assert job.voice_clone_profile_id == "clone_001" - assert job.format == "wav" def test_create_empty_user_id_raises(self): - with pytest.raises(ValueError, match="user_id"): - TTSJob.create(user_id="", input_text="test") - - def test_create_whitespace_user_id_raises(self): - with pytest.raises(ValueError, match="user_id"): + with pytest.raises(ValueError, match="user_id 不能为空"): TTSJob.create(user_id=" ", input_text="test") def test_create_empty_input_text_raises(self): - with pytest.raises(ValueError, match="input_text"): - TTSJob.create(user_id="u", input_text="") + with pytest.raises(ValueError, match="input_text 不能为空"): + TTSJob.create(user_id="u1", input_text=" ") - def test_create_input_text_too_long_raises(self): + def test_create_text_too_long_raises(self): long_text = "a" * 10001 - with pytest.raises(ValueError, match="10000"): - TTSJob.create(user_id="u", input_text=long_text) + with pytest.raises(ValueError, match="input_text 长度不能超过 10000"): + TTSJob.create(user_id="u1", input_text=long_text) - def test_create_input_text_at_limit_ok(self): + def test_create_text_exactly_10000_ok(self): text = "a" * 10000 - job = TTSJob.create(user_id="u", input_text=text) - assert job.input_text == text + job = TTSJob.create(user_id="u1", input_text=text) + assert len(job.input_text) == 10000 def test_create_invalid_format_raises(self): with pytest.raises(ValueError, match="不支持的输出格式"): - TTSJob.create(user_id="u", input_text="t", format="flac") + TTSJob.create(user_id="u1", input_text="test", format="flac") - def test_create_supported_formats(self): - for fmt in ["mp3", "wav", "pcm"]: - job = TTSJob.create(user_id="u", input_text="t", format=fmt) - assert job.format == fmt + def test_create_mp3_format(self): + job = TTSJob.create(user_id="u1", input_text="test", format="mp3") + assert job.format == "mp3" - def test_create_none_metadata_defaults_to_empty_dict(self): - job = TTSJob.create(user_id="u", input_text="t", metadata=None) + def test_create_wav_format(self): + job = TTSJob.create(user_id="u1", input_text="test", format="wav") + assert job.format == "wav" + + def test_create_pcm_format(self): + job = TTSJob.create(user_id="u1", input_text="test", format="pcm") + assert job.format == "pcm" + + def test_create_with_voice_id(self): + job = TTSJob.create(user_id="u1", input_text="test", voice_id="voice-1") + assert job.voice_id == "voice-1" + + def test_create_with_project_id(self): + job = TTSJob.create(user_id="u1", input_text="test", project_id="proj-1") + assert job.project_id == "proj-1" + + def test_create_with_clone_profile(self): + job = TTSJob.create(user_id="u1", input_text="test", voice_clone_profile_id="vc-1") + assert job.voice_clone_profile_id == "vc-1" + + def test_create_with_metadata(self): + meta = {"source": "api", "priority": "high"} + job = TTSJob.create(user_id="u1", input_text="test", metadata=meta) + assert job.metadata == meta + + def test_create_none_metadata_defaults_empty(self): + job = TTSJob.create(user_id="u1", input_text="test", metadata=None) assert job.metadata == {} - def test_create_ids_are_unique(self): - j1 = TTSJob.create(user_id="u", input_text="t") - j2 = TTSJob.create(user_id="u", input_text="t") - assert j1.id != j2.id + def test_create_input_text_stripped(self): + job = TTSJob.create(user_id=" u1 ", input_text=" hello ") + assert job.user_id == "u1" + assert job.input_text == "hello" + + def test_create_custom_max_retries(self): + job = TTSJob.create(user_id="u1", input_text="test", max_retries=5) + assert job.max_retries == 5 -class TestTTSJobStateMachine: - """TTSJob 状态机测试.""" +class TestTTSJobProperties: + """属性测试""" - @pytest.fixture - def pending_job(self): - return TTSJob.create(user_id="user_001", input_text="测试") + def test_is_terminal_pending(self): + job = TTSJob.create(user_id="u1", input_text="test") + assert job.is_terminal is False - def test_initial_status_is_pending(self, pending_job): - assert pending_job.status == TTSJobStatus.PENDING - assert not pending_job.is_terminal + def test_is_terminal_completed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_processing() + job.mark_completed("http://audio.com/a.mp3") + assert job.is_terminal is True - def test_pending_to_processing(self, pending_job): - pending_job.mark_processing() - assert pending_job.status == TTSJobStatus.PROCESSING - assert pending_job.started_at is not None - assert pending_job.error_message == "" + def test_is_terminal_failed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_processing() + job.mark_failed("error") + assert job.is_terminal is True - def test_pending_can_fail_directly(self, pending_job): - """pending 可以直接到 failed（比如入参校验失败）""" - pending_job.mark_failed("校验失败") - assert pending_job.status == TTSJobStatus.FAILED - assert pending_job.error_message == "校验失败" + def test_is_retryable_failed_within_limit(self): + job = TTSJob.create(user_id="u1", input_text="test", max_retries=3) + job.mark_processing() + job.mark_failed("error") + assert job.is_retryable is True - def test_pending_can_be_cancelled(self, pending_job): - pending_job.mark_cancelled() - assert pending_job.status == TTSJobStatus.CANCELLED + def test_is_retryable_failed_at_limit(self): + job = TTSJob.create(user_id="u1", input_text="test", max_retries=1) + job.mark_processing() + job.mark_failed("error") + job.retry_count = 1 + assert job.is_retryable is False - def test_processing_to_completed(self, pending_job): - pending_job.mark_processing() - pending_job.mark_completed(output_audio_url="https://example.com/out.mp3") - assert pending_job.status == TTSJobStatus.COMPLETED - assert pending_job.output_audio_url == "https://example.com/out.mp3" - assert pending_job.completed_at is not None - assert pending_job.error_message == "" + def test_is_completed_true(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_processing() + job.mark_completed("http://audio.com/a.mp3") + assert job.is_completed is True - def test_processing_to_failed(self, pending_job): - pending_job.mark_processing() - pending_job.mark_failed("API 超时") - assert pending_job.status == TTSJobStatus.FAILED - assert pending_job.error_message == "API 超时" + def test_is_completed_no_url(self): + """completed 状态但没有 output_audio_url 为空，is_completed 为 False""" + job = TTSJob.create(user_id="u1", input_text="test") + job.status = TTSJobStatus.COMPLETED + job.output_audio_url = "" + assert job.is_completed is False - def test_processing_can_be_cancelled(self, pending_job): - pending_job.mark_processing() - pending_job.mark_cancelled() - assert pending_job.status == TTSJobStatus.CANCELLED - def test_completed_is_terminal(self, pending_job): - pending_job.mark_processing() - pending_job.mark_completed(output_audio_url="https://example.com/out.mp3") - assert pending_job.is_terminal - assert pending_job.is_completed +class TestTTSJobTransitions: + """状态转换测试""" - def test_failed_is_terminal_but_retryable(self, pending_job): - pending_job.mark_processing() - pending_job.mark_failed("error") - assert pending_job.is_terminal - assert pending_job.is_retryable + def test_pending_to_processing(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to(TTSJobStatus.PROCESSING) + assert job.status == TTSJobStatus.PROCESSING - def test_cancelled_is_terminal_and_not_retryable(self, pending_job): - pending_job.mark_cancelled() - assert pending_job.is_terminal - assert not pending_job.is_retryable + def test_pending_to_failed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to(TTSJobStatus.FAILED) + assert job.status == TTSJobStatus.FAILED - def test_invalid_transition_completed_to_processing_raises(self, pending_job): - pending_job.mark_processing() - pending_job.mark_completed(output_audio_url="https://example.com/out.mp3") + def test_pending_to_cancelled(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to(TTSJobStatus.CANCELLED) + assert job.status == TTSJobStatus.CANCELLED + + def test_processing_to_completed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.COMPLETED) + assert job.status == TTSJobStatus.COMPLETED + + def test_processing_to_failed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.FAILED) + assert job.status == TTSJobStatus.FAILED + + def test_failed_to_pending_retry(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.FAILED) + job.transition_to(TTSJobStatus.PENDING) + assert job.status == TTSJobStatus.PENDING + + def test_invalid_transition_raises(self): + job = TTSJob.create(user_id="u1", input_text="test") with pytest.raises(ValueError, match="非法状态转换"): - pending_job.mark_processing() + job.transition_to(TTSJobStatus.COMPLETED) # pending 不能直接到 completed - def test_invalid_transition_completed_to_failed_raises(self, pending_job): - pending_job.mark_processing() - pending_job.mark_completed(output_audio_url="https://example.com/out.mp3") - with pytest.raises(ValueError, match="非法状态转换"): - pending_job.mark_failed("test") + def test_transition_with_string(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.transition_to("processing") + assert job.status == TTSJobStatus.PROCESSING - def test_cancelled_cannot_transition(self, pending_job): - pending_job.mark_cancelled() - with pytest.raises(ValueError): - pending_job.mark_processing() - with pytest.raises(ValueError): - pending_job.mark_failed("test") - - def test_transition_to_with_string(self, pending_job): - """transition_to 支持字符串参数""" - pending_job.transition_to("processing") - assert pending_job.status == TTSJobStatus.PROCESSING - - def test_transition_to_invalid_string_raises(self, pending_job): + def test_transition_invalid_string_raises(self): + job = TTSJob.create(user_id="u1", input_text="test") with pytest.raises(ValueError, match="无效状态"): - pending_job.transition_to("invalid_status") + job.transition_to("invalid") - def test_state_transition_updates_updated_at(self, pending_job): - old_updated = pending_job.updated_at + def test_transition_updates_updated_at(self): + job = TTSJob.create(user_id="u1", input_text="test") + old = job.updated_at import time - time.sleep(0.001) - pending_job.mark_processing() - assert pending_job.updated_at > old_updated + job.transition_to(TTSJobStatus.PROCESSING) + assert job.updated_at >= old + + +class TestTTSJobMarkMethods: + """便捷标记方法测试""" + + def test_mark_processing(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.error_message = "previous error" + job.mark_processing() + assert job.status == TTSJobStatus.PROCESSING + assert job.started_at is not None + assert job.error_message == "" + + def test_mark_completed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_processing() + job.mark_completed( + "http://example.com/audio.mp3", + output_audio_key="audio/key.mp3", + duration=5.5, + file_size=102400, + ) + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "http://example.com/audio.mp3" + assert job.output_audio_key == "audio/key.mp3" + assert job.duration == 5.5 + assert job.file_size == 102400 + assert job.completed_at is not None + assert job.error_message == "" + + def test_mark_completed_empty_url_raises(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_processing() + with pytest.raises(ValueError, match="output_audio_url 不能为空"): + job.mark_completed(" ") + + def test_mark_failed(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_processing() + job.mark_failed("API 调用超时") + assert job.status == TTSJobStatus.FAILED + assert job.error_message == "API 调用超时" + + def test_mark_cancelled(self): + job = TTSJob.create(user_id="u1", input_text="test") + job.mark_cancelled() + assert job.status == TTSJobStatus.CANCELLED class TestTTSJobRetry: - """TTSJob 重试逻辑测试.""" + """重试逻辑测试""" - def test_failed_can_retry(self): - job = TTSJob.create(user_id="u", input_text="t", max_retries=3) + def test_prepare_retry(self): + job = TTSJob.create(user_id="u1", input_text="test", max_retries=3) job.mark_processing() - job.mark_failed("error") - assert job.is_retryable - assert job.retry_count == 0 - - def test_prepare_retry_resets_to_pending(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - job.mark_failed("error") - + job.mark_failed("超时") job.prepare_retry() assert job.status == TTSJobStatus.PENDING assert job.retry_count == 1 @@ -244,150 +282,62 @@ class TestTTSJobRetry: assert job.started_at is None assert job.completed_at is None - def test_retry_up_to_max_retries(self): - job = TTSJob.create(user_id="u", input_text="t", max_retries=2) - # 第 1 次失败 + 重试 → retry_count=1，还可以重试 + def test_prepare_retry_not_retryable_raises(self): + job = TTSJob.create(user_id="u1", input_text="test", max_retries=0) job.mark_processing() - job.mark_failed("e1") - assert job.is_retryable - job.prepare_retry() - assert job.retry_count == 1 - - # 第 2 次失败 → retry_count=1，还是 failed 状态，还可以重试（max_retries=2） - job.mark_processing() - job.mark_failed("e2") - assert job.is_retryable # retry_count=1 < max_retries=2 - job.prepare_retry() - assert job.retry_count == 2 - - # 第 3 次失败 → retry_count=2，达到上限，不可重试 - job.mark_processing() - job.mark_failed("e3") - assert not job.is_retryable # retry_count=2 == max_retries=2 - - def test_retry_exceed_max_raises(self): - job = TTSJob.create(user_id="u", input_text="t", max_retries=1) - job.mark_processing() - job.mark_failed("e") - job.prepare_retry() # 第 1 次重试，用完了 - - job.mark_processing() - job.mark_failed("e2") + job.mark_failed("error") with pytest.raises(ValueError, match="不可重试"): job.prepare_retry() - def test_pending_not_retryable(self): - job = TTSJob.create(user_id="u", input_text="t") - assert not job.is_retryable - with pytest.raises(ValueError, match="不可重试"): + def test_multiple_retries(self): + job = TTSJob.create(user_id="u1", input_text="test", max_retries=3) + for i in range(3): + job.mark_processing() + job.mark_failed(f"error-{i}") job.prepare_retry() - - def test_completed_not_retryable(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - job.mark_completed(output_audio_url="https://example.com/out.mp3") - assert not job.is_retryable - with pytest.raises(ValueError, match="不可重试"): - job.prepare_retry() - - def test_cancelled_not_retryable(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_cancelled() - assert not job.is_retryable - - -class TestTTSJobMarkCompleted: - """mark_completed 方法测试.""" - - def test_requires_output_url(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - with pytest.raises(ValueError, match="output_audio_url"): - job.mark_completed(output_audio_url="") - - def test_sets_all_fields(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - job.mark_completed( - output_audio_url="https://example.com/out.mp3", - output_audio_key="audio/001.mp3", - duration=30.5, - file_size=102400, - ) - assert job.output_audio_url == "https://example.com/out.mp3" - assert job.output_audio_key == "audio/001.mp3" - assert job.duration == 30.5 - assert job.file_size == 102400 - assert job.completed_at is not None - - def test_strips_whitespace(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - job.mark_completed( - output_audio_url=" https://example.com/out.mp3 ", - output_audio_key=" audio/001.mp3 ", - ) - assert job.output_audio_url == "https://example.com/out.mp3" - assert job.output_audio_key == "audio/001.mp3" - - -class TestTTSJobIsCompleted: - """is_completed 属性测试.""" - - def test_completed_with_url_is_completed(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - job.mark_completed(output_audio_url="https://example.com/out.mp3") - assert job.is_completed - - def test_completed_without_url_not_completed(self): - """极端情况：completed 状态但没有 URL（理论不会发生）""" - job = TTSJob.create(user_id="u", input_text="t") - job.mark_processing() - job.transition_to(TTSJobStatus.COMPLETED) # 直接转，不设 URL - assert not job.is_completed - - def test_pending_not_completed(self): - job = TTSJob.create(user_id="u", input_text="t") - assert not job.is_completed + assert job.retry_count == i + 1 + # 第3次重试后 retry_count=3，等于 max_retries=3，不可再重试 + assert job.is_retryable is False class TestTTSJobToDict: - """to_dict 序列化测试.""" + """to_dict 序列化测试""" - def test_pending_job_to_dict(self): - job = TTSJob.create(user_id="user_001", input_text="测试文本", voice_id="v001") + def test_to_dict_contains_fields(self): + job = TTSJob.create( + user_id="u1", + input_text="test", + voice_id="voice-1", + project_id="proj-1", + ) d = job.to_dict() assert d["id"] == job.id - assert d["user_id"] == "user_001" + assert d["user_id"] == "u1" + assert d["input_text"] == "test" + assert d["voice_id"] == "voice-1" assert d["status"] == "pending" - assert d["input_text"] == "测试文本" - assert d["voice_id"] == "v001" - assert d["retry_count"] == 0 assert d["is_retryable"] is False assert d["is_completed"] is False - assert d["metadata"] == {} + + def test_to_dict_datetime_are_strings(self): + job = TTSJob.create(user_id="u1", input_text="test") + d = job.to_dict() + assert isinstance(d["created_at"], str) + assert isinstance(d["updated_at"], str) + + def test_to_dict_none_datetime(self): + job = TTSJob.create(user_id="u1", input_text="test") + d = job.to_dict() 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_completed_job_to_dict(self): - job = TTSJob.create(user_id="u", input_text="t") + def test_to_dict_after_completion(self): + job = TTSJob.create(user_id="u1", input_text="test") job.mark_processing() - job.mark_completed(output_audio_url="https://example.com/out.mp3", duration=10.0) + job.mark_completed("http://test.mp3", duration=10.0) d = job.to_dict() assert d["status"] == "completed" - assert d["output_audio_url"] == "https://example.com/out.mp3" assert d["duration"] == 10.0 assert d["is_completed"] is True assert d["started_at"] is not None assert d["completed_at"] is not None - - def test_failed_job_to_dict(self): - job = TTSJob.create(user_id="u", input_text="t") - job.mark_failed("出错了") - d = job.to_dict() - assert d["status"] == "failed" - assert d["error_message"] == "出错了" - assert d["is_retryable"] is True diff --git a/tests/unit/test_video_share_domain.py b/tests/unit/test_video_share_domain.py new file mode 100755 index 000000000..2d3d36753 --- /dev/null +++ b/tests/unit/test_video_share_domain.py @@ -0,0 +1,224 @@ +"""VideoShare 领域层单元测试 - video_share.py""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from packages.domain.video_share import ( + VideoShare, + _hash_password, + generate_share_token, +) + + +class TestGenerateShareToken: + """generate_share_token 函数测试""" + + def test_default_length(self): + token = generate_share_token() + assert len(token) == 12 + + def test_custom_length(self): + token = generate_share_token(20) + assert len(token) == 20 + + def test_url_friendly_characters(self): + """token 只包含 URL 友好的字符（没有 l, i, o, 0, 1 等易混字符）""" + token = generate_share_token(100) + # 不应包含易混淆字符 + assert "l" not in token + assert "i" not in token + assert "o" not in token + assert "0" not in token + assert "1" not in token + + def test_randomness(self): + """两次生成的 token 不同（概率上）""" + tokens = {generate_share_token() for _ in range(100)} + # 100 次应该几乎不可能重复 + assert len(tokens) > 95 + + +class TestHashPassword: + """_hash_password 函数测试""" + + def test_empty_password_returns_empty(self): + assert _hash_password("") == "" + + def test_none_password_returns_empty(self): + assert _hash_password(None) == "" # type: ignore + + def test_hash_is_deterministic(self): + """相同密码哈希结果相同""" + h1 = _hash_password("mypassword") + h2 = _hash_password("mypassword") + assert h1 == h2 + + def test_hash_differs_for_different_passwords(self): + """不同密码哈希结果不同""" + h1 = _hash_password("password1") + h2 = _hash_password("password2") + assert h1 != h2 + + def test_hash_is_hex_string(self): + """哈希是 64 位十六进制字符串（SHA-256）""" + h = _hash_password("test") + assert len(h) == 64 + int(h, 16) # 应该能解析为十六进制 + + def test_hash_includes_salt(self): + """加盐后与直接 SHA-256 不同""" + from hashlib import sha256 + direct = sha256("mypass".encode()).hexdigest() + salted = _hash_password("mypass") + assert direct != salted + + +class TestVideoShareCreate: + """VideoShare.create 工厂方法测试""" + + def test_create_basic(self): + share = VideoShare.create(video_id="video-1", user_id="user-1") + assert share.id + assert len(share.id) == 32 + assert share.video_id == "video-1" + assert share.user_id == "user-1" + assert share.share_token + assert len(share.share_token) == 12 + assert share.password_hash is None + assert share.expires_at is None + assert share.view_count == 0 + assert share.download_count == 0 + assert share.is_active is True + assert share.created_at is not None + assert share.updated_at is not None + + def test_create_empty_video_id_raises(self): + with pytest.raises(ValueError, match="video_id cannot be empty"): + VideoShare.create(video_id=" ", user_id="u1") + + def test_create_empty_user_id_raises(self): + with pytest.raises(ValueError, match="user_id cannot be empty"): + VideoShare.create(video_id="v1", user_id=" ") + + def test_create_with_password(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="secret123") + assert share.password_hash is not None + assert share.password_hash != "secret123" # 已哈希 + assert len(share.password_hash) == 64 # SHA-256 hex + + def test_create_with_expires_at(self): + future = datetime.now(timezone.utc) + timedelta(days=7) + share = VideoShare.create(video_id="v1", user_id="u1", expires_at=future) + assert share.expires_at == future + + def test_create_past_expires_at_raises(self): + past = datetime.now(timezone.utc) - timedelta(days=1) + with pytest.raises(ValueError, match="expires_at cannot be in the past"): + VideoShare.create(video_id="v1", user_id="u1", expires_at=past) + + def test_create_fields_stripped(self): + share = VideoShare.create(video_id=" v1 ", user_id=" u1 ") + assert share.video_id == "v1" + assert share.user_id == "u1" + + def test_unique_tokens(self): + """不同分享有不同的 token""" + shares = [VideoShare.create(video_id="v1", user_id="u1") for _ in range(20)] + tokens = [s.share_token for s in shares] + assert len(set(tokens)) == 20 + + +class TestVideoShareProperties: + """属性测试""" + + def test_has_password_true(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="secret") + assert share.has_password is True + + def test_has_password_false(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.has_password is False + + def test_is_expired_no_expiry(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.is_expired is False + + def test_is_expired_future_expiry(self): + future = datetime.now(timezone.utc) + timedelta(days=7) + share = VideoShare.create(video_id="v1", user_id="u1", expires_at=future) + assert share.is_expired is False + + def test_is_expired_past_expiry(self): + # 直接设置 expires_at 为过去时间（绕过 create 的校验） + share = VideoShare.create(video_id="v1", user_id="u1") + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + assert share.is_expired is True + + def test_is_accessible_active_not_expired(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.is_accessible is True + + def test_is_accessible_inactive(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.is_active = False + assert share.is_accessible is False + + def test_is_accessible_expired(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + assert share.is_accessible is False + + def test_is_accessible_inactive_and_expired(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.is_active = False + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + assert share.is_accessible is False + + +class TestVideoSharePassword: + """密码验证测试""" + + def test_verify_password_correct(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="secret123") + assert share.verify_password("secret123") is True + + def test_verify_password_wrong(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="secret123") + assert share.verify_password("wrongpass") is False + + def test_verify_no_password_always_true(self): + """没有设置密码时，任何密码都通过（包括空密码）""" + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.verify_password("") is True + assert share.verify_password("anything") is True + + def test_verify_empty_password_with_password_set(self): + """有密码时，空密码不通过""" + share = VideoShare.create(video_id="v1", user_id="u1", password="secret") + assert share.verify_password("") is False + + +class TestVideoShareCounters: + """计数方法测试""" + + def test_increment_view_count(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.view_count == 0 + share.increment_view_count() + assert share.view_count == 1 + share.increment_view_count() + assert share.view_count == 2 + + def test_increment_download_count(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.download_count == 0 + share.increment_download_count() + assert share.download_count == 1 + + def test_revoke(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.is_active is True + share.revoke() + assert share.is_active is False + assert share.is_accessible is False diff --git a/tests/unit/test_voice_clone_profile_domain.py b/tests/unit/test_voice_clone_profile_domain.py old mode 100644 new mode 100755 index 6bad4ba0a..55c531a52 --- a/tests/unit/test_voice_clone_profile_domain.py +++ b/tests/unit/test_voice_clone_profile_domain.py @@ -1,6 +1,4 @@ -""" -VoiceCloneProfile 音色克隆档案领域模型单元测试 -""" +"""VoiceCloneProfile 领域层单元测试 - voice_clone_profile.py""" import pytest @@ -14,376 +12,351 @@ from packages.domain.voice_clone_profile import ( class TestVoiceCloneStatus: """VoiceCloneStatus 枚举测试""" - def test_status_values(self): + def test_all_statuses_have_values(self): + for s in VoiceCloneStatus: + assert isinstance(s.value, str) + assert s.value + + def test_str_enum(self): assert VoiceCloneStatus.PENDING == "pending" - assert VoiceCloneStatus.PROCESSING == "processing" - assert VoiceCloneStatus.READY == "ready" - assert VoiceCloneStatus.FAILED == "failed" - assert VoiceCloneStatus.DISABLED == "disabled" + assert isinstance(VoiceCloneStatus.PENDING, str) def test_terminal_statuses(self): assert VoiceCloneStatus.READY in TERMINAL_STATUSES assert VoiceCloneStatus.FAILED in TERMINAL_STATUSES assert VoiceCloneStatus.DISABLED in TERMINAL_STATUSES + + def test_non_terminal(self): assert VoiceCloneStatus.PENDING not in TERMINAL_STATUSES assert VoiceCloneStatus.PROCESSING not in TERMINAL_STATUSES class TestVoiceCloneProfileCreate: - """VoiceCloneProfile.create 工厂方法测试""" + """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" + def test_create_basic(self): + profile = VoiceCloneProfile.create(user_id="user-1", name="我的音色") + assert profile.id + assert len(profile.id) == 32 + assert profile.user_id == "user-1" assert profile.name == "我的音色" assert profile.status == VoiceCloneStatus.PENDING + assert profile.description == "" + assert profile.source_audio_url == "" + assert profile.voice_id == "" + assert profile.language == "zh-CN" + assert profile.gender == "unknown" 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 == "描述" + assert profile.metadata == {} def test_create_empty_user_id_raises(self): with pytest.raises(ValueError, match="user_id 不能为空"): - VoiceCloneProfile.create(user_id=" ", name="测试") + VoiceCloneProfile.create(user_id=" ", name="test") def test_create_empty_name_raises(self): with pytest.raises(ValueError, match="name 不能为空"): - VoiceCloneProfile.create(user_id="user1", name=" ") + VoiceCloneProfile.create(user_id="u1", 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) + with pytest.raises(ValueError, match="name 长度不能超过 100"): + VoiceCloneProfile.create(user_id="u1", name=long_name) - def test_create_name_exactly_100_chars_ok(self): + def test_create_name_exactly_100_ok(self): name = "a" * 100 - profile = VoiceCloneProfile.create(user_id="user1", name=name) + profile = VoiceCloneProfile.create(user_id="u1", name=name) assert profile.name == name - def test_create_default_metadata_is_dict(self): - profile = VoiceCloneProfile.create(user_id="u1", name="test") + def test_create_with_description(self): + profile = VoiceCloneProfile.create( + user_id="u1", name="test", description="温暖男声" + ) + assert profile.description == "温暖男声" + + def test_create_with_source_audio(self): + profile = VoiceCloneProfile.create( + user_id="u1", name="test", source_audio_url="http://audio.com/source.wav" + ) + assert profile.source_audio_url == "http://audio.com/source.wav" + + def test_create_with_language(self): + profile = VoiceCloneProfile.create(user_id="u1", name="test", language="en-US") + assert profile.language == "en-US" + + def test_create_gender_lowercased(self): + profile = VoiceCloneProfile.create(user_id="u1", name="test", gender="MALE") + assert profile.gender == "male" + + def test_create_with_metadata(self): + meta = {"source": "upload", "duration": 30} + profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata=meta) + assert profile.metadata == meta + + def test_create_none_metadata_defaults_empty(self): + profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata=None) assert profile.metadata == {} - # 不应该共享同一个 dict - p2 = VoiceCloneProfile.create(user_id="u2", name="test2") - assert profile.metadata is not p2.metadata + + def test_create_fields_stripped(self): + profile = VoiceCloneProfile.create( + user_id=" u1 ", + name=" test ", + description=" desc ", + source_audio_url=" url ", + voice_model=" model ", + language=" zh-CN ", + ) + assert profile.user_id == "u1" + assert profile.name == "test" + assert profile.description == "desc" + assert profile.source_audio_url == "url" + assert profile.voice_model == "model" + assert profile.language == "zh-CN" + + def test_create_custom_max_retries(self): + profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=5) + assert profile.max_retries == 5 class TestVoiceCloneProfileProperties: """属性测试""" + def test_is_terminal_pending(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + assert p.is_terminal is False + def test_is_terminal_ready(self): - profile = VoiceCloneProfile.create(user_id="u1", name="test") - profile.status = VoiceCloneStatus.READY - assert profile.is_terminal is True + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_ready("voice-123") + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_failed("error") + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_disabled() + assert p.is_terminal is True 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 + p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3) + p.mark_processing() + p.mark_failed("error") + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=1) + p.mark_processing() + p.mark_failed("error") + p.retry_count = 1 + assert p.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_true(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_ready("voice-123") + assert p.is_ready is True - 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_no_voice_id(self): + """ready 状态但没有 voice_id，is_ready 为 False""" + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.status = VoiceCloneStatus.READY + p.voice_id = "" + assert p.is_ready is False - 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 + def test_is_ready_not_ready_status(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.voice_id = "some-id" + assert p.is_ready is False # 状态是 PENDING -class TestStateTransitions: +class TestVoiceCloneProfileTransitions: """状态转换测试""" 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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.PROCESSING) + assert p.status == VoiceCloneStatus.PROCESSING def test_pending_to_failed(self): - profile = VoiceCloneProfile.create(user_id="u1", name="test") - profile.transition_to(VoiceCloneStatus.FAILED) - assert profile.status == VoiceCloneStatus.FAILED + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.FAILED) + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.DISABLED) + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.PROCESSING) + p.transition_to(VoiceCloneStatus.READY) + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.PROCESSING) + p.transition_to(VoiceCloneStatus.FAILED) + assert p.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 + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.PROCESSING) + p.transition_to(VoiceCloneStatus.DISABLED) + assert p.status == VoiceCloneStatus.DISABLED 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 + """已就绪音色可以被禁用""" + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_ready("voice-1") + p.transition_to(VoiceCloneStatus.DISABLED) + assert p.status == VoiceCloneStatus.DISABLED + + def test_failed_to_pending_retry(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to(VoiceCloneStatus.PROCESSING) + p.transition_to(VoiceCloneStatus.FAILED) + p.transition_to(VoiceCloneStatus.PENDING) + assert p.status == VoiceCloneStatus.PENDING + + def test_invalid_transition_raises(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + with pytest.raises(ValueError, match="非法状态转换"): + p.transition_to(VoiceCloneStatus.READY) # pending 不能直接到 ready + + def test_disabled_to_pending_raises(self): + """禁用后不能回到 pending""" + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_disabled() + with pytest.raises(ValueError, match="非法状态转换"): + p.transition_to(VoiceCloneStatus.PENDING) def test_transition_with_string(self): - profile = VoiceCloneProfile.create(user_id="u1", name="test") - profile.transition_to("processing") - assert profile.status == VoiceCloneStatus.PROCESSING + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.transition_to("processing") + assert p.status == VoiceCloneStatus.PROCESSING def test_transition_invalid_string_raises(self): - profile = VoiceCloneProfile.create(user_id="u1", name="test") + p = VoiceCloneProfile.create(user_id="u1", name="test") with pytest.raises(ValueError, match="无效状态"): - profile.transition_to("invalid_status") + p.transition_to("invalid") - 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) + def test_transition_updates_updated_at(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + old = p.updated_at + import time + time.sleep(0.001) + p.transition_to(VoiceCloneStatus.PROCESSING) + assert p.updated_at >= old -class TestMarkMethods: - """标记方法测试""" +class TestVoiceCloneProfileMarkMethods: + """便捷标记方法测试""" 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 == "" + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.error_message = "prev error" + p.mark_processing() + assert p.status == VoiceCloneStatus.PROCESSING + assert p.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 == "" + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_ready("voice-abc123") + assert p.status == VoiceCloneStatus.READY + assert p.voice_id == "voice-abc123" + assert p.error_message == "" def test_mark_ready_empty_voice_id_raises(self): - profile = VoiceCloneProfile.create(user_id="u1", name="test") - profile.status = VoiceCloneStatus.PROCESSING + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_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" + p.mark_ready(" ") 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 == "音频质量太差" + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_failed("音频质量太差") + assert p.status == VoiceCloneStatus.FAILED + assert p.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 + def test_mark_disabled(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_disabled() + assert p.status == VoiceCloneStatus.DISABLED -class TestPrepareRetry: - """重试准备测试""" +class TestVoiceCloneProfileRetry: + """重试逻辑测试""" - 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(self): + p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3) + p.mark_processing() + p.mark_failed("超时") + p.voice_id = "partial-id" + p.prepare_retry() + assert p.status == VoiceCloneStatus.PENDING + assert p.retry_count == 1 + assert p.error_message == "" + assert p.voice_id == "" # 重试时清空 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 + def test_prepare_retry_not_retryable_raises(self): + p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=0) + p.mark_processing() + p.mark_failed("error") with pytest.raises(ValueError, match="不可重试"): - profile.prepare_retry() + p.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() + def test_multiple_retries(self): + p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3) + for i in range(3): + p.mark_processing() + p.mark_failed(f"error-{i}") + p.prepare_retry() + assert p.retry_count == i + 1 + assert p.is_retryable is False -class TestToDict: - """序列化测试""" +class TestVoiceCloneProfileToDict: + """to_dict 序列化测试""" - def test_to_dict_basic(self): - profile = VoiceCloneProfile.create( - user_id="user1", - name="测试音色", - description="desc", - max_retries=2, + def test_to_dict_contains_fields(self): + p = VoiceCloneProfile.create( + user_id="u1", + name="我的音色", + description="测试", + language="en-US", + gender="female", ) - d = profile.to_dict() - assert d["id"] == profile.id - assert d["user_id"] == "user1" - assert d["name"] == "测试音色" - assert d["description"] == "desc" + d = p.to_dict() + assert d["id"] == p.id + assert d["user_id"] == "u1" + assert d["name"] == "我的音色" + assert d["description"] == "测试" assert d["status"] == "pending" - assert d["retry_count"] == 0 - assert d["max_retries"] == 2 + assert d["language"] == "en-US" + assert d["gender"] == "female" assert d["is_retryable"] is False assert d["is_ready"] is False + + def test_to_dict_datetime_are_strings(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + d = p.to_dict() 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() + def test_to_dict_after_ready(self): + p = VoiceCloneProfile.create(user_id="u1", name="test") + p.mark_processing() + p.mark_ready("voice-123") + d = p.to_dict() assert d["status"] == "ready" - assert d["voice_id"] == "voice_123" + 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}