beefy-player/tests/test_community.py

"""The community-station providers: relays, status documents and schedules."""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import httpx
import pytest

from beefy_player.providers import PROVIDERS
from beefy_player.providers.community import (
    AnonradioProvider,
    CashmereProvider,
    DatafruitsProvider,
    DublabProvider,
    LotProvider,
    LylProvider,
    RefugeProvider,
    RinseProvider,
    dublab,
)

ICECAST = {
    "icestats": {
        "source": [
            {
                "listenurl": "https://anonradio.net:8443/anonradio",
                "bitrate": 192,
                "server_type": "audio/mpeg",
                "listeners": 13,
                "title": "repeating screwtape from 29-Nov-2023",
            }
        ]
    }
}

RADIO_CO = {
    "status": "online",
    "listeners": 41,
    "current_track": {
        "title": "Guest Show (r) - Ain Autavg",
        "start_time": "2026-11-04T00:00:02+00:00",
        "artwork_url": "https://images.radio.co/station.png",
    },
}


def client_for(payload, status_code: int = 200) -> httpx.AsyncClient:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status_code, json=payload)

    return httpx.AsyncClient(transport=httpx.MockTransport(handler))


def test_every_community_provider_builds_a_playable_station() -> None:
    providers = (
        AnonradioProvider,
        DatafruitsProvider,
        CashmereProvider,
        LylProvider,
        LotProvider,
        RefugeProvider,
        DublabProvider,
        RinseProvider,
    )
    for cls in providers:
        provider = cls()
        stations = provider.fallback_stations()
        assert stations, cls.id
        for station in stations:
            assert station.provider == provider.id
            assert station.playlists, (cls.id, station.id)
            assert all(url.startswith(("http://", "https://")) for url in provider.fallback_urls(station))
        # every one of them spells its name in the same font
        assert len(provider.hero) == 3, cls.id


def test_rinse_is_a_family_of_three_stations() -> None:
    stations = RinseProvider().fallback_stations()
    assert [station.id for station in stations] == ["rinse_uk", "kool", "swu"]
    assert [station.title for station in stations] == ["Rinse UK", "Kool London", "SWU"]
    for station in stations:
        assert "admin.stream.rinse.fm/proxy/" in station.playlists[0]["url"]
        assert station.format_label.endswith("128k")


def test_the_lot_streams_hls_which_the_player_owns() -> None:
    from beefy_player.providers.base import is_direct_stream, is_playlist

    url = LotProvider().fallback_urls(LotProvider().station())[0]
    assert url.endswith(".m3u8")
    assert is_playlist(url) is False
    assert is_direct_stream(url) is True


@pytest.mark.asyncio
async def test_icecast_status_gives_the_station_what_is_on() -> None:
    provider = AnonradioProvider()
    async with client_for(ICECAST) as client:
        station = (await provider.fetch_stations(client))[0]
    assert station.listeners == 13
    assert station.now_playing is not None
    assert station.now_playing.title == "repeating screwtape from 29-Nov-2023"


@pytest.mark.asyncio
async def test_radio_co_status_gives_a_track_and_its_art() -> None:
    provider = RefugeProvider()
    async with client_for(RADIO_CO) as client:
        station = (await provider.fetch_stations(client))[0]
        history = await provider.fetch_history(client, "refuge")
    assert station.listeners == 41
    assert station.now_playing is not None
    assert station.now_playing.album == ""
    assert station.now_playing.art_url.endswith("station.png")
    assert history[0].title == "Ain Autavg"
    # the show's own start time is used, not "just now"
    assert history[0].played_at == datetime(
        2026, 11, 4, 0, 0, 2, tzinfo=timezone.utc
    )


@pytest.mark.asyncio
async def test_history_is_rolling_and_keeps_the_first_sighting() -> None:
    provider = AnonradioProvider()
    async with client_for(ICECAST) as client:
        first = await provider.fetch_history(client, "anonradio")
        again = await provider.fetch_history(client, "anonradio")
    assert len(first) == 1
    assert again[0].played_at == first[0].played_at


@pytest.mark.asyncio
async def test_a_station_without_metadata_still_carries_on() -> None:
    """Rinse, LYL, The Lot and Cashmere publish no status document."""
    provider = RinseProvider()
    async with client_for({}, status_code=500) as client:
        stations = await provider.fetch_stations(client)
        history = await provider.fetch_history(client, "rinse_uk")
    assert len(stations) == 3
    assert all(station.now_playing is None for station in stations)
    assert history == []


def test_dublab_schedule_parses_the_site_calendar() -> None:
    payload = [
        {
            "startTime": "2026-11-03T16:00:00-08:00",
            "endTime": "2026-11-03T18:00:00-08:00",
            "summary": "LIVE // Version Sounds",
        },
        {
            "startTime": "2026-11-03T18:00:00-08:00",
            "endTime": "2026-11-03T20:00:00-08:00",
            "summary": "PRE-REC // Sounds of Now",
        },
    ]
    shows = dublab.parse_schedule(payload)
    assert [show.name for show in shows] == [
        "LIVE // Version Sounds",
        "PRE-REC // Sounds of Now",
    ]
    assert shows[0].start == datetime(
        2026, 11, 3, 16, tzinfo=timezone(timedelta(hours=-8))
    )
    assert dublab.parse_schedule({"nope": True}) == []


def test_the_running_slot_is_marked_live() -> None:
    provider = DublabProvider()
    now = datetime.now(timezone.utc)
    running = dublab.Broadcast(
        name="LIVE // Version Sounds",
        start=now - timedelta(minutes=30),
        end=now + timedelta(minutes=30),
    )
    later = dublab.Broadcast(
        name="Later Show",
        start=now + timedelta(hours=1),
        end=now + timedelta(hours=2),
    )
    marked = provider._mark_live([running, later])
    assert [item.live for item in marked] == [True, False]
    assert marked[0].key() == running.key()


@pytest.mark.asyncio
async def test_dublab_history_follows_the_schedule() -> None:
    provider = DublabProvider()
    now = datetime.now(timezone.utc)
    payload = [
        {
            "startTime": (now - timedelta(minutes=20)).isoformat(),
            "endTime": (now + timedelta(minutes=40)).isoformat(),
            "summary": "LIVE // Version Sounds",
        }
    ]

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=payload)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        await provider.fetch_schedule(client)
        history = await provider.fetch_history(client, "dublab")
    assert history and history[0].title == "LIVE // Version Sounds"


def test_every_catalogue_is_registered_under_its_slug() -> None:
    assert set(PROVIDERS) == {
        "somafm",
        "tilderadio",
        "bassdrive",
        "radioparadise",
        "dublab",
        "rinse",
        "refuge",
        "lyl",
        "lot",
        "radiobrowser",
        "cashmere",
        "datafruits",
        "anonradio",
    }