164 lines
5.0 KiB
Python
164 lines
5.0 KiB
Python
import builtins
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from pam_deploy_graph.agent import PamDeployAgent
|
|
from pam_deploy_graph.fake_runner import FakeActionRunner
|
|
from pam_deploy_graph.interactive import InteractiveCliSession, _build_prompt_input
|
|
|
|
|
|
PARAMS = {
|
|
"HOME_BASE_URL": "https://pam.home.example.com",
|
|
"CLIENT_ID": "client",
|
|
"CLIENT_SECRET": "secret",
|
|
"AIRPORT_CODE": "HET",
|
|
"APP_NAME": "PAM",
|
|
"MODULE_NAME": "Node",
|
|
"VERSION_NUMBER": "2.0.5",
|
|
"ZIP_FILE_PATH": "C:/pkg.zip",
|
|
}
|
|
|
|
|
|
def run_session(session: InteractiveCliSession, inputs: list[str]) -> list[str]:
|
|
output: list[str] = []
|
|
iterator = iter(inputs)
|
|
session.input = lambda _prompt: next(iterator)
|
|
session.output = output.append
|
|
session.run()
|
|
return output
|
|
|
|
|
|
def test_chat_analyzes_natural_language_and_updates_context(tmp_path: Path):
|
|
session = InteractiveCliSession(
|
|
agent=PamDeployAgent(),
|
|
params=PARAMS,
|
|
strategy="fake",
|
|
checkpoint_path=str(tmp_path / "checkpoint.json"),
|
|
)
|
|
|
|
output = run_session(session, ["analyze please use MCP deploy 192.168.1.10", "exit"])
|
|
|
|
assert session.strategy == "hybrid_node_mcp"
|
|
assert session.target_ips == ["192.168.1.10"]
|
|
assert any("执行请输 run" in item for item in output)
|
|
|
|
|
|
def test_chat_run_executes_fake_deploy_and_writes_checkpoint(tmp_path: Path):
|
|
checkpoint = tmp_path / "checkpoint.json"
|
|
session = InteractiveCliSession(
|
|
agent=PamDeployAgent(fake_runner=FakeActionRunner()),
|
|
params=PARAMS,
|
|
strategy="fake",
|
|
checkpoint_path=str(checkpoint),
|
|
)
|
|
|
|
run_session(session, ["run", "yes", "yes", "yes", "exit"])
|
|
|
|
assert checkpoint.exists()
|
|
assert session.state is not None
|
|
assert session.state.pending_confirmation == ""
|
|
assert all(item["status"] == "SUCCESS" for item in session.state.ip_states.values())
|
|
|
|
|
|
def test_chat_approve_then_resume_continues_after_failed_ip(tmp_path: Path):
|
|
fake = FakeActionRunner(
|
|
{
|
|
"verify-ip:192.168.1.10": {
|
|
"ACTION": "verify-ip",
|
|
"IP": "192.168.1.10",
|
|
"SUCCESS": "false",
|
|
"MESSAGE": "health check failed",
|
|
}
|
|
}
|
|
)
|
|
session = InteractiveCliSession(
|
|
agent=PamDeployAgent(fake_runner=fake),
|
|
params=PARAMS,
|
|
strategy="fake",
|
|
checkpoint_path=str(tmp_path / "checkpoint.json"),
|
|
)
|
|
|
|
run_session(session, ["run", "yes", "yes", "yes", "approve", "resume", "exit"])
|
|
|
|
assert session.state is not None
|
|
assert session.state.pending_confirmation == ""
|
|
assert session.state.ip_states["192.168.1.10"]["rollback_status"] == "ROLLBACK_DONE"
|
|
assert session.state.ip_states["192.168.1.11"]["status"] == "SUCCESS"
|
|
|
|
|
|
def test_chat_params_events_and_checkpoint_commands(tmp_path: Path):
|
|
checkpoint = tmp_path / "checkpoint.json"
|
|
session = InteractiveCliSession(
|
|
agent=PamDeployAgent(fake_runner=FakeActionRunner(), action_analysis_enabled=True),
|
|
params=PARAMS,
|
|
strategy="fake",
|
|
checkpoint_path=str(checkpoint),
|
|
)
|
|
|
|
output = run_session(
|
|
session,
|
|
[
|
|
"params",
|
|
"llm action-analysis on",
|
|
"run",
|
|
"yes",
|
|
"yes",
|
|
"yes",
|
|
"events 2",
|
|
"list checkpoints",
|
|
"load checkpoint " + str(checkpoint),
|
|
"exit",
|
|
],
|
|
)
|
|
|
|
assert session.state is not None
|
|
assert any("CLIENT_SECRET: ***" in item for item in output)
|
|
assert any("ACTION_ANALYSIS" in item for item in output)
|
|
assert any("checkpoint 列表" in item for item in output)
|
|
|
|
|
|
def test_chat_can_hot_load_mcp_config(tmp_path: Path):
|
|
mcp_config = tmp_path / "mcp.json"
|
|
mcp_config.write_text('{"transport": "stdio", "command": "python"}', encoding="utf-8")
|
|
session = InteractiveCliSession(
|
|
agent=PamDeployAgent(),
|
|
params=PARAMS,
|
|
strategy="hybrid_node_mcp",
|
|
checkpoint_path=str(tmp_path / "checkpoint.json"),
|
|
)
|
|
|
|
output = run_session(session, ["mcp config " + mcp_config.as_posix(), "exit"])
|
|
|
|
assert session.agent.mcp_runner is not None
|
|
assert session.agent.router.mcp_runner is not None
|
|
assert any("MCP 配置已加载" in item for item in output)
|
|
|
|
|
|
def test_prompt_history_creates_runtime_dir(tmp_path: Path, monkeypatch):
|
|
pytest.importorskip("prompt_toolkit")
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
prompt = _build_prompt_input(builtins.input)
|
|
|
|
assert callable(prompt)
|
|
assert (tmp_path / "runtime").is_dir()
|
|
|
|
|
|
def test_packaged_chat_uses_plain_input_by_default(monkeypatch):
|
|
monkeypatch.setattr("sys.frozen", True, raising=False)
|
|
monkeypatch.delenv("PAM_CHAT_PROMPT_TOOLKIT", raising=False)
|
|
monkeypatch.delenv("PAM_CHAT_SIMPLE_INPUT", raising=False)
|
|
|
|
prompt = _build_prompt_input(builtins.input)
|
|
|
|
assert prompt is builtins.input
|
|
|
|
|
|
def test_simple_input_env_disables_prompt_toolkit(monkeypatch):
|
|
monkeypatch.setenv("PAM_CHAT_SIMPLE_INPUT", "1")
|
|
|
|
prompt = _build_prompt_input(builtins.input)
|
|
|
|
assert prompt is builtins.input
|