diff --git a/tests/test_plugin_helpers.py b/tests/test_plugin_helpers.py new file mode 100644 index 0000000..e891692 --- /dev/null +++ b/tests/test_plugin_helpers.py @@ -0,0 +1,256 @@ +"""Tests for IRC plugin helper functions. + +The pure logic (formatting, API client, admin check) is duplicated across both +plugins. We duplicate the functions here to test without framework dependencies. +""" + +import json +from http.server import BaseHTTPRequestHandler, HTTPServer +from threading import Thread +import urllib.error +import urllib.request + +import pytest + + +# --------------------------------------------------------------------------- +# Duplicated pure functions under test +# Keep in sync with plugins/sopel/ntr_playlist.py and +# plugins/limnoria/NtrPlaylist/plugin.py +# --------------------------------------------------------------------------- + +class ApiError(Exception): + def __init__(self, status_code: int, detail: str): + self.status_code = status_code + self.detail = detail + super().__init__(f"{status_code}: {detail}") + + +def _api_get(base_url: str, path: str) -> dict: + url = f"{base_url.rstrip("/")}{path}" + req = urllib.request.Request(url, headers={"Accept": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + raw = resp.read().decode() + try: + return json.loads(raw) + except (json.JSONDecodeError, ValueError) as exc: + raise ApiError(resp.status, "Invalid API response") from exc + except urllib.error.HTTPError as e: + try: + detail = json.loads(e.read().decode()).get("detail", str(e)) + except (json.JSONDecodeError, ValueError): + detail = str(e) + raise ApiError(e.code, detail) from e + except urllib.error.URLError as e: + raise ApiError(0, "Cannot reach API") from e + + +def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict: + url = f"{base_url.rstrip("/")}{path}" + encoded = None + headers = {"Accept": "application/json", "Authorization": f"Bearer {token}"} + if body is not None: + encoded = json.dumps(body).encode() + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=encoded, headers=headers, method="POST") + try: + with urllib.request.urlopen(req, timeout=10) as resp: + raw = resp.read().decode() + try: + return json.loads(raw) + except (json.JSONDecodeError, ValueError) as exc: + raise ApiError(resp.status, "Invalid API response") from exc + except urllib.error.HTTPError as e: + try: + detail = json.loads(e.read().decode()).get("detail", str(e)) + except (json.JSONDecodeError, ValueError): + detail = str(e) + raise ApiError(e.code, detail) from e + except urllib.error.URLError as e: + raise ApiError(0, "Cannot reach API") from e + + +_MAX_IRC_LINE = 430 + + +def format_track(track: dict) -> str: + pos = track.get("position", 0) + title = track.get("title", "") + artist = track.get("artist", "") + url = track.get("permalink_url", "") + return f"Song #{pos}: {title} by {artist} - {url}" + + +def format_playlist(data: dict) -> str: + episode = data.get("episode_number", "?") + tracks = data.get("tracks", []) + count = len(tracks) + prefix = f"Episode {episode} ({count} tracks): " + parts: list[str] = [] + length = len(prefix) + for t in tracks: + entry = f"{t.get("title", "")} by {t.get("artist", "")}" + sep = ", " if parts else "" + if length + len(sep) + len(entry) + 4 > _MAX_IRC_LINE: + parts.append("...") + break + parts.append(entry) + length += len(sep) + len(entry) + return prefix + ", ".join(parts) + + +# --------------------------------------------------------------------------- +# Test HTTP server +# --------------------------------------------------------------------------- + +class _Handler(BaseHTTPRequestHandler): + response_code = 200 + response_body = "{}" + last_headers: dict = {} + + def do_GET(self): + self._respond() + + def do_POST(self): + _Handler.last_headers = dict(self.headers) + # Consume request body so client connection stays valid + content_length = int(self.headers.get("Content-Length", 0)) + if content_length: + self.rfile.read(content_length) + self._respond() + + def _respond(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 + + +@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() + + +# --------------------------------------------------------------------------- +# format_track +# --------------------------------------------------------------------------- + +class TestFormatTrack: + def test_basic(self): + 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_position_1(self): + track = { + "position": 1, + "title": "A", + "artist": "B", + "permalink_url": "https://example.com", + } + assert format_track(track).startswith("Song #1:") + + def test_missing_fields_uses_defaults(self): + result = format_track({}) + assert result == "Song #0: by - " + + +# --------------------------------------------------------------------------- +# format_playlist +# --------------------------------------------------------------------------- + +class TestFormatPlaylist: + def test_with_tracks(self): + data = { + "episode_number": 530, + "tracks": [ + {"title": "Night Drive", "artist": "SomeArtist"}, + {"title": "Running", "artist": "Purrple Panther"}, + ], + } + assert format_playlist(data) == ( + "Episode 530 (2 tracks): " + "Night Drive by SomeArtist, Running by Purrple Panther" + ) + + def test_empty_tracks(self): + data = {"episode_number": 530, "tracks": []} + assert format_playlist(data) == "Episode 530 (0 tracks): " + + def test_missing_episode(self): + data = {"tracks": [{"title": "A", "artist": "B"}]} + assert format_playlist(data).startswith("Episode ?") + + def test_truncation(self): + tracks = [ + {"title": f"Track{i:03d} With A Longer Name", "artist": f"Artist{i:03d}"} + for i in range(50) + ] + data = {"episode_number": 999, "tracks": tracks} + result = format_playlist(data) + assert len(result) <= _MAX_IRC_LINE + assert result.endswith("...") + + +# --------------------------------------------------------------------------- +# _api_get +# --------------------------------------------------------------------------- + +class TestApiGet: + def test_success(self, mock_server): + base_url, _ = mock_server + _Handler.response_code = 200 + _Handler.response_body = json.dumps({"title": "ok"}) + assert _api_get(base_url, "/test") == {"title": "ok"} + + def test_404_with_detail(self, mock_server): + base_url, _ = 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_unreachable(self): + 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" + + +# --------------------------------------------------------------------------- +# _api_post +# --------------------------------------------------------------------------- + +class TestApiPost: + def test_success_with_token(self, mock_server): + base_url, _ = mock_server + _Handler.response_code = 200 + _Handler.response_body = json.dumps({"status": "refreshed"}) + result = _api_post(base_url, "/admin/refresh", "my-token") + assert result == {"status": "refreshed"} + assert _Handler.last_headers.get("Authorization") == "Bearer my-token" + + def test_post_with_body(self, mock_server): + base_url, _ = mock_server + _Handler.response_code = 200 + _Handler.response_body = json.dumps({"ok": True}) + result = _api_post(base_url, "/endpoint", "tok", body={"full": True}) + assert result == {"ok": True}