279 lines
9.7 KiB
Python
279 lines
9.7 KiB
Python
"""Bassdrive: Icecast status, show titles, the homepage schedule, rolling history."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from beefy_player.models import Broadcast
|
|
from beefy_player.providers.bassdrive import BassdriveProvider, api
|
|
|
|
STATUS = {
|
|
"icestats": {
|
|
"source": [
|
|
{
|
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
|
"bitrate": 128,
|
|
"server_type": "audio/mpeg",
|
|
"listeners": 31,
|
|
"display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]",
|
|
},
|
|
{
|
|
"listenurl": "https://ice.bassdrive.net:443/stream56",
|
|
"bitrate": 56,
|
|
"server_type": "audio/aacp",
|
|
"listeners": 5,
|
|
"display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]",
|
|
},
|
|
{
|
|
"listenurl": "https://ice.bassdrive.net:443/stream32",
|
|
"bitrate": 32,
|
|
"server_type": "audio/aacp",
|
|
"listeners": 2,
|
|
"display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]",
|
|
},
|
|
]
|
|
}
|
|
}
|
|
|
|
# Trimmed from the real homepage: the same markup the site ships.
|
|
SCHEDULE_PAGE = """
|
|
<ul class="schedule">
|
|
<li>
|
|
<a href="radio/get-dipped">
|
|
<p class="title">Get Dipped <span class="location">Los Angeles, CA</span></p>
|
|
<span class="time">
|
|
<local-time datetime="2026-11-03T19:00:00-05:00" hour="numeric"></local-time> -
|
|
<local-time datetime="2026-11-03T21:00:00-05:00" hour="numeric"></local-time>
|
|
</span>
|
|
</a>
|
|
</li>
|
|
<li>
|
|
<a href="radio/drumobsession">
|
|
<p class="title">DrumObsession <span class="location">Poznan, Poland</span></p>
|
|
<span class="time">
|
|
<local-time datetime="2026-11-03T11:00:00-05:00" hour="numeric"></local-time> -
|
|
<local-time datetime="2026-11-03T13:00:00-05:00" hour="numeric"></local-time>
|
|
</span>
|
|
</a>
|
|
</li>
|
|
</ul>
|
|
"""
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fresh_memo():
|
|
api.reset_schedule_memo()
|
|
yield
|
|
api.reset_schedule_memo()
|
|
|
|
|
|
def test_split_title_reads_both_conventions() -> None:
|
|
assert api.split_title("Ben XO :: LIVE :: Night Shift Over [promo]") == (
|
|
"Ben XO",
|
|
"Night Shift Over",
|
|
True,
|
|
)
|
|
# plenty of shows simply append LIVE
|
|
assert api.split_title("Sohl Patrol LIVE") == ("", "Sohl Patrol", True)
|
|
assert api.split_title("Bassdrive Auto DJ") == ("", "Bassdrive Auto DJ", False)
|
|
|
|
|
|
def test_mounts_become_playlists_best_first_and_listeners_add_up() -> None:
|
|
playlists = api.playlists_from_status(STATUS)
|
|
assert [entry["bitrate"] for entry in playlists] == ["128", "56", "32"]
|
|
assert [entry["quality"] for entry in playlists] == ["highest", "high", "low"]
|
|
assert playlists[0]["format"] == "mp3" and playlists[1]["format"] == "aacp"
|
|
assert api.status_listeners(STATUS) == 38
|
|
|
|
|
|
def test_parse_schedule_reads_the_homepage_grid() -> None:
|
|
entries = api.parse_schedule(SCHEDULE_PAGE)
|
|
assert [entry.name for entry in entries] == ["DrumObsession", "Get Dipped"]
|
|
assert entries[0].location == "Poznan, Poland"
|
|
assert entries[0].start == datetime(2026, 11, 3, 11, tzinfo=timezone(timedelta(hours=-5)))
|
|
assert entries[0].end == datetime(2026, 11, 3, 13, tzinfo=timezone(timedelta(hours=-5)))
|
|
# the schedule list itself is not a schedule slot
|
|
assert api.parse_schedule("<ul><li>nothing here</li></ul>") == []
|
|
|
|
|
|
def test_roll_forward_repeats_the_week() -> None:
|
|
start = datetime(2026, 11, 3, 11, tzinfo=timezone(timedelta(hours=-5)))
|
|
entry = Broadcast(
|
|
name="DrumObsession",
|
|
start=start,
|
|
end=start + timedelta(hours=2),
|
|
)
|
|
now = start + timedelta(days=1)
|
|
rolled = api.roll_forward([entry], now)
|
|
assert len(rolled) == 1
|
|
assert rolled[0].start == start + timedelta(days=7)
|
|
assert rolled[0].start >= now
|
|
# an entry that is over and cannot be rolled into the future is dropped
|
|
far = start + timedelta(days=40)
|
|
assert api.roll_forward([entry], far) == []
|
|
|
|
|
|
def test_station_maps_the_title_and_finds_the_live_location() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
slot = Broadcast(
|
|
name="Night Shift Over",
|
|
location="London, UK",
|
|
start=now - timedelta(minutes=30),
|
|
end=now + timedelta(minutes=30),
|
|
)
|
|
station = api.station_from_status(STATUS, [slot])
|
|
assert station is not None
|
|
assert station.id == "bassdrive"
|
|
assert station.dj == "Ben XO"
|
|
assert station.presenter == "live from London, UK"
|
|
assert station.now_playing is not None
|
|
assert station.now_playing.title == "Night Shift Over"
|
|
assert station.now_playing.artist == "Ben XO"
|
|
assert station.listeners == 38
|
|
assert station.format_label == "mp3 · 128k"
|
|
|
|
|
|
def test_a_title_without_a_live_marker_still_counts_when_the_slot_is_running() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
status = {
|
|
"icestats": {
|
|
"source": [
|
|
{
|
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
|
"bitrate": 128,
|
|
"server_type": "audio/mpeg",
|
|
"listeners": 3,
|
|
"display-title": "Sohl Patrol",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
slot = Broadcast(
|
|
name="Sohl Patrol",
|
|
location="White Plains, NY, USA",
|
|
start=now - timedelta(minutes=10),
|
|
end=now + timedelta(minutes=50),
|
|
)
|
|
live = api.live_broadcast(status, [slot], now)
|
|
assert live is not None and live.live is True
|
|
assert live.location == "White Plains, NY, USA"
|
|
|
|
|
|
def test_station_without_title_or_mounts_is_not_a_station() -> None:
|
|
assert api.station_from_status({"icestats": {}}) is None
|
|
|
|
|
|
def client_for(
|
|
status_payload,
|
|
page: str = "",
|
|
*,
|
|
status_code: int = 200,
|
|
page_code: int = 200,
|
|
) -> httpx.AsyncClient:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if str(request.url).endswith("status-json.xsl"):
|
|
return httpx.Response(status_code, json=status_payload)
|
|
return httpx.Response(page_code, text=page)
|
|
|
|
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
|
|
|
|
async def songs_for(payload, station_id: str = "bassdrive"):
|
|
async with client_for(payload) as client:
|
|
return await api.fetch_songs(client, station_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_schedule_puts_the_live_show_first_and_caches_the_grid() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
payload = {
|
|
"icestats": {
|
|
"source": [
|
|
{
|
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
|
"bitrate": 128,
|
|
"server_type": "audio/mpeg",
|
|
"listeners": 9,
|
|
"display-title": "Get Dipped LIVE",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
# the grid is generated relative to now so the live slot is really running
|
|
local = now.astimezone(timezone(timedelta(hours=-5)))
|
|
start = local - timedelta(minutes=15)
|
|
end = local + timedelta(minutes=45)
|
|
page = SCHEDULE_PAGE.replace("2026-11-03T19:00:00-05:00", start.isoformat()).replace(
|
|
"2026-11-03T21:00:00-05:00", end.isoformat()
|
|
)
|
|
async with client_for(payload, page) as client:
|
|
broadcast_list = await api.fetch_schedule(client)
|
|
|
|
assert broadcast_list[0].live is True
|
|
assert broadcast_list[0].name == "Get Dipped"
|
|
assert broadcast_list[0].location == "Los Angeles, CA"
|
|
assert broadcast_list[0].start.hour == start.hour
|
|
assert all(not item.live for item in broadcast_list[1:])
|
|
# the grid was written for offline starts
|
|
assert api.load_cached_schedule()
|
|
# ... and the second fetch reuses the memo instead of the network
|
|
async with client_for(payload, page_code=500) as client:
|
|
assert await api.fetch_schedule(client)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_is_a_rolling_record_of_shows() -> None:
|
|
first = await songs_for(STATUS)
|
|
assert [song.title for song in first] == ["Night Shift Over"]
|
|
|
|
# same show: the first-seen time is kept
|
|
again = await songs_for(STATUS)
|
|
assert again[0].played_at == first[0].played_at
|
|
|
|
other = {
|
|
"icestats": {
|
|
"source": [
|
|
{
|
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
|
"bitrate": 128,
|
|
"server_type": "audio/mpeg",
|
|
"listeners": 1,
|
|
"display-title": "Sohl Patrol LIVE",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
moved = await songs_for(other)
|
|
assert [song.title for song in moved] == ["Sohl Patrol", "Night Shift Over"]
|
|
|
|
|
|
def test_provider_describes_its_catalogue() -> None:
|
|
provider = BassdriveProvider()
|
|
assert provider.has_schedule is True
|
|
assert provider.bandcamp is False
|
|
assert len(provider.hero) == 3
|
|
assert "bright_cyan" in "".join(provider.hero)
|
|
assert provider.hero_lines == (
|
|
"BASSDRIVE",
|
|
"WORLDWIDE DRUM & BASS",
|
|
"LIVE SHOWS 24/7",
|
|
)
|
|
|
|
stations = provider.fallback_stations()
|
|
assert [station.id for station in stations] == ["bassdrive"]
|
|
urls = provider.fallback_urls(stations[0])
|
|
assert urls[0].startswith("https://") and len(urls) == 7
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cold_start_survives_an_unreachable_homepage() -> None:
|
|
"""`fetch_station` asks for the grid; a broken homepage must not stop it."""
|
|
async with client_for(STATUS, page_code=500) as client:
|
|
station = await api.fetch_station(client)
|
|
assert station.id == "bassdrive"
|
|
assert station.presenter == "live now" # live, but the location is unknown
|
|
assert station.now_playing is not None
|