"""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}