e572a26e6f
llm 提示词和规则：新增 progress_complete 判断字段。 deploy.sh / deploy.ps1：poll-* action 入口改为单次查询。 interactive.py：chat 会播报进度更新。 config.txt.example / README / packaging 文档 / Skill 文档：同步进度查询参数和新 workflow 语义。 测试补充了进度重复查询、超时暂停、chat 进度播报。
80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
"""供本地测试和预演使用的 fake action runner。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from .models import ActionResult
|
||
|
||
|
||
class FakeActionRunner:
|
||
"""返回确定性 action 结果，避免测试触碰真实 PAM 环境。"""
|
||
|
||
def __init__(self, fixtures: dict[str, dict[str, Any]] | None = None) -> None:
|
||
"""保存可覆盖默认行为的测试 fixture，并记录调用历史。"""
|
||
self.fixtures = fixtures or {}
|
||
self.calls: list[tuple[str, dict[str, Any]]] = []
|
||
|
||
def run(self, action: str, *, params: dict[str, Any], **kwargs: Any) -> ActionResult:
|
||
"""执行 fake action，优先使用 fixture，否则使用内置默认结果。"""
|
||
self.calls.append((action, kwargs))
|
||
values = self._fixture_for(action, kwargs)
|
||
if not values:
|
||
values = self._default_values(action, kwargs)
|
||
ok = not values.pop("_fail", False)
|
||
return ActionResult(
|
||
action=action,
|
||
backend="fake",
|
||
tool_name=f"fake:{action}",
|
||
ok=ok,
|
||
values=values,
|
||
exit_code=0 if ok else 1,
|
||
raw_output=str(values),
|
||
error_summary="" if ok else str(values.get("MESSAGE", "fake action 执行失败")),
|
||
)
|
||
|
||
def _default_values(self, action: str, kwargs: dict[str, Any]) -> dict[str, Any]:
|
||
"""为常见部署 action 构造稳定的默认返回值。"""
|
||
if action == "get-token":
|
||
return {"ACTION": action, "TOKEN": "***"}
|
||
if action == "upload-package":
|
||
return {"ACTION": action, "HASH_CODE": "fake-hash"}
|
||
if action == "get-node-url":
|
||
return {"ACTION": action, "NODE_URL": "https://fake-node.local"}
|
||
if action == "get-online-ips":
|
||
return {"ACTION": action, "COUNT": "2", "IP": ["192.168.1.10", "192.168.1.11"]}
|
||
if action == "poll-download-progress":
|
||
return {
|
||
"ACTION": action,
|
||
"STEP": "DONE",
|
||
"RATE_OF_PROGRESS": "100",
|
||
"MSG": "success",
|
||
"MESSAGE": "success",
|
||
}
|
||
if action == "upgrade-ip":
|
||
return {"ACTION": action, "IP": kwargs.get("ip", ""), "RESULT": "TASK_CREATED"}
|
||
if action == "poll-upgrade-progress":
|
||
return {
|
||
"ACTION": action,
|
||
"IP": kwargs.get("ip", ""),
|
||
"STEP": "DONE",
|
||
"RATE_OF_PROGRESS": "100",
|
||
"MSG": "success",
|
||
"MESSAGE": "success",
|
||
}
|
||
if action == "start-ip":
|
||
return {"ACTION": action, "IP": kwargs.get("ip", ""), "RESULT": "OK"}
|
||
if action == "verify-ip":
|
||
return {"ACTION": action, "IP": kwargs.get("ip", ""), "SUCCESS": "true", "MESSAGE": "ok"}
|
||
if action == "download-log":
|
||
return {"ACTION": action, "IP": kwargs.get("ip", ""), "LOG_FILE": "logs/fake.zip"}
|
||
return {"ACTION": action, "RESULT": "OK"}
|
||
|
||
def _fixture_for(self, action: str, kwargs: dict[str, Any]) -> dict[str, Any]:
|
||
"""按 action 或 action:ip 查找测试 fixture。"""
|
||
ip = kwargs.get("ip")
|
||
ip_key = f"{action}:{ip}" if ip else ""
|
||
if ip_key and ip_key in self.fixtures:
|
||
return self.fixtures[ip_key].copy()
|
||
return self.fixtures.get(action, {}).copy()
|