Testing

What plain components buy you

Most of what you write needs no kernel in its tests.

Test the component directly

A component bound with provide() is a plain class with a normal constructor. Construct it:

def test_greeting():
    greeter = Greeter(database=FakeDB(), prefix="hi")
    assert greeter.hello("world") == "hi world"

No kernel, no container, no fixtures, no async. This is the whole reason chapter 2 separates the component from its registration. If a test of your business logic needs a kernel, something has leaked.

Test a plugin with a real context

A plugin function does need a context, because taking one is what makes it a plugin. Build a root and mount it:

import asyncio
from plugkit import Context, provide


async def test_admin_registers_its_route():
    root = Context()
    await root.plugin(provide(Server, "server"))
    await root.plugin(admin_api)

    assert "/admin" in root.server.routes

pytest-asyncio with asyncio_mode = "auto" in pyproject.toml lets you write async def test_... with no decorator. plugkit’s own suite uses that.

Waiting correctly

await root.plugin(p) waits for p’s own load. When mounting one plugin unblocks another, await the second one’s fiber:

greeter = await root.plugin(provide(Greeter, "greeter", needs=["database"]))
await root.plugin(provide(Database, "database"))
await greeter                       # now the greeter has loaded

For a test touching several plugins, a settle helper is simpler than tracking each fiber:

async def settle(n=15):
    for _ in range(n):
        await asyncio.sleep(0)

await fiber.dispose() needs no settle — it waits for the unload it cascades to dependents.

Substituting a dependency

Services are found by name, so substitution is mounting a different class under the same name:

async def test_with_a_fake_database():
    root = Context()
    await root.plugin(provide(FakeDatabase, "database"))     # not the real one
    await root.plugin(provide(Greeter, "greeter", needs=["database"]))
    await settle()

    assert root.greeter.hello("x") == "hello x"

Nothing needs a mocking library and nothing patches an import.

Overriding config

with root.config.override({"http": {"timeout": 1}}):
    assert root.config.get("http.timeout") == 1
assert root.config.get("http.timeout") == 30

Readers wake in both directions, so an effect under test sees the value arrive and leave.

Testing that teardown happened

Assert that unload removed everything:

async def test_unload_is_total():
    root = Context()
    await root.plugin(provide(Server, "server"))
    fiber = await root.plugin(admin_api)
    assert "/admin" in root.server.routes

    await fiber.dispose()
    assert "/admin" not in root.server.routes

If you write a plugin that registers something, write this test for it. It is the one that fails when someone later adds a second registration and forgets its disposer.

Observing a failure instead of raising

await root.plugin(p) re-raises whatever p raised. Catch it, then read the state:

from plugkit import FiberState

fiber = root.plugin(might_fail)
with pytest.raises(RuntimeError):
    await fiber
assert fiber.state is FiberState.FAILED

Do not reach for await asyncio.sleep(0) here. A failed load takes three ticks to settle — LOADING, then UNLOADING, then FAILED — and counting ticks is a test that breaks when the transition changes. Awaiting waits for the transition itself.

Testing a tool through the pipeline

ctx.tools.execute never raises. Assert on the result:

from plugkit import PointsService, ToolsService


async def test_the_guard_denies():
    root = Context()
    await root.plugin(PointsService)
    await root.plugin(ToolsService)
    await root.plugin(register_delete_tool)
    await root.plugin(safety_guard)
    await settle()

    result = await root.tools.execute("delete", {"path": "/"})
    assert result.ok is False
    assert result.error["code"] == "DENIED"

Asking the system what it is doing

When a test fails and the reason is not obvious, the usual cause is a plugin that never ran. describe returns a plain snapshot, and format_tree renders it:

from plugkit import describe, format_tree

print(format_tree(describe(root)))
plugkit — 6 fibers, 4 services
├─ [1] PointsService  ACTIVE    provides points
├─ [2] db             ACTIVE    provides database
├─ [3] ToolsService   ACTIVE    provides tools
├─ [4] greeter        ACTIVE    provides greeter
├─ [5] waiting        PENDING   waiting for: cache, queue
└─ [6] broken         FAILED    ConnectionRefusedError: [Errno 61] Connection refused

Line 5 is the one that saves time. A PENDING plugin is the kernel working correctly — it waits for a service that is not there — but from outside it looks identical to a plugin that never mounted. The snapshot names what it is waiting for.

describe is a plain function, not a service. You do not mount it, and it works on any context, including one written before this existed. That matters because you reach for it when something is already wrong.

Assert on the snapshot directly when the shape is the thing under test:

def by_name(snapshot, name):
    return next(f for f in snapshot["fibers"] if f["name"] == name)


async def test_the_greeter_waits_for_its_database():
    root = Context()
    await root.plugin(provide(Greeter, "greeter", needs=["database"]))
    await settle()

    entry = by_name(describe(root), "greeter")
    assert entry["state"] == "PENDING"
    assert entry["missing"] == ["database"]

Everything in a snapshot is a string, number, list or dict, so json.dumps works on it and a snapshot captured in CI renders the same later.

Telling it something it cannot know

The kernel knows a fiber’s state. It does not know a connection pool’s size. Contribute a diagnostic:

from plugkit import DIAGNOSTICS


def database(ctx, config=None):
    pool = Pool()
    ctx.provide("database", pool)
    ctx.points.add(DIAGNOSTICS, lambda: {"pool_size": pool.size}, key="database")

database.inject = ["points"]
describe(root)["diagnostics"]["database"]      # {'pool_size': 4}

It needs PointsService mounted, it disappears when the plugin unloads, and a diagnostic that raises is reported as an error string rather than breaking the snapshot.

What plugkit’s own suite does

Four tests in plugkit’s own suite are patterns you can copy:

Test Pattern
test_component_needs_no_kernel_at_all asserts the components in the file import nothing from plugkit
test_every_export_is_exercised_somewhere walks __all__ and fails on a name no test mentions
test_readme_examples.py runs every README example and asserts the prose’s claims
test_bare_install.py imports every module with the optional packages hidden

The third exists because the README once documented an API that did not exist. If your docs make a claim, run it.

The fourth exists because the development environment has every optional package installed, so the suite cannot see an import that a plain pip install would break. It hides yaml and watchdog from a subprocess and imports the package as a bare user would.