refactor: split web.py — detector and audio-format helpers into isrweb/
Pure move, no behaviour change. web.py 1221 -> 807 lines.
The reason is not file size but isolation of change. CLAUDE.md records
that the next step for detection, should flagging ever need improving, is
a spectral filter or an optional Silero VAD over candidate sections —
that work now lands entirely in isrweb/loudness.py and never has to open
the HTTP layer. The byte-level format code (FLAC frame parsing, CRC-8,
live header patching) is a second self-contained cluster, touched only
when a container format misbehaves.
isrweb/loudness.py 229 detector + its tuning constants
isrweb/audiofile.py 250 filename clock, WAV/FLAC headers, MIME types
web.py 807 cache, listing, handler, main()
Import direction is strictly web.py -> {loudness, audiofile}; the two
leaves import neither each other nor web.py, which is what lets the
detector be imported and tested without an HTTP server.
A package rather than flat modules because `audiofile` is a real PyPI
name and `detect` is generic enough to collide; a local file silently
shadows an installed one, which fails confusingly rather than loudly.
web.py stays at the repo root so `python web.py`, the docker-compose
command and the webui.html lookup are all unchanged.
Each module keeps its own optional numpy/soundfile try-import rather than
sharing one. That block cannot meaningfully diverge — if numpy is
installed every copy says True — and sharing it would stop isr.py being a
standalone file you can drop on a Pi and run.
Tests mirror the modules: test_loudness.py, test_audiofile.py, test_web.py
(+ unchanged test_isr.py). New tests/conftest.py owns the sys.path setup
that was previously an inline hack in test_isr.py — which only worked for
the other modules because that file happened to import first — and
provides the write_wav fixture the split would otherwise have duplicated.
Verified:
* 78 passed, same count as before the split; each test file also passes
standalone.
* All 21 moved functions diffed against HEAD:web.py — byte-identical.
* Import graph asserted acyclic; both leaves import with `web` absent
from sys.modules.
* An analysis cache written *before* the split still hits afterwards,
with identical sections — the one way a "pure" move could quietly
invalidate every user's cache.
* Endpoint smoke (analyze bounds, cache hit, cached_analysis in the
listing, clip Content-Length, 8-bit rejection) and the byte-serving
harness (ranges, download Range, HEAD, 405, live-header patching whole
and straddling) both re-run green.
* Dockerfile COPY set verified by building the image's file tree in a
temp dir and importing web.py with nothing else present.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,30 @@
|
||||
"""Shared test setup.
|
||||
|
||||
Puts the repo root on sys.path so `import isr`, `import web` and
|
||||
`import isrweb.*` work regardless of where pytest is invoked from. This used
|
||||
to be an inline sys.path.insert at the top of test_isr.py, which only
|
||||
happened to work for the other test modules because that file imported first.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import wave
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def write_wav():
|
||||
"""Factory for small 16-bit WAV files. Returns the path written."""
|
||||
def _write(path, seconds=2, framerate=8000, channels=1, sampwidth=2,
|
||||
frames=None):
|
||||
with wave.open(str(path), 'wb') as wf:
|
||||
wf.setnchannels(channels)
|
||||
wf.setsampwidth(sampwidth)
|
||||
wf.setframerate(framerate)
|
||||
wf.writeframes(frames if frames is not None
|
||||
else b'\x00\x00' * (seconds * framerate * channels))
|
||||
return path
|
||||
return _write
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Tests for the filename clock in isrweb/audiofile.py.
|
||||
|
||||
Recordings are named for their *start* time, and that name is the only
|
||||
reliable time anchor the archive has — mtime drifts to the last write.
|
||||
"""
|
||||
|
||||
from isrweb.audiofile import _cut_filename, _recording_start
|
||||
|
||||
|
||||
def test_recording_start_parses_standard_name():
|
||||
from datetime import datetime
|
||||
assert _recording_start("20260523_220000") == datetime(2026, 5, 23, 22, 0, 0)
|
||||
|
||||
|
||||
def test_recording_start_rejects_nonstandard_name():
|
||||
assert _recording_start("radio1_20260523") is None
|
||||
assert _recording_start("notes") is None
|
||||
|
||||
|
||||
def test_cut_filename_uses_wall_clock_span():
|
||||
# Recording started 22:00:00; cut covers 22:31:30 → 22:32:30.
|
||||
name = _cut_filename("20260523_220000", ".flac", 1890, 1950)
|
||||
assert name == "20260523_22-31-30_22-32-30.flac"
|
||||
|
||||
|
||||
def test_cut_filename_rolls_over_the_hour():
|
||||
name = _cut_filename("20260523_220000", ".wav", 3590, 3661)
|
||||
assert name == "20260523_22-59-50_23-01-01.wav"
|
||||
|
||||
|
||||
def test_cut_filename_falls_back_for_nonstandard_name():
|
||||
name = _cut_filename("mixtape", ".mp3", 740, 750.4)
|
||||
assert name == "mixtape_cut_740s-750s.mp3"
|
||||
|
||||
|
||||
+2
-6
@@ -31,14 +31,10 @@ def fixed_clock(dt: datetime):
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Import the module under test
|
||||
# Import the module under test (sys.path is set up in conftest.py)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
import sys
|
||||
import os
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
import isr # noqa: E402 (must come after sys.path manipulation)
|
||||
import isr
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Tests for the adaptive loud-section detector in isrweb/loudness.py."""
|
||||
|
||||
import math
|
||||
import wave
|
||||
|
||||
from isrweb.loudness import _loud_sections, _noise_floor_db, analyze_wav
|
||||
|
||||
WINDOW_DUR = 0.1 # 100 ms windows, as produced by WINDOW_SAMPLES at 48 kHz
|
||||
|
||||
|
||||
def _run(rms, margin_db=12.0, min_gap=2.0, min_duration=0.5):
|
||||
duration = len(rms) * WINDOW_DUR
|
||||
return _loud_sections(rms, WINDOW_DUR, duration, margin_db, min_gap, min_duration)
|
||||
|
||||
|
||||
def test_burst_above_quiet_floor_is_detected():
|
||||
rms = [0.002] * 1200 # 2 min of quiet ambience (−54 dBFS)
|
||||
rms[600:610] = [0.05] * 10 # 1 s burst at −26 dBFS (+28 dB)
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
s = sections[0]
|
||||
assert s['start'] == 60.0
|
||||
assert 60.5 <= s['end'] <= 62.0
|
||||
assert 26.0 <= s['score'] <= 30.0
|
||||
|
||||
|
||||
def test_slow_ambience_swell_is_not_detected():
|
||||
# Level rises 20 dB over 10 minutes and back down — e.g. rain setting in.
|
||||
# The old fixed threshold would have flagged the entire loud half.
|
||||
up = [0.002 * 10 ** (i / 6000 * 1.0) for i in range(6000)] # −54 → −34 dB
|
||||
down = list(reversed(up))
|
||||
assert _run(up + down) == []
|
||||
|
||||
|
||||
def test_burst_still_detected_on_loud_ambience():
|
||||
# Same +28 dB prominence as the quiet test, but on a 20 dB louder floor.
|
||||
rms = [0.02] * 1200
|
||||
rms[600:610] = [0.5] * 10
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 60.0
|
||||
|
||||
|
||||
def test_min_rms_floor_suppresses_blips_in_digital_silence():
|
||||
rms = [0.000001] * 1200
|
||||
rms[600:605] = [0.004] * 5 # −48 dBFS: audible blip, but below MIN_RMS+12dB
|
||||
assert _run(rms) == []
|
||||
rms[600:605] = [0.05] * 5 # −26 dBFS clearly clears the clamped floor
|
||||
assert len(_run(rms)) == 1
|
||||
|
||||
|
||||
def test_min_gap_merges_nearby_bursts():
|
||||
rms = [0.002] * 1200
|
||||
rms[600:605] = [0.05] * 5
|
||||
rms[615:620] = [0.05] * 5 # 1 s gap < 2 s min_gap → one section
|
||||
rms[900:905] = [0.05] * 5 # 28 s away → separate section
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 2
|
||||
assert sections[0]['start'] == 60.0
|
||||
assert sections[0]['end'] >= 61.5
|
||||
assert sections[1]['start'] == 90.0
|
||||
|
||||
|
||||
def test_min_duration_drops_subsecond_blips():
|
||||
# Isolated single-window pops (clicks, single raindrops) spaced wider than
|
||||
# min_gap must not each become their own section — this is what used to
|
||||
# produce thousands of zero-length sections per day.
|
||||
rms = [0.002] * 1200
|
||||
for i in range(600, 660, 30): # 0.1 s blips, 3 s apart (> min_gap)
|
||||
rms[i] = 0.05
|
||||
assert _run(rms) == []
|
||||
# With the filter disabled they are all reported
|
||||
assert len(_run(rms, min_duration=0.0)) == 2
|
||||
|
||||
|
||||
def test_min_duration_keeps_sections_at_or_above_it():
|
||||
rms = [0.002] * 1200
|
||||
rms[600:605] = [0.05] * 5 # exactly 0.5 s
|
||||
sections = _run(rms, min_duration=0.5)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 60.0
|
||||
|
||||
|
||||
def test_min_duration_applies_after_gap_merging():
|
||||
# Two sub-min_duration blips within min_gap merge into one section whose
|
||||
# loud span exceeds min_duration — the merged section must survive.
|
||||
rms = [0.002] * 1200
|
||||
rms[600] = 0.05
|
||||
rms[610] = 0.05 # 1 s apart < 2 s min_gap → merged, 1.1 s span
|
||||
sections = _run(rms, min_duration=1.0)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 60.0
|
||||
assert sections[0]['end'] >= 61.0
|
||||
|
||||
|
||||
def test_slow_swell_scores_near_zero_sharp_burst_scores_high():
|
||||
# A ~10 s swell rises faster than the 30 s floor blocks can track, so it
|
||||
# still flags — but its score must collapse (onset cap) so it ranks at the
|
||||
# bottom, while a sharp burst with the same peak keeps its full score.
|
||||
rms = [0.002] * 1200
|
||||
for k in range(80): # rise −54 → −28 dB over 8 s
|
||||
rms[400 + k] = 0.002 * 10 ** (26 * (k / 80) / 20)
|
||||
rms[480:500] = [0.04] * 20 # hold 2 s at −28 dB
|
||||
for k in range(80): # fall back over 8 s
|
||||
rms[500 + k] = 0.04 * 10 ** (-26 * (k / 80) / 20)
|
||||
rms[900:910] = [0.04] * 10 # sharp 1 s burst, same peak
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 2
|
||||
swell, burst = sections
|
||||
assert swell['start'] < 50.0 < 90.0 <= burst['start']
|
||||
assert burst['score'] >= 24.0 # ≈ full 26 dB prominence
|
||||
assert swell['score'] <= 5.0 # capped by its slow onset
|
||||
assert swell['score'] < burst['score']
|
||||
|
||||
|
||||
def test_burst_at_file_start_keeps_full_score():
|
||||
# No pre-event history to measure a rise against: the onset cap falls back
|
||||
# to prominence above the floor, so events cut off by a file split are not
|
||||
# punished as swells.
|
||||
rms = [0.05] * 10 + [0.002] * 1190
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 0.0
|
||||
assert sections[0]['score'] >= 24.0
|
||||
|
||||
|
||||
def test_section_open_at_eof_ends_at_last_loud_window():
|
||||
# Burst ends 1 s before EOF — less than min_gap, so the section is still
|
||||
# open when the windows run out. It must end where the sound ended, not at
|
||||
# the file duration; otherwise the clip carries a second of trailing
|
||||
# silence and disagrees with every section closed mid-file.
|
||||
rms = [0.002] * 1200
|
||||
rms[1180:1190] = [0.05] * 10 # loud 118.0 → 119.0, file is 120.0 s
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 118.0
|
||||
assert sections[0]['end'] == 119.0
|
||||
|
||||
|
||||
def test_section_running_to_the_very_end_is_capped_at_duration():
|
||||
rms = [0.002] * 1200
|
||||
rms[1190:1200] = [0.05] * 10 # loud right up to EOF
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['end'] == 120.0
|
||||
|
||||
|
||||
def test_noise_floor_tracks_blocks_and_ignores_short_events():
|
||||
quiet_db = 20 * math.log10(0.002)
|
||||
db = [quiet_db] * 1200
|
||||
db[600:650] = [-20.0] * 50 # 5 s event must not raise its own floor
|
||||
floor = _noise_floor_db(db, WINDOW_DUR)
|
||||
assert len(floor) == len(db)
|
||||
assert all(abs(f - quiet_db) < 1.0 for f in floor)
|
||||
|
||||
|
||||
def test_non_16_bit_wav_is_rejected_rather_than_misread(tmp_path, write_wav):
|
||||
# RMS windowing decodes as int16 regardless of the declared sample width,
|
||||
# so an 8-bit file used to produce meaningless sections with no warning.
|
||||
path = tmp_path / 'eight_bit.wav'
|
||||
with wave.open(str(path), 'wb') as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(1)
|
||||
wf.setframerate(8000)
|
||||
wf.writeframes(b'\x80' * 8000)
|
||||
result = analyze_wav(path)
|
||||
assert 'error' in result
|
||||
assert '16-bit' in result['error']
|
||||
|
||||
|
||||
def test_16_bit_wav_still_analyses(tmp_path, write_wav):
|
||||
path = tmp_path / 'ok.wav'
|
||||
write_wav(path)
|
||||
assert 'error' not in analyze_wav(path)
|
||||
|
||||
|
||||
+15
-224
@@ -1,189 +1,21 @@
|
||||
"""Tests for the adaptive loud-section detector in web.py."""
|
||||
"""Tests for the HTTP layer and file listing in web.py."""
|
||||
|
||||
import io
|
||||
import math
|
||||
import wave
|
||||
|
||||
import web
|
||||
from web import _cut_filename, _loud_sections, _noise_floor_db, _recording_start
|
||||
|
||||
WINDOW_DUR = 0.1 # 100 ms windows, as produced by WINDOW_SAMPLES at 48 kHz
|
||||
|
||||
|
||||
def _run(rms, margin_db=12.0, min_gap=2.0, min_duration=0.5):
|
||||
duration = len(rms) * WINDOW_DUR
|
||||
return _loud_sections(rms, WINDOW_DUR, duration, margin_db, min_gap, min_duration)
|
||||
def test_listing_is_ordered_by_the_filename_clock_not_mtime(tmp_path, write_wav):
|
||||
# mtime is the last write (≈ the end of the recording) and drifts from the
|
||||
# name on any copy or round-trip. Here the older recording is touched last.
|
||||
import os
|
||||
early, late = tmp_path / '20260523_080000.wav', tmp_path / '20260523_090000.wav'
|
||||
write_wav(early)
|
||||
write_wav(late)
|
||||
os.utime(late, (1000, 1000)) # newest recording, oldest mtime
|
||||
|
||||
|
||||
def test_burst_above_quiet_floor_is_detected():
|
||||
rms = [0.002] * 1200 # 2 min of quiet ambience (−54 dBFS)
|
||||
rms[600:610] = [0.05] * 10 # 1 s burst at −26 dBFS (+28 dB)
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
s = sections[0]
|
||||
assert s['start'] == 60.0
|
||||
assert 60.5 <= s['end'] <= 62.0
|
||||
assert 26.0 <= s['score'] <= 30.0
|
||||
|
||||
|
||||
def test_slow_ambience_swell_is_not_detected():
|
||||
# Level rises 20 dB over 10 minutes and back down — e.g. rain setting in.
|
||||
# The old fixed threshold would have flagged the entire loud half.
|
||||
up = [0.002 * 10 ** (i / 6000 * 1.0) for i in range(6000)] # −54 → −34 dB
|
||||
down = list(reversed(up))
|
||||
assert _run(up + down) == []
|
||||
|
||||
|
||||
def test_burst_still_detected_on_loud_ambience():
|
||||
# Same +28 dB prominence as the quiet test, but on a 20 dB louder floor.
|
||||
rms = [0.02] * 1200
|
||||
rms[600:610] = [0.5] * 10
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 60.0
|
||||
|
||||
|
||||
def test_min_rms_floor_suppresses_blips_in_digital_silence():
|
||||
rms = [0.000001] * 1200
|
||||
rms[600:605] = [0.004] * 5 # −48 dBFS: audible blip, but below MIN_RMS+12dB
|
||||
assert _run(rms) == []
|
||||
rms[600:605] = [0.05] * 5 # −26 dBFS clearly clears the clamped floor
|
||||
assert len(_run(rms)) == 1
|
||||
|
||||
|
||||
def test_min_gap_merges_nearby_bursts():
|
||||
rms = [0.002] * 1200
|
||||
rms[600:605] = [0.05] * 5
|
||||
rms[615:620] = [0.05] * 5 # 1 s gap < 2 s min_gap → one section
|
||||
rms[900:905] = [0.05] * 5 # 28 s away → separate section
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 2
|
||||
assert sections[0]['start'] == 60.0
|
||||
assert sections[0]['end'] >= 61.5
|
||||
assert sections[1]['start'] == 90.0
|
||||
|
||||
|
||||
def test_min_duration_drops_subsecond_blips():
|
||||
# Isolated single-window pops (clicks, single raindrops) spaced wider than
|
||||
# min_gap must not each become their own section — this is what used to
|
||||
# produce thousands of zero-length sections per day.
|
||||
rms = [0.002] * 1200
|
||||
for i in range(600, 660, 30): # 0.1 s blips, 3 s apart (> min_gap)
|
||||
rms[i] = 0.05
|
||||
assert _run(rms) == []
|
||||
# With the filter disabled they are all reported
|
||||
assert len(_run(rms, min_duration=0.0)) == 2
|
||||
|
||||
|
||||
def test_min_duration_keeps_sections_at_or_above_it():
|
||||
rms = [0.002] * 1200
|
||||
rms[600:605] = [0.05] * 5 # exactly 0.5 s
|
||||
sections = _run(rms, min_duration=0.5)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 60.0
|
||||
|
||||
|
||||
def test_min_duration_applies_after_gap_merging():
|
||||
# Two sub-min_duration blips within min_gap merge into one section whose
|
||||
# loud span exceeds min_duration — the merged section must survive.
|
||||
rms = [0.002] * 1200
|
||||
rms[600] = 0.05
|
||||
rms[610] = 0.05 # 1 s apart < 2 s min_gap → merged, 1.1 s span
|
||||
sections = _run(rms, min_duration=1.0)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 60.0
|
||||
assert sections[0]['end'] >= 61.0
|
||||
|
||||
|
||||
def test_slow_swell_scores_near_zero_sharp_burst_scores_high():
|
||||
# A ~10 s swell rises faster than the 30 s floor blocks can track, so it
|
||||
# still flags — but its score must collapse (onset cap) so it ranks at the
|
||||
# bottom, while a sharp burst with the same peak keeps its full score.
|
||||
rms = [0.002] * 1200
|
||||
for k in range(80): # rise −54 → −28 dB over 8 s
|
||||
rms[400 + k] = 0.002 * 10 ** (26 * (k / 80) / 20)
|
||||
rms[480:500] = [0.04] * 20 # hold 2 s at −28 dB
|
||||
for k in range(80): # fall back over 8 s
|
||||
rms[500 + k] = 0.04 * 10 ** (-26 * (k / 80) / 20)
|
||||
rms[900:910] = [0.04] * 10 # sharp 1 s burst, same peak
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 2
|
||||
swell, burst = sections
|
||||
assert swell['start'] < 50.0 < 90.0 <= burst['start']
|
||||
assert burst['score'] >= 24.0 # ≈ full 26 dB prominence
|
||||
assert swell['score'] <= 5.0 # capped by its slow onset
|
||||
assert swell['score'] < burst['score']
|
||||
|
||||
|
||||
def test_burst_at_file_start_keeps_full_score():
|
||||
# No pre-event history to measure a rise against: the onset cap falls back
|
||||
# to prominence above the floor, so events cut off by a file split are not
|
||||
# punished as swells.
|
||||
rms = [0.05] * 10 + [0.002] * 1190
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 0.0
|
||||
assert sections[0]['score'] >= 24.0
|
||||
|
||||
|
||||
def test_section_open_at_eof_ends_at_last_loud_window():
|
||||
# Burst ends 1 s before EOF — less than min_gap, so the section is still
|
||||
# open when the windows run out. It must end where the sound ended, not at
|
||||
# the file duration; otherwise the clip carries a second of trailing
|
||||
# silence and disagrees with every section closed mid-file.
|
||||
rms = [0.002] * 1200
|
||||
rms[1180:1190] = [0.05] * 10 # loud 118.0 → 119.0, file is 120.0 s
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['start'] == 118.0
|
||||
assert sections[0]['end'] == 119.0
|
||||
|
||||
|
||||
def test_section_running_to_the_very_end_is_capped_at_duration():
|
||||
rms = [0.002] * 1200
|
||||
rms[1190:1200] = [0.05] * 10 # loud right up to EOF
|
||||
sections = _run(rms)
|
||||
assert len(sections) == 1
|
||||
assert sections[0]['end'] == 120.0
|
||||
|
||||
|
||||
def test_noise_floor_tracks_blocks_and_ignores_short_events():
|
||||
quiet_db = 20 * math.log10(0.002)
|
||||
db = [quiet_db] * 1200
|
||||
db[600:650] = [-20.0] * 50 # 5 s event must not raise its own floor
|
||||
floor = _noise_floor_db(db, WINDOW_DUR)
|
||||
assert len(floor) == len(db)
|
||||
assert all(abs(f - quiet_db) < 1.0 for f in floor)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Filename parsing / cut naming
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_recording_start_parses_standard_name():
|
||||
from datetime import datetime
|
||||
assert _recording_start("20260523_220000") == datetime(2026, 5, 23, 22, 0, 0)
|
||||
|
||||
|
||||
def test_recording_start_rejects_nonstandard_name():
|
||||
assert _recording_start("radio1_20260523") is None
|
||||
assert _recording_start("notes") is None
|
||||
|
||||
|
||||
def test_cut_filename_uses_wall_clock_span():
|
||||
# Recording started 22:00:00; cut covers 22:31:30 → 22:32:30.
|
||||
name = _cut_filename("20260523_220000", ".flac", 1890, 1950)
|
||||
assert name == "20260523_22-31-30_22-32-30.flac"
|
||||
|
||||
|
||||
def test_cut_filename_rolls_over_the_hour():
|
||||
name = _cut_filename("20260523_220000", ".wav", 3590, 3661)
|
||||
assert name == "20260523_22-59-50_23-01-01.wav"
|
||||
|
||||
|
||||
def test_cut_filename_falls_back_for_nonstandard_name():
|
||||
name = _cut_filename("mixtape", ".mp3", 740, 750.4)
|
||||
assert name == "mixtape_cut_740s-750s.mp3"
|
||||
names = [f['name'] for f in web.list_files(str(tmp_path))]
|
||||
assert names == ['20260523_090000.wav', '20260523_080000.wav']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -211,52 +43,11 @@ class _OfflineHandler(web._Handler):
|
||||
pass
|
||||
|
||||
|
||||
def _write_wav(path, seconds=2, framerate=8000):
|
||||
with wave.open(str(path), 'wb') as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(framerate)
|
||||
wf.writeframes(b'\x00\x00' * (seconds * framerate))
|
||||
|
||||
|
||||
def test_listing_is_ordered_by_the_filename_clock_not_mtime(tmp_path):
|
||||
# mtime is the last write (≈ the end of the recording) and drifts from the
|
||||
# name on any copy or round-trip. Here the older recording is touched last.
|
||||
import os
|
||||
early, late = tmp_path / '20260523_080000.wav', tmp_path / '20260523_090000.wav'
|
||||
_write_wav(early)
|
||||
_write_wav(late)
|
||||
os.utime(late, (1000, 1000)) # newest recording, oldest mtime
|
||||
|
||||
names = [f['name'] for f in web.list_files(str(tmp_path))]
|
||||
assert names == ['20260523_090000.wav', '20260523_080000.wav']
|
||||
|
||||
|
||||
def test_non_16_bit_wav_is_rejected_rather_than_misread(tmp_path):
|
||||
# RMS windowing decodes as int16 regardless of the declared sample width,
|
||||
# so an 8-bit file used to produce meaningless sections with no warning.
|
||||
path = tmp_path / 'eight_bit.wav'
|
||||
with wave.open(str(path), 'wb') as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(1)
|
||||
wf.setframerate(8000)
|
||||
wf.writeframes(b'\x80' * 8000)
|
||||
result = web.analyze_wav(path)
|
||||
assert 'error' in result
|
||||
assert '16-bit' in result['error']
|
||||
|
||||
|
||||
def test_16_bit_wav_still_analyses(tmp_path):
|
||||
path = tmp_path / 'ok.wav'
|
||||
_write_wav(path)
|
||||
assert 'error' not in web.analyze_wav(path)
|
||||
|
||||
|
||||
def test_clip_failure_after_headers_closes_connection_instead_of_writing_json(tmp_path):
|
||||
def test_clip_failure_after_headers_closes_connection_instead_of_writing_json(tmp_path, write_wav):
|
||||
# Once the audio headers are out, an error body would be a second response
|
||||
# on a keep-alive connection that already promised audio/wav plus an exact
|
||||
# Content-Length — desyncing every later request on that connection.
|
||||
_write_wav(tmp_path / 'rec.wav')
|
||||
write_wav(tmp_path / 'rec.wav')
|
||||
h = _OfflineHandler(tmp_path)
|
||||
|
||||
def boom(path, start, end):
|
||||
@@ -271,8 +62,8 @@ def test_clip_failure_after_headers_closes_connection_instead_of_writing_json(tm
|
||||
assert b'error' not in h.wfile.getvalue() # no JSON body
|
||||
|
||||
|
||||
def test_clip_failure_before_headers_still_reports_json(tmp_path):
|
||||
_write_wav(tmp_path / 'rec.wav')
|
||||
def test_clip_failure_before_headers_still_reports_json(tmp_path, write_wav):
|
||||
write_wav(tmp_path / 'rec.wav')
|
||||
h = _OfflineHandler(tmp_path)
|
||||
|
||||
def boom(path, start, end):
|
||||
|
||||
Reference in New Issue
Block a user