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


# ─────────────────────────────────────────────────────────────────────────────
# 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 research packs are local-first: no shell, no network.
    assert m.permissions.shell is False
    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