14e297a488
- 新增 pam_deploy_graph 包，包含 Agent runtime、ActionRouter、脚本/MCP/fake runner - 支持 hybrid_node_mcp 策略：PAM_HOME 走脚本 action，PAM_NODE 走 MCP - 支持 script_only 离线策略，全部 action 走现有脚本 action - 新增 LLM structured output 骨架和规则 fallback，支持意图识别、参数抽取、计划生成 - 新增 LangGraph StateGraph 工厂和 MCP client adapter - 新增 CLI：preview、analyze、run-global、run-deploy - 增加 fake 完整部署流程、单 IP 失败待回滚确认状态和报告输出 - 增加单元测试覆盖路由、parser、runner、Skill 加载、LLM 输出、MCP adapter 和 LangGraph 图 - 更新 README，记录当前代码骨架、进度、使用方式和下一步计划
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
from dataclasses import asdict
|
|
|
|
from pam_deploy_graph.agent import PamDeployAgent
|
|
from pam_deploy_graph.checkpoint_store import redact_mapping
|
|
from pam_deploy_graph.llm.rule_based import RuleBasedLlmClient
|
|
from pam_deploy_graph.llm.validators import validate_deploy_plan
|
|
from pam_deploy_graph.models import LlmDeployPlan
|
|
|
|
|
|
def test_understand_request_prefers_hybrid_for_mcp():
|
|
result = RuleBasedLlmClient().understand_request("请用 MCP 部署 HET")
|
|
assert result.intent == "deploy"
|
|
assert result.mode_preference == "MCP"
|
|
assert result.strategy_preference == "hybrid_node_mcp"
|
|
|
|
|
|
def test_extract_params_from_key_value_text():
|
|
result = RuleBasedLlmClient().extract_params(
|
|
"HOME_BASE_URL=https://x CLIENT_ID=id CLIENT_SECRET=s AIRPORT_CODE=HET "
|
|
"APP_NAME=PAM MODULE_NAME=Node VERSION_NUMBER=2.0.5 ZIP_FILE_PATH=C:/pkg.zip"
|
|
)
|
|
assert result.extracted_params["AIRPORT_CODE"] == "HET"
|
|
assert result.missing_required_params == []
|
|
assert "CLIENT_SECRET" in result.sensitive_fields_present
|
|
|
|
|
|
def test_analyze_request_returns_structured_objects():
|
|
agent = PamDeployAgent()
|
|
result = agent.analyze_request(
|
|
"不要动环境，预演部署",
|
|
{
|
|
"HOME_BASE_URL": "https://x",
|
|
"CLIENT_ID": "id",
|
|
"CLIENT_SECRET": "s",
|
|
"AIRPORT_CODE": "HET",
|
|
"APP_NAME": "PAM",
|
|
"MODULE_NAME": "Node",
|
|
"VERSION_NUMBER": "2.0.5",
|
|
"ZIP_FILE_PATH": "C:/pkg.zip",
|
|
},
|
|
)
|
|
payload = {key: asdict(value) for key, value in result.items()}
|
|
assert payload["intent"]["intent"] == "preview"
|
|
assert payload["plan"]["execution_strategy"] == "hybrid_node_mcp"
|
|
|
|
|
|
def test_analyze_payload_can_be_redacted():
|
|
agent = PamDeployAgent()
|
|
result = agent.analyze_request(
|
|
"帮我部署",
|
|
{
|
|
"HOME_BASE_URL": "https://x",
|
|
"CLIENT_ID": "id",
|
|
"CLIENT_SECRET": "super-secret",
|
|
"AIRPORT_CODE": "HET",
|
|
"APP_NAME": "PAM",
|
|
"MODULE_NAME": "Node",
|
|
"VERSION_NUMBER": "2.0.5",
|
|
"ZIP_FILE_PATH": "C:/pkg.zip",
|
|
},
|
|
)
|
|
payload = redact_mapping({key: asdict(value) for key, value in result.items()})
|
|
assert payload["params"]["extracted_params"]["CLIENT_SECRET"] == "***"
|
|
|
|
|
|
def test_plan_guardrails_reject_executable_text():
|
|
plan = LlmDeployPlan(summary="run bash ./deploy.sh", planned_actions=["get-token"])
|
|
try:
|
|
validate_deploy_plan(plan)
|
|
except ValueError as exc:
|
|
assert "forbidden" in str(exc)
|
|
else:
|
|
raise AssertionError("expected guardrail failure")
|