cottongin/NtR-soudcloud-fetcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: add IRC bot plugins design and implementation plan

Browse Source 
Made-with: Cursor
This commit is contained in:
cottongin
2026-03-12 03:20:45 -04:00
parent 5c227766f1
commit b63c851d14
2 changed files with 915 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
docs/plans/2026-03-12-irc-plugins-design.md Normal file
View File

@@ -0,0 +1,157 @@
# NtR IRC Bot Plugins — Design Document
> **Date**: 2026-03-12
> **Status**: Approved
## Purpose
Two functionally-identical IRC bot plugins (Sopel and Limnoria) that query the NtR SoundCloud Fetcher API to serve playlist and track information in IRC channels.
## Approach
Fully independent plugins with no shared code. Each plugin is a self-contained unit you drop into the bot's plugin directory. Both follow identical internal structure and function naming so a diff between them shows only framework-specific glue.
## Directory Layout
```
plugins/
├── sopel/
│ └── ntr_playlist.py # Single-file Sopel plugin
└── limnoria/
└── NtrPlaylist/ # Limnoria package plugin
├── __init__.py # Plugin metadata
├── config.py # Registry values
├── plugin.py # Command handlers
└── test.py # Test stub
```
Sopel: copy `ntr_playlist.py` into `~/.sopel/plugins/`.
Limnoria: copy `NtrPlaylist/` into `limnoria-bot/plugins/`.
## Commands
| Trigger | API Call | Auth | Output |
|---------|----------|------|--------|
| `!1`, `!2`, ... `!N` | `GET /playlist/{n}` | None | Single track |
| `!song <episode> <position>` | `GET /shows/by-episode/{episode}` → filter by position | None | Single track |
| `!refresh` | `POST /admin/refresh` (bearer token) | Admin nicks | Refresh result |
| `!status` | `GET /health` | None | Full health info |
| `!playlist` | `GET /playlist` | None | Comma-separated track list |
| `!playlist <episode>` | `GET /shows/by-episode/{episode}` | None | Comma-separated track list |
### Number commands (`!1`, `!2`, etc.)
Neither framework natively registers bare-digit commands. Both plugins use a regex/rule pattern to match messages starting with `!` followed by one or more digits and nothing else.
- Sopel: `@plugin.rule(r'^!(\d+)$')`
- Limnoria: `doPrivmsg` override checking against `r'^!(\d+)$'`
### Output Formats
**Single track** (`!N`, `!song`):
```
Song #3: Night Drive by SomeArtist - https://soundcloud.com/someartist/night-drive
```
**Playlist** (`!playlist`):
```
Episode 530 (9 tracks): Night Drive by SomeArtist, Running Through My Mind by Purrple Panther, ...
```
**Status** (`!status`):
```
Status: OK | Poller: alive | Last fetch: 2026-03-12T02:00:00+00:00 | Tracks this week: 9
```
**Refresh** (`!refresh`):
```
Refreshed — 9 tracks
```
**Errors** (pass through API detail):
```
No track at position 15
```
## Configuration
Both plugins expose the same three settings:
| Setting | Type | Default | Description |
|---------|------|---------|-------------|
| `api_base_url` | string | `http://127.0.0.1:8000` | NtR API base URL (no trailing slash) |
| `admin_token` | string | *(empty)* | Bearer token for `POST /admin/refresh` |
| `admin_nicks` | list of strings | *(empty)* | IRC nicknames allowed to run `!refresh` |
### Sopel (INI)
```ini
[ntr_playlist]
api_base_url = http://127.0.0.1:8000
admin_token = secret-token-here
admin_nicks =
NicktheRat
SomeOtherAdmin
```
Defined via `StaticSection` with `ValidatedAttribute` and `ListAttribute`.
### Limnoria (registry)
```
config plugins.NtrPlaylist.apiBaseUrl http://127.0.0.1:8000
config plugins.NtrPlaylist.adminToken secret-token-here
config plugins.NtrPlaylist.adminNicks NicktheRat SomeOtherAdmin
```
`adminToken` marked `private=True`. `adminNicks` uses `SpaceSeparatedListOfStrings`.
### Admin Check
Compare the triggering user's IRC nickname against `admin_nicks`, case-insensitive. Framework-level admin/owner systems are not used. This keeps admin management identical and portable between plugins.
## HTTP Client
Both plugins use `urllib.request` from stdlib — no external dependencies beyond the bot framework.
### Internal Helpers
**`_api_get(base_url, path)`** → `dict`
- `GET {base_url}{path}` with `Accept: application/json`
- 10-second timeout
- On success: parse and return JSON
- On HTTP error: parse JSON error body, raise `ApiError(status_code, detail)`
- On connection failure: raise `ApiError(0, "Cannot reach API")`
**`_api_post(base_url, path, token, body=None)`** → `dict`
- Same as above but `POST` with `Authorization: Bearer {token}` and `Content-Type: application/json`
- Used only by `!refresh`
### Error Handling
```
try:
data = _api_get(base_url, f"/playlist/{position}")
reply with formatted track
except ApiError as e:
reply with e.detail
```
No retries at the plugin level. The API is local; IRC users expect immediate responses. If the API is down, the user sees "Cannot reach API".
No async concerns — both frameworks run command handlers synchronously. A blocking 10-second timeout is acceptable.
## Design Decisions
| Decision | Rationale |
|----------|-----------|
| Independent plugins, no shared code | Each plugin is one self-contained directory. No deployment headaches with shared imports. |
| Regex for `!N` commands | Frameworks don't support bare-digit command names. Regex catch-all is the cleanest solution. |
| Plugin-config admin nicks only | Portable between frameworks. No coupling to Sopel's `core.admins` or Limnoria's capability system. |
| No caching | API is local, responses are fast. Simplifies code and guarantees fresh data. |
| stdlib `urllib.request` | Zero external dependencies. Both frameworks already have Python 3.11+. |
| `!song` fetches full show, filters client-side | API has no single-track-by-episode endpoint. The response is small enough that client-side filtering is fine. |
## Dependencies
None beyond the bot framework itself and Python 3.11+ stdlib.

758
docs/plans/2026-03-12-irc-plugins-implementation.md Normal file
View File

@@ -0,0 +1,758 @@
# IRC Bot Plugins Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Build two functionally-identical IRC bot plugins (Sopel and Limnoria) that query the NtR SoundCloud Fetcher API.
**Architecture:** Fully independent plugins, no shared code. Each plugin has its own API client (`urllib.request`), formatting functions, and admin-nick check. Sopel is a single-file plugin; Limnoria is a four-file package. Both produce identical IRC output.
**Tech Stack:** Python 3.11+ stdlib only (`urllib.request`, `json`, `re`). Sopel 7.0+ API. Limnoria 2025.07.18+ API. pytest for helper tests.
**Design doc:** `docs/plans/2026-03-12-irc-plugins-design.md`
**API reference:** `docs/api.md`
---
### Task 1: Sopel Plugin — Complete Implementation
**Files:**
- Create: `plugins/sopel/ntr_playlist.py`
**Step 1: Create the plugin file**
Create `plugins/sopel/ntr_playlist.py` with the full contents below. This is a single-file Sopel plugin covering config, API helpers, formatting, admin check, and all five commands.
```python
"""NtR Playlist — Sopel plugin for NicktheRat SoundCloud playlists."""
import json
import urllib.error
import urllib.request
from sopel import plugin
from sopel.config import types
# --- Configuration -----------------------------------------------------------
class NtrPlaylistSection(types.StaticSection):
api_base_url = types.ValidatedAttribute(
"api_base_url", str, default="http://127.0.0.1:8000",
)
admin_token = types.ValidatedAttribute("admin_token", str, default="")
admin_nicks = types.ListAttribute("admin_nicks")
def setup(bot):
bot.settings.define_section("ntr_playlist", NtrPlaylistSection)
def configure(config):
config.define_section("ntr_playlist", NtrPlaylistSection, validate=False)
# --- API helpers -------------------------------------------------------------
class ApiError(Exception):
def __init__(self, status_code: int, detail: str):
self.status_code = status_code
self.detail = detail
super().__init__(detail)
def _api_get(base_url: str, path: str) -> dict:
req = urllib.request.Request(
f"{base_url}{path}",
headers={"Accept": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
try:
body = json.loads(exc.read().decode())
detail = body.get("detail", str(exc))
except Exception:
detail = str(exc)
raise ApiError(exc.code, detail) from exc
except urllib.error.URLError as exc:
raise ApiError(0, "Cannot reach API") from exc
def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
req = urllib.request.Request(
f"{base_url}{path}",
data=json.dumps(body).encode() if body else None,
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
try:
body = json.loads(exc.read().decode())
detail = body.get("detail", str(exc))
except Exception:
detail = str(exc)
raise ApiError(exc.code, detail) from exc
except urllib.error.URLError as exc:
raise ApiError(0, "Cannot reach API") from exc
# --- Formatting --------------------------------------------------------------
def format_track(track: dict) -> str:
return (
f"Song #{track['position']}: {track['title']} "
f"by {track['artist']} - {track['permalink_url']}"
)
def format_playlist(data: dict) -> str:
ep = data.get("episode_number", "?")
tracks = data.get("tracks", [])
items = ", ".join(f"{t['title']} by {t['artist']}" for t in tracks)
return f"Episode {ep} ({len(tracks)} tracks): {items}"
# --- Admin check -------------------------------------------------------------
def _is_admin(bot, nick: str) -> bool:
nicks = bot.settings.ntr_playlist.admin_nicks or []
return nick.lower() in [n.lower() for n in nicks]
# --- Commands ----------------------------------------------------------------
@plugin.rule(r"^!(\d+)$")
def track_by_number(bot, trigger):
"""Fetch a track by position number (!1, !2, etc.)."""
position = trigger.group(1)
base_url = bot.settings.ntr_playlist.api_base_url
try:
data = _api_get(base_url, f"/playlist/{position}")
bot.say(format_track(data))
except ApiError as exc:
bot.say(exc.detail)
@plugin.command("song")
def song(bot, trigger):
"""Fetch a track from a specific episode: !song <episode> <position>."""
raw = trigger.group(2)
if not raw:
bot.say("Usage: !song <episode> <position>")
return
parts = raw.strip().split()
if len(parts) != 2:
bot.say("Usage: !song <episode> <position>")
return
try:
episode, position = int(parts[0]), int(parts[1])
except ValueError:
bot.say("Usage: !song <episode> <position>")
return
base_url = bot.settings.ntr_playlist.api_base_url
try:
data = _api_get(base_url, f"/shows/by-episode/{episode}")
track = next(
(t for t in data.get("tracks", []) if t["position"] == position),
None,
)
if track is None:
bot.say(f"No track at position {position} in episode {episode}")
return
bot.say(format_track(track))
except ApiError as exc:
bot.say(exc.detail)
@plugin.command("playlist")
def playlist_cmd(bot, trigger):
"""Current playlist, or a specific episode: !playlist [episode]."""
raw = trigger.group(2)
base_url = bot.settings.ntr_playlist.api_base_url
try:
if raw and raw.strip():
episode = int(raw.strip())
data = _api_get(base_url, f"/shows/by-episode/{episode}")
else:
data = _api_get(base_url, "/playlist")
bot.say(format_playlist(data))
except ValueError:
bot.say("Usage: !playlist [episode]")
except ApiError as exc:
bot.say(exc.detail)
@plugin.command("status")
def status_cmd(bot, trigger):
"""Show API health status."""
base_url = bot.settings.ntr_playlist.api_base_url
try:
data = _api_get(base_url, "/health")
poller = "alive" if data.get("poller_alive") else "dead"
last_fetch = data.get("last_fetch") or "never"
count = data.get("current_week_track_count", 0)
bot.say(
f"Status: {data['status'].upper()} | Poller: {poller} "
f"| Last fetch: {last_fetch} | Tracks this week: {count}"
)
except ApiError as exc:
bot.say(exc.detail)
@plugin.command("refresh")
def refresh_cmd(bot, trigger):
"""Manually refresh the playlist (admin only)."""
if not _is_admin(bot, trigger.nick):
bot.say("You don't have permission to use this command.")
return
base_url = bot.settings.ntr_playlist.api_base_url
token = bot.settings.ntr_playlist.admin_token
if not token:
bot.say("Admin token not configured.")
return
try:
data = _api_post(base_url, "/admin/refresh", token)
count = data.get("track_count", "?")
bot.say(f"Refreshed \u2014 {count} tracks")
except ApiError as exc:
bot.say(f"Refresh failed: {exc.detail}")
```
**Step 2: Verify syntax**
Run: `python -c "import ast; ast.parse(open('plugins/sopel/ntr_playlist.py').read()); print('OK')"`
Expected: `OK`
**Step 3: Commit**
```bash
git add plugins/sopel/ntr_playlist.py
git commit -m "feat(sopel): add NtR playlist IRC plugin"
```
---
### Task 2: Limnoria Plugin — Scaffold
**Files:**
- Create: `plugins/limnoria/NtrPlaylist/__init__.py`
- Create: `plugins/limnoria/NtrPlaylist/config.py`
- Create: `plugins/limnoria/NtrPlaylist/test.py`
**Step 1: Create `__init__.py`**
```python
"""NtR Playlist — Limnoria plugin for NicktheRat SoundCloud playlists."""
import importlib
from . import config, plugin
importlib.reload(config)
importlib.reload(plugin)
Class = plugin.Class
configure = config.configure
__version__ = "0.1.0"
__author__ = "NtR SoundCloud Fetcher"
```
**Step 2: Create `config.py`**
```python
from supybot import conf, registry
def configure(advanced):
conf.registerPlugin("NtrPlaylist", True)
NtrPlaylist = conf.registerPlugin("NtrPlaylist")
conf.registerGlobalValue(
NtrPlaylist,
"apiBaseUrl",
registry.String(
"http://127.0.0.1:8000",
"""Base URL for the NtR SoundCloud Fetcher API (no trailing slash).""",
),
)
conf.registerGlobalValue(
NtrPlaylist,
"adminToken",
registry.String(
"",
"""Bearer token for admin API endpoints.""",
private=True,
),
)
conf.registerGlobalValue(
NtrPlaylist,
"adminNicks",
registry.SpaceSeparatedListOfStrings(
[],
"""IRC nicknames allowed to run admin commands (space-separated).""",
),
)
```
**Step 3: Create `test.py`**
```python
from supybot.test import PluginTestCase
class NtrPlaylistTestCase(PluginTestCase):
plugins = ("NtrPlaylist",)
```
**Step 4: Verify syntax on all three files**
Run: `for f in plugins/limnoria/NtrPlaylist/__init__.py plugins/limnoria/NtrPlaylist/config.py plugins/limnoria/NtrPlaylist/test.py; do python -c "import ast; ast.parse(open('$f').read()); print('$f OK')"; done`
Expected: all three print OK
**Step 5: Commit**
```bash
git add plugins/limnoria/NtrPlaylist/
git commit -m "feat(limnoria): scaffold NtrPlaylist plugin package"
```
---
### Task 3: Limnoria Plugin — Commands
**Files:**
- Create: `plugins/limnoria/NtrPlaylist/plugin.py`
**Step 1: Create `plugin.py`**
This file contains the API helpers, formatting functions, and all command handlers — functionally identical to the Sopel plugin.
```python
"""NtR Playlist — Limnoria command handlers."""
import json
import re
import urllib.error
import urllib.request
from supybot import callbacks
from supybot.commands import optional, wrap
# --- API helpers -------------------------------------------------------------
class ApiError(Exception):
def __init__(self, status_code: int, detail: str):
self.status_code = status_code
self.detail = detail
super().__init__(detail)
def _api_get(base_url: str, path: str) -> dict:
req = urllib.request.Request(
f"{base_url}{path}",
headers={"Accept": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
try:
body = json.loads(exc.read().decode())
detail = body.get("detail", str(exc))
except Exception:
detail = str(exc)
raise ApiError(exc.code, detail) from exc
except urllib.error.URLError as exc:
raise ApiError(0, "Cannot reach API") from exc
def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
req = urllib.request.Request(
f"{base_url}{path}",
data=json.dumps(body).encode() if body else None,
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
try:
body = json.loads(exc.read().decode())
detail = body.get("detail", str(exc))
except Exception:
detail = str(exc)
raise ApiError(exc.code, detail) from exc
except urllib.error.URLError as exc:
raise ApiError(0, "Cannot reach API") from exc
# --- Formatting --------------------------------------------------------------
def format_track(track: dict) -> str:
return (
f"Song #{track['position']}: {track['title']} "
f"by {track['artist']} - {track['permalink_url']}"
)
def format_playlist(data: dict) -> str:
ep = data.get("episode_number", "?")
tracks = data.get("tracks", [])
items = ", ".join(f"{t['title']} by {t['artist']}" for t in tracks)
return f"Episode {ep} ({len(tracks)} tracks): {items}"
# --- Plugin ------------------------------------------------------------------
_NUMBER_RE = re.compile(r"^!(\d+)$")
class NtrPlaylist(callbacks.Plugin):
"""Query the NtR SoundCloud Fetcher API from IRC."""
def _is_admin(self, nick: str) -> bool:
nicks = self.registryValue("adminNicks")
return nick.lower() in [n.lower() for n in nicks]
def doPrivmsg(self, irc, msg):
channel = msg.args[0] if msg.args else None
if not channel or not irc.isChannel(channel):
return
text = msg.args[1] if len(msg.args) > 1 else ""
match = _NUMBER_RE.match(text)
if not match:
return
position = match.group(1)
base_url = self.registryValue("apiBaseUrl")
try:
data = _api_get(base_url, f"/playlist/{position}")
irc.reply(format_track(data))
except ApiError as exc:
irc.reply(exc.detail)
@wrap(["int", "int"])
def song(self, irc, msg, args, episode, position):
"""<episode> <position>
Returns a track from a specific episode's playlist.
"""
base_url = self.registryValue("apiBaseUrl")
try:
data = _api_get(base_url, f"/shows/by-episode/{episode}")
track = next(
(t for t in data.get("tracks", []) if t["position"] == position),
None,
)
if track is None:
irc.reply(f"No track at position {position} in episode {episode}")
return
irc.reply(format_track(track))
except ApiError as exc:
irc.reply(exc.detail)
@wrap([optional("int")])
def playlist(self, irc, msg, args, episode):
"""[<episode>]
Returns the playlist for the current show, or a specific episode.
"""
base_url = self.registryValue("apiBaseUrl")
try:
if episode is not None:
data = _api_get(base_url, f"/shows/by-episode/{episode}")
else:
data = _api_get(base_url, "/playlist")
irc.reply(format_playlist(data))
except ApiError as exc:
irc.reply(exc.detail)
@wrap
def status(self, irc, msg, args):
"""takes no arguments
Returns the current API status.
"""
base_url = self.registryValue("apiBaseUrl")
try:
data = _api_get(base_url, "/health")
poller = "alive" if data.get("poller_alive") else "dead"
last_fetch = data.get("last_fetch") or "never"
count = data.get("current_week_track_count", 0)
irc.reply(
f"Status: {data['status'].upper()} | Poller: {poller} "
f"| Last fetch: {last_fetch} | Tracks this week: {count}"
)
except ApiError as exc:
irc.reply(exc.detail)
@wrap
def refresh(self, irc, msg, args):
"""takes no arguments
Triggers a manual playlist refresh. Admin only.
"""
if not self._is_admin(msg.nick):
irc.reply("You don't have permission to use this command.")
return
base_url = self.registryValue("apiBaseUrl")
token = self.registryValue("adminToken")
if not token:
irc.reply("Admin token not configured.")
return
try:
data = _api_post(base_url, "/admin/refresh", token)
count = data.get("track_count", "?")
irc.reply(f"Refreshed \u2014 {count} tracks")
except ApiError as exc:
irc.reply(f"Refresh failed: {exc.detail}")
Class = NtrPlaylist
```
**Step 2: Verify syntax**
Run: `python -c "import ast; ast.parse(open('plugins/limnoria/NtrPlaylist/plugin.py').read()); print('OK')"`
Expected: `OK`
**Step 3: Commit**
```bash
git add plugins/limnoria/NtrPlaylist/plugin.py
git commit -m "feat(limnoria): add NtrPlaylist command handlers"
```
---
### Task 4: Tests for Formatting and Admin Logic
Both plugins have identical formatting functions. We test the logic once using the Sopel plugin's module (it's a plain Python file with no import-time side effects from Sopel beyond the decorators, which we can handle).
Since importing the Sopel plugin requires `sopel` to be installed (due to top-level imports), we write standalone tests that duplicate the pure functions. This avoids adding `sopel` or `limnoria` as test dependencies for the main project.
**Files:**
- Create: `tests/test_plugin_helpers.py`
**Step 1: Write the tests**
```python
"""Tests for IRC plugin helper functions (formatting, admin check, API errors).
These test the pure logic shared across both Sopel and Limnoria plugins.
The functions are duplicated here to avoid importing framework-dependent modules.
"""
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
import pytest
# --- Duplicated pure functions under test ------------------------------------
# Kept in sync with plugins/sopel/ntr_playlist.py and
# plugins/limnoria/NtrPlaylist/plugin.py
import urllib.error
import urllib.request
class ApiError(Exception):
def __init__(self, status_code: int, detail: str):
self.status_code = status_code
self.detail = detail
super().__init__(detail)
def _api_get(base_url: str, path: str) -> dict:
req = urllib.request.Request(
f"{base_url}{path}",
headers={"Accept": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
try:
body = json.loads(exc.read().decode())
detail = body.get("detail", str(exc))
except Exception:
detail = str(exc)
raise ApiError(exc.code, detail) from exc
except urllib.error.URLError as exc:
raise ApiError(0, "Cannot reach API") from exc
def format_track(track: dict) -> str:
return (
f"Song #{track['position']}: {track['title']} "
f"by {track['artist']} - {track['permalink_url']}"
)
def format_playlist(data: dict) -> str:
ep = data.get("episode_number", "?")
tracks = data.get("tracks", [])
items = ", ".join(f"{t['title']} by {t['artist']}" for t in tracks)
return f"Episode {ep} ({len(tracks)} tracks): {items}"
# --- format_track tests -----------------------------------------------------
def test_format_track_basic():
track = {
"position": 3,
"title": "Night Drive",
"artist": "SomeArtist",
"permalink_url": "https://soundcloud.com/someartist/night-drive",
}
assert format_track(track) == (
"Song #3: Night Drive by SomeArtist "
"- https://soundcloud.com/someartist/night-drive"
)
def test_format_track_position_1():
track = {
"position": 1,
"title": "A",
"artist": "B",
"permalink_url": "https://example.com",
}
result = format_track(track)
assert result.startswith("Song #1:")
# --- format_playlist tests --------------------------------------------------
def test_format_playlist_with_tracks():
data = {
"episode_number": 530,
"tracks": [
{"title": "Night Drive", "artist": "SomeArtist"},
{"title": "Running", "artist": "Purrple Panther"},
],
}
result = format_playlist(data)
assert result == (
"Episode 530 (2 tracks): "
"Night Drive by SomeArtist, Running by Purrple Panther"
)
def test_format_playlist_empty():
data = {"episode_number": 530, "tracks": []}
assert format_playlist(data) == "Episode 530 (0 tracks): "
def test_format_playlist_missing_episode():
data = {"tracks": [{"title": "A", "artist": "B"}]}
result = format_playlist(data)
assert result.startswith("Episode ?")
# --- _api_get tests (with a real HTTP server) --------------------------------
class _Handler(BaseHTTPRequestHandler):
response_code = 200
response_body = "{}"
def do_GET(self):
self.send_response(self.response_code)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(self.response_body.encode())
def log_message(self, *_args):
pass # silence logs
@pytest.fixture()
def mock_server():
server = HTTPServer(("127.0.0.1", 0), _Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
host, port = server.server_address
yield f"http://{host}:{port}", server
server.shutdown()
def test_api_get_success(mock_server):
base_url, srv = mock_server
_Handler.response_code = 200
_Handler.response_body = json.dumps({"title": "ok"})
result = _api_get(base_url, "/test")
assert result == {"title": "ok"}
def test_api_get_404(mock_server):
base_url, srv = mock_server
_Handler.response_code = 404
_Handler.response_body = json.dumps({"detail": "Not found"})
with pytest.raises(ApiError) as exc_info:
_api_get(base_url, "/missing")
assert exc_info.value.status_code == 404
assert exc_info.value.detail == "Not found"
def test_api_get_unreachable():
with pytest.raises(ApiError) as exc_info:
_api_get("http://127.0.0.1:1", "/nope")
assert exc_info.value.status_code == 0
assert exc_info.value.detail == "Cannot reach API"
```
**Step 2: Run the tests**
Run: `pytest tests/test_plugin_helpers.py -v`
Expected: all tests pass
**Step 3: Commit**
```bash
git add tests/test_plugin_helpers.py
git commit -m "test: add tests for IRC plugin formatting and API helpers"
```
---
### Task 5: Final Verification and Docs
**Step 1: Run full test suite**
Run: `pytest -v`
Expected: all tests pass (existing + new)
**Step 2: Run ruff on plugin files**
Run: `ruff check plugins/ tests/test_plugin_helpers.py`
Expected: no errors (or only pre-existing ones outside these files)
**Step 3: Commit any fixes from linting**
If ruff found issues, fix and commit:
```bash
git add -u
git commit -m "style: fix lint issues in IRC plugins"
```