xiaoxia-saas/tests/unit/test_job_domain.py

"""
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