From b9d3d7e8090d5b0f1f0b45bcc6460ca369a160f1 Mon Sep 17 00:00:00 2001
From: beefcarver <beefcarver@tilde.club>
Date: Sun, 1 Nov 2026 09:45:47 -0500
Subject: [PATCH] test(spectrum): cover monitor recovery across tuning

---
 tests/test_spectrum.py | 121 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 121 insertions(+)

diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py
index 0d4cf6f..b742b39 100644
--- a/tests/test_spectrum.py
+++ b/tests/test_spectrum.py
@@ -324,3 +324,124 @@ def test_dead_monitor_falls_back_instead_of_flat_bars(tmp_path) -> None:
     finally:
         engine._stop_proc(proc)
         engine.stop()
+
+
+def test_monitor_survives_silence_after_audio(tmp_path, monkeypatch) -> None:
+    # Tuning a station leaves mpv's sink silent while the new stream buffers.
+    # That gap must not be mistaken for a dead monitor: only a capture that has
+    # never delivered audio is written off, otherwise SYNC would be lost (and
+    # replaced by the drifting stream tap) for the rest of the session.
+    script = tmp_path / "parec"
+    script.write_text(
+        "#!/usr/bin/env python3\n"
+        "import math, struct, sys, time\n"
+        "sr = 44100\n"
+        "n = 2048\n"
+        "tone = struct.pack('<%dh' % n, *["
+        "int(0.6 * 32767 * math.sin(2 * math.pi * 1000 * i / sr)) for i in range(n)])\n"
+        "silence = b'\\0' * (2 * n)\n"
+        "out = sys.stdout.buffer\n"
+        "for _ in range(40):\n"
+        "    out.write(tone)\n"
+        "out.flush()\n"
+        "for _ in range(400):\n"
+        "    out.write(silence)\n"
+        "out.flush()\n"
+        "time.sleep(30)\n"
+    )
+    script.chmod(0o755)
+
+    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(ffmpeg_path="")
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/live")
+        assert wait_for_spectrum(engine, "monitor").status == "monitor"
+
+        # The script now feeds far more than SILENT_MONITOR_BLOCKS of digital
+        # silence.  Give the analyser time to consume it, then make sure the
+        # capture is still alive and still SYNC.
+        time.sleep(1.0)
+        snap = engine.snapshot()
+        assert snap.status == "monitor", snap
+        assert snap.label == "SYNC"
+        assert engine.monitor_available is True
+        assert engine._monitor_failed is False
+    finally:
+        engine.stop()
+
+
+class SequencedTap(FakeTap):
+    """A tap that hands out a fixed list of buffers, then goes quiet."""
+
+    def __init__(self, chunks: list[bytes], **kwargs) -> None:
+        super().__init__(**kwargs)
+        self._chunks = list(chunks)
+
+    def read(self, timeout: float | None = 1.0) -> bytes | None:
+        if self.stopped:
+            return None
+        return self._chunks.pop(0) if self._chunks else None
+
+
+def test_tap_survives_silence_after_audio(monkeypatch) -> None:
+    # A real tap that has already captured audio is allowed to go quiet while
+    # a new station buffers; only a tap that never produced audio means the
+    # System Audio Recording permission is missing.
+    engine = SpectrumEngine(ffmpeg_path="")
+    engine._parec = None
+    engine._darwin = True
+    monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
+    chunks = [tap_pcm(1000, 4096)] + [b"\x00" * (4096 * 8)] * 200
+    tap = SequencedTap(chunks)
+    monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
+
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/live")
+        assert wait_for_spectrum(engine, "monitor").status == "monitor"
+        time.sleep(0.5)
+        snap = engine.snapshot()
+        assert snap.status == "monitor", snap
+        assert engine.monitor_available is True
+        assert engine._monitor_failed is False
+        assert tap.stopped is False
+    finally:
+        engine.stop()
+
+
+def test_tuning_clears_a_monitor_failure_and_retries_sync(tmp_path, monkeypatch) -> None:
+    # Once an output capture has failed (or been given up on after a tuning
+    # gap), a station change must clear that state and try SYNC again instead
+    # of staying on the drifting FFT stream for the rest of the session.
+    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(ffmpeg_path="")
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/a")
+        assert wait_for_spectrum(engine, "monitor").status == "monitor"
+
+        # Simulate the latch that a silent retry used to leave behind.
+        engine._monitor_failed = True
+        engine._tap_retry_at = time.monotonic() + 60.0
+
+        for index in range(10):
+            engine.set_source(f"https://example.invalid/b{index}")
+            time.sleep(0.02)
+
+        snap = wait_for_spectrum(engine, "monitor")
+        assert snap.status == "monitor", snap
+        assert engine._monitor_failed is False
+        assert engine.monitor_available is True
+        assert engine._monitor_worked is True
+    finally:
+        engine.stop()