lambdagent/tests/test_agentpack_e2e.py

"""
Phase I dogfood — AgentPack end-to-end integration tests (no LLM).

Exercises the full HTTP path that a desktop user takes:

  install built-in pack ──► list ──► get detail
        │
        └──► create-agent ──► GET agent (verify config)
                                    │
                                    └──► from_config() ──► agent compiles ✓
        │
        └──► uninstall pack ──► agent STILL exists (no cascade delete)

NOTE on install: POST /agentpacks/install is loopback-only (127.0.0.1).
TestClient sends requests as host "testclient", so the loopback check
returns 404. We test the loopback gate separately (test_install_loopback_blocked)
and use install_from_zip() directly for the pack-setup fixture — consistent
with test_agentpack.py::test_create_agent_from_pack_endpoint. All subsequent
API calls (list / get / create-agent / delete) go through HTTP.

Real LLM execution (review_report.md verification) lives in
scripts/dogfood_agentpack.py and is marked skip here.
"""
from __future__ import annotations

import json
import os
import zipfile

import pytest
import yaml

os.environ.setdefault("AGENTPAAS_DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("AGENTPAAS_TESTING", "1")

# ── Helpers ───────────────────────────────────────────────────────────────────

def _pack_zip_from_disk(tmp_path, pack_id: str) -> str:
    """Zip the on-disk agentexample pack so install_from_zip() can consume it."""
    repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    pack_dir = os.path.join(repo_root, "agentexample", "agentpacks", pack_id)
    if not os.path.isdir(pack_dir):
        pytest.skip(f"built-in pack not found on disk: {pack_dir}")

    zip_path = str(tmp_path / f"{pack_id}.zip")
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
        for dirpath, _, filenames in os.walk(pack_dir):
            for fname in filenames:
                full = os.path.join(dirpath, fname)
                arcname = os.path.relpath(full, pack_dir)
                zf.write(full, arcname)
    return zip_path


def _install_pack(packs_dir: str, zip_path: str):
    """Install a pack directly (bypasses loopback gate — tested separately)."""
    from agentpaas.engine.agentpack_store import install_from_zip
    return install_from_zip(zip_path, packs_dir)


@pytest.fixture()
def http_client(tmp_path, monkeypatch):
    """TestClient + fresh in-memory DB + one tenant/user/key + reviewer pack pre-installed."""
    import secrets
    from fastapi.testclient import TestClient
    from agentpaas.api.app import app
    from agentpaas.api.middleware.auth import hash_key
    from agentpaas.config import settings
    from agentpaas.db.models import Database, gen_id, now_utc
    import agentpaas.db.session as _session_mod

    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    monkeypatch.setattr(settings, "agentpacks_dir", str(packs_dir))

    prev_db = _session_mod._db
    _session_mod._db = Database("sqlite:///:memory:")
    db = _session_mod._db

    tid = gen_id("tn_")
    uid = gen_id("usr_")
    raw_key = f"ap_{secrets.token_hex(16)}"
    now = now_utc()
    db.execute(
        "INSERT INTO tenants (id, name, plan, status, created_at) "
        "VALUES (?, 'test', 'free', 'active', ?)", (tid, now),
    )
    db.execute(
        "INSERT INTO users (id, tenant_id, email, role, created_at) "
        "VALUES (?, ?, '', 'admin', ?)", (uid, tid, now),
    )
    db.execute(
        "INSERT INTO api_keys "
        "(id, tenant_id, user_id, key_hash, key_prefix, name, scopes, "
        " rate_limit, status, created_at) "
        "VALUES (?, ?, ?, ?, ?, 'test', ?, 600, 'active', ?)",
        (gen_id("key_"), tid, uid, hash_key(raw_key), raw_key[:8],
         json.dumps(["agents:*", "keys:*"]), now),
    )
    db.commit()

    # Pre-install the built-in reviewer pack directly (bypass loopback gate)
    zip_path = _pack_zip_from_disk(tmp_path, REVIEWER_ID)
    pack = _install_pack(str(packs_dir), zip_path)

    with TestClient(app) as client:
        yield client, raw_key, str(packs_dir), pack

    _session_mod._db = prev_db


# ── Auth helper ───────────────────────────────────────────────────────────────

def _auth(key: str) -> dict:
    return {"Authorization": f"Bearer {key}"}


# ── Tests ─────────────────────────────────────────────────────────────────────

REVIEWER_ID = "research.top-journal-reviewer"


def test_install_loopback_blocked(tmp_path, monkeypatch):
    """POST /agentpacks/install from non-loopback → 404 (loopback gate).

    TestClient host = 'testclient', not 127.0.0.1 — this is the gate we want.
    """
    import secrets
    from fastapi.testclient import TestClient
    from agentpaas.api.app import app
    from agentpaas.config import settings

    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    monkeypatch.setattr(settings, "agentpacks_dir", str(packs_dir))

    zip_path = _pack_zip_from_disk(tmp_path, REVIEWER_ID)
    with TestClient(app) as client:
        r = client.post(
            "/api/v1/agentpacks/install",
            json={"zip_path": zip_path},
        )
    assert r.status_code == 404, r.text[:200]


def test_install_builtin_pack_via_api(tmp_path, http_client):
    """Pre-installed reviewer pack is visible via GET /agentpacks after direct install."""
    client, key, packs_dir, pack = http_client
    # pack was installed in fixture — verify it via the HTTP list
    r = client.get("/api/v1/agentpacks", headers=_auth(key))
    assert r.status_code == 200, r.text[:300]
    body = r.json()
    assert body["count"] >= 1
    ids = [p["id"] for p in body["agentpacks"]]
    assert REVIEWER_ID in ids
    # spot-check detail fields
    detail = next(p for p in body["agentpacks"] if p["id"] == REVIEWER_ID)
    assert detail["version"] == "0.1.0"
    assert detail["domain"] == "research"
    assert detail["permissions"]["shell"] is False
    assert detail["permissions"]["network"] is False


def test_installed_pack_appears_in_list(tmp_path, http_client):
    """GET /agentpacks → pre-installed pack shows up."""
    client, key, packs_dir, pack = http_client
    r = client.get("/api/v1/agentpacks", headers=_auth(key))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["count"] >= 1
    ids = [p["id"] for p in body["agentpacks"]]
    assert REVIEWER_ID in ids


def test_get_installed_pack_detail(tmp_path, http_client):
    """GET /agentpacks/{id} → correct detail including permission_summary."""
    client, key, packs_dir, pack = http_client
    r = client.get(f"/api/v1/agentpacks/{REVIEWER_ID}", headers=_auth(key))
    assert r.status_code == 200, r.text
    p = r.json()
    assert p["id"] == REVIEWER_ID
    assert len(p["permission_summary"]) > 0
    assert "professor" in p["audience"] or "phd_student" in p["audience"]


def test_create_agent_from_installed_pack(tmp_path, http_client):
    """POST create-agent → 201, agent_id returned."""
    client, key, packs_dir, pack = http_client
    r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "我的审稿助手", "description": "dogfood test"},
        headers=_auth(key),
    )
    assert r.status_code == 201, r.text[:300]
    body = r.json()
    assert body["ok"] is True
    assert body["agent_id"].startswith("ag_")
    assert body["pack"]["id"] == REVIEWER_ID


def test_created_agent_has_correct_config(tmp_path, http_client):
    """GET /agents/{id} → agent_dir=pack.path, _config_dir correct."""
    client, key, packs_dir, pack = http_client
    create_r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "审稿助手"},
        headers=_auth(key),
    )
    agent_id = create_r.json()["agent_id"]

    r = client.get(f"/api/v1/agents/{agent_id}", headers=_auth(key))
    assert r.status_code == 200, r.text
    agent = r.json()

    assert agent["agent_dir"] == pack.path
    assert agent["agent_template"] == REVIEWER_ID

    config = agent["config"]
    # _config_dir must equal pack.path so sub-agent refs resolve correctly
    assert config.get("_config_dir") == pack.path


def test_agent_config_compiles_with_from_config(tmp_path, http_client):
    """Critical seam: config stored in DB → from_config() → agent term.

    Mirrors what _execute_agent does (dump to tmp YAML, compile).
    If this passes, the agent can actually run (LLM aside).
    """
    import tempfile
    from lambdagent.fromconfig import from_config as lambdagent_from_config

    client, key, packs_dir, pack = http_client
    create_r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "审稿助手"},
        headers=_auth(key),
    )
    agent_id = create_r.json()["agent_id"]
    config = client.get(f"/api/v1/agents/{agent_id}", headers=_auth(key)).json()["config"]

    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".yml", delete=False, encoding="utf-8"
    ) as f:
        yaml.dump(config, f, allow_unicode=True)
        tmp_yml = f.name
    try:
        term = lambdagent_from_config(tmp_yml)
        assert term is not None
    finally:
        os.unlink(tmp_yml)


def test_agent_appears_in_agents_list(tmp_path, http_client):
    """GET /agents after create-agent → new agent visible."""
    client, key, packs_dir, pack = http_client
    create_r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "审稿助手"},
        headers=_auth(key),
    )
    agent_id = create_r.json()["agent_id"]

    r = client.get("/api/v1/agents", headers=_auth(key))
    assert r.status_code == 200
    agent_ids = [a["id"] for a in r.json()["agents"]]
    assert agent_id in agent_ids


def test_uninstall_pack_does_not_delete_agents(tmp_path, http_client):
    """DELETE /agentpacks/{id} removes the pack; derived agent persists.

    Agents are data, not derived assets. Deleting the template pack MUST NOT
    cascade-delete agents (they may have run history and workspace artifacts).
    """
    client, key, packs_dir, pack = http_client
    create_r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "审稿助手"},
        headers=_auth(key),
    )
    agent_id = create_r.json()["agent_id"]

    del_r = client.delete(f"/api/v1/agentpacks/{REVIEWER_ID}", headers=_auth(key))
    assert del_r.status_code == 200
    assert del_r.json()["ok"] is True

    # Pack is gone
    list_r = client.get("/api/v1/agentpacks", headers=_auth(key))
    assert list_r.json()["count"] == 0

    # Agent STILL exists
    agent_r = client.get(f"/api/v1/agents/{agent_id}", headers=_auth(key))
    assert agent_r.status_code == 200
    assert agent_r.json()["name"] == "审稿助手"


def test_get_unknown_pack_returns_404(tmp_path, http_client):
    """GET /agentpacks/does.not.exist → 404."""
    client, key, packs_dir, pack = http_client
    r = client.get("/api/v1/agentpacks/does.not.exist", headers=_auth(key))
    assert r.status_code == 404


def test_create_agent_from_uninstalled_pack_returns_404(tmp_path, http_client):
    """POST create-agent for a pack that was never installed → 404."""
    client, key, packs_dir, pack = http_client
    r = client.post(
        "/api/v1/agentpacks/does.not.exist/create-agent",
        json={"name": "x"},
        headers=_auth(key),
    )
    assert r.status_code == 404


@pytest.mark.skip(reason="requires real LLM (ANTHROPIC_API_KEY). Run: python scripts/dogfood_agentpack.py")
def test_run_agent_produces_review_report(tmp_path, http_client):
    """Smoke test: run reviewer pack agent with a short abstract.

    Skipped in CI — needs a live LLM. Use scripts/dogfood_agentpack.py for the
    full end-to-end dogfood (workspace/review_report.md verification).
    """
    client, key, packs_dir, pack = http_client
    create_r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "审稿助手"},
        headers=_auth(key),
    )
    agent_id = create_r.json()["agent_id"]

    abstract = (
        "Title: Quantum Speedup for Matrix Multiplication.\n"
        "Abstract: We present a quantum algorithm achieving O(n^{1.5}) "
        "matrix multiplication, improving on the classical Strassen bound O(n^{2.37}).\n"
        "Verify this claim and produce a review report."
    )
    run_r = client.post(
        f"/api/v1/agents/{agent_id}/run",
        json={"input": abstract},
        headers=_auth(key),
        timeout=300,
    )
    assert run_r.status_code == 200, run_r.text[:500]
    body = run_r.json()
    workspace = body.get("workspace_path", "")
    assert workspace, "workspace_path missing from response"
    assert os.path.isfile(os.path.join(workspace, "review_report.md")), \
        f"review_report.md not found in {workspace}"
    assert os.path.isfile(os.path.join(workspace, "review_result.json")), \
        f"review_result.json not found in {workspace}"