zzstoatzz.io / plyr.fm
audio streaming app plyr.fm
38
fork

Configure Feed

Select the types of activity you want to include in your feed.

test: cover webm/ogg record uploads in the right test suite (#1254)

my previous PR (#1253) added webm/ogg parameterization to
backend/tests/test_integration_upload.py — which turns out to be a
LEGACY file that the CI integration workflow does not run. the
actual suite is backend/tests/integration/ and CI invokes it via
\`pytest tests/integration -m integration -v\` (see
.github/workflows/integration-tests.yml). i caught this only after
manually dispatching the integration workflow and reading the output
logs to verify my tests executed — they did not appear at all, and
the 12 tests that ran were all under tests/integration/.

this PR corrects the placement:

1. reverts backend/tests/test_integration_upload.py to its pre-#1253
state (wav-only, as before — nothing is gained by adding cases to
a file CI won't run)

2. adds backend/tests/integration/test_record_formats.py with two
new integration tests: test_upload_webm and test_upload_ogg. both
use the existing user1_client + upload-then-delete pattern from
test_lossless_uploads.py so they fit the surrounding idiom.

3. adds drone_webm and drone_ogg fixtures to the integration
conftest.py. they use a new save_drone_as_opus helper that forces
libopus — save_drone_as with ffmpeg defaults produced
FLAC-in-ogg (the ffprobe-confirmed default codec ffmpeg picks for
a bare .ogg extension), which isn't what browser MediaRecorder
actually emits. opus-in-{webm,ogg} matches real-world output.

4. save_drone_as_opus lives in tests/integration/utils/audio.py
alongside the existing save_drone / save_drone_as helpers. it
skips with a helpful message if ffmpeg is unavailable.

verified locally:
- ffprobe confirms both fixtures are opus (19925B webm, 18994B ogg)
- both tests collect and skip cleanly without tokens (2 skipped)
- just backend lint + just backend test still pass (765 passed, 15
skipped — +2 from this PR)

next time the CI integration workflow runs (automatic on backend/src
changes, manual dispatch otherwise) it will exercise the full
/record upload path end-to-end against staging: post webm/ogg →
transcoder → mp3 → verify track.file_type == "mp3" → delete.

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

authored by

nate nowack
Claude Opus 4.6 (1M context) and committed by
GitHub 91b1496b 3810f495

+148 -91
+20 -1
backend/tests/integration/conftest.py
··· 17 17 18 18 import pytest 19 19 20 - from .utils.audio import save_drone, save_drone_as 20 + from .utils.audio import save_drone, save_drone_as, save_drone_as_opus 21 21 22 22 if TYPE_CHECKING: 23 23 from plyrfm import AsyncPlyrClient ··· 172 172 path = tmp_path / "drone_a4.aif" 173 173 save_drone_as(path, "A4", duration_sec=2.0) 174 174 yield path 175 + 176 + 177 + # /record page format fixtures — browser MediaRecorder output shapes 178 + 179 + 180 + @pytest.fixture 181 + def drone_webm(tmp_path: Path) -> Generator[Path, None, None]: 182 + """generate a 2-second A4 drone as opus-in-webm (Chrome MediaRecorder).""" 183 + path = tmp_path / "drone_a4.webm" 184 + save_drone_as_opus(path, "A4", duration_sec=2.0) 185 + yield path 186 + 187 + 188 + @pytest.fixture 189 + def drone_ogg(tmp_path: Path) -> Generator[Path, None, None]: 190 + """generate a 2-second A4 drone as opus-in-ogg (Firefox MediaRecorder).""" 191 + path = tmp_path / "drone_a4.ogg" 192 + save_drone_as_opus(path, "A4", duration_sec=2.0) 193 + yield path
+65
backend/tests/integration/test_record_formats.py
··· 1 + """integration tests for /record page MediaRecorder output formats. 2 + 3 + the /record page posts whatever the browser's MediaRecorder API gives us — 4 + opus-in-webm on Chromium, opus-in-mp4 on Safari, opus-in-ogg on Firefox. 5 + both webm and ogg containers are backend-side transcoded to mp3 via the 6 + transcoder service; these tests exercise that full path against the real 7 + staging API so an AudioFormat enum or transcoder regression surfaces in 8 + CI rather than in production. 9 + 10 + requires: 11 + - ffmpeg installed (for generating opus-encoded fixtures) 12 + """ 13 + 14 + from __future__ import annotations 15 + 16 + from typing import TYPE_CHECKING 17 + 18 + import pytest 19 + 20 + if TYPE_CHECKING: 21 + from pathlib import Path 22 + 23 + from plyrfm import AsyncPlyrClient 24 + 25 + pytestmark = [pytest.mark.integration, pytest.mark.timeout(180)] 26 + 27 + 28 + async def test_upload_webm(user1_client: AsyncPlyrClient, drone_webm: Path): 29 + """upload opus-in-webm (Chromium MediaRecorder) — should transcode to MP3.""" 30 + client = user1_client 31 + 32 + result = await client.upload( 33 + drone_webm, 34 + "Test WEBM Upload", 35 + tags={"integration-test", "record", "webm"}, 36 + ) 37 + track_id = result.track_id 38 + assert track_id is not None 39 + 40 + try: 41 + track = await client.get_track(track_id) 42 + assert track.title == "Test WEBM Upload" 43 + assert track.file_type == "mp3", f"expected mp3, got {track.file_type}" 44 + finally: 45 + await client.delete(track_id) 46 + 47 + 48 + async def test_upload_ogg(user1_client: AsyncPlyrClient, drone_ogg: Path): 49 + """upload opus-in-ogg (Firefox MediaRecorder) — should transcode to MP3.""" 50 + client = user1_client 51 + 52 + result = await client.upload( 53 + drone_ogg, 54 + "Test OGG Upload", 55 + tags={"integration-test", "record", "ogg"}, 56 + ) 57 + track_id = result.track_id 58 + assert track_id is not None 59 + 60 + try: 61 + track = await client.get_track(track_id) 62 + assert track.title == "Test OGG Upload" 63 + assert track.file_type == "mp3", f"expected mp3, got {track.file_type}" 64 + finally: 65 + await client.delete(track_id)
+51
backend/tests/integration/utils/audio.py
··· 212 212 wav_path.unlink(missing_ok=True) 213 213 214 214 return path 215 + 216 + 217 + def save_drone_as_opus( 218 + path: Path, 219 + note: str = "A4", 220 + duration_sec: float = 2.0, 221 + ) -> Path: 222 + """generate an opus-encoded drone wrapped in the container inferred 223 + from the file extension. 224 + 225 + supports .webm and .ogg — these are the containers browser 226 + MediaRecorder emits on Chrome/Firefox respectively. unlike 227 + save_drone_as (which lets ffmpeg pick a default codec and ends up 228 + with FLAC-in-ogg), this forces libopus so the resulting file matches 229 + what the /record page actually posts. 230 + """ 231 + import shutil 232 + import subprocess 233 + 234 + if not shutil.which("ffmpeg"): 235 + msg = "ffmpeg not found - required for opus/webm/ogg format tests" 236 + raise FileNotFoundError(msg) 237 + 238 + suffix = path.suffix.lower() 239 + if suffix not in {".webm", ".ogg"}: 240 + msg = f"save_drone_as_opus expects .webm or .ogg, got {suffix}" 241 + raise ValueError(msg) 242 + 243 + wav_path = path.with_suffix(".wav") 244 + save_drone(wav_path, note, duration_sec) 245 + try: 246 + subprocess.run( 247 + [ 248 + "ffmpeg", 249 + "-y", 250 + "-hide_banner", 251 + "-loglevel", 252 + "error", 253 + "-i", 254 + str(wav_path), 255 + "-c:a", 256 + "libopus", 257 + str(path), 258 + ], 259 + check=True, 260 + capture_output=True, 261 + ) 262 + finally: 263 + wav_path.unlink(missing_ok=True) 264 + 265 + return path
+12 -90
backend/tests/test_integration_upload.py
··· 11 11 import json 12 12 import os 13 13 import struct 14 - import subprocess 15 14 import tempfile 16 15 from collections.abc import Generator 17 16 from pathlib import Path ··· 21 20 22 21 API_URL = os.getenv("PLYR_API_URL", "http://localhost:8001") 23 22 TOKEN = os.getenv("PLYR_TOKEN") or os.getenv("PLYRFM_API_TOKEN") 24 - 25 - # formats exercised by the integration suite — each variant runs the full 26 - # upload → process → verify → delete flow. webm and ogg are the formats 27 - # produced by browser MediaRecorder on Chrome/Firefox and are routed 28 - # through the transcoder service on upload, so parameterizing covers the 29 - # /record page end-to-end. 30 - AUDIO_FORMATS: dict[str, str] = { 31 - "wav": "audio/wav", 32 - "webm": "audio/webm", 33 - "ogg": "audio/ogg", 34 - } 35 23 36 24 37 25 def generate_wav_file(duration_seconds: float = 1.0, sample_rate: int = 44100) -> bytes: ··· 65 53 return header + audio_data 66 54 67 55 68 - def _ffmpeg_generate_tone(target_fmt: str, duration: float = 2.0) -> bytes: 69 - """use ffmpeg's lavfi sine generator to produce a short tone in the 70 - target container format. 71 - 72 - webm and ogg both use opus via libopus and can stream to pipe:1. 73 - generating a real sine wave (not silence from a pre-built wav) gives 74 - mutagen enough structure to parse duration, which matches what a real 75 - browser MediaRecorder produces much more closely than a minimal 76 - silent file would. mp4 (m4a) can't write to a pipe because the 77 - container needs seekable output, so it's not supported here. 78 - """ 79 - codec_by_fmt = { 80 - "webm": (["-c:a", "libopus"], "webm"), 81 - "ogg": (["-c:a", "libopus"], "ogg"), 82 - } 83 - if target_fmt not in codec_by_fmt: 84 - raise ValueError(f"no ffmpeg pipeline for format: {target_fmt}") 85 - codec_args, fmt_flag = codec_by_fmt[target_fmt] 86 - try: 87 - result = subprocess.run( 88 - [ 89 - "ffmpeg", 90 - "-y", 91 - "-hide_banner", 92 - "-loglevel", 93 - "error", 94 - "-f", 95 - "lavfi", 96 - "-i", 97 - f"sine=frequency=440:duration={duration}", 98 - *codec_args, 99 - "-f", 100 - fmt_flag, 101 - "pipe:1", 102 - ], 103 - capture_output=True, 104 - check=False, 105 - ) 106 - except FileNotFoundError: 107 - pytest.skip( 108 - f"ffmpeg not installed — required to generate test audio for {target_fmt}" 109 - ) 110 - if result.returncode != 0: 111 - raise RuntimeError( 112 - f"ffmpeg generation of {target_fmt} failed: " 113 - f"{result.stderr.decode(errors='replace')[:500]}" 114 - ) 115 - return result.stdout 116 - 117 - 118 - def _generate_audio(fmt: str) -> bytes: 119 - """generate a short test audio file in the requested format.""" 120 - if fmt == "wav": 121 - return generate_wav_file(duration_seconds=1.0) 122 - return _ffmpeg_generate_tone(fmt) 56 + @pytest.fixture 57 + def test_audio_file() -> Generator[Path, None, None]: 58 + """create a temporary test audio file.""" 59 + wav_data = generate_wav_file(duration_seconds=1.0) 123 60 124 - 125 - @pytest.fixture(params=list(AUDIO_FORMATS.keys())) 126 - def test_audio_file( 127 - request: pytest.FixtureRequest, 128 - ) -> Generator[tuple[Path, str, str], None, None]: 129 - """create a temporary test audio file in each supported format.""" 130 - fmt = request.param 131 - mime = AUDIO_FORMATS[fmt] 132 - audio_bytes = _generate_audio(fmt) 133 - 134 - with tempfile.NamedTemporaryFile(suffix=f".{fmt}", delete=False) as f: 135 - f.write(audio_bytes) 61 + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: 62 + f.write(wav_data) 136 63 path = Path(f.name) 137 64 138 - yield path, fmt, mime 65 + yield path 139 66 140 67 # cleanup 141 68 path.unlink(missing_ok=True) 142 69 143 70 144 71 @pytest.mark.integration 145 - async def test_upload_and_delete_track(test_audio_file: tuple[Path, str, str]): 146 - """integration test: upload a track, wait for processing, then delete it. 147 - 148 - parameterized across wav / webm / ogg — webm and ogg exercise the 149 - transcoder path that the /record page depends on. 150 - """ 151 - audio_path, fmt, mime = test_audio_file 72 + async def test_upload_and_delete_track(test_audio_file: Path): 73 + """integration test: upload a track, wait for processing, then delete it.""" 152 74 if not TOKEN: 153 75 pytest.skip("PLYR_TOKEN or PLYRFM_API_TOKEN not set") 154 76 ··· 165 87 print(f"authenticated as: {user['handle']}") 166 88 167 89 # 2. upload track 168 - with open(audio_path, "rb") as f: 169 - files = {"file": (f"test_integration.{fmt}", f, mime)} 170 - data = {"title": f"Integration Test Track [{fmt}] (DELETE ME)"} 90 + with open(test_audio_file, "rb") as f: 91 + files = {"file": ("test_integration.wav", f, "audio/wav")} 92 + data = {"title": "Integration Test Track (DELETE ME)"} 171 93 172 94 upload_response = await client.post( 173 95 f"{API_URL}/tracks/",