Kotoba

Where to start

What Kotoba isWho she is and where she runs, in one page
InstallingOne package, two commands
First runA key, a model, her first words
The approval gateHow she asks before she acts
The two voice modesLocal voice, or the agent tunnel
The soul fileChange who she is
RoadmapWhat grows next, and what was cut on purpose

Or a section

↑↓ move openesc close124 pages
All pages

Plugins7 of 7

Testing your plugin

About 3 minutes to read

On this page

Your plugin lives outside the repository, so it needs its own tests. The good news is that most of it tests like ordinary Python: execute() is a coroutine you can call, and the module's attributes are just values.

The part worth real effort is the wiring — that the loader actually finds it, registers it under the toolset you expect, and offers it in work mode. That is the half a unit test cannot see, and the half that breaks on an upgrade.


Layout

kotoba-plugin-hackernews/
  hackernews/
    top.py               <- the tool; copied to ~/.kotoba/plugins/hackernews/ to install
  tests/
    test_hn_top.py
  README.md

hackernews/ is the directory a user copies into ~/.kotoba/plugins/. The wiring test below copies it into a temporary plugins directory and loads it there — do not just point KOTOBA_PLUGINS_PATH at your repository root, because the loader would treat tests/ as a plugin too and execute your test files at import.

You need pytest and kotoba-companion in the environment:

bash
pip install pytest kotoba-companion

The suite

tests/test_hn_top.py, for the tool built in Your first plugin. It runs green as written.

python
"""Tests for the hn_top plugin."""
from __future__ import annotations

import asyncio
import inspect
import json
import shutil
import subprocess
import sys
from pathlib import Path

PLUGIN_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PLUGIN_ROOT / "hackernews"))

import top  # noqa: E402


# --- the contract -------------------------------------------------------------

def test_schema_is_a_function_tool_with_a_name():
    assert top.SCHEMA["type"] == "function"
    assert top.SCHEMA["name"] == "hn_top"
    assert top.SCHEMA["description"].strip()
    assert top.SCHEMA["parameters"]["type"] == "object"


def test_risk_is_one_the_registry_understands():
    assert top.RISK in {"read", "write", "exec", "network"}


def test_execute_is_a_coroutine_function():
    """A plain def raises TypeError out of the tool runner instead of failing gracefully."""
    assert inspect.iscoroutinefunction(top.execute)


def test_expressions_are_real_faces():
    from kotoba.core.emotions import VALID_EMOTIONS
    assert set(top.EXPRESSIONS) <= {"focus", "done", "fail"}
    assert set(top.EXPRESSIONS.values()) <= VALID_EMOTIONS


def test_check_is_cheap_and_answers_a_bool():
    assert top.check() in (True, False)


# --- the behaviour ------------------------------------------------------------

def test_it_returns_a_sentence_when_the_network_fails(monkeypatch):
    """An exception reaches her as 'The tool errored.' with no log line, so it must not escape."""
    import httpx

    def boom(*a, **k):
        raise httpx.ConnectError("no route")

    monkeypatch.setattr(httpx.AsyncClient, "get", boom)
    out = asyncio.run(top.execute({}, None))
    assert "couldn't reach" in out.lower()


def test_bad_arguments_never_raise(monkeypatch):
    """The model writes `args`. Nothing validates it against your schema before you see it."""
    import httpx

    monkeypatch.setattr(httpx.AsyncClient, "get",
                        lambda *a, **k: (_ for _ in ()).throw(httpx.ConnectError("no route")))
    for args in (None, {}, {"count": "not a number"}, {"count": -5}, {"count": 9999}):
        out = asyncio.run(top.execute(args, None))
        assert isinstance(out, str) and out.strip()


# --- the wiring ---------------------------------------------------------------

_PROBE = """
import json
import kotoba.tools
import kotoba.tools.registry as reg
from kotoba.core.plugins import loaded_plugins
spec = reg.registry().get("hn_top")
print(json.dumps({
    "loaded": loaded_plugins(),
    "toolset": spec and spec.toolset,
    "risk": spec and spec.risk,
    "in_work": "hn_top" in {s.get("name") for s in
                            reg.schemas_for("work", {"read", "write", "exec", "network"})},
}))
"""


def _install(tmp_path):
    """Install the plugin the way a user does: copy the directory into a plugins folder. Copied to a
    temp dir rather than pointing the loader at the repository, which would also scan tests/."""
    root = tmp_path / "plugins"
    root.mkdir()
    shutil.copytree(PLUGIN_ROOT / "hackernews", root / "hackernews")
    return {
        "PATH": "/usr/bin:/bin",
        "HOME": str(tmp_path),
        "KOTOBA_HOME": str(tmp_path / "kotoba"),
        "KOTOBA_PLUGINS_PATH": str(root),
        "KOTOBA_SETTINGS": str(tmp_path / "settings.yaml"),
        "DATABASE_URL": "sqlite:///" + str(tmp_path / "test.db"),
    }


def test_it_loads_as_a_plugin(tmp_path):
    """A child process with its own home: never point this at the real ~/.kotoba."""
    env = _install(tmp_path)
    out = subprocess.run([sys.executable, "-c", _PROBE],
                         capture_output=True, text=True, timeout=180, env=env)
    assert out.returncode == 0, out.stderr[-2000:]
    data = json.loads(out.stdout.strip().splitlines()[-1])

    assert data["loaded"]["hackernews"]["tools"] == ["hn_top"]
    assert data["toolset"] == "plugin:hackernews"
    assert data["risk"] == "network"
    assert data["in_work"] is True


def test_the_name_is_not_already_taken(tmp_path):
    """Registration is refused if a built-in owns the name — and disabling the plugin would then
    deregister the built-in."""
    probe = "import kotoba.tools; from kotoba.tools.registry import registry; print('hn_top' in registry())"
    (tmp_path / "plugins").mkdir()          # empty: nothing of yours is loaded
    env = {"PATH": "/usr/bin:/bin", "HOME": str(tmp_path),
           "KOTOBA_HOME": str(tmp_path / "kotoba"),
           "KOTOBA_PLUGINS_PATH": str(tmp_path / "plugins"),
           "KOTOBA_SETTINGS": str(tmp_path / "settings.yaml"),
           "DATABASE_URL": "sqlite:///" + str(tmp_path / "test.db")}
    out = subprocess.run([sys.executable, "-c", probe],
                         capture_output=True, text=True, timeout=180, env=env)
    assert out.returncode == 0, out.stderr[-2000:]
    assert out.stdout.strip().endswith("False"), "a built-in already owns the name hn_top"
$ pytest -q
.........                                                                [100%]
9 passed

Why the wiring test runs in a child process

Two reasons, both practical.

kotoba.tools runs discover() at import, once per process. There is no second chance: whatever KOTOBA_PLUGINS_PATH said at that moment is what got loaded. A subprocess with a controlled environment is the only clean way to test discovery, and it is how the project's own plugin tests do it (api/tests/test_disabled_plugin_never_runs.py).

It also keeps the test honest about the environment a real install has. The child gets PATH, HOME, and the three Kotoba variables — nothing else. If your plugin quietly depends on something in your shell, the test finds out.

Give the child its own KOTOBA_HOME and its own DATABASE_URL. Without them the test writes into the real ~/.kotoba and the real database. That is the one mistake in this file that costs somebody their data.


Testing against the loop, not just against your function

execute() returning a string is not the same as the loop accepting it. The runner adds three verdicts your function never sees — empty output, an exception, and a timeout — and all three come back as ok=False. Test through it:

python
def test_the_loop_accepts_the_result():
    import asyncio
    import kotoba.tools
    from kotoba.tools import ToolContext
    from kotoba.core.loop import execute_with_heartbeat

    async def run():
        ctx = ToolContext(db=None, mode="work", channel="text")
        return await execute_with_heartbeat("hn_top", {"count": 2}, asyncio.Queue(), {}, ctx)

    ok, out = asyncio.run(run())
    assert ok is True
    assert out.strip()

That test only works in a process where your plugin was discovered, so it needs KOTOBA_PLUGINS_PATH set before kotoba.tools is first imported. Either fold it into the subprocess probe, or run the suite with the variable already pointing at a directory holding a copy of your plugin and nothing else:

bash
mkdir -p /tmp/kotoba-test/plugins && cp -r hackernews /tmp/kotoba-test/plugins/
KOTOBA_PLUGINS_PATH=/tmp/kotoba-test/plugins KOTOBA_HOME=/tmp/kotoba-test \
  DATABASE_URL=sqlite:////tmp/kotoba-test/test.db pytest -q

What to guard against an upgrade

Your plugin reads things out of kotoba that are not a stability-guaranteed public API. The following are worth a test each, because they are the ones that would break silently:

AssumptionThe test
The loader still finds folder plugins where you put themtest_it_loads_as_a_plugin
Your toolset name is still plugin:<dir>the toolset assertion in it
Your tool is still offered in work modethe in_work assertion
Your EXPRESSIONS values are still real facestest_expressions_are_real_faces
Your RISK is still a value the registry knowstest_risk_is_one_the_registry_understands
Nothing new in the tree took your tool's nametest_the_name_is_not_already_taken
Any helper you import (validate_within_dir, note_tool_refusal, ToolResult) still existsan import at the top of a test

That last one is the cheapest insurance in the list. A one-line import test turns an upgrade that moved a helper into a red test instead of a tool that quietly stopped loading.


Where your tests live relative to an upgrade

Your repository is yours; pip install -U kotoba-companion does not touch it, and neither does it touch ~/.kotoba/plugins/. What changes underneath you is the package your tests import. So:

  • Run your suite after every Kotoba upgrade, not only after your own changes.
  • If you ship a distribution, put a floor in its dependencieskotoba-companion>=<the version you tested> — so pip refuses an environment you have never run against.
  • Keep the wiring test in CI against the latest release, and you find out that the contract moved before your users do.