zhou
40f641611b
1. 将CliRunner默认执行策略从sequential改为dependency 2. 新增RunReport的任务状态查询和时长统计方法 3. 实现task装饰器并补充executor参数文档 4. 新增进程池执行器支持CPU密集型任务 5. 新增Graph.chain链式构建和add_subgraph子图合并功能 6. 新增流式任务传递、进程池执行、命名空间等多类测试用例 7. 补充tests目录路径导入配置
27 lines
536 B
Python
27 lines
536 B
Python
"""进程池测试辅助：模块级函数（须可 pickle）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
|
|
def cpu_heavy(n: int) -> int:
|
|
"""CPU 密集型计算（求平方和）。"""
|
|
return sum(i * i for i in range(n))
|
|
|
|
|
|
def add(a: int, b: int) -> int:
|
|
"""简单加法。"""
|
|
return a + b
|
|
|
|
|
|
def sub(a: int, b: int) -> int:
|
|
"""简单减法。"""
|
|
return a - b
|
|
|
|
|
|
def slow_sleep(seconds: float) -> int:
|
|
"""睡眠指定秒数，用于测试超时。"""
|
|
time.sleep(seconds)
|
|
return int(seconds)
|