zhou
5c8ec281ff
主要变更： 1. 替换旧retries参数为RetryPolicy配置 2. 重构条件函数，支持上下文参数与动态依赖判断 3. 更新上下文注入逻辑，支持软依赖与更清晰的注入描述 4. 新增sglang CLI命令与相关配置 5. 格式化代码统一列表与参数写法 6. 更新文档与测试用例适配新API
64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
"""TaskSpec / TaskResult 数据结构测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from pyflowx.task import RetryPolicy, TaskResult, TaskSpec, TaskStatus
|
|
|
|
|
|
def _fn() -> None:
|
|
return None
|
|
|
|
|
|
def test_spec_empty_name_rejected() -> None:
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
TaskSpec("", _fn)
|
|
|
|
|
|
def test_spec_negative_max_attempts_rejected() -> None:
|
|
with pytest.raises(ValueError, match="max_attempts"):
|
|
TaskSpec("a", _fn, retry=RetryPolicy(max_attempts=0))
|
|
|
|
|
|
def test_spec_zero_timeout_rejected() -> None:
|
|
with pytest.raises(ValueError, match="timeout"):
|
|
TaskSpec("a", _fn, timeout=0)
|
|
|
|
|
|
def test_spec_self_dependency_rejected() -> None:
|
|
with pytest.raises(ValueError, match="depend on itself"):
|
|
TaskSpec("a", _fn, depends_on=("a",))
|
|
|
|
|
|
def test_task_result_duration_none_when_not_started() -> None:
|
|
spec: TaskSpec[None] = TaskSpec("a", _fn)
|
|
result: TaskResult[None] = TaskResult(spec=spec)
|
|
assert result.duration is None
|
|
|
|
|
|
def test_task_result_duration_when_partial() -> None:
|
|
spec: TaskSpec[None] = TaskSpec("a", _fn)
|
|
result: TaskResult[None] = TaskResult(spec=spec, started_at=datetime.now())
|
|
# started_at 已设但 finished_at 未设 -> None
|
|
assert result.duration is None
|
|
|
|
|
|
def test_task_result_duration_computed() -> None:
|
|
spec: TaskSpec[None] = TaskSpec("a", _fn)
|
|
start = datetime(2024, 1, 1, 0, 0, 0)
|
|
end = datetime(2024, 1, 1, 0, 0, 5)
|
|
result: TaskResult[None] = TaskResult(spec=spec, started_at=start, finished_at=end)
|
|
assert result.duration == 5.0
|
|
|
|
|
|
def test_task_result_default_status() -> None:
|
|
spec: TaskSpec[None] = TaskSpec("a", _fn)
|
|
result: TaskResult[None] = TaskResult(spec=spec)
|
|
assert result.status == TaskStatus.PENDING
|
|
assert result.value is None
|
|
assert result.error is None
|
|
assert result.attempts == 0
|