agent_deply/tests/test_interactive_cli.py

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_run_prints_action_progress(tmp_path: Path):
    checkpoint = tmp_path / "checkpoint.json"
    session = InteractiveCliSession(
        agent=PamDeployAgent(fake_runner=FakeActionRunner()),
        params=PARAMS,
        strategy="fake",
        checkpoint_path=str(checkpoint),
    )

    output = run_session(session, ["run", "yes", "yes", "yes", "exit"])

    assert any("开始执行 action: get-token" in item for item in output)
    assert any("完成 action: verify-ip" in item for item in output)


def test_chat_greeting_does_not_trigger_structured_analysis(tmp_path: Path):
    session = InteractiveCliSession(
        agent=PamDeployAgent(),
        params=PARAMS,
        strategy="fake",
        checkpoint_path=str(tmp_path / "checkpoint.json"),
    )

    output = run_session(session, ["你好", "exit"])

    assert session.last_analysis is None
    assert any("可以输入 help 查看命令" in item for item in output)
    assert not any("已生成结构化理解" in item for item in output)


def test_chat_preflight_blocks_missing_zip_path_before_confirm(tmp_path: Path):
    missing_package = tmp_path / "missing.zip"
    session = InteractiveCliSession(
        agent=PamDeployAgent(),
        params={**PARAMS, "ZIP_FILE_PATH": str(missing_package)},
        strategy="script_only",
        checkpoint_path=str(tmp_path / "checkpoint.json"),
    )

    output = run_session(session, ["run", "exit"])

    assert session.state is None
    assert any("执行前检查未通过" in item for item in output)
    assert any("ZIP_FILE_PATH 不存在" in item for item in output)


def test_chat_action_failure_does_not_report_langgraph_unavailable(tmp_path: Path):
    fake = FakeActionRunner({"upload-package": {"ACTION": "upload-package"}})
    session = InteractiveCliSession(
        agent=PamDeployAgent(fake_runner=fake),
        params=PARAMS,
        strategy="fake",
        checkpoint_path=str(tmp_path / "checkpoint.json"),
    )

    output = run_session(session, ["run", "yes", "yes", "yes", "exit"])

    assert session.state is not None
    assert session.state.last_failed_step == "upload-package"
    assert any("执行已停止" in item for item in output)
    assert not any("LangGraph 确认运行器不可用" in item for item in output)


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()