Added module testing documentation.

2026-04-22 21:48:55 -04:00
parent 6e1596f3a6
commit eb0d4e37b5
3 changed files with 540 additions and 1 deletions
+534
@@ -0,0 +1,534 @@
# 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:
```bash
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`:
```python
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`.
```toml
[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:
```python
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):
```python
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:
```python
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:
```python
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:
```python
@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:
```python
@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:
```python
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:
```python
@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:
```python
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:
```python
@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:
```python
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](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:
```python
@on_route("/status")
async def status(ctx: RouteContext) -> dict[str, Any]:
return {"online": True}
```
A test for this route:
```python
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.
### 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:
```python
@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:
```python
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.
```python
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`](https://github.com/pnuckowski/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:
```python
@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:
```python
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](https://github.com/pnuckowski/aioresponses) 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`](https://docs.pytest.org/en/stable/how-to/monkeypatch.html) fixture with [`AsyncMock`](https://docs.python.org/3/library/unittest.mock.html#unittest.mock.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:
```python
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:
```python
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:
```python
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](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:
```python
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:
```python
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](Modules-Owncast-API#admin-api-ctxadmin_client) for the method list.
+4
@@ -249,3 +249,7 @@ async def quotes_page(ctx: RouteContext) -> web.Response:
``` ```
For more complex HTML, use `ctx.templates.render()` with Jinja2 templates from a package module's `templates/` directory. Templates can extend Owlbot's built-in `base.html` for a Bootstrap layout. See [HTTP Routes](Modules-Routes) for details. For more complex HTML, use `ctx.templates.render()` with Jinja2 templates from a package module's `templates/` directory. Templates can extend Owlbot's built-in `base.html` for a Bootstrap layout. See [HTTP Routes](Modules-Routes) for details.
## Testing
Owlbot includes a pytest plugin with fixtures and helpers for testing modules without a running Owncast server. Tests run against the real dispatch pipeline with in-memory storage and recording stubs. See [Testing](Modules-Testing) for setup, available fixtures, and examples.
+1
@@ -27,3 +27,4 @@
* [Owncast API](Modules-Owncast-API) * [Owncast API](Modules-Owncast-API)
* [HTTP Client](Modules-HTTP-Client) * [HTTP Client](Modules-HTTP-Client)
* [Logger](Modules-Logger) * [Logger](Modules-Logger)
* [Testing](Modules-Testing)