""" Job 领域模型单元测试 """ import time from datetime import datetime, timezone from unittest.mock import patch import pytest from packages.domain.job import ( TERMINAL_STATUSES, Job, JobStatus, JobType, ) class TestJobType: """JobType 枚举测试""" def test_job_type_values(self): """测试所有 JobType 值""" assert JobType.VIDEO_COMPOSE == "video_compose" assert JobType.RENDER_EDIT_PLAN == "render_edit_plan" assert JobType.ASSET_INGEST == "asset_ingest" assert JobType.CLASSIFICATION == "classification" assert JobType.VOICE_EXTRACTION == "voice_extraction" assert JobType.GENERATION == "generation" def test_job_type_is_string(self): """测试 StrEnum 行为""" assert isinstance(JobType.VIDEO_COMPOSE, str) assert JobType.VIDEO_COMPOSE == "video_compose" class TestJobStatus: """JobStatus 枚举测试""" def test_job_status_values(self): """测试所有 JobStatus 值""" assert JobStatus.PENDING == "pending" assert JobStatus.RUNNING == "running" assert JobStatus.SUCCESS == "success" assert JobStatus.FAILED == "failed" assert JobStatus.CANCELLED == "cancelled" def test_terminal_statuses(self): """测试终态集合""" assert JobStatus.SUCCESS in TERMINAL_STATUSES assert JobStatus.FAILED in TERMINAL_STATUSES assert JobStatus.CANCELLED in TERMINAL_STATUSES assert JobStatus.PENDING not in TERMINAL_STATUSES assert JobStatus.RUNNING not in TERMINAL_STATUSES class TestJobCreate: """Job 创建测试""" def test_create_basic_job(self): """测试创建基本任务""" job = Job.create( project_id="proj-123", job_type=JobType.VIDEO_COMPOSE, ) assert job.id is not None assert len(job.id) > 0 assert job.project_id == "proj-123" assert job.job_type == JobType.VIDEO_COMPOSE assert job.status == JobStatus.PENDING assert job.progress == 0.0 assert job.payload == {} assert job.result == {} assert job.error_message == "" assert job.retry_count == 0 assert job.max_retries == 3 assert job.created_at is not None assert job.updated_at is not None assert job.started_at is None assert job.completed_at is None def test_create_with_all_params(self): """测试创建带所有参数的任务""" job = Job.create( project_id="proj-456", job_type=JobType.GENERATION, payload={"key": "value"}, source_id="src-789", created_by_user_id="user-001", max_retries=5, ) assert job.project_id == "proj-456" assert job.job_type == JobType.GENERATION assert job.payload == {"key": "value"} assert job.source_id == "src-789" assert job.created_by_user_id == "user-001" assert job.max_retries == 5 def test_create_with_string_job_type(self): """测试用字符串创建任务""" job = Job.create( project_id="proj-123", job_type="video_compose", ) assert job.job_type == JobType.VIDEO_COMPOSE def test_create_with_invalid_job_type(self): """测试无效任务类型""" with pytest.raises(ValueError, match="不支持的任务类型"): Job.create( project_id="proj-123", job_type="invalid_type", ) def test_create_empty_project_id(self): """测试空 project_id""" with pytest.raises(ValueError, match="project_id 不能为空"): Job.create( project_id="", job_type=JobType.VIDEO_COMPOSE, ) def test_create_whitespace_project_id(self): """测试空白 project_id 被 strip 后为空""" with pytest.raises(ValueError, match="project_id 不能为空"): Job.create( project_id=" ", job_type=JobType.VIDEO_COMPOSE, ) def test_create_strips_strings(self): """测试字符串字段被 strip""" job = Job.create( project_id=" proj-123 ", job_type=JobType.VIDEO_COMPOSE, source_id=" src-456 ", created_by_user_id=" user-789 ", ) assert job.project_id == "proj-123" assert job.source_id == "src-456" assert job.created_by_user_id == "user-789" def test_create_default_payload(self): """测试 None payload 默认化为空 dict""" job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE, payload=None) assert job.payload == {} def test_create_generates_unique_ids(self): """测试每次创建生成不同的 ID""" job1 = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) job2 = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) assert job1.id != job2.id def test_create_sets_timestamps(self): """测试创建时设置时间戳""" before = datetime.now(timezone.utc) time.sleep(0.01) job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) time.sleep(0.01) after = datetime.now(timezone.utc) assert before < job.created_at < after assert before < job.updated_at < after class TestJobStateTransitions: """Job 状态转换测试""" @pytest.fixture def new_job(self): return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) # ===== Pending → Running ===== def test_pending_to_running(self, new_job): """测试 pending → running""" assert new_job.status == JobStatus.PENDING new_job.mark_running() assert new_job.status == JobStatus.RUNNING assert new_job.started_at is not None assert new_job.completed_at is None assert not new_job.is_terminal def test_pending_to_running_with_stage(self, new_job): """测试 pending → running 带阶段描述""" new_job.mark_running(stage="初始化") assert new_job.current_stage == "初始化" # ===== Pending → Success ===== def test_pending_to_success(self, new_job): """测试 pending → success(直接成功)""" new_job.mark_success() assert new_job.status == JobStatus.SUCCESS assert new_job.progress == 100.0 assert new_job.current_stage == "完成" assert new_job.completed_at is not None assert new_job.is_terminal def test_pending_to_success_with_result(self, new_job): """测试 pending → success 带结果""" result = {"output_url": "http://example.com/video.mp4"} new_job.mark_success(result=result) assert new_job.result == result # ===== Pending → Cancelled ===== def test_pending_to_cancelled(self, new_job): """测试 pending → cancelled""" new_job.mark_cancelled() assert new_job.status == JobStatus.CANCELLED assert new_job.current_stage == "已取消" assert new_job.is_terminal # ===== Running → Success ===== def test_running_to_success(self, new_job): """测试 running → success""" new_job.mark_running() new_job.mark_success() assert new_job.status == JobStatus.SUCCESS assert new_job.completed_at is not None assert new_job.progress == 100.0 assert new_job.is_terminal def test_running_to_success_preserves_started_at(self, new_job): """测试 running → success 保留 started_at""" new_job.mark_running() started_at = new_job.started_at new_job.mark_success() assert new_job.started_at == started_at # ===== Running → Failed ===== def test_running_to_failed(self, new_job): """测试 running → failed""" new_job.mark_running() new_job.mark_failed("Something went wrong") assert new_job.status == JobStatus.FAILED assert new_job.error_message == "Something went wrong" assert new_job.current_stage == "失败" assert new_job.completed_at is not None assert new_job.is_terminal # ===== Running → Cancelled ===== def test_running_to_cancelled(self, new_job): """测试 running → cancelled""" new_job.mark_running() new_job.mark_cancelled() assert new_job.status == JobStatus.CANCELLED assert new_job.is_terminal # ===== Failed → Pending (Retry) ===== def test_failed_to_pending_retry(self, new_job): """测试 failed → pending(重试)""" new_job.mark_running() new_job.mark_failed("error") assert new_job.retry_count == 0 new_job.prepare_retry() assert new_job.status == JobStatus.PENDING assert new_job.retry_count == 1 assert new_job.progress == 0.0 assert new_job.error_message == "" assert new_job.started_at is None assert new_job.completed_at is None assert new_job.celery_task_id == "" assert "第 1 次重试" in new_job.current_stage def test_retry_up_to_max_retries(self, new_job): """测试最多重试 max_retries 次""" new_job.max_retries = 2 new_job.mark_running() # 第一次失败重试 new_job.mark_failed("error 1") assert new_job.is_retryable # 失败后可重试 new_job.prepare_retry() assert new_job.retry_count == 1 # 第二次失败重试 new_job.mark_running() new_job.mark_failed("error 2") assert new_job.is_retryable # retry_count=1 < max_retries=2 new_job.prepare_retry() assert new_job.retry_count == 2 # 第三次失败后不可重试(retry_count == max_retries) new_job.mark_running() new_job.mark_failed("error 3") assert not new_job.is_retryable # 达到上限 with pytest.raises(ValueError, match="任务不可重试"): new_job.prepare_retry() def test_retry_not_from_failed(self, new_job): """测试非 failed 状态不可重试""" with pytest.raises(ValueError, match="任务不可重试"): new_job.prepare_retry() # pending 状态 # ===== 非法状态转换 ===== def test_invalid_transition_success_to_running(self, new_job): """测试 success → running 非法""" new_job.mark_success() with pytest.raises(ValueError, match="非法状态转换"): new_job.mark_running() def test_invalid_transition_cancelled_to_running(self, new_job): """测试 cancelled → running 非法""" new_job.mark_cancelled() with pytest.raises(ValueError, match="非法状态转换"): new_job.mark_running() def test_invalid_transition_pending_to_failed(self, new_job): """测试 pending → failed 非法(必须经过 running)""" with pytest.raises(ValueError, match="非法状态转换"): new_job.mark_failed("test error") def test_invalid_status_string(self, new_job): """测试无效状态字符串""" with pytest.raises(ValueError, match="无效状态"): new_job.transition_to("invalid_status") class TestJobProperties: """Job 属性测试""" @pytest.fixture def new_job(self): return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) def test_is_terminal_pending(self, new_job): """测试 pending 不是终态""" assert not new_job.is_terminal def test_is_terminal_running(self, new_job): """测试 running 不是终态""" new_job.mark_running() assert not new_job.is_terminal def test_is_terminal_success(self, new_job): """测试 success 是终态""" new_job.mark_success() assert new_job.is_terminal def test_is_terminal_failed(self, new_job): """测试 failed 是终态""" new_job.mark_running() new_job.mark_failed("error") assert new_job.is_terminal def test_is_terminal_cancelled(self, new_job): """测试 cancelled 是终态""" new_job.mark_cancelled() assert new_job.is_terminal def test_is_retryable_failed_under_limit(self, new_job): """测试失败且未达上限时可重试""" new_job.mark_running() new_job.mark_failed("error") assert new_job.is_retryable def test_is_retryable_failed_at_limit(self, new_job): """测试失败且达上限时不可重试""" new_job.max_retries = 0 new_job.mark_running() new_job.mark_failed("error") assert not new_job.is_retryable def test_is_retryable_not_failed(self, new_job): """测试非失败状态不可重试""" assert not new_job.is_retryable # pending new_job.mark_running() assert not new_job.is_retryable # running new_job.mark_success() assert not new_job.is_retryable # success class TestJobProgress: """Job 进度更新测试""" @pytest.fixture def running_job(self): job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) job.mark_running() return job def test_update_progress_normal(self, running_job): """测试正常更新进度""" running_job.update_progress(50.0, stage="处理中") assert running_job.progress == 50.0 assert running_job.current_stage == "处理中" def test_update_progress_zero(self, running_job): """测试更新进度为 0""" running_job.update_progress(0.0) assert running_job.progress == 0.0 def test_update_progress_hundred(self, running_job): """测试更新进度为 100""" running_job.update_progress(100.0) assert running_job.progress == 100.0 def test_update_progress_negative(self, running_job): """测试负进度报错""" with pytest.raises(ValueError, match="进度必须在 0~100 之间"): running_job.update_progress(-1.0) def test_update_progress_over_hundred(self, running_job): """测试超过 100 的进度报错""" with pytest.raises(ValueError, match="进度必须在 0~100 之间"): running_job.update_progress(101.0) def test_update_progress_without_stage(self, running_job): """测试更新进度但不改变阶段""" running_job.current_stage = "初始阶段" running_job.update_progress(30.0) assert running_job.progress == 30.0 assert running_job.current_stage == "初始阶段" # 保留原值 def test_update_progress_updates_updated_at(self, running_job): """测试更新进度会更新 updated_at""" old_updated = running_job.updated_at time.sleep(0.01) running_job.update_progress(50.0) assert running_job.updated_at > old_updated class TestJobToDict: """Job 序列化测试""" def test_to_dict_pending_job(self): """测试 pending 状态的 Job 序列化为字典""" job = Job.create( project_id="proj-123", job_type=JobType.VIDEO_COMPOSE, payload={"input": "data"}, source_id="src-456", ) d = job.to_dict() assert d["id"] == job.id assert d["project_id"] == "proj-123" assert d["job_type"] == "video_compose" assert d["status"] == "pending" assert d["progress"] == 0.0 assert d["payload"] == {"input": "data"} assert d["result"] == {} assert d["error_message"] == "" assert d["retry_count"] == 0 assert d["max_retries"] == 3 assert d["source_id"] == "src-456" assert d["is_retryable"] is False assert d["started_at"] is None assert d["completed_at"] is None assert d["created_at"] is not None assert d["updated_at"] is not None def test_to_dict_completed_job(self): """测试完成状态的 Job 序列化为字典""" job = Job.create(project_id="proj-123", job_type=JobType.GENERATION) job.mark_running() job.mark_success(result={"output": "result"}) d = job.to_dict() assert d["status"] == "success" assert d["progress"] == 100.0 assert d["result"] == {"output": "result"} assert d["started_at"] is not None assert d["completed_at"] is not None def test_to_dict_failed_job(self): """测试失败状态的 Job 序列化为字典""" job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) job.mark_running() job.mark_failed("timeout error") d = job.to_dict() assert d["status"] == "failed" assert d["error_message"] == "timeout error" assert d["is_retryable"] is True class TestTransitionTimestamps: """状态转换时间戳测试""" @pytest.fixture def new_job(self): return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE) def test_mark_running_sets_started_at(self, new_job): """测试 mark_running 设置 started_at""" assert new_job.started_at is None new_job.mark_running() assert new_job.started_at is not None assert isinstance(new_job.started_at, datetime) assert new_job.started_at.tzinfo is not None def test_mark_running_twice_preserves_started_at(self, new_job): """测试再次 mark_running 不覆盖 started_at""" # 先手动转换到 running new_job.transition_to(JobStatus.RUNNING) first_started = new_job.started_at # 不能直接再调 mark_running(会报错),但可以验证 started_at 不被重复设置 # transition_to 已经处理了 started_at is None 的逻辑 assert first_started == new_job.started_at def test_mark_success_sets_completed_at(self, new_job): """测试 mark_success 设置 completed_at""" new_job.mark_running() assert new_job.completed_at is None new_job.mark_success() assert new_job.completed_at is not None def test_mark_failed_sets_completed_at(self, new_job): """测试 mark_failed 设置 completed_at""" new_job.mark_running() assert new_job.completed_at is None new_job.mark_failed("error") assert new_job.completed_at is not None def test_transition_updates_updated_at(self, new_job): """测试每次状态转换都更新 updated_at""" old_updated = new_job.updated_at time.sleep(0.01) new_job.mark_running() assert new_job.updated_at > old_updated