""" Job 应用层用例单元测试（第十八波） 覆盖： - CreateJobUseCase - SubmitJobUseCase - UpdateJobProgressUseCase - CompleteJobUseCase - FailJobUseCase - RetryJobUseCase - CancelJobUseCase - GetJobUseCase - ListJobsUseCase - GetJobStatisticsUseCase """ from unittest.mock import MagicMock import pytest from packages.application.jobs import ( CancelJobUseCase, CompleteJobCommand, CompleteJobUseCase, CreateJobCommand, CreateJobUseCase, FailJobCommand, FailJobUseCase, GetJobStatisticsUseCase, GetJobUseCase, ListJobsUseCase, RetryJobUseCase, SubmitJobUseCase, UpdateJobProgressCommand, UpdateJobProgressUseCase, ) from packages.domain.job import Job, JobStatus, JobType @pytest.fixture def mock_repo(): return MagicMock() def make_job( status=JobStatus.PENDING, job_type=JobType.VIDEO_COMPOSE, project_id="proj-1", **kwargs, ): job = Job.create( project_id=project_id, job_type=job_type, **kwargs, ) # 绕过状态机直接设置状态（测试构造用） if status != JobStatus.PENDING: object.__setattr__(job, "status", status) return job # ============================================================ # CreateJobUseCase # ============================================================ class TestCreateJobUseCase: """CreateJobUseCase 创建任务""" def test_create_success(self, mock_repo): """正常创建任务""" mock_repo.create.side_effect = lambda j: j cmd = CreateJobCommand( project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, payload={"key": "val"}, source_id="src-1", created_by_user_id="user-1", max_retries=5, ) uc = CreateJobUseCase(mock_repo) job = uc.execute(cmd) assert job.project_id == "proj-1" assert job.job_type == JobType.VIDEO_COMPOSE assert job.payload == {"key": "val"} assert job.source_id == "src-1" assert job.created_by_user_id == "user-1" assert job.max_retries == 5 assert job.status == JobStatus.PENDING assert job.progress == 0.0 mock_repo.create.assert_called_once() def test_create_default_values(self, mock_repo): """默认参数""" mock_repo.create.side_effect = lambda j: j cmd = CreateJobCommand(project_id="proj-1", job_type="video_compose") uc = CreateJobUseCase(mock_repo) job = uc.execute(cmd) assert job.payload == {} assert job.source_id == "" assert job.created_by_user_id == "" assert job.max_retries == 3 def test_create_string_job_type(self, mock_repo): """字符串类型的 job_type 也支持""" mock_repo.create.side_effect = lambda j: j cmd = CreateJobCommand(project_id="proj-1", job_type="asset_ingest") uc = CreateJobUseCase(mock_repo) job = uc.execute(cmd) assert job.job_type == JobType.ASSET_INGEST # ============================================================ # SubmitJobUseCase # ============================================================ class TestSubmitJobUseCase: """SubmitJobUseCase 提交任务""" def test_submit_success(self, mock_repo): """正常提交 pending 任务""" job = make_job(status=JobStatus.PENDING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j uc = SubmitJobUseCase(mock_repo) result = uc.execute(job.id, celery_task_id="celery-123") assert result.status == JobStatus.RUNNING assert result.celery_task_id == "celery-123" assert result.current_stage == "已提交，等待执行" assert result.started_at is not None mock_repo.update.assert_called_once() def test_submit_job_not_found(self, mock_repo): """任务不存在""" mock_repo.get.return_value = None uc = SubmitJobUseCase(mock_repo) with pytest.raises(ValueError, match="任务不存在"): uc.execute("nonexistent") def test_submit_already_running(self, mock_repo): """已经是 running 状态不能再提交""" job = make_job(status=JobStatus.RUNNING) mock_repo.get.return_value = job uc = SubmitJobUseCase(mock_repo) with pytest.raises(ValueError, match="只有 pending 状态"): uc.execute(job.id) def test_submit_without_celery_id(self, mock_repo): """不传 celery_task_id 也可以""" job = make_job(status=JobStatus.PENDING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j uc = SubmitJobUseCase(mock_repo) result = uc.execute(job.id) assert result.status == JobStatus.RUNNING assert result.celery_task_id == "" # ============================================================ # UpdateJobProgressUseCase # ============================================================ class TestUpdateJobProgressUseCase: """UpdateJobProgressUseCase 更新进度""" def test_update_progress_success(self, mock_repo): """正常更新进度""" job = make_job(status=JobStatus.RUNNING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j cmd = UpdateJobProgressCommand(job_id=job.id, progress=50.0, current_stage="处理中") uc = UpdateJobProgressUseCase(mock_repo) result = uc.execute(cmd) assert result.progress == 50.0 assert result.current_stage == "处理中" mock_repo.update.assert_called_once() def test_update_progress_job_not_found(self, mock_repo): """任务不存在""" mock_repo.get.return_value = None cmd = UpdateJobProgressCommand(job_id="nope", progress=50.0) uc = UpdateJobProgressUseCase(mock_repo) with pytest.raises(ValueError, match="任务不存在"): uc.execute(cmd) def test_update_progress_not_running(self, mock_repo): """非 running 状态不能更新进度""" job = make_job(status=JobStatus.PENDING) mock_repo.get.return_value = job cmd = UpdateJobProgressCommand(job_id=job.id, progress=50.0) uc = UpdateJobProgressUseCase(mock_repo) with pytest.raises(ValueError, match="只有 running 状态"): uc.execute(cmd) # ============================================================ # CompleteJobUseCase # ============================================================ class TestCompleteJobUseCase: """CompleteJobUseCase 完成任务""" def test_complete_from_running(self, mock_repo): """从 running 状态完成""" job = make_job(status=JobStatus.RUNNING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j cmd = CompleteJobCommand(job_id=job.id, result={"output": "ok"}) uc = CompleteJobUseCase(mock_repo) result = uc.execute(cmd) assert result.status == JobStatus.SUCCESS assert result.progress == 100.0 assert result.result == {"output": "ok"} assert result.completed_at is not None mock_repo.update.assert_called_once() def test_complete_from_pending(self, mock_repo): """从 pending 状态也可以直接完成""" job = make_job(status=JobStatus.PENDING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j cmd = CompleteJobCommand(job_id=job.id) uc = CompleteJobUseCase(mock_repo) result = uc.execute(cmd) assert result.status == JobStatus.SUCCESS assert result.progress == 100.0 def test_complete_job_not_found(self, mock_repo): """任务不存在""" mock_repo.get.return_value = None cmd = CompleteJobCommand(job_id="nope") uc = CompleteJobUseCase(mock_repo) with pytest.raises(ValueError, match="任务不存在"): uc.execute(cmd) def test_complete_already_failed(self, mock_repo): """已失败的任务不能直接标记完成""" job = make_job(status=JobStatus.FAILED) job.error_message = "some error" mock_repo.get.return_value = job cmd = CompleteJobCommand(job_id=job.id) uc = CompleteJobUseCase(mock_repo) with pytest.raises(ValueError, match="只有 running/pending"): uc.execute(cmd) # ============================================================ # FailJobUseCase # ============================================================ class TestFailJobUseCase: """FailJobUseCase 失败任务""" def test_fail_from_running(self, mock_repo): """从 running 状态失败""" job = make_job(status=JobStatus.RUNNING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j cmd = FailJobCommand(job_id=job.id, error_message="网络超时") uc = FailJobUseCase(mock_repo) result = uc.execute(cmd) assert result.status == JobStatus.FAILED assert result.error_message == "网络超时" assert result.current_stage == "失败" assert result.completed_at is not None mock_repo.update.assert_called_once() def test_fail_pending_rejected_by_domain(self, mock_repo): """pending 状态不能直接失败（领域状态机约束）""" job = make_job(status=JobStatus.PENDING) mock_repo.get.return_value = job cmd = FailJobCommand(job_id=job.id, error_message="资源不足") uc = FailJobUseCase(mock_repo) with pytest.raises(ValueError, match="非法状态转换"): uc.execute(cmd) def test_fail_job_not_found(self, mock_repo): """任务不存在""" mock_repo.get.return_value = None cmd = FailJobCommand(job_id="nope", error_message="err") uc = FailJobUseCase(mock_repo) with pytest.raises(ValueError, match="任务不存在"): uc.execute(cmd) # ============================================================ # RetryJobUseCase # ============================================================ class TestRetryJobUseCase: """RetryJobUseCase 重试任务""" def test_retry_success(self, mock_repo): """失败任务重试成功""" job = make_job(status=JobStatus.FAILED, max_retries=3) job.retry_count = 0 job.error_message = "timeout" mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j uc = RetryJobUseCase(mock_repo) result = uc.execute(job.id) assert result.status == JobStatus.PENDING assert result.retry_count == 1 assert result.progress == 0.0 assert result.error_message == "" assert result.started_at is None assert result.completed_at is None assert result.celery_task_id == "" mock_repo.update.assert_called_once() def test_retry_job_not_found(self, mock_repo): """任务不存在""" mock_repo.get.return_value = None uc = RetryJobUseCase(mock_repo) with pytest.raises(ValueError, match="任务不存在"): uc.execute("nope") def test_retry_exceeds_max_retries(self, mock_repo): """超过最大重试次数不可重试""" job = make_job(status=JobStatus.FAILED, max_retries=3) job.retry_count = 3 mock_repo.get.return_value = job uc = RetryJobUseCase(mock_repo) with pytest.raises(ValueError, match="不可重试"): uc.execute(job.id) # ============================================================ # CancelJobUseCase # ============================================================ class TestCancelJobUseCase: """CancelJobUseCase 取消任务""" def test_cancel_pending(self, mock_repo): """取消 pending 任务""" job = make_job(status=JobStatus.PENDING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j uc = CancelJobUseCase(mock_repo) result = uc.execute(job.id) assert result.status == JobStatus.CANCELLED assert result.current_stage == "已取消" mock_repo.update.assert_called_once() def test_cancel_running(self, mock_repo): """取消 running 任务""" job = make_job(status=JobStatus.RUNNING) mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j uc = CancelJobUseCase(mock_repo) result = uc.execute(job.id) assert result.status == JobStatus.CANCELLED def test_cancel_job_not_found(self, mock_repo): """任务不存在""" mock_repo.get.return_value = None uc = CancelJobUseCase(mock_repo) with pytest.raises(ValueError, match="任务不存在"): uc.execute("nope") def test_cancel_already_success(self, mock_repo): """已成功的任务不能取消""" job = make_job(status=JobStatus.SUCCESS) mock_repo.get.return_value = job uc = CancelJobUseCase(mock_repo) with pytest.raises(ValueError, match="终态"): uc.execute(job.id) def test_cancel_already_failed(self, mock_repo): """已失败的任务不能取消（走重试）""" job = make_job(status=JobStatus.FAILED) job.error_message = "err" mock_repo.get.return_value = job uc = CancelJobUseCase(mock_repo) with pytest.raises(ValueError, match="终态"): uc.execute(job.id) # ============================================================ # GetJobUseCase # ============================================================ class TestGetJobUseCase: """GetJobUseCase 获取任务""" def test_get_existing(self, mock_repo): """获取存在的任务""" job = make_job() mock_repo.get.return_value = job uc = GetJobUseCase(mock_repo) result = uc.execute(job.id) assert result is job mock_repo.get.assert_called_once_with(job.id) def test_get_not_found(self, mock_repo): """获取不存在的任务返回 None""" mock_repo.get.return_value = None uc = GetJobUseCase(mock_repo) result = uc.execute("nope") assert result is None # ============================================================ # ListJobsUseCase # ============================================================ class TestListJobsUseCase: """ListJobsUseCase 列出任务""" def test_list_by_project(self, mock_repo): """按项目列出""" jobs = [make_job(), make_job()] mock_repo.list_by_project.return_value = jobs uc = ListJobsUseCase(mock_repo) result = uc.execute(project_id="proj-1") assert len(result) == 2 mock_repo.list_by_project.assert_called_once() def test_list_by_project_with_filters(self, mock_repo): """按项目 + 类型 + 状态过滤""" mock_repo.list_by_project.return_value = [] uc = ListJobsUseCase(mock_repo) uc.execute( project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, status=JobStatus.RUNNING, limit=20, offset=10, ) mock_repo.list_by_project.assert_called_once_with( "proj-1", job_type=JobType.VIDEO_COMPOSE, status=JobStatus.RUNNING, limit=20, offset=10, ) def test_list_by_user(self, mock_repo): """按用户列出""" jobs = [make_job()] mock_repo.list_by_user.return_value = jobs uc = ListJobsUseCase(mock_repo) result = uc.execute(user_id="user-1") assert len(result) == 1 mock_repo.list_by_user.assert_called_once() def test_list_no_filter_raises(self, mock_repo): """不指定 project_id 或 user_id 报错""" uc = ListJobsUseCase(mock_repo) with pytest.raises(ValueError, match="必须指定"): uc.execute() def test_list_project_takes_precedence(self, mock_repo): """同时传 project_id 和 user_id，优先按项目查""" mock_repo.list_by_project.return_value = [] uc = ListJobsUseCase(mock_repo) uc.execute(project_id="proj-1", user_id="user-1") mock_repo.list_by_project.assert_called_once() mock_repo.list_by_user.assert_not_called() # ============================================================ # GetJobStatisticsUseCase # ============================================================ class TestGetJobStatisticsUseCase: """GetJobStatisticsUseCase 任务统计""" def test_stats_counts(self, mock_repo): """统计各状态数量""" mock_repo.count_by_project.side_effect = lambda pid, status=None: { None: 10, # total JobStatus.PENDING: 2, JobStatus.RUNNING: 3, JobStatus.SUCCESS: 4, JobStatus.FAILED: 1, }[status] uc = GetJobStatisticsUseCase(mock_repo) stats = uc.execute("proj-1") assert stats["project_id"] == "proj-1" assert stats["total"] == 10 assert stats["pending"] == 2 assert stats["running"] == 3 assert stats["success"] == 4 assert stats["failed"] == 1 # 总共调用 5 次 count_by_project assert mock_repo.count_by_project.call_count == 5 # ============================================================ # Command 对象 # ============================================================ class TestCommandObjects: """命令对象基本属性""" def test_create_job_command_defaults(self): cmd = CreateJobCommand(project_id="p1", job_type=JobType.VIDEO_COMPOSE) assert cmd.payload == {} assert cmd.source_id == "" assert cmd.created_by_user_id == "" assert cmd.max_retries == 3 def test_update_progress_command_defaults(self): cmd = UpdateJobProgressCommand(job_id="j1", progress=50.0) assert cmd.current_stage == "" def test_complete_job_command_defaults(self): cmd = CompleteJobCommand(job_id="j1") assert cmd.result == {} def test_fail_job_command(self): cmd = FailJobCommand(job_id="j1", error_message="err") assert cmd.error_message == "err"