tests/unit/domain/test_job.py

@@ -0,0 +1,523 @@
+"""job 单测.
+
+domain 层统一异步任务纯逻辑模块，0 外部依赖。
+覆盖：枚举、create工厂、状态机、进度更新、重试机制、序列化。
+"""
+
+from __future__ import annotations
+
+from packages.domain.job import Job, JobStatus, JobType, TERMINAL_STATUSES
+
+
+class TestJobType:
+    """JobType 枚举测试."""
+
+    def test_six_types(self):
+        """六种任务类型."""
+        assert len(JobType) == 6
+
+    def test_video_compose(self):
+        assert JobType.VIDEO_COMPOSE == "video_compose"
+
+    def test_render_edit_plan(self):
+        assert JobType.RENDER_EDIT_PLAN == "render_edit_plan"
+
+    def test_asset_ingest(self):
+        assert JobType.ASSET_INGEST == "asset_ingest"
+
+    def test_classification(self):
+        assert JobType.CLASSIFICATION == "classification"
+
+    def test_voice_extraction(self):
+        assert JobType.VOICE_EXTRACTION == "voice_extraction"
+
+    def test_generation(self):
+        assert JobType.GENERATION == "generation"
+
+
+class TestJobStatus:
+    """JobStatus 枚举测试."""
+
+    def test_five_statuses(self):
+        assert len(JobStatus) == 5
+
+    def test_pending(self):
+        assert JobStatus.PENDING == "pending"
+
+    def test_running(self):
+        assert JobStatus.RUNNING == "running"
+
+    def test_success(self):
+        assert JobStatus.SUCCESS == "success"
+
+    def test_failed(self):
+        assert JobStatus.FAILED == "failed"
+
+    def test_cancelled(self):
+        assert JobStatus.CANCELLED == "cancelled"
+
+
+class TestTerminalStatuses:
+    """终态集合测试."""
+
+    def test_success_is_terminal(self):
+        assert JobStatus.SUCCESS in TERMINAL_STATUSES
+
+    def test_failed_is_terminal(self):
+        assert JobStatus.FAILED in TERMINAL_STATUSES
+
+    def test_cancelled_is_terminal(self):
+        assert JobStatus.CANCELLED in TERMINAL_STATUSES
+
+    def test_pending_not_terminal(self):
+        assert JobStatus.PENDING not in TERMINAL_STATUSES
+
+    def test_running_not_terminal(self):
+        assert JobStatus.RUNNING not in TERMINAL_STATUSES
+
+
+class TestJobCreate:
+    """Job.create 工厂测试."""
+
+    def test_create_minimal(self):
+        """最简创建."""
+        job = Job.create(project_id="proj1", job_type=JobType.VIDEO_COMPOSE)
+        assert job.project_id == "proj1"
+        assert job.job_type == JobType.VIDEO_COMPOSE
+        assert job.status == JobStatus.PENDING
+        assert job.progress == 0.0
+        assert job.current_stage == ""
+        assert job.payload == {}
+        assert job.result == {}
+        assert job.error_message == ""
+        assert job.retry_count == 0
+        assert job.max_retries == 3
+        assert job.source_id == ""
+        assert job.created_by_user_id == ""
+        assert job.started_at is None
+        assert job.completed_at is None
+        assert isinstance(job.id, str)
+        assert len(job.id) > 0
+
+    def test_create_with_string_job_type(self):
+        """用字符串传 job_type."""
+        job = Job.create(project_id="p1", job_type="video_compose")
+        assert job.job_type == JobType.VIDEO_COMPOSE
+
+    def test_create_invalid_job_type_string(self):
+        """无效的 job_type 字符串."""
+        try:
+            Job.create(project_id="p1", job_type="invalid_type")
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "不支持的任务类型" in str(e)
+
+    def test_create_full(self):
+        """带全部字段."""
+        job = Job.create(
+            project_id="  proj1  ",
+            job_type=JobType.CLASSIFICATION,
+            payload={"asset_id": "a1"},
+            source_id="  src1  ",
+            created_by_user_id="  user1  ",
+            max_retries=5,
+        )
+        assert job.project_id == "proj1"  # strip
+        assert job.job_type == JobType.CLASSIFICATION
+        assert job.payload == {"asset_id": "a1"}
+        assert job.source_id == "src1"  # strip
+        assert job.created_by_user_id == "user1"  # strip
+        assert job.max_retries == 5
+
+    def test_create_empty_project_id(self):
+        """空 project_id 无效."""
+        try:
+            Job.create(project_id="", job_type=JobType.VIDEO_COMPOSE)
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "project_id" in str(e)
+
+    def test_create_whitespace_project_id(self):
+        """空白 project_id 无效."""
+        try:
+            Job.create(project_id="   ", job_type=JobType.VIDEO_COMPOSE)
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "project_id" in str(e)
+
+    def test_create_payload_none_defaults_empty(self):
+        """payload=None 默认为空 dict."""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, payload=None)
+        assert job.payload == {}
+
+    def test_create_unique_id(self):
+        """不同任务 id 不同."""
+        j1 = Job.create("p", JobType.VIDEO_COMPOSE)
+        j2 = Job.create("p", JobType.VIDEO_COMPOSE)
+        assert j1.id != j2.id
+
+    def test_create_has_timestamps(self):
+        """有创建和更新时间."""
+        job = Job.create("p", JobType.VIDEO_COMPOSE)
+        assert job.created_at is not None
+        assert job.updated_at is not None
+
+
+class TestJobIsTerminal:
+    """is_terminal 属性测试."""
+
+    def _make_job(self):
+        return Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+
+    def test_pending_not_terminal(self):
+        job = self._make_job()
+        assert job.is_terminal is False
+
+    def test_running_not_terminal(self):
+        job = self._make_job()
+        job.mark_running()
+        assert job.is_terminal is False
+
+    def test_success_is_terminal(self):
+        job = self._make_job()
+        job.mark_running()
+        job.mark_success()
+        assert job.is_terminal is True
+
+    def test_failed_is_terminal(self):
+        job = self._make_job()
+        job.mark_running()
+        job.mark_failed("error")
+        assert job.is_terminal is True
+
+    def test_cancelled_is_terminal(self):
+        job = self._make_job()
+        job.mark_cancelled()
+        assert job.is_terminal is True
+
+
+class TestJobStatusTransitions:
+    """状态转换测试."""
+
+    def _make_job(self):
+        return Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+
+    def test_pending_to_running(self):
+        """pending → running."""
+        job = self._make_job()
+        job.transition_to(JobStatus.RUNNING)
+        assert job.status == JobStatus.RUNNING
+
+    def test_pending_to_success(self):
+        """pending → success（瞬时任务）."""
+        job = self._make_job()
+        job.transition_to(JobStatus.SUCCESS)
+        assert job.status == JobStatus.SUCCESS
+
+    def test_pending_to_cancelled(self):
+        """pending → cancelled."""
+        job = self._make_job()
+        job.transition_to(JobStatus.CANCELLED)
+        assert job.status == JobStatus.CANCELLED
+
+    def test_running_to_success(self):
+        """running → success."""
+        job = self._make_job()
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.SUCCESS)
+        assert job.status == JobStatus.SUCCESS
+
+    def test_running_to_failed(self):
+        """running → failed."""
+        job = self._make_job()
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.FAILED)
+        assert job.status == JobStatus.FAILED
+
+    def test_running_to_cancelled(self):
+        """running → cancelled."""
+        job = self._make_job()
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.CANCELLED)
+        assert job.status == JobStatus.CANCELLED
+
+    def test_failed_to_pending_retry(self):
+        """failed → pending（重试）."""
+        job = self._make_job()
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.FAILED)
+        job.transition_to(JobStatus.PENDING)
+        assert job.status == JobStatus.PENDING
+
+    def test_cannot_success_to_running(self):
+        """success 不能回 running."""
+        job = self._make_job()
+        job.mark_running()
+        job.mark_success()
+        try:
+            job.transition_to(JobStatus.RUNNING)
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "非法状态转换" in str(e)
+
+    def test_cannot_pending_to_failed_directly(self):
+        """pending 不能直接到 failed（必须经过 running）."""
+        job = self._make_job()
+        try:
+            job.transition_to(JobStatus.FAILED)
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "非法状态转换" in str(e)
+
+    def test_transition_with_string(self):
+        """用字符串传状态."""
+        job = self._make_job()
+        job.transition_to("running")
+        assert job.status == JobStatus.RUNNING
+
+    def test_transition_invalid_string(self):
+        """无效状态字符串."""
+        job = self._make_job()
+        try:
+            job.transition_to("invalid")
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "无效状态" in str(e)
+
+    def test_transition_sets_started_at(self):
+        """第一次到 running 设置 started_at."""
+        job = self._make_job()
+        assert job.started_at is None
+        job.transition_to(JobStatus.RUNNING)
+        assert job.started_at is not None
+
+    def test_transition_sets_completed_at_on_success(self):
+        """success 设置 completed_at."""
+        job = self._make_job()
+        job.mark_running()
+        assert job.completed_at is None
+        job.mark_success()
+        assert job.completed_at is not None
+
+    def test_transition_sets_completed_at_on_failed(self):
+        """failed 设置 completed_at."""
+        job = self._make_job()
+        job.mark_running()
+        job.mark_failed("err")
+        assert job.completed_at is not None
+
+
+class TestJobMarkMethods:
+    """便捷标记方法测试."""
+
+    def _make_job(self):
+        return Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+
+    def test_mark_running(self):
+        job = self._make_job()
+        job.mark_running()
+        assert job.status == JobStatus.RUNNING
+
+    def test_mark_running_with_stage(self):
+        job = self._make_job()
+        job.mark_running(stage="正在合成视频")
+        assert job.status == JobStatus.RUNNING
+        assert job.current_stage == "正在合成视频"
+
+    def test_mark_success(self):
+        job = self._make_job()
+        job.mark_running()
+        job.mark_success()
+        assert job.status == JobStatus.SUCCESS
+        assert job.progress == 100.0
+        assert job.current_stage == "完成"
+
+    def test_mark_success_with_result(self):
+        job = self._make_job()
+        job.mark_running()
+        job.mark_success(result={"video_url": "http://x/v.mp4"})
+        assert job.result == {"video_url": "http://x/v.mp4"}
+
+    def test_mark_failed(self):
+        job = self._make_job()
+        job.mark_running()
+        job.mark_failed("网络超时")
+        assert job.status == JobStatus.FAILED
+        assert job.error_message == "网络超时"
+        assert job.current_stage == "失败"
+
+    def test_mark_cancelled(self):
+        job = self._make_job()
+        job.mark_cancelled()
+        assert job.status == JobStatus.CANCELLED
+        assert job.current_stage == "已取消"
+
+
+class TestJobProgress:
+    """进度更新测试."""
+
+    def _make_job(self):
+        return Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+
+    def test_update_progress(self):
+        """正常更新进度."""
+        job = self._make_job()
+        job.update_progress(50.0)
+        assert job.progress == 50.0
+
+    def test_update_progress_with_stage(self):
+        """更新进度同时更新阶段."""
+        job = self._make_job()
+        job.update_progress(30.0, stage="合成中")
+        assert job.progress == 30.0
+        assert job.current_stage == "合成中"
+
+    def test_update_progress_zero(self):
+        """0% 合法."""
+        job = self._make_job()
+        job.update_progress(0.0)
+        assert job.progress == 0.0
+
+    def test_update_progress_100(self):
+        """100% 合法."""
+        job = self._make_job()
+        job.update_progress(100.0)
+        assert job.progress == 100.0
+
+    def test_update_progress_negative(self):
+        """负进度无效."""
+        job = self._make_job()
+        try:
+            job.update_progress(-1.0)
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "进度" in str(e)
+
+    def test_update_progress_over_100(self):
+        """超过100%无效."""
+        job = self._make_job()
+        try:
+            job.update_progress(101.0)
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "进度" in str(e)
+
+    def test_update_progress_updates_timestamp(self):
+        """更新进度时更新 updated_at."""
+        job = self._make_job()
+        old = job.updated_at
+        job.update_progress(50.0)
+        assert job.updated_at >= old
+
+
+class TestJobRetry:
+    """重试机制测试."""
+
+    def _make_failed_job(self, max_retries=3):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=max_retries)
+        job.mark_running()
+        job.mark_failed("error")
+        return job
+
+    def test_is_retryable_true(self):
+        """失败且未超过重试次数时可重试."""
+        job = self._make_failed_job(max_retries=3)
+        assert job.is_retryable is True
+
+    def test_is_retryable_false_not_failed(self):
+        """非失败状态不可重试."""
+        job = Job.create("p", JobType.VIDEO_COMPOSE)
+        assert job.is_retryable is False
+
+    def test_is_retryable_false_exceeded(self):
+        """超过重试次数不可重试."""
+        job = self._make_failed_job(max_retries=0)
+        assert job.is_retryable is False
+
+    def test_prepare_retry(self):
+        """准备重试."""
+        job = self._make_failed_job(max_retries=3)
+        job.prepare_retry()
+        assert job.status == JobStatus.PENDING
+        assert job.retry_count == 1
+        assert job.progress == 0.0
+        assert "第 1 次重试" in job.current_stage
+        assert job.error_message == ""
+        assert job.started_at is None
+        assert job.completed_at is None
+        assert job.celery_task_id == ""
+
+    def test_prepare_retry_multiple_times(self):
+        """多次重试."""
+        job = self._make_failed_job(max_retries=3)
+        job.prepare_retry()
+        job.mark_running()
+        job.mark_failed("err2")
+        assert job.retry_count == 1
+        job.prepare_retry()
+        assert job.retry_count == 2
+        assert "第 2 次重试" in job.current_stage
+
+    def test_prepare_retry_not_retryable(self):
+        """不可重试时报错."""
+        job = self._make_failed_job(max_retries=0)
+        try:
+            job.prepare_retry()
+            raise AssertionError("unexpected success")
+        except ValueError as e:
+            assert "不可重试" in str(e)
+
+
+class TestJobToDict:
+    """to_dict 序列化测试."""
+
+    def test_to_dict_keys(self):
+        """包含所有必要字段."""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        d = job.to_dict()
+        assert isinstance(d, dict)
+        keys = [
+            "id",
+            "project_id",
+            "job_type",
+            "status",
+            "progress",
+            "payload",
+            "result",
+            "error_message",
+            "retry_count",
+            "max_retries",
+            "source_id",
+            "is_retryable",
+            "created_at",
+            "updated_at",
+        ]
+        for k in keys:
+            assert k in d, f"缺少字段: {k}"
+
+    def test_to_dict_enums_as_strings(self):
+        """枚举值序列化为字符串."""
+        job = Job.create(project_id="p1", job_type=JobType.CLASSIFICATION)
+        d = job.to_dict()
+        assert d["job_type"] == "classification"
+        assert d["status"] == "pending"
+
+    def test_to_dict_none_timestamps(self):
+        """None 时间戳序列化为 None."""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        d = job.to_dict()
+        assert d["started_at"] is None
+        assert d["completed_at"] is None
+
+    def test_to_dict_after_success(self):
+        """成功后的序列化."""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running("test")
+        job.mark_success({"url": "http://x"})
+        d = job.to_dict()
+        assert d["status"] == "success"
+        assert d["progress"] == 100.0
+        assert d["result"] == {"url": "http://x"}
+        assert d["started_at"] is not None
+        assert d["completed_at"] is not None
+        assert d["is_retryable"] is False