test: add test suite for player, api and widgets

tests/test_spectrum.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import numpy as np
+
+from somafm_tui import spectrum as spectrum_module
+from somafm_tui.spectrum import Frame, SpectrumEngine
+
+PCM_SCRIPT = """#!/usr/bin/env python3
+import math, struct, sys, time
+sr = 44100
+samples = [int(0.6 * 32767 * math.sin(2 * math.pi * 1000 * i / sr)) for i in range(4096)]
+chunk = struct.pack("<%dh" % len(samples), *samples)
+out = sys.stdout.buffer
+try:
+    while True:
+        out.write(chunk)
+        out.flush()
+        time.sleep(0.08)
+except (BrokenPipeError, KeyboardInterrupt):
+    pass
+"""
+
+
+def write_pcm_script(path: Path) -> None:
+    path.write_text(PCM_SCRIPT)
+    path.chmod(0o755)
+
+
+def wait_for_spectrum(engine: SpectrumEngine, status: str, timeout: float = 6.0):
+    deadline = time.monotonic() + timeout
+    snap = engine.snapshot()
+    while time.monotonic() < deadline:
+        snap = engine.snapshot()
+        if snap.status == status and sum(snap.bands) > 1.0:
+            break
+        time.sleep(0.05)
+    return snap
+
+
+def test_fft_detects_a_1khz_tone() -> None:
+    engine = SpectrumEngine(ffmpeg_path="")
+    sample_rate = 44100
+    t = np.arange(8192) / sample_rate
+    tone = np.sin(2 * np.pi * 1000 * t) * 0.5
+    engine.feed_pcm((tone * 32767).astype("<i2").tobytes())
+
+    snap = engine.snapshot()
+    assert snap.status == "live"
+    assert snap.label == "FFT"
+    assert len(snap.bands) == 72
+    top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
+    expected = 72 * np.log(1000.0 / 35.0) / np.log(16000.0 / 35.0)
+    assert abs(top - expected) <= 2
+    assert snap.bands[top] > 0.8
+    assert snap.peaks[top] >= snap.bands[top] - 1e-6
+    assert any(abs(value) > 0.3 for value in snap.wave)
+    assert 0.0 < snap.level <= 1.0
+
+
+def test_frame_labels() -> None:
+    assert Frame((), (), (), 0.0, "monitor").label == "SYNC"
+    assert Frame((), (), (), 0.0, "live").label == "FFT"
+    assert Frame((), (), (), 0.0, "sim").label == "SIM"
+    assert Frame((), (), (), 0.0, "idle").label == "IDLE"
+    assert Frame((), (), (), 0.0, "off").label == "OFF"
+    assert Frame((), (), (), 0.0, "monitor").is_live is True
+
+
+def test_simulation_fallback() -> None:
+    engine = SpectrumEngine(ffmpeg_path="", source_mode="sim")
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/demo")
+        deadline = time.monotonic() + 5
+        snap = engine.snapshot()
+        while time.monotonic() < deadline:
+            snap = engine.snapshot()
+            if snap.status == "sim" and max(snap.bands) > 0.2:
+                break
+            time.sleep(0.05)
+        assert snap.status == "sim"
+        assert snap.label == "SIM"
+        assert max(snap.bands) > 0.2
+        assert 0.0 <= snap.level <= 1.0
+    finally:
+        engine.stop()
+
+
+def test_monitor_capture_prefers_real_output(tmp_path, monkeypatch) -> None:
+    script = tmp_path / "parec"
+    write_pcm_script(script)
+
+    def fake_which(name: str) -> str | None:
+        return str(script) if name == "parec" else None
+
+    monkeypatch.setattr(spectrum_module.shutil, "which", fake_which)
+    engine = SpectrumEngine(source_mode="auto")
+    assert engine.monitor_available is True
+
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/live")
+        snap = wait_for_spectrum(engine, "monitor")
+        assert snap.status == "monitor", snap
+        assert snap.label == "SYNC"
+
+        top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
+        assert 36 <= top <= 42
+        assert snap.bands[top] > 0.5
+
+        engine.set_source(None)
+        time.sleep(0.3)
+        assert engine.snapshot().status == "idle"
+    finally:
+        engine.stop()
+
+
+def test_stream_tap_pipeline(tmp_path) -> None:
+    script = tmp_path / "fake-ffmpeg"
+    write_pcm_script(script)
+
+    engine = SpectrumEngine(ffmpeg_path=str(script), source_mode="stream")
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/stream")
+        snap = wait_for_spectrum(engine, "live")
+        assert snap.status == "live", snap
+        assert snap.label == "FFT"
+        top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
+        assert 36 <= top <= 42
+        assert snap.bands[top] > 0.5
+
+        engine.set_source(None)
+        time.sleep(0.3)
+        assert engine.snapshot().status == "idle"
+    finally:
+        engine.stop()