Testing plugins

Pytest layout, async setup, testcontainers, KernelMock, FakeLLMRouter, and golden files.

Every plugin must ship its own tests. The repo's testing_kit package provides the building blocks; this guide shows how to wire them together.

Pytest configuration

In your plugin's pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

asyncio_mode = "auto" lets you write async def test_... without sprinkling @pytest.mark.asyncio everywhere. The async event loop is created per-test by default.

Run your package's tests in isolation:

uv run pytest packages/<your_plugin> -v

Or via the repo Makefile:

make test           # whole workspace
make test-quiet     # CI style
make test-cov       # coverage report

Layout convention

packages/<your_plugin>/
├── tests/
│   ├── conftest.py
│   ├── test_smoke.py            # imports + manifest parse only
│   ├── test_<plugin>.py         # unit tests
│   └── golden/                  # JSON golden fixtures
│       └── happy_path.json

A test_smoke.py that just imports the plugin and loads its manifest is cheap, fast, and catches broken manifests at PR time:

def test_imports() -> None:
    import skill_my  # noqa: F401


def test_manifest_attached() -> None:
    from skill_my import MySkill
    assert MySkill.manifest is not None
    assert MySkill.manifest.plugin.id == "my"

Unit tests with KernelMock

testing_kit.KernelMock is an in-memory implementation of every SkillContext sub-API. Construct it with a project id and (optionally) a pre-built Project, Organization, or TriggerEvent:

from uuid import uuid4

from testing_kit import KernelMock

from skill_my import MySkill


async def test_executes_happy_path() -> None:
    mock = KernelMock(project_id=uuid4())
    result = await MySkill().execute(mock)
    assert result.status == "completed"

KernelMock tracks every side effect on public attributes:

Attribute	What it records
actions_called	Every ctx.actions.execute invocation
sub_skills_called	Every ctx.run_skill invocation
approvals_requested	Every ctx.approve call
entities._entities	All entities created/upserted (internal, exposed via API)
metrics_recorded	Custom metrics emitted via telemetry
spans_recorded	OTel spans opened during the run

You can assert against these directly:

assert any(
    call["action"].kind == "pull_request" and call["payload"]["body"]
    for call in mock.actions_called
)

Mocking LLM agents — FakeLLMRouter

testing_kit.FakeLLMRouter records every ctx.agents.run call and returns canned outputs:

from testing_kit import FakeLLMRouter, KernelMock


async def test_router_records_calls() -> None:
    mock = KernelMock(project_id=uuid4())
    router = FakeLLMRouter()
    router.register_fixture("classify_finding", {"severity": "high"})
    mock._fake_llm_router = router  # type: ignore[attr-defined]

    await MySkill().execute(mock)

    router.assert_called_with("classify_finding")
    assert router.calls[0].input["text"] == "..."

FakeAgentResult (the object returned from run) satisfies the AgentResult Protocol — your skill code does not need to be aware of the difference.

Reset between test cases:

@pytest.fixture(autouse=True)
def reset_router() -> Iterator[None]:
    yield
    # Fixtures + calls are cleared by re-instantiating, or by:
    router.reset()

Mocking outbound HTTP — connectors

Wrap httpx.AsyncClient with a MockTransport:

import httpx
import pytest


async def test_health_check(monkeypatch: pytest.MonkeyPatch) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/v1/me"
        return httpx.Response(200, json={"id": "u_1"})

    monkeypatch.setattr(
        "httpx.AsyncClient",
        lambda **kw: httpx.AsyncClient(transport=httpx.MockTransport(handler), **kw),
    )

    conn = MyConnector()
    await conn.setup(project=..., config=..., credentials=...)
    assert (await conn.health_check()).status == "ok"

For richer matching, respx works too — pick whichever your team prefers, but never hit the real external API in unit tests. The CI sandbox has no outbound network.

Integration tests — Postgres testcontainers

For tests that exercise the kernel's real services (entity service, approvals, plugin registry), spin up a Postgres 16 testcontainer. The repo convention uses asyncio.run + create_async_engine rather than the legacy synchronous psycopg2 driver. The pattern, from packages/kernel/tests/test_entity_service.py:

import asyncio
from collections.abc import AsyncIterator, Iterator

import pytest
from sqlalchemy.engine.url import make_url
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from testcontainers.postgres import PostgresContainer

from kernel.db.base import Base


def _to_async_url(sync_url: str) -> str:
    return (
        make_url(sync_url)
        .set(drivername="postgresql+asyncpg")
        .render_as_string(hide_password=False)
    )


@pytest.fixture(scope="module")
def pg_url() -> Iterator[str]:
    with PostgresContainer("postgres:16-alpine") as pg:
        async_url = _to_async_url(pg.get_connection_url())

        async def _create_schema() -> None:
            engine = create_async_engine(async_url)
            async with engine.begin() as conn:
                await conn.run_sync(Base.metadata.create_all)
            await engine.dispose()

        asyncio.run(_create_schema())
        yield async_url


@pytest.fixture()
async def session(pg_url: str) -> AsyncIterator[AsyncSession]:
    eng = create_async_engine(pg_url)
    factory = async_sessionmaker(eng, expire_on_commit=False)
    async with factory() as s:
        yield s
        await s.rollback()
    await eng.dispose()

Why this matters:

• testcontainers-python exposes a sync URL (postgresql://). The kernel uses asyncpg; we rewrite the driver portion before passing it to SQLAlchemy.
• Schema creation runs once per module on a freshly built event loop via asyncio.run. Each test gets its own short-lived engine + session that rolls back at the end.
• Do not install or import psycopg2 for integration tests — asyncpg is the project standard.

Golden datasets

testing_kit.golden provides snapshot helpers for deterministic skill outputs:

from pathlib import Path

from testing_kit import assert_golden, load_golden


def test_review_output_stable(tmp_path: Path) -> None:
    actual = {"findings": [...]}
    assert_golden(actual, golden_path="tests/golden/review_v1.json")

To regenerate the golden file when the output legitimately changes:

UPDATE_GOLDENS=1 uv run pytest packages/skill_my -v

This rewrites the JSON on disk; review the diff before committing. Use goldens for:

• Stable LLM-derived structures (when paired with FakeLLMRouter fixtures so the LLM is deterministic).
• Connector normalization outputs (raw payload → NormalizedEntity).
• Pack parser output for a checked-in sample file.

The reusable Skill contract test

Inherit from SkillContractTest (re-exported by both skill_base and testing_kit) for the standard battery of structural assertions the runtime relies on:

from skill_base import SkillContractTest

from skill_my import MySkill


class TestMySkillContract(SkillContractTest):
    __test__ = True       # base class sets this False so pytest doesn't collect the base
    skill_cls = MySkill

The base class asserts that:

• manifest is a SkillManifest instance.
• state_kind is None or a str.
• execute(self, ctx) exists, is a coroutine function, and takes exactly one arg.
• execute is annotated to return SkillResult.
• If state_kind is set, on_state_event is a coroutine function.

If you change the Skill Protocol, every plugin's contract test will fail loudly — by design.

Coverage expectations

The repo target is ≥ 80% line coverage on touched modules; the base packages aim for 100%. For a plugin, prefer to:

• Cover the manifest happy path + at least one validation failure (e.g. malformed config).
• Cover every branch in execute / handle_webhook / parse — including error and short-circuit paths.
• Cover both auto-approved and human-approval branches if ctx.approve is used.
• Cover idempotency: call execute twice with the same input and assert no duplicate side effects.

Generate coverage locally:

make test-cov

Common pitfalls

Symptom	Likely cause
pytest hangs on async test	Missing asyncio_mode = "auto" in pyproject.toml
ImportError: psycopg2 during integration test	Using the sync URL from testcontainers — rewrite to asyncpg
ValidationError: extra fields not permitted on manifest	YAML has a typo (extra="forbid" is set everywhere)
AssertionError from SkillContractTest.test_manifest_attached	Class is missing @skill(...) decorator
FileNotFoundError: manifest.yaml	Decorator path is relative — but class lives in a different file than expected
httpx.ConnectError in CI	A real outbound call slipped through — mock the transport