Commit 90e2cea6 by PLN (Algolia)

take-rec: arm, verify and summarise an Ardour take from the stem files

Verify reads the capture files, never a self-report: missing or flat
per orbit. Never binds UDP 3820 (gig-log owns Ardour's OSC feedback);
reads gig-log's rec lines instead. The master-rec toggle is only sent
when that stream says OFF, the orbit list comes from the session XML,
and halt stops the transport a stop must never forget.
parent 8831e216
"""Tests for tools/take-rec.py — the REC button's brain.
The three properties that actually matter:
1. STRIP MAPPING: the session XML's route order is the one guess in the
file, so it must survive busses and non-Tidal routes — an off-by-one
here arms the WRONG track, which is worse than arming none.
2. VERIFY IS FILES, NOT HOPES: a missing stem must say MISSING even when
every UI would say fine (the D8 case), and a growing file must say ok.
3. NO PORT WARS: nothing here may bind UDP 3820 — gig-log owns it.
"""
import importlib.util
import json
import os
import sys
import threading
import time
from pathlib import Path
import pytest
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent))
spec = importlib.util.spec_from_file_location(
"_take_rec", HERE.parent / "take-rec.py")
tr = importlib.util.module_from_spec(spec)
spec.loader.exec_module(tr)
# ---- 1. strip mapping -------------------------------------------------------
ARDOUR_XML = """<?xml version="1.0"?>
<Mixer>
<Routes>
<Route name="Click"/>
<Route name="Tidal 01"/><Route name="Tidal 02"/>
<Route name="Mixbus A"/>
<Route name="Tidal 08"/>
</Routes>
</Mixer>
"""
def test_strip_ids_skip_non_tidal_routes(tmp_path):
sess = tmp_path / "Tidal Live" # the .ardour file is <dirname>.ardour
sess.mkdir()
(sess / "Tidal Live.ardour").write_text(ARDOUR_XML)
ids = tr.tidal_strip_ids(sess)
# 1-based over ALL routes: Click=1, Tidal 01=2, Tidal 02=3, Mixbus=4, 08=5
assert ids == {1: 2, 2: 3, 8: 5}
def test_strip_ids_absent_session_is_empty_not_crash(tmp_path):
assert tr.tidal_strip_ids(tmp_path) == {}
# ---- 2. verify reads files --------------------------------------------------
def make_session(tmp_path, orbit_files):
sess = tmp_path / "Sess"
af = sess / "interchange" / "Sess" / "audiofiles"
af.mkdir(parents=True)
now = time.time()
for name in orbit_files:
f = af / name
f.write_bytes(b"\0" * 1000)
os.utime(f, (now, now))
return sess
def test_verify_screams_missing_orbit_per_number(tmp_path, capsys):
sess = make_session(tmp_path, ["Take9_Tidal 01-1%L.wav"])
rc = tr.verify([1, 8], session=sess, delta_s=0.1)
captured = capsys.readouterr()
out, err = captured.out, captured.err
assert rc == 1
# orbit 1's file exists but nothing grows it — STALE is the honest verdict
assert "STALE" in out and "orbit 1" in out
assert "orbit 8: MISSING" in out
assert "2/2 orbit(s) NOT recording" in err
def test_verify_sees_growth_not_stale(tmp_path, capsys):
sess = make_session(tmp_path, ["Take9_Tidal 03-1%L.wav"])
def grow():
time.sleep(0.05)
f = sess / "interchange" / "Sess" / "audiofiles" / "Take9_Tidal 03-1%L.wav"
with open(f, "ab") as fh:
fh.write(b"\0" * 5000)
threading.Thread(target=grow).start()
rc = tr.verify([3], session=sess, delta_s=0.2)
out = capsys.readouterr().out
assert rc == 0
assert "ok" in out and "+5000" in out
def test_verify_ignores_old_takes(tmp_path, capsys):
sess = make_session(tmp_path, ["Take3_Tidal 02-1%L.wav"])
old = sess / "interchange" / "Sess" / "audiofiles" / "Take3_Tidal 02-1%L.wav"
os.utime(old, (time.time() - 3600, time.time() - 3600)) # an hour ago
rc = tr.verify([2], session=sess, delta_s=0.1)
assert rc == 1 # stale files are not recording
# ---- gig-log lens (no port binding) ------------------------------------------
def test_latest_rec_line_reads_newest_log(tmp_path, monkeypatch):
old = tmp_path / "gig-20260101-000000.jsonl"
new = tmp_path / "gig-20260102-000000.jsonl"
old.write_text(json.dumps({"k": "rec", "rec": True, "t": 1.0}) + "\n")
new.write_text(json.dumps({"k": "cc", "v": 3}) + "\n"
+ json.dumps({"k": "rec", "rec": False, "t": 2.0}) + "\n")
os.utime(old, (1_000, 1_000)) # mtime picks the file; ties are
os.utime(new, (2_000, 2_000)) # otherwise filesystem luck
monkeypatch.setattr(tr, "LOG_DIR", tmp_path)
d = tr.latest_rec_line()
assert d is not None and d["rec"] is False and d["t"] == 2.0
def test_latest_rec_line_no_logs_is_none(tmp_path, monkeypatch):
monkeypatch.setattr(tr, "LOG_DIR", tmp_path)
assert tr.latest_rec_line() is None
def test_nothing_binds_udp_3820():
"""The one hard rule: gig-log owns 3820. The mechanical guard is not the
string (docs mention the port) but the SOCKET calls: this file sends,
it never binds."""
src = (HERE.parent / "take-rec.py").read_text()
assert ".bind(" not in src
assert ".sendto(" in src # the only socket verb allowed
# ---- summary -----------------------------------------------------------------
def test_summary_reports_take_and_missing_orbits(tmp_path, capsys):
sess = make_session(tmp_path, ["Take12_Tidal 01-1%L.wav",
"Take12_Tidal 02-1%L.wav"])
rc = tr.summary(session=sess)
out = capsys.readouterr().out
assert "Take12" in out and "orbits 2" in out
assert rc == 1 # 12 of 14 missing
def test_summary_no_takes(tmp_path, capsys):
(tmp_path / "Sess").mkdir()
rc = tr.summary(session=tmp_path / "Sess")
assert rc == 1
assert "no takes" in capsys.readouterr().err
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment