agent_deply/tests/test_llm_structured.py

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.openai_compatible import OpenAICompatibleLlmClient
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")


def test_openai_compatible_client_uses_base_url_api_key_and_prompt():
    calls = []

    def transport(url, headers, payload, timeout_sec):
        calls.append((url, headers, payload, timeout_sec))
        return {
            "choices": [
                {
                    "message": {
                        "content": (
                            '{"intent":"deploy","mode_preference":"MCP",'
                            '"strategy_preference":"hybrid_node_mcp","confidence":0.9,'
                            '"reasons":["ok"]}'
                        )
                    }
                }
            ]
        }

    client = OpenAICompatibleLlmClient(
        base_url="https://llm.example/v1",
        api_key="secret-key",
        model="model-a",
        transport=transport,
    )

    result = client.understand_request("请用 MCP 部署")

    assert result.intent == "deploy"
    assert calls[0][0] == "https://llm.example/v1/chat/completions"
    assert calls[0][1]["Authorization"] == "Bearer secret-key"
    assert calls[0][2]["model"] == "model-a"
    assert "只输出一个 JSON 对象" in calls[0][2]["messages"][0]["content"]


def test_openai_compatible_client_does_not_send_base_secret():
    calls = []

    def transport(url, headers, payload, timeout_sec):
        calls.append(payload)
        return {
            "choices": [
                {
                    "message": {
                        "content": (
                            '{"extracted_params":{"AIRPORT_CODE":"HET"},'
                            '"extracted_control":{},'
                            '"missing_required_params":[],'
                            '"ambiguous_fields":[]}'
                        )
                    }
                }
            ]
        }

    client = OpenAICompatibleLlmClient(
        base_url="https://llm.example/v1",
        api_key="secret-key",
        model="model-a",
        transport=transport,
    )

    result = client.extract_params("机场 HET", {"CLIENT_SECRET": "real-secret", "CLIENT_ID": "id"})

    serialized_prompt = str(calls[0])
    assert "real-secret" not in serialized_prompt
    assert result.extracted_params["CLIENT_SECRET"] == "real-secret"