3

Modules Testing

Logan Fick edited this page 2026-05-04 21:48:06 -04:00

Table of Contents

• Modules - Testing
• Getting Started
• Writing Tests
• Testing Commands
• Testing Events
• Testing Routes
• Testing Protected Routes
• Using Storage
• Using Module Config
• Testing HTTP Calls
• Simulating Owncast Responses
• Stubbing in a test
• Stubbing values seen by @on_setup
• Class-wide stubbing
• Reference
• Fixture Reference
• Configuration fixtures
• Individual fixtures
• Event Factories
• make_user
• Available Factories
• Recording Stubs
• RecordingOwncastClient
• RecordingOwncastAdminClient

Modules - Testing

Owlbot includes a pytest plugin for testing modules without a running Owncast server. Handlers run through the real command and event dispatchers, storage is backed by in-memory SQLite, and Owncast API calls are recorded rather than sent.

Getting Started

Install Owlbot with the testing extra:

uv pip install owlbot[testing] \
  --default-index https://git.logal.dev/api/packages/LogalDeveloper/pypi/simple/ \
  --index https://pypi.org/simple

This pulls in pytest, pytest-asyncio, pytest-aiohttp, and aioresponses.

Enable the plugin from your test suite's conftest.py:

pytest_plugins = ["owlbot.testing.plugin"]

The examples below assume pytest-asyncio's auto mode, which lets async tests run without per-test decorators. If you don't already have a mode configured, this is a reasonable default for your pyproject.toml. Strict mode works too; just mark each async test with @pytest.mark.asyncio.

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

Then set module_pkg to the imported module you want to test and apply module_lifecycle to your test. Here is a complete test file for a module with a !greet command:

import pytest
from types import ModuleType

from owlbot.api import EventType
from owlbot.testing import RecordingOwncastClient, make_chat_event

import mymodule  # your @on_command-decorated module


@pytest.fixture
def module_pkg() -> ModuleType:
    return mymodule


@pytest.mark.usefixtures("module_lifecycle")
async def test_greet(
    event_dispatcher,
    owncast_client: RecordingOwncastClient,
) -> None:
    event = make_chat_event(raw_body="!greet")
    await event_dispatcher.dispatch(EventType.CHAT, event)
    assert owncast_client.calls[0].kwargs["body"] == "Hello!"

module_pkg points at the module under test. module_lifecycle registers the module's handlers and runs @on_setup before the test body. make_chat_event builds a realistic event. owncast_client.calls is the recorded call log that assertions read from.

Writing Tests

Patterns for testing each kind of handler against the real dispatch pipeline.

Testing Commands

Command tests dispatch chat events through the full event and command pipeline, then assert on what the recording client observed. Each entry in owncast_client.calls is a RecordedCall with .name (the method name) and .kwargs (the keyword arguments it was called with):

from owlbot.api import EventType
from owlbot.testing import make_chat_event


@pytest.mark.usefixtures("module_lifecycle")
async def test_ping(event_dispatcher, owncast_client) -> None:
    event = make_chat_event(raw_body="!ping")
    await event_dispatcher.dispatch(EventType.CHAT, event)
    assert owncast_client.calls[0].kwargs["body"] == "pong"

For permission-gated commands, build the invoking user with the appropriate flag and attach them to the chat event. is_moderator=True adds the MODERATOR scope; is_authenticated=True marks the user as authenticated in Owncast:

from owlbot.testing import make_user


@pytest.mark.usefixtures("module_lifecycle")
async def test_ban(event_dispatcher, owncast_client) -> None:
    user = make_user(is_moderator=True)
    event = make_chat_event(raw_body="!ban someone", user=user)
    await event_dispatcher.dispatch(EventType.CHAT, event)
    assert owncast_client.calls[0].kwargs["body"] == "someone has been banned."


@pytest.mark.usefixtures("module_lifecycle")
async def test_vote(event_dispatcher, owncast_client) -> None:
    user = make_user(is_authenticated=True)
    event = make_chat_event(raw_body="!vote yes", user=user)
    await event_dispatcher.dispatch(EventType.CHAT, event)
    assert owncast_client.calls[0].kwargs["body"] == "Vote recorded."

If @on_setup calls the Owncast client (fetching stream status, updating the stream title, sending a message, etc.), those calls appear in the log before anything the command produced. This module's @on_setup hook registers a !ping command and fetches stream status:

async def ping(ctx: CommandContext) -> None:
    await ctx.owncast_client.send_message("pong")


@on_setup
async def setup(ctx: ModuleContext) -> None:
    ctx.commands.register("ping", ping)
    status = await ctx.owncast_client.get_status()
    ctx.state["online"] = status.get("online", False)

A test for a command on this module indexes past the setup call:

@pytest.mark.usefixtures("module_lifecycle")
async def test_ping_after_setup(event_dispatcher, owncast_client) -> None:
    event = make_chat_event(raw_body="!ping")
    await event_dispatcher.dispatch(EventType.CHAT, event)
    # calls[0] is the get_status from @on_setup; calls[1] is the !ping response.
    assert owncast_client.calls[1].kwargs["body"] == "pong"

Testing Events

Event tests dispatch an event through the same pipeline and assert on what the handler did. A module that announces when the stream goes live:

@on_event(EventType.STREAM_STARTED)
async def announce_live(ctx: EventContext[StreamStartedEvent]) -> None:
    await ctx.owncast_client.send_message(f"{ctx.event.stream_title} is live!")

A test for this handler:

from owlbot.api import EventType
from owlbot.testing import make_stream_started_event


@pytest.mark.usefixtures("module_lifecycle")
async def test_announce_live(event_dispatcher, owncast_client) -> None:
    event = make_stream_started_event(stream_title="Game Night")
    await event_dispatcher.dispatch(EventType.STREAM_STARTED, event)
    assert owncast_client.calls[0].kwargs["body"] == "Game Night is live!"

Events that carry a user accept one built with make_user, so factories nest naturally. A welcome handler:

@on_event(EventType.USER_JOINED)
async def welcome(ctx: EventContext[UserJoinedEvent]) -> None:
    await ctx.owncast_client.send_message(f"Welcome, {ctx.event.user.display_name}!")

A test for the welcome handler:

from owlbot.testing import make_user, make_user_joined_event


@pytest.mark.usefixtures("module_lifecycle")
async def test_welcome(event_dispatcher, owncast_client) -> None:
    user = make_user(display_name="Alice")
    event = make_user_joined_event(user=user)
    await event_dispatcher.dispatch(EventType.USER_JOINED, event)
    assert owncast_client.calls[0].kwargs["body"] == "Welcome, Alice!"

A single handler can register for multiple event types:

@on_event(EventType.STREAM_STARTED, EventType.STREAM_STOPPED)
async def log_state(ctx: EventContext[StreamStartedEvent | StreamStoppedEvent]) -> None:
    state = "online" if isinstance(ctx.event, StreamStartedEvent) else "offline"
    await ctx.owncast_client.send_message(f"Stream {state}")

A test dispatches each type separately:

from owlbot.testing import make_stream_stopped_event


@pytest.mark.usefixtures("module_lifecycle")
async def test_log_state(event_dispatcher, owncast_client) -> None:
    await event_dispatcher.dispatch(
        EventType.STREAM_STARTED, make_stream_started_event()
    )
    await event_dispatcher.dispatch(
        EventType.STREAM_STOPPED, make_stream_stopped_event()
    )
    assert owncast_client.calls[0].kwargs["body"] == "Stream online"
    assert owncast_client.calls[1].kwargs["body"] == "Stream offline"

See Modules - Events for the full event-type reference, including each event's dataclass fields.

Testing Routes

The route_client fixture is an aiohttp.TestClient wired to the route dispatcher, and tests request routes the same way production clients do. A module with a status route:

@on_route("/status")
async def status(ctx: RouteContext) -> dict[str, Any]:
    return {"online": True}

A test for this route:

from aiohttp.test_utils import TestClient
from owlbot.api import ModuleContext


@pytest.mark.usefixtures("module_lifecycle")
async def test_status_route(
    module_context: ModuleContext,
    route_client: TestClient,
) -> None:
    resp = await route_client.get(
        f"/owlbot/{module_context.module_name}/status"
    )
    assert resp.status == 200
    data = await resp.json()
    assert data["online"] is True

Route URLs follow the /owlbot/<module_name>/<path> pattern used in production. Setting module_pkg also wires a real ModuleTemplates into module_context.templates, so route handlers that call ctx.templates.render(...) work without any extra setup.

Testing Protected Routes

The route_client fixture includes Owlbot's connect routes and keeps cookies between requests, so protected route tests can use the same flow as a browser. Dispatch a command or user-bearing event for a test user, redeem the generated session URL with route_client, then request the protected route:

from aiohttp.test_utils import TestClient
from owlbot.api import CommandContext, EventType, ModuleContext, RouteContext
from owlbot.testing import make_chat_event, make_user


async def test_protected_route(
    event_dispatcher,
    module_context: ModuleContext,
    route_client: TestClient,
) -> None:
    session_urls: list[str] = []

    async def handler(ctx: RouteContext) -> dict[str, str]:
        assert ctx.session is not None
        return {"user_id": ctx.session.user.id}

    async def send_link(ctx: CommandContext) -> None:
        session_urls.append(ctx.session_url_for("/secure"))

    module_context.routes.register(
        "/secure",
        handler,
        requires_authenticated=True,
    )
    module_context.commands.register("secure", send_link)

    event = make_chat_event(
        raw_body="!secure",
        user=make_user(id="alice", is_authenticated=True),
    )
    await event_dispatcher.dispatch(EventType.CHAT, event)

    session_url = session_urls[0]
    path = session_url.removeprefix(module_context.config.public_base_url)

    redeem = await route_client.get(path, allow_redirects=False)
    assert redeem.status == 302

    resp = await route_client.get("/owlbot/test_module/secure")
    assert resp.status == 200
    assert await resp.json() == {"user_id": "alice"}

Use make_user(is_authenticated=True) or make_user(is_moderator=True) to build users that pass the matching route guards. If you only need to assert that a protected page rejects an unconnected browser, request the route directly without redeeming a session URL first.

Using Storage

The storage fixture provides a real ModuleStorage backed by an in-memory SQLite database. It behaves identically to production storage but does not write to disk, so each test starts with a clean database. A module that creates its schema at startup:

@on_setup
async def setup(ctx: ModuleContext) -> None:
    await ctx.storage.execute("""
        CREATE TABLE items (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL
        )
    """)

A test applies module_lifecycle so setup creates the schema before the test body runs:

from owlbot.api import ModuleStorage


@pytest.mark.usefixtures("module_lifecycle")
async def test_items_schema(storage: ModuleStorage) -> None:
    await storage.execute("INSERT INTO items (name) VALUES (?)", ("first",))
    row = await storage.fetch_one("SELECT name FROM items WHERE id = 1")
    assert row is not None
    assert row["name"] == "first"

Using Module Config

The module_config fixture returns a real owlbot.api.config.ModuleConfig scoped to the test module. It is backed by a real Config loaded from a YAML file that the plugin writes to a per-test tmp_path directory, so get(), set(), register_defaults(), and as_dict() all exercise the same code path that production uses, and set() persists through to the file on disk.

The default seed sets owncast.url to http://localhost:8080 and owlbot.public_base_url to http://localhost:8081. Override the config_data fixture to change the seed (for example, to preload a modules.<name> section or a different public_base_url).

OWLBOT_* environment variables are cleared for the duration of each test so CI-level overrides do not bleed into the fixture.

from owlbot.api.config import ModuleConfig

async def test_reads_config(module_config: ModuleConfig) -> None:
    module_config.set("greeting", "Hi!")
    assert module_config.get("greeting") == "Hi!"

Testing HTTP Calls

Handlers that call ctx.http.get/post/put/delete/request are tested against aioresponses, which intercepts every outbound request at the aiohttp level. The plugin wires this in through two fixtures:

• http yields a real, started HttpClient (the same type as ctx.http in production).
• mocked_http yields an aioresponses controller that intercepts the client's requests.

Any test that uses http (or module_context, which depends on it) has interception active for the duration of the test. Requests to URLs you have not registered raise aiohttp.ClientConnectionError, so a forgotten mock cannot leak real network traffic. Loopback addresses (127.0.0.1, localhost) pass through unmocked so the route_client fixture can reach its own test server.

A module whose command fetches data from an external API:

@on_command("weather")
async def weather(ctx: CommandContext) -> None:
    resp = await ctx.http.get(f"https://api.example.com/weather/{ctx.args}")
    data = await resp.json()
    await ctx.owncast_client.send_message(f"Temp: {data['temp']}F")

A test registers a canned response on mocked_http before dispatching:

from aioresponses import aioresponses

from owlbot.api import EventType
from owlbot.registries.events import EventDispatcher
from owlbot.testing import RecordingOwncastClient, make_chat_event


@pytest.mark.usefixtures("module_lifecycle")
async def test_reports_temperature(
    event_dispatcher: EventDispatcher,
    owncast_client: RecordingOwncastClient,
    mocked_http: aioresponses,
) -> None:
    mocked_http.get(
        "https://api.example.com/weather/SF",
        payload={"temp": 72},
    )
    event = make_chat_event(raw_body="!weather SF")
    await event_dispatcher.dispatch(EventType.CHAT, event)
    assert owncast_client.calls[0].kwargs["body"] == "Temp: 72F"

See the aioresponses documentation for the full API: error simulation with status=, exception=, response headers, repeat matches, and regex URL patterns.

Simulating Owncast Responses

Recording stubs return empty defaults ("", [], {}) from every method. That is enough to assert "my handler called X" but not enough to test "my handler reacts correctly to what Owncast returned." Use pytest's monkeypatch fixture with AsyncMock to stub a method for one test, or override the owncast_client fixture for class-wide behavior. Pass side_effect= instead of return_value= to raise instead of returning (e.g., to simulate OwncastError).

Stubbing in a test

Patch the method on the recording stub before dispatching. The patch applies inside the test body, so dispatch calls made after it pick up the programmed value:

from unittest.mock import AsyncMock

from owlbot.api import EventType
from owlbot.testing import RecordingOwncastClient, make_chat_event


@pytest.mark.usefixtures("module_lifecycle")
async def test_status_reports_online(
    event_dispatcher,
    owncast_client: RecordingOwncastClient,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        owncast_client,
        "get_status",
        AsyncMock(return_value={"online": True}),
    )
    event = make_chat_event(raw_body="!status")
    await event_dispatcher.dispatch(EventType.CHAT, event)
    assert owncast_client.calls[0].kwargs["body"] == "Stream is live"

Stubbing values seen by @on_setup

@on_setup runs inside module_lifecycle's setup phase, before the test body. To stub a value that setup must observe, skip module_lifecycle and call the module's setup function directly with the patch applied first:

from owlbot.api import ModuleContext

from mymodule import setup


async def test_setup_caches_live_status(
    module_context: ModuleContext,
    owncast_client: RecordingOwncastClient,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        owncast_client,
        "get_status",
        AsyncMock(return_value={"online": True}),
    )
    await setup(module_context)
    assert module_context.state["online"] is True

If @on_setup starts async work (background tasks, timers), pair the setup call with a matching await teardown(module_context) (or module-specific shutdown) at the end of the test so it does not leak tasks.

Class-wide stubbing

When every test in a class needs the same canned behavior, override the owncast_client fixture with a subclass of RecordingOwncastClient. Grouping the tests in a class scopes the fixture override to those tests without affecting the rest of the suite:

from typing import Any

from owlbot.api import EventType
from owlbot.testing import RecordingOwncastClient, make_chat_event


class OnlineOwncastClient(RecordingOwncastClient):
    async def get_status(self) -> dict[str, Any]:
        return {"online": True}


@pytest.mark.usefixtures("module_lifecycle")
class TestStreamOnline:
    @pytest.fixture
    def owncast_client(self) -> OnlineOwncastClient:
        return OnlineOwncastClient()

    async def test_status_reports_live(
        self,
        event_dispatcher,
        owncast_client: OnlineOwncastClient,
    ) -> None:
        event = make_chat_event(raw_body="!status")
        await event_dispatcher.dispatch(EventType.CHAT, event)
        assert owncast_client.calls[0].kwargs["body"] == "Stream is live"

Reference

Factory, stub, fixture, and config reference material.

Fixture Reference

The plugin exposes two categories of fixtures: configuration fixtures that change defaults (override these in your test file) and individual fixtures your tests consume directly.

Configuration fixtures

Override these in your test file to change the defaults.

Fixture	Default	Description
module_pkg	None	Package of the module under test. Setting it auto-wires module_name, templates, and handler registration in module_lifecycle.
module_name	"test_module"	Canonical module name. Derived from module_pkg when set (last dotted segment). Override directly to set a name without importing a package.
admin_client_enabled	False	Whether the admin client is wired into the module context.
command_prefix	"!"	Prefix character for command parsing.
handler_timeout	30.0	Timeout (seconds) for handler execution.
config_data	{"owncast": {"url": "http://localhost:8080"}, "owlbot": {"public_base_url": "http://localhost:8081"}}	Seed YAML written to config_path before Config loads. Override to preload module sections, change public_base_url, etc.

Individual fixtures

Fixture	Type	Description
storage	ModuleStorage	In-memory SQLite database, closed automatically after the test.
templates	ModuleTemplates	Jinja2 renderer derived from module_pkg. Falls back to a core-only renderer when module_pkg is unset.
owncast_client	RecordingOwncastClient	Recording stub for the Owncast client.
admin_client	RecordingOwncastAdminClient	Recording stub for the Owncast admin client. Requires admin_client_enabled=True; raises pytest.UsageError otherwise.
config_path	Path	Location of the seed YAML file under tmp_path.
config	Config	Real Config loaded from config_path, with OWLBOT_* env vars cleared.
module_config	ModuleConfig	Real ModuleConfig scoped to module_name. set() persists to config_path.
http	HttpClient	Real HttpClient with requests intercepted by mocked_http.
mocked_http	aioresponses	Controller for registering canned HTTP responses.
module_context	ModuleContext	Fully wired context with storage, stubs, and dispatchers.
event_dispatcher	EventDispatcher	Dispatcher backing the context. Use this to fire events.
command_dispatcher	CommandDispatcher	Dispatcher backing the context.
route_dispatcher	RouteDispatcher	Dispatcher backing the context.
route_client	aiohttp.TestClient	HTTP test client for route handlers.
module_lifecycle	None	Registers every @on_command, @on_event, and @on_route handler from module_pkg, runs every @on_setup hook, and runs every @on_teardown hook on test exit. Requires module_pkg. Apply via @pytest.mark.usefixtures on integration tests.
registered_handlers	None	Handler registration only, without running @on_setup/@on_teardown. Use when setup has side effects the test cannot tolerate; prefer module_lifecycle otherwise.

Event Factories

Factory functions build event and user objects with sensible defaults. All parameters are keyword-only.

make_user

Parameter	Default	Description
id	"test-user-id"	User identifier.
display_name	"TestUser"	Display name shown in chat.
display_color	0	Color index.
created_at	None	Account creation timestamp.
previous_names	[]	Previous display names.
name_changed_at	None	Timestamp of last name change.
is_bot	False	Whether the user is a bot.
is_authenticated	False	Whether the user is authenticated.
is_moderator	False	When True, adds "MODERATOR" to scopes.
scopes	frozenset()	Explicit scope set. Unioned with moderator scope if both given.

Available Factories

Function	Returns
make_chat_event	ChatEvent
make_user_joined_event	UserJoinedEvent
make_user_parted_event	UserPartedEvent
make_name_changed_event	NameChangedEvent
make_stream_started_event	StreamStartedEvent
make_stream_stopped_event	StreamStoppedEvent
make_stream_status	StreamStatus
make_stream_title_updated_event	StreamTitleUpdatedEvent
make_visibility_update_event	VisibilityUpdateEvent

A few things to note:

• timestamp: every event factory accepts one (default None). make_stream_status is the exception; it's a value type used to populate StreamTitleUpdatedEvent.status and does not carry a timestamp.
• raw_body vs body in make_chat_event: raw_body is the primary text parameter because that is what the command dispatcher parses. body defaults to a literal copy of raw_body and is not Owncast-rendered HTML, so tests that exercise the rendered form (HTML-stripped text, rendered emotes, etc.) must pass body= explicitly.
• is_visible defaults: make_chat_event defaults to True (chat messages are normally shown); make_visibility_update_event defaults to False (visibility updates are primarily used to hide messages).

See Modules - Events for the full event dataclass field reference.

Recording Stubs

Recording stubs mirror the real client method surfaces. Each public async method appends a RecordedCall(name, kwargs) to calls and returns a type-appropriate empty default ("" for strings, {} for dicts, [] for lists). Assertions usually read the fields by name:

from owlbot.testing import RecordingOwncastClient

async def test_sends_message(
    owncast_client: RecordingOwncastClient,
) -> None:
    await owncast_client.send_message("hello")
    assert owncast_client.calls[0].name == "send_message"
    assert owncast_client.calls[0].kwargs["body"] == "hello"

RecordedCall is a NamedTuple, so it also compares equal to a plain (name, kwargs) tuple if you want to assert both fields in one line:

assert owncast_client.calls[0] == ("send_message", {"body": "hello", "unsanitized": False})

RecordingOwncastClient

Method	Description
base_url	Property returning the base URL passed at construction (default "http://localhost:8080"; seeded from config.owncast_url by the default owncast_client fixture).
get_status()	Returns {}.
send_message(body, *, unsanitized=False)	Returns "".
send_system_message(body, *, unsanitized=False)	Returns "".
send_action(body, *, unsanitized=False)	Returns "".
send_system_message_to_client(client_id, body, *, unsanitized=False)	Returns "".
set_message_visibility(message_ids, *, visible)	Returns "".
get_chat_history()	Returns [].
get_connected_clients()	Returns [].
set_stream_title(title)	Returns "".

RecordingOwncastAdminClient

RecordingOwncastAdminClient mirrors the full OwncastAdminClient surface with the same recording behavior. See Modules - Owncast API for the method list.

• Home
• Configuration
• Command Reference
• Logging

---

Built-in Modules

• Clips
• Custom Commands
• Emoji Wall
• Polls
• Quotes
• Timers

---

Creating New Modules

• Overview
• Lifecycle
• Context
• Events
• Commands
• Routes
• Storage
• Configuration
• Owncast API
• HTTP Client
• Logger
• Testing