NtR-soudcloud-fetcher/tests/test_plugin_helpers.py

"""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) + 5 > _MAX_IRC_LINE:  # +5 for ", ..."
            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_many_tracks(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("...")

    def test_truncation_long_single_track(self):
        tracks = [
            {"title": "A" * 200, "artist": "B" * 200},
            {"title": "Second", "artist": "Track"},
        ]
        data = {"episode_number": 1, "tracks": tracks}
        result = format_playlist(data)
        assert len(result) <= _MAX_IRC_LINE


# ---------------------------------------------------------------------------
# _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}