xiaoxia
30457629da
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Failing after 1m20s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 48s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 51s
CI/CD Pipeline / Unit Tests (push) Successful in 4m54s
CI/CD Pipeline / Integration Tests (push) Successful in 2m6s
CI/CD Pipeline / Frontend Lint (push) Successful in 28s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 55s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 12m7s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m15s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m6s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m34s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m29s
CI/CD Pipeline / ACR Image Cleanup (push) Failing after 5s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m50s
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
558 lines
20 KiB
Python
Executable File
558 lines
20 KiB
Python
Executable File
"""GenerationTask 领域模型状态机单元测试.
|
|
|
|
覆盖：
|
|
- 初始状态为 pending
|
|
- mark_processing: pending → running
|
|
- mark_completed: running → completed
|
|
- mark_failed: pending/running → failed
|
|
- mark_cancelled: pending/running → cancelled
|
|
- mark_pending_from_failed: failed → pending（重试）
|
|
- 非法状态转换抛出 ValueError
|
|
- is_terminal / is_completed / is_failed / is_running 属性
|
|
- 状态转换时的时间戳设置
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from packages.domain.generation_task import (
|
|
TERMINAL_STATUSES,
|
|
GenerationTask,
|
|
GenerationTaskStatus,
|
|
)
|
|
|
|
|
|
def _make_task(**overrides) -> GenerationTask:
|
|
"""创建一个测试用的 GenerationTask。"""
|
|
defaults = dict(
|
|
id="task-test-001",
|
|
project_id="proj-1",
|
|
asset_library_id="lib-1",
|
|
)
|
|
defaults.update(overrides)
|
|
return GenerationTask(**defaults)
|
|
|
|
|
|
# ── 初始状态 ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestInitialState:
|
|
"""测试初始状态。"""
|
|
|
|
def test_default_status_is_pending(self) -> None:
|
|
"""新创建的任务默认状态为 pending。"""
|
|
task = _make_task()
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
assert task.progress == 0.0
|
|
assert task.result_count == 0
|
|
assert task.error_message == ""
|
|
assert task.started_at is None
|
|
assert task.completed_at is None
|
|
|
|
def test_create_factory_returns_pending(self) -> None:
|
|
"""GenerationTask.create() 返回的任务状态为 pending。"""
|
|
task = GenerationTask.create(
|
|
project_id="proj-1",
|
|
asset_library_id="lib-1",
|
|
created_by_user_id="user-1",
|
|
)
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
assert task.progress == 0.0
|
|
assert task.result_count == 0
|
|
|
|
def test_is_not_terminal_initially(self) -> None:
|
|
"""初始状态不是终态。"""
|
|
task = _make_task()
|
|
assert not task.is_terminal
|
|
assert not task.is_completed
|
|
assert not task.is_failed
|
|
assert not task.is_running
|
|
|
|
def test_terminal_statuses_constant(self) -> None:
|
|
"""终态集合包含 completed / failed / cancelled。"""
|
|
assert GenerationTaskStatus.COMPLETED in TERMINAL_STATUSES
|
|
assert GenerationTaskStatus.FAILED in TERMINAL_STATUSES
|
|
assert GenerationTaskStatus.CANCELLED in TERMINAL_STATUSES
|
|
assert GenerationTaskStatus.PENDING not in TERMINAL_STATUSES
|
|
assert GenerationTaskStatus.RUNNING not in TERMINAL_STATUSES
|
|
|
|
|
|
# ── mark_processing ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestMarkProcessing:
|
|
"""测试 pending → running 转换。"""
|
|
|
|
def test_pending_to_running_success(self) -> None:
|
|
"""pending 状态的任务可以标记为 running。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
assert task.status == GenerationTaskStatus.RUNNING
|
|
assert task.is_running
|
|
assert task.started_at is not None
|
|
assert task.error_message == ""
|
|
|
|
def test_started_at_is_set(self) -> None:
|
|
"""mark_processing 设置 started_at 时间戳。"""
|
|
task = _make_task()
|
|
assert task.started_at is None
|
|
task.mark_processing()
|
|
assert task.started_at is not None
|
|
|
|
def test_error_message_cleared(self) -> None:
|
|
"""mark_processing 清除 error_message（如果有的话）。"""
|
|
task = _make_task()
|
|
# 注意：pending 状态通常没有 error_message，这里验证确保被清除
|
|
task.error_message = "some old error"
|
|
# 直接设置状态绕过校验（模拟异常场景）
|
|
task.status = GenerationTaskStatus.PENDING
|
|
task.mark_processing()
|
|
assert task.error_message == ""
|
|
|
|
def test_running_to_running_raises(self) -> None:
|
|
"""running 状态不能再次 mark_processing。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_processing()
|
|
|
|
def test_completed_to_running_raises(self) -> None:
|
|
"""completed 状态不能回到 running。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_processing()
|
|
|
|
def test_failed_to_running_raises(self) -> None:
|
|
"""failed 状态不能直接到 running（应先重置为 pending）。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("some error")
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_processing()
|
|
|
|
|
|
# ── mark_completed ───────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestMarkCompleted:
|
|
"""测试 running → completed 转换。"""
|
|
|
|
def test_running_to_completed_success(self) -> None:
|
|
"""running 状态的任务可以标记为 completed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
assert task.status == GenerationTaskStatus.COMPLETED
|
|
assert task.is_completed
|
|
assert task.is_terminal
|
|
assert task.completed_at is not None
|
|
|
|
def test_progress_set_to_100(self) -> None:
|
|
"""mark_completed 设置 progress 为 100.0。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.progress = 50.0 # 模拟中间进度
|
|
task.mark_completed()
|
|
assert task.progress == 100.0
|
|
|
|
def test_default_result_count_is_1(self) -> None:
|
|
"""默认 result_count 为 1。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
assert task.result_count == 1
|
|
|
|
def test_custom_result_count(self) -> None:
|
|
"""可以指定 result_count。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed(result_count=5)
|
|
assert task.result_count == 5
|
|
|
|
def test_error_message_cleared(self) -> None:
|
|
"""mark_completed 清除 error_message。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.error_message = "temporary error"
|
|
task.mark_completed()
|
|
assert task.error_message == ""
|
|
|
|
def test_completed_at_is_set(self) -> None:
|
|
"""mark_completed 设置 completed_at。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
assert task.completed_at is None
|
|
task.mark_completed()
|
|
assert task.completed_at is not None
|
|
|
|
def test_pending_to_completed_raises(self) -> None:
|
|
"""pending 状态不能直接到 completed。"""
|
|
task = _make_task()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_completed()
|
|
|
|
def test_completed_to_completed_raises(self) -> None:
|
|
"""completed 状态不能再次 mark_completed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_completed()
|
|
|
|
def test_failed_to_completed_raises(self) -> None:
|
|
"""failed 状态不能直接到 completed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("error")
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_completed()
|
|
|
|
|
|
# ── mark_failed ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestMarkFailed:
|
|
"""测试 pending/running → failed 转换。"""
|
|
|
|
def test_pending_to_failed_success(self) -> None:
|
|
"""pending 状态可以直接标记为 failed。"""
|
|
task = _make_task()
|
|
task.mark_failed("资源不足")
|
|
assert task.status == GenerationTaskStatus.FAILED
|
|
assert task.is_failed
|
|
assert task.is_terminal
|
|
assert task.error_message == "资源不足"
|
|
assert task.completed_at is not None
|
|
|
|
def test_running_to_failed_success(self) -> None:
|
|
"""running 状态可以标记为 failed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("生成失败：FFmpeg 错误")
|
|
assert task.status == GenerationTaskStatus.FAILED
|
|
assert task.is_failed
|
|
assert task.is_terminal
|
|
assert task.error_message == "生成失败：FFmpeg 错误"
|
|
assert task.completed_at is not None
|
|
|
|
def test_completed_to_failed_raises(self) -> None:
|
|
"""completed 状态不能标记为 failed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_failed("late error")
|
|
|
|
def test_failed_to_failed_raises(self) -> None:
|
|
"""failed 状态不能再次 mark_failed。"""
|
|
task = _make_task()
|
|
task.mark_failed("first error")
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_failed("second error")
|
|
|
|
def test_error_message_preserved(self) -> None:
|
|
"""错误信息被正确保存。"""
|
|
task = _make_task()
|
|
error_msg = "FFmpeg returned non-zero exit status 1"
|
|
task.mark_failed(error_msg)
|
|
assert task.error_message == error_msg
|
|
|
|
|
|
# ── mark_cancelled ───────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestMarkCancelled:
|
|
"""测试 pending/running → cancelled 转换。"""
|
|
|
|
def test_pending_to_cancelled_success(self) -> None:
|
|
"""pending 状态可以取消。"""
|
|
task = _make_task()
|
|
task.mark_cancelled()
|
|
assert task.status == GenerationTaskStatus.CANCELLED
|
|
assert task.is_terminal
|
|
assert task.completed_at is not None
|
|
|
|
def test_running_to_cancelled_success(self) -> None:
|
|
"""running 状态可以取消。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_cancelled()
|
|
assert task.status == GenerationTaskStatus.CANCELLED
|
|
assert task.is_terminal
|
|
|
|
def test_completed_to_cancelled_raises(self) -> None:
|
|
"""completed 状态不能取消。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_cancelled()
|
|
|
|
def test_failed_to_cancelled_raises(self) -> None:
|
|
"""failed 状态不能取消。"""
|
|
task = _make_task()
|
|
task.mark_failed("some error")
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.mark_cancelled()
|
|
|
|
|
|
# ── mark_pending_from_failed (重试) ─────────────────────────────────────────
|
|
|
|
|
|
class TestMarkPendingFromFailed:
|
|
"""测试 failed → pending（重试）转换。"""
|
|
|
|
def test_failed_to_pending_success(self) -> None:
|
|
"""failed 状态可以重置为 pending（用于重试）。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("临时错误")
|
|
task.mark_pending_from_failed()
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
assert not task.is_terminal
|
|
assert task.error_message == ""
|
|
assert task.started_at is None
|
|
assert task.completed_at is None
|
|
assert task.progress == 0.0
|
|
assert task.result_count == 0
|
|
|
|
def test_pending_to_pending_raises(self) -> None:
|
|
"""pending 状态不能调用 mark_pending_from_failed。"""
|
|
task = _make_task()
|
|
with pytest.raises(ValueError, match="只有 failed 状态"):
|
|
task.mark_pending_from_failed()
|
|
|
|
def test_running_to_pending_raises(self) -> None:
|
|
"""running 状态不能调用 mark_pending_from_failed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
with pytest.raises(ValueError, match="只有 failed 状态"):
|
|
task.mark_pending_from_failed()
|
|
|
|
def test_completed_to_pending_raises(self) -> None:
|
|
"""completed 状态不能调用 mark_pending_from_failed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
with pytest.raises(ValueError, match="只有 failed 状态"):
|
|
task.mark_pending_from_failed()
|
|
|
|
|
|
# ── transition_to 通用方法 ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestTransitionTo:
|
|
"""测试通用的 transition_to 方法。"""
|
|
|
|
def test_string_status_conversion(self) -> None:
|
|
"""可以传入字符串形式的状态。"""
|
|
task = _make_task()
|
|
task.transition_to("running")
|
|
assert task.status == GenerationTaskStatus.RUNNING
|
|
|
|
def test_invalid_string_raises(self) -> None:
|
|
"""无效的状态字符串抛出 ValueError。"""
|
|
task = _make_task()
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
task.transition_to("invalid_status")
|
|
|
|
def test_enum_status(self) -> None:
|
|
"""可以传入枚举形式的状态。"""
|
|
task = _make_task()
|
|
task.transition_to(GenerationTaskStatus.RUNNING)
|
|
assert task.status == GenerationTaskStatus.RUNNING
|
|
|
|
def test_error_message_includes_allowed_statuses(self) -> None:
|
|
"""错误信息包含允许的状态列表。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
with pytest.raises(ValueError) as exc_info:
|
|
task.transition_to(GenerationTaskStatus.RUNNING)
|
|
assert "completed" in str(exc_info.value)
|
|
assert "running" in str(exc_info.value)
|
|
|
|
|
|
# ── 完整流转路径 ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestFullFlow:
|
|
"""测试完整的状态流转路径。"""
|
|
|
|
def test_happy_path(self) -> None:
|
|
"""正常路径：pending → running → completed。"""
|
|
task = _make_task()
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
assert not task.is_terminal
|
|
|
|
task.mark_processing()
|
|
assert task.status == GenerationTaskStatus.RUNNING
|
|
assert task.started_at is not None
|
|
assert not task.is_terminal
|
|
|
|
task.mark_completed(result_count=3)
|
|
assert task.status == GenerationTaskStatus.COMPLETED
|
|
assert task.is_completed
|
|
assert task.is_terminal
|
|
assert task.completed_at is not None
|
|
assert task.result_count == 3
|
|
assert task.progress == 100.0
|
|
|
|
def test_failure_path_from_running(self) -> None:
|
|
"""失败路径：pending → running → failed。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
assert task.is_running
|
|
|
|
task.mark_failed("网络超时")
|
|
assert task.is_failed
|
|
assert task.is_terminal
|
|
assert task.error_message == "网络超时"
|
|
assert task.completed_at is not None
|
|
|
|
def test_failure_path_from_pending(self) -> None:
|
|
"""失败路径：pending → failed（启动前校验失败等）。"""
|
|
task = _make_task()
|
|
task.mark_failed("参数校验失败")
|
|
assert task.is_failed
|
|
assert task.is_terminal
|
|
|
|
def test_retry_path(self) -> None:
|
|
"""重试路径：pending → running → failed → pending → running → completed。"""
|
|
task = _make_task()
|
|
|
|
# 第一次尝试失败
|
|
task.mark_processing()
|
|
task.mark_failed("临时错误")
|
|
assert task.is_failed
|
|
|
|
# 重试
|
|
task.mark_pending_from_failed()
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
assert task.error_message == ""
|
|
|
|
# 第二次成功
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
assert task.is_completed
|
|
|
|
def test_cancel_from_pending(self) -> None:
|
|
"""取消路径：pending → cancelled。"""
|
|
task = _make_task()
|
|
task.mark_cancelled()
|
|
assert task.status == GenerationTaskStatus.CANCELLED
|
|
assert task.is_terminal
|
|
|
|
def test_cancel_from_running(self) -> None:
|
|
"""取消路径：pending → running → cancelled。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_cancelled()
|
|
assert task.status == GenerationTaskStatus.CANCELLED
|
|
assert task.is_terminal
|
|
|
|
|
|
# ── 错误信息与重试（任务中心升级） ──────────────────────────────────────────
|
|
|
|
|
|
class TestErrorInfo:
|
|
"""测试 error_info 结构化错误信息。"""
|
|
|
|
def test_mark_failed_default_error_info(self) -> None:
|
|
"""mark_failed 不传 error_info 时自动生成默认结构。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("something went wrong")
|
|
assert task.is_failed
|
|
assert task.error_message == "something went wrong"
|
|
assert task.error_info["error_type"] == "UnknownError"
|
|
assert task.error_info["message"] == "something went wrong"
|
|
assert "failed_at" in task.error_info
|
|
|
|
def test_mark_failed_with_custom_error_info(self) -> None:
|
|
"""mark_failed 传自定义 error_info。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
info = {
|
|
"error_type": "FFmpegError",
|
|
"message": "Invalid data found",
|
|
"stack_trace": "Traceback...",
|
|
"stage": "render",
|
|
"failed_at": "2026-01-01T00:00:00+00:00",
|
|
}
|
|
task.mark_failed("Invalid data found", error_info=info)
|
|
assert task.error_info == info
|
|
|
|
def test_error_info_cleared_on_retry(self) -> None:
|
|
"""重试时 error_info 被清空。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("oops")
|
|
assert task.error_info # 失败时有值
|
|
task.mark_pending_from_failed()
|
|
assert task.error_info == {}
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
|
|
|
|
class TestRetryCount:
|
|
"""测试 retry_count 重试次数。"""
|
|
|
|
def test_default_retry_count_is_zero(self) -> None:
|
|
"""新任务 retry_count 默认 0。"""
|
|
task = _make_task()
|
|
assert task.retry_count == 0
|
|
|
|
def test_retry_increments_count(self) -> None:
|
|
"""每次失败后重试，retry_count +1。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_failed("fail 1")
|
|
task.mark_pending_from_failed()
|
|
assert task.retry_count == 1
|
|
|
|
task.mark_processing()
|
|
task.mark_failed("fail 2")
|
|
task.mark_pending_from_failed()
|
|
assert task.retry_count == 2
|
|
|
|
def test_completed_does_not_affect_retry_count(self) -> None:
|
|
"""正常完成不改变 retry_count。"""
|
|
task = _make_task()
|
|
task.mark_processing()
|
|
task.mark_completed()
|
|
assert task.retry_count == 0
|
|
|
|
|
|
class TestAutoRetryConfig:
|
|
"""测试自动重试配置。"""
|
|
|
|
def test_default_auto_retry_disabled(self) -> None:
|
|
"""默认关闭自动重试。"""
|
|
task = _make_task()
|
|
assert task.auto_retry_enabled is False
|
|
assert task.auto_retry_max == 0
|
|
|
|
def test_create_with_auto_retry(self) -> None:
|
|
"""create 工厂方法支持 auto_retry 参数。"""
|
|
task = GenerationTask.create(
|
|
project_id="proj-1",
|
|
asset_library_id="lib-1",
|
|
auto_retry_enabled=True,
|
|
auto_retry_max=3,
|
|
)
|
|
assert task.auto_retry_enabled is True
|
|
assert task.auto_retry_max == 3
|
|
|
|
def test_auto_retry_max_default_zero(self) -> None:
|
|
"""auto_retry_max 默认 0 表示不自动重试。"""
|
|
task = GenerationTask.create(
|
|
project_id="proj-1",
|
|
asset_library_id="lib-1",
|
|
auto_retry_enabled=True,
|
|
)
|
|
assert task.auto_retry_enabled is True
|
|
assert task.auto_retry_max == 0
|