diff --git a/tests/unit/test_jobs.py b/tests/unit/test_jobs.py new file mode 100755 index 000000000..61ba82790 --- /dev/null +++ b/tests/unit/test_jobs.py @@ -0,0 +1,578 @@ +""" +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"