JOJO
f77f1d67a8
- F1: RuntimeService.list_runs 公共 Run 发现（工作区/会话/状态筛选）； task_public_payload 序列化收敛至 models.py 核心层单一实现（Web/公共入口同构， helpers.py 旧实现删除，无新旧双轨） - F2: tasks/api.py 7 处、api_v1.py 3 处、chat/approval.py 6 处全部转调 runtime_service；HTTP 轮询透传 window_start 缺口水位；死导入清零 - F3: _resources_for_query 补 principal workspace_id 一致性校验 （跨工作区查询必须重新认证） - F4: config 新增 ASTRION_IGNORE_DOTENV 逃生门（.env 对 ASTRION_DATA_ROOT 的 优先覆盖是刻意设计，保留）；验收测试自包含（假模型 127.0.0.1:9 快失败）+ 隔离生效断言；lifecycle 改 api_request_start 装配证据断言； 新增审批等待链验收（approval 档真实工具循环：等待→公共入口批准→继续）； chain 首段改双客户端场景（B 经 list_runs 发现 A 的 Run 并观察/取消） - execution_plane/base.py 注释修正：命令校验/路径授权仍在旧链路（审核 §4） - 全量 75 测试：失败恰为 4 项存量，与改造无关
71 lines
2.9 KiB
Python
71 lines
2.9 KiB
Python
"""Gateway 第 1/2 步验收：独立任务生命周期 + 极简协议全链路（G4/G13）。
|
||
|
||
验收标准（来自 cache_research/gateway/gateway_current_state.md §7.2）：
|
||
- 进程全程不创建 Flask app、不注册蓝图、不初始化 SocketIO 服务；
|
||
- RuntimeService 真实受理（create_task → task_id）、任务线程真实启动；
|
||
- 资源装配真实执行（get_user_resources 由 RuntimeIdentity 驱动，不重 mock）；
|
||
- 事件流按 idx/offset 协议可读；任务可取消；门闸最终释放；
|
||
- 模型调用允许失败（外部依赖非验收对象）——装配与生命周期必须真实。
|
||
|
||
隔离设计：真实检查体在 test/runtime_standalone_checks.py，由本文件以
|
||
**子进程**方式执行。原因：config/paths.py 的 DATA_DIR / DEPLOY_CONFIG_DIR
|
||
是 import 时固化的模块级常量；unittest discover 全量运行时排在前面的测试
|
||
模块会先 import config 使常量固化，进程内设环境变量已无效，被测代码会落到
|
||
非隔离目录（甚至用户真实运行态目录）执行任务。子进程内 import 顺序完全可控，
|
||
隔离（ASTRION_DATA_ROOT + DEPLOY_CONFIG_DIR 指向临时目录）100% 可靠，
|
||
单跑与全量行为一致。
|
||
"""
|
||
import subprocess
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
_CHECKS_SCRIPT = Path(__file__).resolve().parent / "runtime_standalone_checks.py"
|
||
_PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def _run_check(mode: str) -> str:
|
||
proc = subprocess.run(
|
||
[sys.executable, str(_CHECKS_SCRIPT), mode],
|
||
capture_output=True,
|
||
text=True,
|
||
cwd=str(_PROJECT_ROOT),
|
||
timeout=150,
|
||
)
|
||
output = f"STDOUT:\n{proc.stdout[-4000:]}\nSTDERR:\n{proc.stderr[-4000:]}"
|
||
if proc.returncode != 0:
|
||
raise AssertionError(f"独立验收子进程失败 (mode={mode}, rc={proc.returncode}):\n{output}")
|
||
return output
|
||
|
||
|
||
class StandaloneRuntimeLifecycleTest(unittest.TestCase):
|
||
"""无 Flask app 的独立 Gateway 生命周期验收（子进程执行）。"""
|
||
|
||
def test_create_observe_cancel_without_web_app(self):
|
||
_run_check("lifecycle")
|
||
|
||
|
||
class ProtocolSmokeChainTest(unittest.TestCase):
|
||
"""第 2 步验收：极简协议客户端视角的全链路（工作区→会话→运行→停止→历史→审批）。"""
|
||
|
||
def test_full_chain_run_history_offset_approval(self):
|
||
_run_check("chain")
|
||
|
||
|
||
class ExecutionPlaneFakeBackendTest(unittest.TestCase):
|
||
"""第 4 步验收：Runtime 工具编排层 + 替身执行器可独立测试（无真实副作用）。"""
|
||
|
||
def test_runtime_with_fake_execution_backend(self):
|
||
_run_check("fake_exec")
|
||
|
||
|
||
class ApprovalWaitChainTest(unittest.TestCase):
|
||
"""审核 F4 交互覆盖：执行中审批等待→公共入口回答→执行继续（真实工具编排层）。"""
|
||
|
||
def test_approval_wait_resolve_continue(self):
|
||
_run_check("approval_wait")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|