xiaoxia-saas/tests/unit/test_job_service.py

"""Unit tests for JobService — Phase 8 任务 2.10.

测试 Job 领域模型、Use Cases 和 Service 层。
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

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

# ── Fixtures ──────────────────────────────────────────────────────────────────


class FakeJobRepo:
    """内存中的 JobRepository 测试替身。"""

    def __init__(self):
        self._store: dict[str, Job] = {}

    def create(self, job: Job) -> Job:
        self._store[job.id] = job
        return job

    def get(self, job_id: str) -> Job | None:
        return self._store.get(job_id)

    def update(self, job: Job) -> Job:
        if job.id not in self._store:
            raise ValueError(f"Job {job.id} not found")
        self._store[job.id] = job
        return job

    def list_by_project(self, project_id, *, job_type=None, status=None, limit=50, offset=0):
        results = [
            j
            for j in self._store.values()
            if j.project_id == project_id
            and (job_type is None or j.job_type == job_type or j.job_type == JobType(job_type))
            and (status is None or j.status == status or j.status == JobStatus(status))
        ]
        return results[offset : offset + limit]

    def list_by_user(self, user_id, *, job_type=None, status=None, limit=50, offset=0):
        results = [
            j
            for j in self._store.values()
            if j.created_by_user_id == user_id
            and (job_type is None or j.job_type == job_type or j.job_type == JobType(job_type))
            and (status is None or j.status == status or j.status == JobStatus(status))
        ]
        return results[offset : offset + limit]

    def count_by_project(self, project_id, *, status=None):
        return len(
            [
                j
                for j in self._store.values()
                if j.project_id == project_id
                and (status is None or j.status == status or j.status == JobStatus(status))
            ]
        )

    def find_active_by_source(self, source_id, job_type):
        jt = job_type.value if isinstance(job_type, JobType) else job_type
        for j in self._store.values():
            if (
                j.source_id == source_id
                and j.job_type.value == jt
                and j.status in (JobStatus.PENDING, JobStatus.RUNNING)
            ):
                return j
        return None


@pytest.fixture
def repo():
    return FakeJobRepo()


# ── Job 领域模型测试 ──────────────────────────────────────────────────────────


class TestJobDomainModel:
    def test_create_job(self):
        job = Job.create(
            project_id="proj-1",
            job_type=JobType.VIDEO_COMPOSE,
            payload={"plan_id": "plan-1"},
            source_id="plan-1",
            created_by_user_id="user-1",
        )
        assert job.project_id == "proj-1"
        assert job.job_type == JobType.VIDEO_COMPOSE
        assert job.status == JobStatus.PENDING
        assert job.progress == 0.0
        assert job.retry_count == 0
        assert job.max_retries == 3
        assert job.payload == {"plan_id": "plan-1"}

    def test_create_job_with_string_type(self):
        job = Job.create(project_id="proj-1", job_type="render_edit_plan")
        assert job.job_type == JobType.RENDER_EDIT_PLAN

    def test_create_job_invalid_type_raises(self):
        with pytest.raises(ValueError, match="不支持的任务类型"):
            Job.create(project_id="proj-1", job_type="invalid_type")

    def test_create_job_empty_project_id_raises(self):
        with pytest.raises(ValueError, match="project_id 不能为空"):
            Job.create(project_id="  ", job_type=JobType.VIDEO_COMPOSE)

    def test_is_terminal(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        assert not job.is_terminal

        job.status = JobStatus.SUCCESS
        assert job.is_terminal

        job.status = JobStatus.FAILED
        assert job.is_terminal

        job.status = JobStatus.CANCELLED
        assert job.is_terminal

    def test_is_retryable(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=2)
        job.status = JobStatus.FAILED
        assert job.is_retryable

        job.retry_count = 2
        assert not job.is_retryable

    def test_mark_running(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        job.mark_running("初始化中")
        assert job.status == JobStatus.RUNNING
        assert job.current_stage == "初始化中"
        assert job.started_at is not None

    def test_mark_success(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        job.mark_running()
        job.mark_success(result={"output_url": "https://example.com/video.mp4"})
        assert job.status == JobStatus.SUCCESS
        assert job.progress == 100.0
        assert job.current_stage == "完成"
        assert job.result == {"output_url": "https://example.com/video.mp4"}
        assert job.completed_at is not None

    def test_mark_failed(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        job.mark_running()
        job.mark_failed("FFmpeg 执行失败")
        assert job.status == JobStatus.FAILED
        assert job.error_message == "FFmpeg 执行失败"
        assert job.completed_at is not None

    def test_mark_cancelled(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        job.mark_cancelled()
        assert job.status == JobStatus.CANCELLED
        assert job.current_stage == "已取消"

    def test_update_progress(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        job.update_progress(50.0, "合成中")
        assert job.progress == 50.0
        assert job.current_stage == "合成中"

    def test_update_progress_invalid_value_raises(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
            job.update_progress(150.0)

    def test_prepare_retry(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=3)
        job.mark_running()
        job.mark_failed("临时错误")
        job.prepare_retry()
        assert job.status == JobStatus.PENDING
        assert job.retry_count == 1
        assert job.progress == 0.0
        assert job.error_message == ""
        assert "第 1 次重试" in job.current_stage

    def test_prepare_retry_exceeded_raises(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=1)
        job.mark_running()
        job.mark_failed("错误")
        job.prepare_retry()  # retry_count = 1, max_retries = 1 → not retryable
        with pytest.raises(ValueError, match="任务不可重试"):
            job.prepare_retry()

    def test_invalid_transition_raises(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        job.mark_running()
        job.mark_success()
        with pytest.raises(ValueError, match="非法状态转换"):
            job.transition_to(JobStatus.RUNNING)  # success → running 不合法（终态不可转换）

    def test_to_dict(self):
        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        d = job.to_dict()
        assert d["project_id"] == "proj-1"
        assert d["job_type"] == "video_compose"
        assert d["status"] == "pending"
        assert "is_retryable" in d


# ── Use Case 测试 ─────────────────────────────────────────────────────────────


class TestCreateJobUseCase:
    def test_execute(self, repo):
        use_case = CreateJobUseCase(repo)
        cmd = CreateJobCommand(
            project_id="proj-1",
            job_type=JobType.VIDEO_COMPOSE,
            payload={"plan_id": "plan-1"},
            source_id="plan-1",
            created_by_user_id="user-1",
        )
        job = use_case.execute(cmd)
        assert job.id in repo._store
        assert job.status == JobStatus.PENDING

    def test_execute_with_string_type(self, repo):
        use_case = CreateJobUseCase(repo)
        cmd = CreateJobCommand(
            project_id="proj-1",
            job_type="video_compose",
        )
        job = use_case.execute(cmd)
        assert job.job_type == JobType.VIDEO_COMPOSE


class TestSubmitJobUseCase:
    def test_submit_pending_job(self, repo):
        # 先创建
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))

        # 提交
        submit_uc = SubmitJobUseCase(repo)
        submitted = submit_uc.execute(job.id, celery_task_id="celery-abc")
        assert submitted.status == JobStatus.RUNNING
        assert submitted.celery_task_id == "celery-abc"
        assert submitted.started_at is not None

    def test_submit_non_pending_raises(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        job.mark_running()
        repo.update(job)

        submit_uc = SubmitJobUseCase(repo)
        with pytest.raises(ValueError, match="只有 pending 状态"):
            submit_uc.execute(job.id)

    def test_submit_nonexistent_raises(self, repo):
        submit_uc = SubmitJobUseCase(repo)
        with pytest.raises(ValueError, match="任务不存在"):
            submit_uc.execute("nonexistent-id")


class TestUpdateJobProgressUseCase:
    def test_update_running_job(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        job.mark_running()
        repo.update(job)

        progress_uc = UpdateJobProgressUseCase(repo)
        updated = progress_uc.execute(UpdateJobProgressCommand(job_id=job.id, progress=75.0, current_stage="渲染中"))
        assert updated.progress == 75.0
        assert updated.current_stage == "渲染中"

    def test_update_non_running_raises(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))

        progress_uc = UpdateJobProgressUseCase(repo)
        with pytest.raises(ValueError, match="只有 running 状态"):
            progress_uc.execute(UpdateJobProgressCommand(job_id=job.id, progress=50.0))


class TestCompleteJobUseCase:
    def test_complete_running_job(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        job.mark_running()
        repo.update(job)

        complete_uc = CompleteJobUseCase(repo)
        completed = complete_uc.execute(CompleteJobCommand(job_id=job.id, result={"url": "https://example.com/v.mp4"}))
        assert completed.status == JobStatus.SUCCESS
        assert completed.progress == 100.0
        assert completed.result == {"url": "https://example.com/v.mp4"}

    def test_complete_non_running_raises(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        job.mark_running()
        job.mark_success()  # 已经是终态

        complete_uc = CompleteJobUseCase(repo)
        with pytest.raises(ValueError, match="只有 running/pending 状态"):
            complete_uc.execute(CompleteJobCommand(job_id=job.id))


class TestFailJobUseCase:
    def test_fail_job(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        job.mark_running()
        repo.update(job)

        fail_uc = FailJobUseCase(repo)
        failed = fail_uc.execute(FailJobCommand(job_id=job.id, error_message="磁盘空间不足"))
        assert failed.status == JobStatus.FAILED
        assert failed.error_message == "磁盘空间不足"


class TestRetryJobUseCase:
    def test_retry_failed_job(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=3))
        job.mark_running()
        job.mark_failed("网络超时")
        repo.update(job)

        retry_uc = RetryJobUseCase(repo)
        retried = retry_uc.execute(job.id)
        assert retried.status == JobStatus.PENDING
        assert retried.retry_count == 1
        assert retried.error_message == ""

    def test_retry_non_failed_raises(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))

        retry_uc = RetryJobUseCase(repo)
        with pytest.raises(ValueError, match="任务不可重试"):
            retry_uc.execute(job.id)


class TestCancelJobUseCase:
    def test_cancel_pending_job(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))

        cancel_uc = CancelJobUseCase(repo)
        cancelled = cancel_uc.execute(job.id)
        assert cancelled.status == JobStatus.CANCELLED

    def test_cancel_completed_raises(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        job.mark_running()
        job.mark_success()
        repo.update(job)

        cancel_uc = CancelJobUseCase(repo)
        with pytest.raises(ValueError, match="已处于终态"):
            cancel_uc.execute(job.id)


class TestGetJobUseCase:
    def test_get_existing_job(self, repo):
        create_uc = CreateJobUseCase(repo)
        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))

        get_uc = GetJobUseCase(repo)
        fetched = get_uc.execute(job.id)
        assert fetched is not None
        assert fetched.id == job.id

    def test_get_nonexistent_returns_none(self, repo):
        get_uc = GetJobUseCase(repo)
        assert get_uc.execute("nonexistent") is None


class TestListJobsUseCase:
    def test_list_by_project(self, repo):
        create_uc = CreateJobUseCase(repo)
        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.RENDER_EDIT_PLAN))
        create_uc.execute(CreateJobCommand(project_id="proj-2", job_type=JobType.VIDEO_COMPOSE))

        list_uc = ListJobsUseCase(repo)
        jobs = list_uc.execute(project_id="proj-1")
        assert len(jobs) == 2

    def test_list_by_project_with_type_filter(self, repo):
        create_uc = CreateJobUseCase(repo)
        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.RENDER_EDIT_PLAN))

        list_uc = ListJobsUseCase(repo)
        jobs = list_uc.execute(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
        assert len(jobs) == 1
        assert jobs[0].job_type == JobType.VIDEO_COMPOSE

    def test_list_by_project_with_status_filter(self, repo):
        create_uc = CreateJobUseCase(repo)
        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        j2 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        j2.mark_running()
        repo.update(j2)

        list_uc = ListJobsUseCase(repo)
        running_jobs = list_uc.execute(project_id="proj-1", status=JobStatus.RUNNING)
        assert len(running_jobs) == 1

    def test_list_by_user(self, repo):
        create_uc = CreateJobUseCase(repo)
        create_uc.execute(
            CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, created_by_user_id="user-1")
        )
        create_uc.execute(
            CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, created_by_user_id="user-2")
        )

        list_uc = ListJobsUseCase(repo)
        jobs = list_uc.execute(user_id="user-1")
        assert len(jobs) == 1

    def test_list_without_project_or_user_raises(self, repo):
        list_uc = ListJobsUseCase(repo)
        with pytest.raises(ValueError, match="必须指定"):
            list_uc.execute()


class TestGetJobStatisticsUseCase:
    def test_statistics(self, repo):
        create_uc = CreateJobUseCase(repo)
        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        j2 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        j2.mark_running()
        repo.update(j2)
        j3 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        j3.mark_running()
        j3.mark_success()
        repo.update(j3)
        j4 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
        j4.mark_running()
        j4.mark_failed("err")
        repo.update(j4)

        stats_uc = GetJobStatisticsUseCase(repo)
        stats = stats_uc.execute("proj-1")
        assert stats["total"] == 4
        assert stats["pending"] == 1
        assert stats["running"] == 1
        assert stats["success"] == 1
        assert stats["failed"] == 1


# ── JobService 集成测试 ───────────────────────────────────────────────────────


class TestJobService:
    def test_create_compose_job(self, repo):
        from app.services.job_service import JobService

        svc = JobService(repo)
        job = svc.create_compose_job(
            project_id="proj-1",
            plan_id="plan-1",
            user_id="user-1",
        )
        assert job.job_type == JobType.VIDEO_COMPOSE
        assert job.payload == {"plan_id": "plan-1"}
        assert job.source_id == "plan-1"
        assert job.created_by_user_id == "user-1"

    def test_create_render_job(self, repo):
        from app.services.job_service import JobService

        svc = JobService(repo)
        job = svc.create_render_job(
            project_id="proj-1",
            plan_id="plan-1",
            user_id="user-1",
        )
        assert job.job_type == JobType.RENDER_EDIT_PLAN

    def test_submit_compose_if_not_exists_creates_new(self, repo):
        from app.services.job_service import JobService

        svc = JobService(repo)
        job, created = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-1")
        assert created is True
        assert job.status == JobStatus.RUNNING
        assert job.celery_task_id == "celery-1"

    def test_submit_compose_if_not_exists_returns_existing(self, repo):
        from app.services.job_service import JobService

        svc = JobService(repo)
        job1, created1 = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-1")
        job2, created2 = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-2")
        assert created2 is False
        assert job2.id == job1.id

    def test_full_lifecycle(self, repo):
        """完整生命周期测试：创建 → 提交 → 更新进度 → 完成。"""
        from app.services.job_service import JobService

        svc = JobService(repo)

        # 创建
        job = svc.create_compose_job("proj-1", "plan-1", "user-1")
        assert job.status == JobStatus.PENDING

        # 提交
        job = svc.submit_job(job.id, "celery-xyz")
        assert job.status == JobStatus.RUNNING
        assert job.celery_task_id == "celery-xyz"

        # 更新进度
        job = svc.update_progress(job.id, 50.0, "合成中")
        assert job.progress == 50.0

        job = svc.update_progress(job.id, 80.0, "上传结果")
        assert job.progress == 80.0

        # 完成
        job = svc.complete_job(job.id, result={"url": "https://example.com/v.mp4"})
        assert job.status == JobStatus.SUCCESS
        assert job.progress == 100.0

    def test_full_lifecycle_with_retry(self, repo):
        """带重试的完整生命周期测试。"""
        from app.services.job_service import JobService

        svc = JobService(repo)

        # 创建并执行
        job = svc.create_compose_job("proj-1", "plan-1", "user-1", max_retries=2)
        job = svc.submit_job(job.id, "celery-1")

        # 失败
        job = svc.fail_job(job.id, "临时网络错误")
        assert job.status == JobStatus.FAILED

        # 重试
        job = svc.retry_job(job.id)
        assert job.status == JobStatus.PENDING
        assert job.retry_count == 1

        # 再次提交
        job = svc.submit_job(job.id, "celery-2")
        assert job.status == JobStatus.RUNNING

        # 成功
        job = svc.complete_job(job.id, result={"url": "ok"})
        assert job.status == JobStatus.SUCCESS