xiaoxia-saas/tests/unit/domain/test_job.py

"""job 单测.

domain 层统一异步任务纯逻辑模块，0 外部依赖。
覆盖：枚举、create工厂、状态机、进度更新、重试机制、序列化。
"""

from __future__ import annotations

from packages.domain.job import TERMINAL_STATUSES, Job, JobStatus, JobType


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