lambdagent/tests/test_builtin_packs.py

"""
tests/test_builtin_packs.py — Unit tests for builtin_packs.ensure_builtin_packs_installed().

Verifies:
  - All 3 research packs are installed into a fresh directory.
  - Second call is idempotent (no re-install, no error).
  - Missing source dir is handled gracefully (no exception).
  - Per-pack failures don't abort the remaining installs.
"""
from __future__ import annotations

import os
from pathlib import Path
from unittest.mock import patch

import pytest

from agentpaas.engine.builtin_packs import (
    BUILTIN_PACK_IDS,
    ensure_builtin_packs_installed,
)
from agentpaas.engine.agentpack_store import get_installed, list_installed


# ── helpers ──────────────────────────────────────────────────────────────────


def _packs_src() -> Path | None:
    """Locate agentexample/agentpacks/ relative to repo root."""
    here = Path(__file__).resolve().parent.parent  # repo root
    candidate = here / "agentexample" / "agentpacks"
    return candidate if candidate.is_dir() else None


# ── tests ─────────────────────────────────────────────────────────────────────


@pytest.mark.skipif(
    _packs_src() is None,
    reason="agentexample/agentpacks/ not present (running outside repo)",
)
def test_ensure_installs_all_builtin_packs(tmp_path):
    """Fresh packs_dir → all 3 built-in packs installed."""
    packs_dir = str(tmp_path / "packs")
    ensure_builtin_packs_installed(packs_dir)

    installed_ids = {p.id for p in list_installed(packs_dir)}
    for pack_id in BUILTIN_PACK_IDS:
        assert pack_id in installed_ids, f"pack {pack_id} was not installed"


@pytest.mark.skipif(
    _packs_src() is None,
    reason="agentexample/agentpacks/ not present",
)
def test_ensure_idempotent(tmp_path):
    """Calling twice does not raise and does not corrupt the installs."""
    packs_dir = str(tmp_path / "packs")

    ensure_builtin_packs_installed(packs_dir)
    first_count = len(list_installed(packs_dir))

    ensure_builtin_packs_installed(packs_dir)  # second call
    second_count = len(list_installed(packs_dir))

    assert first_count == second_count == len(BUILTIN_PACK_IDS)


def test_ensure_graceful_when_no_source_dir(tmp_path):
    """If agentexample/agentpacks/ cannot be found, no exception is raised."""
    packs_dir = str(tmp_path / "packs")

    with patch(
        "agentpaas.engine.builtin_packs._find_packs_src",
        return_value=None,
    ):
        # Must not raise
        ensure_builtin_packs_installed(packs_dir)

    # Nothing installed (source was unavailable)
    assert list_installed(packs_dir) == []


@pytest.mark.skipif(
    _packs_src() is None,
    reason="agentexample/agentpacks/ not present",
)
def test_ensure_continues_after_per_pack_failure(tmp_path, monkeypatch):
    """A failure in the first pack does not block the remaining two."""
    packs_dir = str(tmp_path / "packs")
    first_pack = BUILTIN_PACK_IDS[0]
    rest = BUILTIN_PACK_IDS[1:]

    import agentpaas.engine.agentpack_store as _store

    original_get = _store.get_installed
    call_count = {"n": 0}

    def patched_get(d, pid, v=None):
        call_count["n"] += 1
        if pid == first_pack and call_count["n"] == 1:
            raise RuntimeError("simulated failure for first pack")
        return original_get(d, pid, v)

    # Patch the function on the source module so the local import inside
    # ensure_builtin_packs_installed() picks up the patched version.
    monkeypatch.setattr(_store, "get_installed", patched_get)

    # Must not raise even though first pack fails
    ensure_builtin_packs_installed(packs_dir)

    installed_ids = {p.id for p in list_installed(packs_dir)}
    for pack_id in rest:
        assert pack_id in installed_ids, (
            f"pack {pack_id} should have been installed despite earlier failure"
        )