lambdagent/tests/test_agentpack.py

"""
Tests for the AgentPack feature (M2 Phase D).

Covers:
  - lambdagent.agentpack: manifest validation (kernel, pure)
  - agentpaas.engine.agentpack_store: safe install / list / uninstall
  - zip-slip defense
  - SR-003 shell refusal for third-party packs

Spec: docs/AGENTPACK_SPEC.md.
"""
from __future__ import annotations

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

import io
import zipfile

import pytest
import yaml

from lambdagent.agentpack import (
    AgentPackManifest,
    PackPermissions,
    AgentPackError,
    load_manifest,
    validate_manifest_dict,
    enforce_config_permissions,
)
from agentpaas.engine.agentpack_store import (
    install_from_zip,
    list_installed,
    get_installed,
    uninstall,
    AgentPackInstallError,
)


# ─────────────────────────────────────────────────────────────────────────────
# Kernel: manifest validation
# ─────────────────────────────────────────────────────────────────────────────

_VALID = {
    "id": "research.top-journal-reviewer",
    "name": "Top Journal Reviewer",
    "version": "0.1.0",
    "domain": "research",
    "entrypoint": "agents/reviewer.yml",
    "permissions": {"network": False, "shell": False, "file_write": "workspace"},
    "audience": ["professor", "phd_student"],
    "model": {"recommended": ["claude-code/sonnet"]},
}


def test_manifest_valid():
    m = validate_manifest_dict(_VALID)
    assert m.id == "research.top-journal-reviewer"
    assert m.version == "0.1.0"
    assert m.permissions.file_write == "workspace"
    assert m.permissions.shell is False
    assert "professor" in m.audience
    assert m.model_recommended == ["claude-code/sonnet"]


@pytest.mark.parametrize("missing", ["id", "name", "version", "domain", "entrypoint", "permissions"])
def test_manifest_missing_required_field(missing):
    bad = dict(_VALID)
    bad.pop(missing)
    with pytest.raises(AgentPackError):
        validate_manifest_dict(bad)


def test_manifest_bad_id_rejected():
    bad = dict(_VALID, id="Bad ID With Spaces!")
    with pytest.raises(AgentPackError):
        validate_manifest_dict(bad)


def test_manifest_bad_version_rejected():
    bad = dict(_VALID, version="v1")
    with pytest.raises(AgentPackError):
        validate_manifest_dict(bad)


def test_manifest_entrypoint_escape_rejected():
    bad = dict(_VALID, entrypoint="../../etc/passwd")
    with pytest.raises(AgentPackError):
        validate_manifest_dict(bad)


def test_manifest_bad_file_write_scope_rejected():
    bad = dict(_VALID, permissions={"file_write": "everywhere"})
    with pytest.raises(AgentPackError):
        validate_manifest_dict(bad)


def test_permissions_deny_by_default():
    # Empty permissions block → safe defaults.
    m = validate_manifest_dict(dict(_VALID, permissions={}))
    assert m.permissions.network is False
    assert m.permissions.shell is False
    assert m.permissions.file_write == "workspace"


def test_permission_summary_zh():
    m = validate_manifest_dict(_VALID)
    s = m.permissions.summary_zh()
    assert "读取本地知识库" in s
    assert "不执行 shell" in s
    assert "不访问网络" in s


def test_permissions_are_enforced_on_runtime_config():
    config = {
        "type": "react",
        "systemPrompt": "test",
        "mcp": {
            "localTools": [
                "Bash", "RunTests", "WebSearch", "ReadFile", "WriteFile",
                "KBSearch", "KBAdd", "terminate",
            ],
            "onlineTool": {"remote": ["dangerous_remote_tool"]},
        },
    }
    effective, report = enforce_config_permissions(config, PackPermissions(
        network=False, shell=False, file_write="workspace",
        read_knowledge=True, read_filesystem=False,
    ))

    assert effective["mcp"]["localTools"] == ["WriteFile", "KBSearch", "terminate"]
    assert "onlineTool" not in effective["mcp"]
    assert set(report["removed_tools"]) == {
        "Bash", "RunTests", "WebSearch", "ReadFile", "KBAdd",
    }
    assert effective["_agentpack_permissions"]["file_write"] == "workspace"


def test_gateway_wraps_validated_builtin_tools_case_insensitively():
    from lambdagent.tool_gateway import GatedTool, GatewayPolicy, ToolGateway
    from lambdagent.validated_tool import ShellToolInput, ValidatedTool

    executed = []
    bash = ValidatedTool("Bash", lambda value: executed.append(value) or "ran", ShellToolInput)
    gated = ToolGateway(GatewayPolicy()).wrap(bash)

    assert isinstance(gated, GatedTool)
    result = gated.apply('{"command":"rm -rf /"}')
    assert result.startswith("[BLOCKED]")
    assert executed == []


# ─────────────────────────────────────────────────────────────────────────────
# Helpers to build pack zips
# ─────────────────────────────────────────────────────────────────────────────

def _make_pack_zip(tmp_path, manifest_dict, *, entrypoint_body="type: react\nname: x\nsystemPrompt: hi\n", nested=False, extra_members=None):
    """Build a .zip containing manifest.yml + the entrypoint config.
    If nested=True, wraps everything in a top-level dir."""
    prefix = "mypack/" if nested else ""
    zpath = tmp_path / "pack.zip"
    with zipfile.ZipFile(zpath, "w") as zf:
        zf.writestr(prefix + "manifest.yml", yaml.safe_dump(manifest_dict, allow_unicode=True))
        zf.writestr(prefix + manifest_dict["entrypoint"], entrypoint_body)
        for name, body in (extra_members or {}).items():
            zf.writestr(prefix + name, body)
    return str(zpath)


# ─────────────────────────────────────────────────────────────────────────────
# Store: install / list / get / uninstall
# ─────────────────────────────────────────────────────────────────────────────

def test_install_list_get_uninstall_roundtrip(tmp_path):
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    zp = _make_pack_zip(tmp_path, _VALID)

    pack = install_from_zip(zp, str(packs_dir))
    assert pack.id == "research.top-journal-reviewer"
    assert pack.version == "0.1.0"
    assert os.path.isfile(os.path.join(pack.path, "manifest.yml"))
    assert os.path.isfile(os.path.join(pack.path, "agents", "reviewer.yml"))

    listed = list_installed(str(packs_dir))
    assert len(listed) == 1
    assert listed[0].id == pack.id

    got = get_installed(str(packs_dir), pack.id)
    assert got is not None and got.version == "0.1.0"

    n = uninstall(str(packs_dir), pack.id)
    assert n == 1
    assert list_installed(str(packs_dir)) == []


def test_install_nested_pack_zip(tmp_path):
    """A zip with everything under a single top-level dir still installs."""
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    zp = _make_pack_zip(tmp_path, _VALID, nested=True)
    pack = install_from_zip(zp, str(packs_dir))
    assert pack.id == "research.top-journal-reviewer"


def test_install_rejects_missing_entrypoint(tmp_path):
    """Manifest references agents/reviewer.yml but the zip doesn't contain it."""
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    zpath = tmp_path / "pack.zip"
    with zipfile.ZipFile(zpath, "w") as zf:
        zf.writestr("manifest.yml", yaml.safe_dump(_VALID))
        # entrypoint deliberately absent
    with pytest.raises((AgentPackInstallError, AgentPackError)):
        install_from_zip(str(zpath), str(packs_dir))


def test_install_refuses_third_party_shell(tmp_path):
    """SR-003: a pack declaring shell:true is refused unless allow_shell."""
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    shell_manifest = dict(_VALID, id="research.shelly",
                          permissions={"shell": True, "file_write": "workspace"})
    zp = _make_pack_zip(tmp_path, shell_manifest)

    with pytest.raises(AgentPackInstallError) as exc:
        install_from_zip(zp, str(packs_dir))
    assert "shell" in str(exc.value).lower()

    # With allow_shell it goes through.
    pack = install_from_zip(zp, str(packs_dir), allow_shell=True)
    assert pack.manifest.permissions.shell is True


def test_install_rejects_zip_slip(tmp_path):
    """A malicious member with ../ must be rejected (SPEC §6 zip-slip)."""
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    zpath = tmp_path / "evil.zip"
    with zipfile.ZipFile(zpath, "w") as zf:
        zf.writestr("manifest.yml", yaml.safe_dump(_VALID))
        zf.writestr("agents/reviewer.yml", "type: react\n")
        zf.writestr("../../escape.txt", "pwned")
    with pytest.raises(AgentPackInstallError) as exc:
        install_from_zip(str(zpath), str(packs_dir))
    assert "escape" in str(exc.value).lower() or "unsafe" in str(exc.value).lower()


def test_uninstall_nonexistent_returns_zero(tmp_path):
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    assert uninstall(str(packs_dir), "does.not.exist") == 0


# ─────────────────────────────────────────────────────────────────────────────
# Built-in packs (FR-008) — regression guard: the 3 bundled research packs
# must always load, validate, and compile.
# ─────────────────────────────────────────────────────────────────────────────

_BUNDLED_IDS = {
    "research.top-journal-reviewer",
    "research.literature-mapper",
    "research.grant-planner",
}


def _bundled_pack_dirs():
    import glob
    repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    return sorted(glob.glob(os.path.join(repo_root, "agentexample", "agentpacks", "*", "manifest.yml")))


def test_bundled_packs_present():
    found = {load_manifest(os.path.dirname(p)).id for p in _bundled_pack_dirs()}
    assert _BUNDLED_IDS.issubset(found), f"missing bundled packs: {_BUNDLED_IDS - found}"


@pytest.mark.parametrize("manifest_path", _bundled_pack_dirs())
def test_bundled_pack_valid_and_compiles(manifest_path):
    """Each bundled pack: manifest validates, entrypoint exists, and the
    entrypoint config compiles via from_config (it's a real runnable agent)."""
    from lambdagent import from_config

    pack_dir = os.path.dirname(manifest_path)
    m = load_manifest(pack_dir)
    assert os.path.isfile(m.entrypoint_path()), m.entrypoint
    # Built-in packs are local-first: no network.
    assert m.permissions.network is False
    # The entrypoint must compile to a runnable term.
    term = from_config(m.entrypoint_path())
    assert term is not None
    # Policy: bundled packs default to no shell. The one exception — a
    # Claude-Code-style workspace assistant — may declare shell:true, but
    # ONLY if its agent guards it with dangerousCommandBlock (runtime safety
    # net for arbitrary Bash). 杜绝"内置包随便要 shell 又不设防护"。
    if m.permissions.shell:
        import yaml as _yaml
        cfg = _yaml.safe_load(open(m.entrypoint_path(), encoding="utf-8"))
        guard = cfg.get("guard") or {}
        assert guard.get("dangerousCommandBlock") is True, (
            f"bundled pack {m.id} 声明 shell:true 但未设 dangerousCommandBlock 防护"
        )


# ─────────────────────────────────────────────────────────────────────────────
# Phase G: agentpack → agent creation endpoint integration
# ─────────────────────────────────────────────────────────────────────────────

def test_create_agent_from_pack_endpoint(tmp_path, monkeypatch):
    """End-to-end: install a synthetic pack, POST /agentpacks/{id}/create-agent,
    verify the new agent row carries pack.path as agent_dir and pack.id as
    agent_template."""
    import json
    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

    # Redirect settings.agentpacks_dir into tmp so we don't touch ~/...
    packs_dir = tmp_path / "agentpacks"
    packs_dir.mkdir()
    monkeypatch.setattr(settings, "agentpacks_dir", str(packs_dir))

    # Install a pack via the public function.
    pack_zip = _make_pack_zip(tmp_path, _VALID)
    pack = install_from_zip(pack_zip, str(packs_dir))
    assert pack.id == "research.top-journal-reviewer"

    # Fresh in-memory DB + a real tenant + key.
    prev_db = _session_mod._db
    _session_mod._db = Database("sqlite:///:memory:")
    db = _session_mod._db
    try:
        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()

        with TestClient(app) as client:
            r = client.post(
                f"/api/v1/agentpacks/{pack.id}/create-agent",
                json={"name": "My Reviewer", "description": "from a pack"},
                headers={"Authorization": f"Bearer {raw_key}"},
            )
            assert r.status_code == 201, r.text[:300]
            body = r.json()
            assert body["ok"] is True
            agent_id = body["agent_id"]
            assert body["pack"]["id"] == pack.id
            assert body["pack"]["version"] == pack.version

        # Verify DB state: agent_dir = pack.path, agent_template = pack.id.
        row = db.fetchone(
            "SELECT agent_dir, agent_template, name FROM agents WHERE id = ?",
            (agent_id,),
        )
        assert row["agent_dir"] == pack.path
        assert row["agent_template"] == pack.id
        assert row["name"] == "My Reviewer"

        # The initial version's config carries _config_dir = pack.path so
        # any sub-agent ref inside the pack resolves correctly.
        ver = db.fetchone(
            "SELECT config FROM agent_versions WHERE agent_id = ? AND version = 1",
            (agent_id,),
        )
        cfg = json.loads(ver["config"])
        assert cfg.get("_config_dir") == pack.path
    finally:
        _session_mod._db = prev_db


def test_create_agent_from_unknown_pack_returns_404(tmp_path, monkeypatch):
    """POST against a pack_id that isn't installed → 404."""
    from fastapi.testclient import TestClient

    from agentpaas.api.app import app
    from agentpaas.config import settings

    monkeypatch.setattr(settings, "agentpacks_dir", str(tmp_path / "empty"))
    (tmp_path / "empty").mkdir()

    with TestClient(app) as client:
        # No auth — gate-blocking 401 is the first dependency that fires,
        # which is fine; the assertion we care about is "no 5xx, no
        # accidental success". 401 covers both paths cleanly.
        r = client.post(
            "/api/v1/agentpacks/does.not.exist/create-agent",
            json={"name": "x"},
        )
        assert r.status_code in (401, 404)