Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4437a08d4f | |||
| f5f54b98b6 |
@@ -14,6 +14,7 @@ from urllib.parse import urlparse
|
|||||||
|
|
||||||
import oss2
|
import oss2
|
||||||
from worker_app.celery_app import celery_app
|
from worker_app.celery_app import celery_app
|
||||||
|
from worker_app.db import SessionLocal
|
||||||
|
|
||||||
OUTPUT_WIDTH = 1280
|
OUTPUT_WIDTH = 1280
|
||||||
OUTPUT_HEIGHT = 720
|
OUTPUT_HEIGHT = 720
|
||||||
@@ -28,6 +29,58 @@ PUBLIC_API_BASE_URL = os.getenv("PUBLIC_API_BASE_URL", "https://api.xiaoxiajianj
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
# ── 状态更新辅助函数 ──────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
|
||||||
|
"""更新 GenerationTask 状态(独立 session,异常不向外抛出)。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
task_id: 任务 ID
|
||||||
|
status_action: 状态动作名,如 "mark_processing" / "mark_completed" / "mark_failed"
|
||||||
|
**kwargs: 传递给对应方法的参数
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
True 表示更新成功,False 表示更新失败
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
|
||||||
|
SQLAlchemyGenerationTaskRepository,
|
||||||
|
)
|
||||||
|
|
||||||
|
session = SessionLocal()
|
||||||
|
try:
|
||||||
|
repo = SQLAlchemyGenerationTaskRepository(session)
|
||||||
|
task = repo.get(task_id)
|
||||||
|
if task is None:
|
||||||
|
logger.warning("更新任务状态失败:任务不存在 task_id=%s", task_id)
|
||||||
|
return False
|
||||||
|
|
||||||
|
action = getattr(task, status_action, None)
|
||||||
|
if action is None:
|
||||||
|
logger.warning("未知的状态动作: %s", status_action)
|
||||||
|
return False
|
||||||
|
|
||||||
|
action(**kwargs)
|
||||||
|
repo.update(task)
|
||||||
|
logger.info("GenerationTask 状态更新成功: task_id=%s action=%s", task_id, status_action)
|
||||||
|
return True
|
||||||
|
finally:
|
||||||
|
session.close()
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(
|
||||||
|
"更新 GenerationTask 状态异常: task_id=%s action=%s error=%s",
|
||||||
|
task_id,
|
||||||
|
status_action,
|
||||||
|
e,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
# ── FFmpeg / OSS helpers ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
def _run_ffmpeg(command: list[str]) -> None:
|
def _run_ffmpeg(command: list[str]) -> None:
|
||||||
"""执行 FFmpeg 命令"""
|
"""执行 FFmpeg 命令"""
|
||||||
subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603
|
subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603
|
||||||
@@ -155,8 +208,6 @@ def _download_library_assets(
|
|||||||
"""
|
"""
|
||||||
# 导入模型和会话
|
# 导入模型和会话
|
||||||
try:
|
try:
|
||||||
from worker_app.db import SessionLocal
|
|
||||||
|
|
||||||
from packages.adapters.sqlalchemy_impl.models import AssetModel
|
from packages.adapters.sqlalchemy_impl.models import AssetModel
|
||||||
|
|
||||||
session = SessionLocal()
|
session = SessionLocal()
|
||||||
@@ -231,6 +282,9 @@ def _process_with_editing_mode(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Celery Task ──────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
|
@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
|
||||||
def generate_video(self, task_id: str) -> dict:
|
def generate_video(self, task_id: str) -> dict:
|
||||||
"""
|
"""
|
||||||
@@ -242,19 +296,21 @@ def generate_video(self, task_id: str) -> dict:
|
|||||||
Returns:
|
Returns:
|
||||||
生成结果字典
|
生成结果字典
|
||||||
"""
|
"""
|
||||||
from worker_app.db import SessionLocal
|
from packages.domain import EditingMode
|
||||||
|
|
||||||
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
|
logger.info("开始生成视频任务: task_id=%s", task_id)
|
||||||
SQLAlchemyGenerationTaskRepository,
|
|
||||||
)
|
|
||||||
from packages.domain import EditingMode, GeneratedVideo, GenerationTaskStatus
|
|
||||||
|
|
||||||
# 从数据库加载任务信息
|
# 从数据库加载任务信息
|
||||||
session = SessionLocal()
|
session = SessionLocal()
|
||||||
try:
|
try:
|
||||||
|
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
|
||||||
|
SQLAlchemyGenerationTaskRepository,
|
||||||
|
)
|
||||||
|
|
||||||
task_repo = SQLAlchemyGenerationTaskRepository(session)
|
task_repo = SQLAlchemyGenerationTaskRepository(session)
|
||||||
gen_task = task_repo.get(task_id)
|
gen_task = task_repo.get(task_id)
|
||||||
if gen_task is None:
|
if gen_task is None:
|
||||||
|
logger.error("生成任务不存在: task_id=%s", task_id)
|
||||||
return {"status": "failed", "error": f"generation task {task_id} not found"}
|
return {"status": "failed", "error": f"generation task {task_id} not found"}
|
||||||
project_id = gen_task.project_id
|
project_id = gen_task.project_id
|
||||||
asset_library_id = gen_task.asset_library_id
|
asset_library_id = gen_task.asset_library_id
|
||||||
@@ -265,6 +321,9 @@ def generate_video(self, task_id: str) -> dict:
|
|||||||
finally:
|
finally:
|
||||||
session.close()
|
session.close()
|
||||||
|
|
||||||
|
# 标记任务为 running
|
||||||
|
_update_task_status(task_id, "mark_processing")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
editing_mode = EditingMode(mode)
|
editing_mode = EditingMode(mode)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
@@ -315,7 +374,7 @@ def generate_video(self, task_id: str) -> dict:
|
|||||||
file_url = f"{GENERATED_FILES_URL_PREFIX}/{task_id}/{output_name}"
|
file_url = f"{GENERATED_FILES_URL_PREFIX}/{task_id}/{output_name}"
|
||||||
|
|
||||||
# 创建 GeneratedVideo 记录 + 查重
|
# 创建 GeneratedVideo 记录 + 查重
|
||||||
_create_video_record_and_dedup(
|
video_count = _create_video_record_and_dedup(
|
||||||
task_id=task_id,
|
task_id=task_id,
|
||||||
project_id=project_id,
|
project_id=project_id,
|
||||||
batch_id=batch_id,
|
batch_id=batch_id,
|
||||||
@@ -326,6 +385,11 @@ def generate_video(self, task_id: str) -> dict:
|
|||||||
mode=editing_mode.value,
|
mode=editing_mode.value,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# 标记任务为 completed
|
||||||
|
_update_task_status(task_id, "mark_completed", result_count=video_count or 1)
|
||||||
|
|
||||||
|
logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size)
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"status": "completed",
|
"status": "completed",
|
||||||
"task_id": task_id,
|
"task_id": task_id,
|
||||||
@@ -337,7 +401,9 @@ def generate_video(self, task_id: str) -> dict:
|
|||||||
"mode": editing_mode.value,
|
"mode": editing_mode.value,
|
||||||
}
|
}
|
||||||
except Exception as error:
|
except Exception as error:
|
||||||
logger.error(f"Video generation failed: {error}")
|
logger.error(f"Video generation failed: {error}", exc_info=True)
|
||||||
|
# 标记任务为 failed
|
||||||
|
_update_task_status(task_id, "mark_failed", error_message=str(error))
|
||||||
return {
|
return {
|
||||||
"status": "failed",
|
"status": "failed",
|
||||||
"task_id": task_id,
|
"task_id": task_id,
|
||||||
@@ -355,12 +421,15 @@ def _create_video_record_and_dedup(
|
|||||||
duration: float,
|
duration: float,
|
||||||
video_path: str,
|
video_path: str,
|
||||||
mode: str,
|
mode: str,
|
||||||
) -> None:
|
) -> int:
|
||||||
"""创建 GeneratedVideo 记录,计算指纹并执行查重(历史 + 批次)。"""
|
"""创建 GeneratedVideo 记录,计算指纹并执行查重(历史 + 批次)。
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
创建的视频记录数量(1 表示成功,0 表示失败)
|
||||||
|
"""
|
||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
|
|
||||||
from video_processing.dedup import VideoDeduplicator
|
from video_processing.dedup import VideoDeduplicator
|
||||||
from worker_app.db import SessionLocal
|
|
||||||
|
|
||||||
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
|
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
|
||||||
SQLAlchemyGeneratedVideoRepository,
|
SQLAlchemyGeneratedVideoRepository,
|
||||||
@@ -395,7 +464,7 @@ def _create_video_record_and_dedup(
|
|||||||
except Exception as fp_err:
|
except Exception as fp_err:
|
||||||
logger.warning(f"Fingerprint computation failed for {video_id}: {fp_err}")
|
logger.warning(f"Fingerprint computation failed for {video_id}: {fp_err}")
|
||||||
session.commit()
|
session.commit()
|
||||||
return
|
return 1
|
||||||
|
|
||||||
generated_video.video_fingerprint = fingerprint.to_dict()
|
generated_video.video_fingerprint = fingerprint.to_dict()
|
||||||
|
|
||||||
@@ -420,8 +489,10 @@ def _create_video_record_and_dedup(
|
|||||||
video_repo.update(generated_video)
|
video_repo.update(generated_video)
|
||||||
session.commit()
|
session.commit()
|
||||||
logger.info(f"GeneratedVideo record created: {video_id} (task={task_id}, dup={generated_video.is_duplicate})")
|
logger.info(f"GeneratedVideo record created: {video_id} (task={task_id}, dup={generated_video.is_duplicate})")
|
||||||
|
return 1
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Failed to create video record / dedup for task {task_id}: {e}")
|
logger.error(f"Failed to create video record / dedup for task {task_id}: {e}")
|
||||||
session.rollback()
|
session.rollback()
|
||||||
|
return 0
|
||||||
finally:
|
finally:
|
||||||
session.close()
|
session.close()
|
||||||
|
|||||||
@@ -1,3 +1,11 @@
|
|||||||
|
"""GenerationTask 领域模型 — 视频生成任务.
|
||||||
|
|
||||||
|
状态机:
|
||||||
|
pending → running → completed
|
||||||
|
↘ failed → pending (重试)
|
||||||
|
↘ cancelled
|
||||||
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
@@ -17,11 +25,43 @@ from uuid import uuid4
|
|||||||
|
|
||||||
|
|
||||||
class GenerationTaskStatus(StrEnum):
|
class GenerationTaskStatus(StrEnum):
|
||||||
|
"""生成任务状态枚举。"""
|
||||||
|
|
||||||
PENDING = "pending"
|
PENDING = "pending"
|
||||||
|
"""待处理(任务已创建,等待执行)"""
|
||||||
|
|
||||||
RUNNING = "running"
|
RUNNING = "running"
|
||||||
|
"""运行中(正在生成视频)"""
|
||||||
|
|
||||||
COMPLETED = "completed"
|
COMPLETED = "completed"
|
||||||
|
"""已完成(视频生成成功)"""
|
||||||
|
|
||||||
FAILED = "failed"
|
FAILED = "failed"
|
||||||
|
"""失败(生成失败)"""
|
||||||
|
|
||||||
CANCELLED = "cancelled"
|
CANCELLED = "cancelled"
|
||||||
|
"""已取消(用户取消或系统取消)"""
|
||||||
|
|
||||||
|
|
||||||
|
# 终态集合
|
||||||
|
TERMINAL_STATUSES = frozenset(
|
||||||
|
{GenerationTaskStatus.COMPLETED, GenerationTaskStatus.FAILED, GenerationTaskStatus.CANCELLED}
|
||||||
|
)
|
||||||
|
|
||||||
|
# 合法状态转换
|
||||||
|
_VALID_TRANSITIONS: dict[GenerationTaskStatus, set[GenerationTaskStatus]] = {
|
||||||
|
GenerationTaskStatus.PENDING: {
|
||||||
|
GenerationTaskStatus.RUNNING,
|
||||||
|
GenerationTaskStatus.FAILED,
|
||||||
|
GenerationTaskStatus.CANCELLED,
|
||||||
|
},
|
||||||
|
GenerationTaskStatus.RUNNING: {
|
||||||
|
GenerationTaskStatus.COMPLETED,
|
||||||
|
GenerationTaskStatus.FAILED,
|
||||||
|
GenerationTaskStatus.CANCELLED,
|
||||||
|
},
|
||||||
|
GenerationTaskStatus.FAILED: {GenerationTaskStatus.PENDING}, # 重试回到 pending
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@dataclass(slots=True)
|
@dataclass(slots=True)
|
||||||
@@ -83,3 +123,123 @@ class GenerationTask:
|
|||||||
asset_select_mode=asset_select_mode,
|
asset_select_mode=asset_select_mode,
|
||||||
batch_id=batch_id,
|
batch_id=batch_id,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# ── 状态查询 ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_terminal(self) -> bool:
|
||||||
|
"""是否处于终态(completed / failed / cancelled)。"""
|
||||||
|
return self.status in TERMINAL_STATUSES
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_completed(self) -> bool:
|
||||||
|
"""是否已完成。"""
|
||||||
|
return self.status == GenerationTaskStatus.COMPLETED
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_failed(self) -> bool:
|
||||||
|
"""是否失败。"""
|
||||||
|
return self.status == GenerationTaskStatus.FAILED
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_running(self) -> bool:
|
||||||
|
"""是否运行中。"""
|
||||||
|
return self.status == GenerationTaskStatus.RUNNING
|
||||||
|
|
||||||
|
# ── 状态转换 ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def transition_to(self, new_status: GenerationTaskStatus | str) -> None:
|
||||||
|
"""执行状态转换。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
new_status: 目标状态
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: 非法状态转换
|
||||||
|
"""
|
||||||
|
if isinstance(new_status, str):
|
||||||
|
try:
|
||||||
|
new_status = GenerationTaskStatus(new_status)
|
||||||
|
except ValueError:
|
||||||
|
raise ValueError(f"无效状态: {new_status}")
|
||||||
|
|
||||||
|
allowed = _VALID_TRANSITIONS.get(self.status, set())
|
||||||
|
if new_status not in allowed:
|
||||||
|
raise ValueError(
|
||||||
|
f"非法状态转换: {self.status.value} → {new_status.value},"
|
||||||
|
f"允许: {{{', '.join(sorted(s.value for s in allowed))}}}"
|
||||||
|
)
|
||||||
|
|
||||||
|
self.status = new_status
|
||||||
|
|
||||||
|
def mark_processing(self) -> None:
|
||||||
|
"""标记为处理中(pending → running)。
|
||||||
|
|
||||||
|
设置 started_at,清除 error_message。
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: 当前状态不允许转换到 running
|
||||||
|
"""
|
||||||
|
self.transition_to(GenerationTaskStatus.RUNNING)
|
||||||
|
self.started_at = datetime.now(timezone.utc)
|
||||||
|
self.error_message = ""
|
||||||
|
|
||||||
|
def mark_completed(self, result_count: int = 1) -> None:
|
||||||
|
"""标记为已完成(running → completed)。
|
||||||
|
|
||||||
|
设置 completed_at、progress=100.0、result_count,清除 error_message。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
result_count: 生成的视频数量,默认为 1
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: 当前状态不允许转换到 completed
|
||||||
|
"""
|
||||||
|
self.transition_to(GenerationTaskStatus.COMPLETED)
|
||||||
|
self.completed_at = datetime.now(timezone.utc)
|
||||||
|
self.progress = 100.0
|
||||||
|
self.result_count = result_count
|
||||||
|
self.error_message = ""
|
||||||
|
|
||||||
|
def mark_failed(self, error_message: str) -> None:
|
||||||
|
"""标记为失败(pending / running → failed)。
|
||||||
|
|
||||||
|
设置 error_message、completed_at。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
error_message: 错误信息
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: 当前状态不允许转换到 failed
|
||||||
|
"""
|
||||||
|
self.transition_to(GenerationTaskStatus.FAILED)
|
||||||
|
self.error_message = error_message
|
||||||
|
self.completed_at = datetime.now(timezone.utc)
|
||||||
|
|
||||||
|
def mark_cancelled(self) -> None:
|
||||||
|
"""标记为已取消(pending / running → cancelled)。
|
||||||
|
|
||||||
|
设置 completed_at。
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: 当前状态不允许转换到 cancelled
|
||||||
|
"""
|
||||||
|
self.transition_to(GenerationTaskStatus.CANCELLED)
|
||||||
|
self.completed_at = datetime.now(timezone.utc)
|
||||||
|
|
||||||
|
def mark_pending_from_failed(self) -> None:
|
||||||
|
"""从失败状态重置为待处理(用于重试)。
|
||||||
|
|
||||||
|
清除 error_message、started_at、completed_at、progress。
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: 当前状态不是 failed
|
||||||
|
"""
|
||||||
|
if self.status != GenerationTaskStatus.FAILED:
|
||||||
|
raise ValueError(f"只有 failed 状态的任务可以重置为 pending,当前状态: {self.status.value}")
|
||||||
|
self.transition_to(GenerationTaskStatus.PENDING)
|
||||||
|
self.error_message = ""
|
||||||
|
self.started_at = None
|
||||||
|
self.completed_at = None
|
||||||
|
self.progress = 0.0
|
||||||
|
self.result_count = 0
|
||||||
|
|||||||
@@ -0,0 +1,455 @@
|
|||||||
|
"""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
|
||||||
Reference in New Issue
Block a user