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))
    # 自动目录开辟隔离到 tmp（否则 create-agent 测试会写真实 Workspace/）
    monkeypatch.setattr(settings, "workspace_base", str(tmp_path / "Workspace"))

    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}"


# ── Pack update (zip upload → agents on latest config) ───────────────────────
# AUDIT 后新增功能: POST /agentpacks/{id}/update-upload。HTTP 层的 loopback
# 门单测 + 核心传播逻辑直测（与 install 的测试策略一致）。

def _pack_zip_with_prompt(tmp_path, pack_id: str, new_prompt: str,
                          new_version: str = "") -> str:
    """复制内置包源码，改 entrypoint 的 systemPrompt（和可选版本号），打成 zip。"""
    import shutil
    repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    src = os.path.join(repo_root, "agentexample", "agentpacks", pack_id)
    if not os.path.isdir(src):
        pytest.skip(f"built-in pack not found on disk: {src}")
    work = str(tmp_path / "pack_v2_src")
    shutil.copytree(src, work)

    with open(os.path.join(work, "manifest.yml"), encoding="utf-8") as f:
        manifest = yaml.safe_load(f)
    if new_version:
        manifest["version"] = new_version
        with open(os.path.join(work, "manifest.yml"), "w", encoding="utf-8") as f:
            yaml.safe_dump(manifest, f, allow_unicode=True)

    entry = os.path.join(work, manifest["entrypoint"])
    with open(entry, encoding="utf-8") as f:
        cfg = yaml.safe_load(f)
    cfg["systemPrompt"] = new_prompt
    with open(entry, "w", encoding="utf-8") as f:
        yaml.safe_dump(cfg, f, allow_unicode=True)

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


def test_update_upload_loopback_blocked(http_client, tmp_path):
    """非 loopback 调 update-upload → 404（同 /install 的 oracle 防御）。"""
    client, key, packs_dir, pack = http_client
    zip_path = _pack_zip_with_prompt(tmp_path, REVIEWER_ID, "x")
    with open(zip_path, "rb") as f:
        r = client.post(
            f"/api/v1/agentpacks/{REVIEWER_ID}/update-upload",
            files={"file": ("pack.zip", f, "application/zip")},
            headers=_auth(key),
        )
    assert r.status_code == 404


def test_peek_manifest_id_check(tmp_path, http_client):
    """peek_manifest_from_zip 返回正确 id（更新端点据此拒绝错包）。"""
    from agentpaas.engine.agentpack_store import peek_manifest_from_zip
    zip_path = _pack_zip_with_prompt(tmp_path, REVIEWER_ID, "NEW PROMPT")
    m = peek_manifest_from_zip(zip_path)
    assert m.id == REVIEWER_ID
    # zip 不会被安装
    assert not os.path.exists(
        os.path.join(str(tmp_path), "agentpacks", REVIEWER_ID, "0.9.9"))


def test_pack_update_refreshes_agents(http_client, tmp_path):
    """更新包 → 由它创建的智能体生成新版本、配置换新、可回滚。"""
    from agentpaas.api.v1.agentpacks import _refresh_agents_from_pack

    client, key, packs_dir, pack = http_client

    # 1. 从包创建智能体（HTTP）
    r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "更新测试-审稿"},
        headers=_auth(key),
    )
    assert r.status_code == 201, r.text[:300]
    agent_id = r.json()["agent_id"]

    g0 = client.get(f"/api/v1/agents/{agent_id}", headers=_auth(key)).json()
    old_prompt = g0["config"]["systemPrompt"]
    assert g0["current_version"] == 1

    # 2. 构造新版包（新版本号 + 新 systemPrompt）并安装
    NEW_PROMPT = "你是更新后的审稿专家 v2。"
    zip_path = _pack_zip_with_prompt(
        tmp_path, REVIEWER_ID, NEW_PROMPT, new_version="0.9.9")
    new_pack = _install_pack(packs_dir, zip_path)
    assert new_pack.version == "0.9.9"

    # 3. 传播到智能体
    refreshed = _refresh_agents_from_pack(new_pack)
    assert any(a["agent_id"] == agent_id and a["version"] == 2
               for a in refreshed)

    # 4. 智能体现在跑最新配置
    g1 = client.get(f"/api/v1/agents/{agent_id}", headers=_auth(key)).json()
    assert g1["current_version"] == 2
    assert g1["config"]["systemPrompt"] == NEW_PROMPT
    assert g1["config"]["systemPrompt"] != old_prompt
    # agent_dir 跟到新版本安装路径
    assert g1["agent_dir"].endswith("0.9.9")

    # 5. 旧配置仍可回滚
    rb = client.post(
        f"/api/v1/agents/{agent_id}/rollback",
        json={"target_version": 1},
        headers=_auth(key),
    )
    assert rb.status_code == 200, rb.text[:300]
    g2 = client.get(f"/api/v1/agents/{agent_id}", headers=_auth(key)).json()
    assert g2["config"]["systemPrompt"] == old_prompt


def test_create_agent_from_pack_auto_provisions_dirs(http_client, tmp_path):
    """从包创建智能体也自动开辟 Workspace/<名>/{data,workspace} 目录。"""
    client, key, packs_dir, pack = http_client
    r = client.post(
        f"/api/v1/agentpacks/{REVIEWER_ID}/create-agent",
        json={"name": "目录开辟-审稿"},
        headers=_auth(key),
    )
    assert r.status_code == 201, r.text[:300]
    body = r.json()
    assert body["work_dir"].endswith("目录开辟-审稿")
    assert body["source_dir"] == os.path.join(body["work_dir"], "data")
    assert body["run_dir"] == os.path.join(body["work_dir"], "workspace")
    assert os.path.isdir(body["source_dir"])
    assert os.path.isdir(body["run_dir"])

    # GET 回读一致
    g = client.get(f"/api/v1/agents/{body['agent_id']}", headers=_auth(key)).json()
    assert g["work_dir"] == body["work_dir"]
    assert g["run_dir"] == body["run_dir"]


def test_run_with_context_work_dir_override(tmp_path, http_client, monkeypatch):
    """工作区对话模式回归：run 带 context.work_dir 应覆盖工作目录、不报 500。

    曾因 run handler 用裸 `os`（本文件无模块级 import os，仅 `import os as _os`）
    在 work_dir override 分支 NameError 崩 500——且该分支只在 context.work_dir
    有值时触发，普通 run 测试覆盖不到。本例正好走这条分支。
    """
    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"]

    captured = {}

    def _stub_exec(config, input_text, **kw):
        captured["work_dir"] = kw.get("work_dir")
        captured["source_dir"] = kw.get("source_dir")
        captured["inplace_dir"] = kw.get("inplace_dir")
        return ("摘要：ok", {"workspace_path": "", "input_tokens": 1,
                             "output_tokens": 1, "steps": 1, "cost_usd": 0.0,
                             "cache_read_tokens": 0, "cache_creation_tokens": 0})

    import agentpaas.api.v1.agents as _agents_mod
    monkeypatch.setattr(_agents_mod, "_execute_agent", _stub_exec)

    folder = str(tmp_path)
    r = client.post(
        f"/api/v1/agents/{agent_id}/run",
        json={"input": "总结一下", "context": {"work_dir": folder}},
        headers=_auth(key),
    )
    assert r.status_code != 500, f"work_dir override 崩了: {r.text[:300]}"
    assert r.status_code == 200, r.text[:300]
    # override 生效：_execute_agent 收到的 work_dir/source_dir 都是所选文件夹
    assert captured.get("work_dir") == folder
    assert captured.get("source_dir") == folder
    # 工作区模式:inplace_dir 传入 → 模型 [工作目录] 会被设为 F(就地操作)
    assert captured.get("inplace_dir") == folder