Files
shawn 349139ad1f
Lint & Validate / lint (push) Has been cancelled
P5-R3: fix count-in recursion + test buffer loop size
2026-05-19 21:08:15 -04:00

1643 lines
60 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tests for the backing track player module.
Covers: types, audio loader, playlist manager, transport,
metronome, and the main backing track player orchestrator.
"""
from __future__ import annotations
import io
import json
import struct
import tempfile
import wave
from pathlib import Path
import numpy as np
import pytest
from src.backing.types import (
PlayMode,
TrackState,
Track,
Playlist,
BackingPlayerConfig,
)
from src.backing.loader import AudioLoader, AudioData, load_audio, SUPPORTED_EXTENSIONS
from src.backing.playlist import PlaylistManager
from src.backing.transport import JACKTransport, TransportState, JACKTransportState
from src.backing.metronome import Metronome, ClickSound
from src.backing.player import BackingTrackPlayer
# ═══════════════════════════════════════════════════════════════════════════════
# Test helpers
# ═══════════════════════════════════════════════════════════════════════════════
def _make_wav(
duration: float = 1.0,
sample_rate: int = 48000,
channels: int = 1,
frequency: float = 440.0,
) -> bytes:
"""Generate a WAV file in memory as bytes."""
n_samples = int(duration * sample_rate)
t = np.linspace(0, duration, n_samples, endpoint=False, dtype=np.float32)
if channels == 1:
samples = (np.sin(2 * np.pi * frequency * t) * 0.5).astype(np.float32)
else:
left = (np.sin(2 * np.pi * frequency * t) * 0.5).astype(np.float32)
right = (np.sin(2 * np.pi * (frequency * 1.5) * t) * 0.5).astype(np.float32)
samples = np.column_stack([left, right])
# Convert to int16 for WAV
int_samples = (samples * 32767).astype(np.int16)
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
wf.setnchannels(channels)
wf.setsampwidth(2) # 16-bit
wf.setframerate(sample_rate)
wf.writeframes(int_samples.tobytes())
return buf.getvalue()
def _write_temp_wav(
duration: float = 1.0,
sample_rate: int = 48000,
channels: int = 1,
frequency: float = 440.0,
suffix: str = ".wav",
) -> Path:
"""Write a temporary WAV file and return its path."""
data = _make_wav(duration, sample_rate, channels, frequency)
tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
tmp.write(data)
tmp.close()
return Path(tmp.name)
# ═══════════════════════════════════════════════════════════════════════════════
# Types tests
# ═══════════════════════════════════════════════════════════════════════════════
class TestPlayMode:
def test_enum_values(self):
assert PlayMode.ONE_SHOT == "one_shot"
assert PlayMode.LOOP == "loop"
assert PlayMode.SEGUE == "segue"
def test_str_enum_coercion(self):
assert PlayMode("loop") == PlayMode.LOOP
assert PlayMode("one_shot") == PlayMode.ONE_SHOT
class TestTrackState:
def test_enum_values(self):
assert TrackState.IDLE == "idle"
assert TrackState.PLAYING == "playing"
assert TrackState.PAUSED == "paused"
assert TrackState.STOPPED == "stopped"
assert TrackState.FINISHED == "finished"
assert TrackState.ERROR == "error"
class TestTrack:
def test_defaults(self):
t = Track(id="t1")
assert t.id == "t1"
assert t.title == ""
assert t.volume == 1.0
assert t.pan == 0.5
assert t.muted is False
assert t.soloed is False
assert t.play_mode == PlayMode.ONE_SHOT
assert t.state == TrackState.IDLE
assert t.duration == 0.0
assert t.loaded is False
def test_file_name(self):
t = Track(id="t1", file_path="/music/song.wav")
assert t.file_name == "song.wav"
def test_file_name_empty(self):
t = Track(id="t2")
assert t.file_name == ""
def test_has_loop_region(self):
t = Track(id="t1", loop_start=1.0, loop_end=5.0)
assert t.has_loop_region is True
def test_no_loop_region(self):
t = Track(id="t1", loop_start=1.0, loop_end=0.0)
assert t.has_loop_region is False
def test_effective_loop_end(self):
t = Track(id="t1", duration=60.0, loop_end=30.0)
assert t.effective_loop_end == 30.0
def test_effective_loop_end_no_set(self):
t = Track(id="t1", duration=60.0, loop_end=0.0)
assert t.effective_loop_end == 60.0
class TestPlaylist:
def test_empty(self):
pl = Playlist(name="Test")
assert pl.name == "Test"
assert pl.track_count == 0
assert pl.current_track is None
def test_add_tracks(self):
pl = Playlist(name="Setlist")
pl.tracks = [
Track(id="t0", title="Song A"),
Track(id="t1", title="Song B"),
Track(id="t2", title="Song C"),
]
assert pl.track_count == 3
def test_current_track(self):
pl = Playlist(name="Setlist", current_index=1)
pl.tracks = [Track(id="t0"), Track(id="t1"), Track(id="t2")]
assert pl.current_track.id == "t1"
def test_current_track_none(self):
pl = Playlist(name="Empty", current_index=-1)
assert pl.current_track is None
def test_next_track(self):
pl = Playlist(name="Setlist", current_index=0)
pl.tracks = [Track(id="t0"), Track(id="t1"), Track(id="t2")]
n = pl.next_track()
assert n.id == "t1"
assert pl.current_index == 1
def test_next_track_at_end(self):
pl = Playlist(name="Setlist", current_index=2)
pl.tracks = [Track(id="t0"), Track(id="t1"), Track(id="t2")]
assert pl.next_track() is None
def test_prev_track(self):
pl = Playlist(name="Setlist", current_index=1)
pl.tracks = [Track(id="t0"), Track(id="t1"), Track(id="t2")]
p = pl.prev_track()
assert p.id == "t0"
assert pl.current_index == 0
def test_prev_track_at_start(self):
pl = Playlist(name="Setlist", current_index=0)
pl.tracks = [Track(id="t0"), Track(id="t1")]
assert pl.prev_track() is None
def test_get_track(self):
pl = Playlist(name="Setlist")
pl.tracks = [Track(id="t0"), Track(id="t1")]
assert pl.get_track(0).id == "t0"
assert pl.get_track(1).id == "t1"
assert pl.get_track(2) is None
assert pl.get_track(-1) is None
def test_total_duration(self):
pl = Playlist(name="Dur")
pl.tracks = [
Track(id="t0", duration=10.0),
Track(id="t1", duration=20.0),
Track(id="t2", duration=5.0),
]
assert pl.total_duration == 35.0
class TestBackingPlayerConfig:
def test_defaults(self):
cfg = BackingPlayerConfig()
assert cfg.sample_rate == 48000
assert cfg.buffer_size == 256
assert cfg.channels == 2
assert cfg.jack_client_name == "pi_mixer_backing"
assert cfg.jack_transport_enabled is True
assert cfg.metronome_enabled is True
assert cfg.count_in_bars == 2
assert cfg.time_signature_num == 4
assert cfg.time_signature_den == 4
assert cfg.auto_advance is True
# ═══════════════════════════════════════════════════════════════════════════════
# AudioLoader tests
# ═══════════════════════════════════════════════════════════════════════════════
class TestAudioLoader:
def test_load_wav_mono(self):
"""Load a mono WAV file."""
wav = _write_temp_wav(duration=1.0, sample_rate=44100, channels=1)
try:
audio = AudioLoader.load(wav)
assert isinstance(audio, AudioData)
assert audio.sample_rate == 44100
assert audio.channels == 1
assert abs(audio.duration - 1.0) < 0.05
assert audio.num_frames > 0
assert audio.samples.ndim == 1
assert audio.samples.dtype == np.float32
assert not audio.is_stereo
finally:
wav.unlink()
def test_load_wav_stereo(self):
"""Load a stereo WAV file."""
wav = _write_temp_wav(duration=0.5, sample_rate=48000, channels=2)
try:
audio = AudioLoader.load(wav)
assert audio.channels == 2
assert audio.is_stereo
assert audio.samples.ndim == 2
assert audio.samples.shape[1] == 2
finally:
wav.unlink()
def test_load_nonexistent_file(self):
"""Loading a nonexistent file raises FileNotFoundError."""
with pytest.raises(FileNotFoundError):
AudioLoader.load("/tmp/nonexistent_file_xyz123.wav")
def test_load_unsupported_format(self):
"""Loading an unsupported format raises ValueError."""
tmp = tempfile.NamedTemporaryFile(suffix=".xyz", delete=False)
tmp.close()
try:
with pytest.raises(ValueError, match="Unsupported audio format"):
AudioLoader.load(tmp.name)
finally:
Path(tmp.name).unlink()
def test_probe_wav(self):
"""Probe WAV metadata without loading samples."""
wav = _write_temp_wav(duration=2.0, sample_rate=44100, channels=1)
try:
info = AudioLoader.probe(wav)
assert info["sample_rate"] == 44100
assert info["channels"] == 1
assert abs(info["duration"] - 2.0) < 0.05
finally:
wav.unlink()
def test_to_mono(self):
"""Stereo → mono mixdown."""
wav = _write_temp_wav(duration=0.2, sample_rate=48000, channels=2)
try:
audio = AudioLoader.load(wav)
mono = audio.to_mono()
assert mono.ndim == 1
assert len(mono) == audio.num_frames
finally:
wav.unlink()
def test_load_with_resample(self):
"""Load and resample to a target rate."""
wav = _write_temp_wav(duration=0.5, sample_rate=22050, channels=1)
try:
audio = AudioLoader.load(wav, target_sr=44100)
assert audio.sample_rate == 44100
assert audio.num_frames > 0
finally:
wav.unlink()
def test_audio_data_properties(self):
"""AudioData properties work correctly."""
samples = np.zeros((1000,), dtype=np.float32)
ad = AudioData(
samples=samples,
sample_rate=48000,
channels=1,
duration=1000 / 48000,
file_path="/tmp/test.wav",
format_name="WAV",
)
assert ad.num_frames == 1000
assert not ad.is_stereo
def test_convenience_load_audio(self):
"""load_audio convenience function works."""
wav = _write_temp_wav(duration=0.1, sample_rate=44100, channels=1)
try:
audio = load_audio(wav)
assert isinstance(audio, AudioData)
finally:
wav.unlink()
def test_supported_extensions(self):
"""All required formats are in supported extensions."""
required = {".wav", ".flac", ".mp3", ".aiff", ".aif"}
assert required.issubset(SUPPORTED_EXTENSIONS)
# ═══════════════════════════════════════════════════════════════════════════════
# PlaylistManager tests
# ═══════════════════════════════════════════════════════════════════════════════
class TestPlaylistManager:
def test_create_and_get(self):
pm = PlaylistManager()
pl = pm.create("My Set")
assert pl.name == "My Set"
assert pm.get("My Set") is pl
assert pm.active is pl
def test_list_names(self):
pm = PlaylistManager()
pm.create("A")
pm.create("B")
assert set(pm.list_names()) == {"A", "B"}
def test_delete(self):
pm = PlaylistManager()
pm.create("Keep")
pm.create("Delete")
assert pm.delete("Delete") is True
assert pm.get("Delete") is None
assert pm.get("Keep") is not None
def test_delete_nonexistent(self):
pm = PlaylistManager()
assert pm.delete("Nope") is False
def test_delete_active(self):
pm = PlaylistManager()
pm.create("Active")
assert pm.delete("Active") is False # Can't delete active
def test_set_active(self):
pm = PlaylistManager()
pm.create("First")
pm.create("Second")
assert pm.active.name == "First"
pm.set_active("Second")
assert pm.active.name == "Second"
def test_set_active_nonexistent(self):
pm = PlaylistManager()
assert pm.set_active("Nope") is False
def test_add_track(self):
pm = PlaylistManager()
pm.create("List")
tid = pm.add_track("/music/song.wav", title="Song 1")
assert tid == "bt_000"
assert pm.active.track_count == 1
def test_remove_track(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/a.wav")
pm.add_track("/music/b.wav")
pm.add_track("/music/c.wav")
assert pm.remove_track("bt_001") is True
assert pm.active.track_count == 2
def test_move_track(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/a.wav", title="A")
pm.add_track("/music/b.wav", title="B")
pm.add_track("/music/c.wav", title="C")
# Move C (index 2) to front
pm.move_track("bt_002", 0)
tracks = pm.active.tracks
assert tracks[0].id == "bt_002"
assert tracks[1].id == "bt_000"
def test_clear(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/a.wav")
pm.add_track("/music/b.wav")
assert pm.clear() == 2
assert pm.active.track_count == 0
assert pm.active.current_index == -1
def test_track_controls(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/song.wav")
assert pm.set_volume("bt_000", 0.75)
assert pm.get_track("bt_000").volume == 0.75
assert pm.set_pan("bt_000", 0.3)
assert pm.get_track("bt_000").pan == 0.3
assert pm.set_mute("bt_000", True)
assert pm.get_track("bt_000").muted is True
assert pm.set_solo("bt_000", True)
assert pm.get_track("bt_000").soloed is True
assert pm.set_play_mode("bt_000", PlayMode.LOOP)
assert pm.get_track("bt_000").play_mode == PlayMode.LOOP
def test_set_volume_clamped(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/song.wav")
pm.set_volume("bt_000", 2.0)
assert pm.get_track("bt_000").volume == 1.0
pm.set_volume("bt_000", -0.5)
assert pm.get_track("bt_000").volume == 0.0
def test_select_track(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/a.wav", title="A")
pm.add_track("/music/b.wav", title="B")
t = pm.select_track(1)
assert t.title == "B"
assert pm.active.current_index == 1
def test_select_invalid(self):
pm = PlaylistManager()
pm.create("List")
assert pm.select_track(999) is None
def test_next_prev_track(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/a.wav")
pm.add_track("/music/b.wav")
pm.add_track("/music/c.wav")
pm.select_track(1)
n = pm.next_track()
assert n.id == "bt_002"
p = pm.prev_track()
assert p.id == "bt_001"
def test_cue_and_loop_region(self):
pm = PlaylistManager()
pm.create("List")
pm.add_track("/music/song.wav")
pm.set_cue_point("bt_000", 5.0)
assert pm.get_track("bt_000").cue_point == 5.0
pm.set_loop_region("bt_000", 2.0, 8.0)
t = pm.get_track("bt_000")
assert t.loop_start == 2.0
assert t.loop_end == 8.0
def test_serialization(self):
pm = PlaylistManager()
pm.create("Gig")
pm.add_track("/music/a.wav", title="Opener", play_mode=PlayMode.ONE_SHOT)
pm.add_track("/music/b.wav", title="Closer", play_mode=PlayMode.LOOP)
pm.set_volume("bt_000", 0.8)
pm.set_pan("bt_001", 0.3)
pm.set_mute("bt_000", True)
d = pm.to_dict()
assert d["name"] == "Gig"
assert len(d["tracks"]) == 2
assert d["tracks"][0]["volume"] == 0.8
assert d["tracks"][0]["muted"] is True
assert d["tracks"][1]["pan"] == 0.3
# Round-trip
pm2 = PlaylistManager()
pm2.from_dict(d)
assert pm2.get("Gig").track_count == 2
assert pm2.get_track("bt_000").volume == 0.8
assert pm2.get_track("bt_000").muted is True
assert pm2.get_track("bt_001").pan == 0.3
def test_json_serialization(self):
pm = PlaylistManager()
pm.create("Test")
pm.add_track("/music/a.wav", title="A")
json_str = pm.to_json()
data = json.loads(json_str)
assert data["name"] == "Test"
assert len(data["tracks"]) == 1
pm2 = PlaylistManager()
pm2.from_json(json_str)
assert pm2.get("Test").track_count == 1
# ═══════════════════════════════════════════════════════════════════════════════
# JACK Transport tests
# ═══════════════════════════════════════════════════════════════════════════════
class TestJACKTransport:
def test_init(self):
"""JACKTransport initializes with safe defaults (JACK may be absent)."""
jt = JACKTransport(sample_rate=48000)
assert jt._sample_rate == 48000
# available may be True or False depending on environment
assert isinstance(jt.state, JACKTransportState)
def test_state_defaults(self):
"""Transport state has sensible defaults."""
jt = JACKTransport()
st = jt.state
assert st.sample_rate == 48000
assert st.tempo == 120.0
assert isinstance(st.state, TransportState)
def test_get_state_returns_valid(self):
"""get_state always returns a valid JACKTransportState."""
jt = JACKTransport()
st = jt.get_state()
assert isinstance(st, JACKTransportState)
assert st.tempo > 0
assert st.sample_rate > 0
def test_start_stop_no_jack(self):
"""start/stop don't crash without JACK."""
jt = JACKTransport()
# These should not raise
jt.start()
jt.stop()
jt.toggle()
def test_locate_no_jack(self):
"""locate updates internal state even without JACK."""
jt = JACKTransport(sample_rate=48000)
jt.locate(48000) # 1 second
st = jt.get_state()
# If JACK is not available, internal state should be updated
if not jt.available:
assert jt.state.frame == 48000
def test_set_tempo_no_jack(self):
"""set_tempo updates internal state without JACK."""
jt = JACKTransport()
jt.set_tempo(140.0)
if not jt.available:
assert jt.state.tempo == 140.0
def test_callback_registration(self):
"""Callbacks can be registered and removed."""
jt = JACKTransport()
calls = []
def cb(st):
calls.append(st)
jt.on_transport_change(cb)
assert len(jt._callbacks) == 1
jt.remove_callback(cb)
assert len(jt._callbacks) == 0
def test_locate_seconds(self):
jt = JACKTransport(sample_rate=48000)
jt.locate_seconds(2.5)
if not jt.available:
assert jt.state.frame == 120000 # 2.5 * 48000
class TestJACKTransportState:
def test_defaults(self):
st = JACKTransportState()
assert st.state == TransportState.UNKNOWN
assert st.tempo == 120.0
assert st.position == 0.0
assert st.frame == 0
def test_custom_values(self):
st = JACKTransportState(
state=TransportState.ROLLING,
tempo=140.0,
position=10.0,
frame=480000,
sample_rate=48000,
bar=3,
beat=2,
)
assert st.state == TransportState.ROLLING
assert st.tempo == 140.0
assert st.bar == 3
assert st.beat == 2
# ═══════════════════════════════════════════════════════════════════════════════
# Metronome tests
# ═══════════════════════════════════════════════════════════════════════════════
class TestClickSound:
def test_generate_sine(self):
cs = ClickSound(frequency=1000.0, duration_ms=15.0, volume=0.5, waveform="sine")
samples = cs.generate(48000)
assert samples.dtype == np.float32
assert len(samples) > 0
assert np.max(np.abs(samples)) <= 0.5
def test_generate_square(self):
cs = ClickSound(waveform="square")
samples = cs.generate(48000)
assert samples.dtype == np.float32
assert len(samples) > 0
def test_generate_noise(self):
cs = ClickSound(waveform="noise")
samples = cs.generate(48000)
assert len(samples) > 0
def test_generate_click(self):
cs = ClickSound(waveform="click")
samples = cs.generate(48000)
assert len(samples) > 0
assert abs(samples[0]) > 0 # first sample should be non-zero
def test_zero_duration_handled(self):
cs = ClickSound(duration_ms=0.0)
samples = cs.generate(48000)
assert len(samples) > 0 # At least 1 sample
def test_low_volume(self):
cs = ClickSound(volume=0.01)
samples = cs.generate(48000)
assert np.max(np.abs(samples)) <= 0.01
class TestMetronome:
def test_init(self):
m = Metronome(sample_rate=48000)
assert m.sample_rate == 48000
assert m.enabled is True
assert m.time_sig_num == 4
assert m.count_in_bars == 2
def test_enable_disable(self):
m = Metronome()
m.enabled = False
assert m.enabled is False
m.enabled = True
assert m.enabled is True
def test_set_volume(self):
m = Metronome()
m.set_volume(0.3)
assert m.strong_click.volume == 0.3
assert m.weak_click.volume == pytest.approx(0.24, rel=0.1)
def test_generate_click_strong(self):
"""Beat 1 generates strong click."""
m = Metronome(sample_rate=48000)
click = m.generate_click(beat=1, bar=1)
assert len(click) > 0
assert click.dtype == np.float32
def test_generate_click_weak(self):
"""Beat 2 generates weak click."""
m = Metronome(sample_rate=48000)
click = m.generate_click(beat=2, bar=1)
assert len(click) > 0
def test_generate_click_disabled(self):
"""Disabled metronome returns empty click."""
m = Metronome()
m.enabled = False
click = m.generate_click(beat=1, bar=1)
assert len(click) == 0
def test_generate_click_count_in_bypasses_enabled(self):
"""Count-in clicks are generated even when metronome is disabled."""
m = Metronome()
m.enabled = False
click = m.generate_click(beat=1, bar=1, is_count_in=True)
assert len(click) > 0
def test_generate_clicks_for_position(self):
"""Generate clicks for a buffer at a given position."""
m = Metronome(sample_rate=48000)
buf = m.generate_clicks_for_position(
position_seconds=0.0,
tempo=120.0,
bar=1,
beat=1,
buffer_size=48000, # 1 second of audio
)
assert buf.dtype == np.float32
assert len(buf) == 48000
# At 120 BPM, 0.5s per beat. In a 1s buffer starting at beat 1,
# we should have a click at the start (beat 1, already past) and
# possibly one at beat 2 (offset 0.5s)
# The click for beat 2 should appear around sample 24000
assert np.any(np.abs(buf[23000:25000]) > 0.001)
def test_generate_clicks_disabled(self):
"""Disabled metronome returns silence."""
m = Metronome(sample_rate=48000)
m.enabled = False
buf = m.generate_clicks_for_position(0, 120, 1, 1, 1024)
assert np.all(buf == 0.0)
def test_count_in(self):
"""Count-in generates clicks for a set number of bars."""
m = Metronome(sample_rate=48000)
m.count_in_bars = 2
m.time_sig_num = 4 # 4/4: 8 beats of count-in
m.start_count_in(bars=2)
assert m.is_count_in_active() is True
# Generate enough buffers to exhaust the count-in
# 8 beats * 24000 samples/beat / 12000 buffer = 16 buffers needed
buf_size = 12000 # 0.25s at 48k
for _ in range(20): # 20 > 16, safe margin
buf = m.generate_clicks_for_position(0, 120, 1, 1, buf_size)
# Count-in should be complete
assert m.is_count_in_active() is False
def test_count_in_stop(self):
"""Stopping count-in mid-way works."""
m = Metronome()
m.start_count_in()
assert m.is_count_in_active() is True
m.stop_count_in()
assert m.is_count_in_active() is False
def test_stereo_clicks(self):
"""Stereo output has correct shape."""
m = Metronome(sample_rate=48000)
buf = m.generate_stereo_clicks(0.0, 120.0, 1, 1, 1024)
assert buf.shape == (1024, 2)
assert buf.dtype == np.float32
def test_tempo_affects_click_spacing(self):
"""Faster tempo = clicks closer together."""
m = Metronome(sample_rate=48000)
# At 240 BPM, ~0.25s per beat
buf_fast = m.generate_clicks_for_position(0.0, 240.0, 1, 1, 12000)
# At 60 BPM, ~1.0s per beat
buf_slow = m.generate_clicks_for_position(0.0, 60.0, 1, 1, 12000)
# Both should be valid
assert len(buf_fast) == 12000
assert len(buf_slow) == 12000
def test_buffer_boundary_clicks(self):
"""Clicks that land at buffer boundaries are handled."""
m = Metronome(sample_rate=48000)
# Very small buffer
buf = m.generate_clicks_for_position(0.0, 120.0, 1, 1, 128)
assert len(buf) == 128
assert buf.dtype == np.float32
def test_very_slow_tempo(self):
"""Very slow tempo (no beats in buffer) returns silence."""
m = Metronome(sample_rate=48000)
buf = m.generate_clicks_for_position(0.1, 10.0, 1, 1, 1024)
assert len(buf) == 1024
# ═══════════════════════════════════════════════════════════════════════════════
# BackingTrackPlayer tests
# ═══════════════════════════════════════════════════════════════════════════════
class TestBackingTrackPlayer:
def test_init(self):
"""Player initializes without errors."""
cfg = BackingPlayerConfig(jack_transport_enabled=False)
player = BackingTrackPlayer(cfg)
assert player.config == cfg
assert player.is_playing is False
assert player.is_paused is False
assert player.active_track is None
assert player.position == 0.0
player.shutdown()
def test_init_defaults(self):
"""Player works with default config."""
player = BackingTrackPlayer()
assert player.config.sample_rate == 48000
player.shutdown()
def test_load_track(self):
"""Loading a track updates metadata."""
wav = _write_temp_wav(duration=0.5, sample_rate=44100, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Test Track")
assert player.load_track("bt_000")
track = player.playlists.get_track("bt_000")
assert track.loaded is True
assert track.state == TrackState.IDLE
assert track.duration > 0
assert track.sample_rate > 0
assert track.channels > 0
finally:
wav.unlink()
player.shutdown()
def test_load_nonexistent_file(self):
"""Loading a nonexistent file returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/nonexistent/file.wav")
assert player.load_track("bt_000") is False
t = player.playlists.get_track("bt_000")
assert t.state == TrackState.ERROR
player.shutdown()
def test_play_without_loaded_track(self):
"""Playing with no loaded tracks returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/ghost.wav", title="Ghost")
# Track exists in playlist but audio is not loaded
assert player.play(count_in=False) is False
player.shutdown()
def test_play_stop(self):
"""Play and stop cycle works."""
wav = _write_temp_wav(duration=0.3, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
assert player.load_track("bt_000")
assert player.play(count_in=False) is True
assert player.is_playing is True
assert player.active_track.id == "bt_000"
assert player.stop() is True
assert player.is_playing is False
finally:
wav.unlink()
player.shutdown()
def test_pause_resume(self):
"""Pause and resume works."""
wav = _write_temp_wav(duration=0.5, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.play(count_in=False)
assert player.pause() is True
assert player.is_paused is True
assert player.is_playing is False
assert player.resume() is True
assert player.is_paused is False
assert player.is_playing is True
finally:
wav.unlink()
player.shutdown()
def test_toggle_play(self):
"""Toggle cycles through play/pause states."""
wav = _write_temp_wav(duration=0.5, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
# Start
player.play(count_in=False)
assert player.is_playing
# Pause
player.pause()
assert player.is_paused
# Resume (toggle calls resume since paused)
player.resume()
assert player.is_playing
finally:
wav.unlink()
player.shutdown()
def test_seek(self):
"""Seeking within a track works."""
wav = _write_temp_wav(duration=1.0, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.play(count_in=False)
assert player.seek(0.5)
assert pytest.approx(player.position, rel=0.1) == 0.5
finally:
wav.unlink()
player.shutdown()
def test_seek_beyond_end_clamped(self):
"""Seeking beyond track duration clamps to duration."""
wav = _write_temp_wav(duration=0.5, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.play(count_in=False)
player.seek(999.0)
assert player.position <= 1.0 # should be clamped
finally:
wav.unlink()
player.shutdown()
def test_volume_control(self):
"""Per-track volume can be set."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
assert player.set_volume("bt_000", 0.75)
t = player.playlists.get_track("bt_000")
assert t.volume == 0.75
player.shutdown()
def test_mute_control(self):
"""Per-track mute."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
assert player.set_mute("bt_000", True)
assert player.playlists.get_track("bt_000").muted is True
player.shutdown()
def test_pan_control(self):
"""Per-track pan."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
assert player.set_pan("bt_000", 0.2)
assert player.playlists.get_track("bt_000").pan == 0.2
player.shutdown()
def test_solo_control(self):
"""Per-track solo."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
assert player.set_solo("bt_000", True)
assert player.playlists.get_track("bt_000").soloed is True
player.shutdown()
def test_play_mode(self):
"""Setting play mode on a track."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
assert player.set_play_mode("bt_000", PlayMode.LOOP)
assert player.playlists.get_track("bt_000").play_mode == PlayMode.LOOP
player.shutdown()
def test_set_tempo(self):
"""Setting tempo works."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.set_tempo(140.0)
assert player.transport.state.tempo == 140.0
player.shutdown()
def test_metronome_enable(self):
"""Metronome can be enabled/disabled."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.set_metronome_enabled(False)
assert player.metronome.enabled is False
player.set_metronome_enabled(True)
assert player.metronome.enabled is True
player.shutdown()
def test_click_volume(self):
"""Click volume can be adjusted."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.set_click_volume(0.25)
assert player.config.click_volume == 0.25
player.shutdown()
def test_next_track(self):
"""Next track navigation works."""
wav1 = _write_temp_wav(duration=0.1, frequency=440.0)
wav2 = _write_temp_wav(duration=0.1, frequency=880.0)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav1), title="Song A")
player.playlists.add_track(str(wav2), title="Song B")
player.load_track("bt_000")
player.load_track("bt_001")
player.play(count_in=False)
assert player.active_track.id == "bt_000"
player.next_track()
assert player.is_playing
assert player.active_track.id == "bt_001"
finally:
wav1.unlink()
wav2.unlink()
player.shutdown()
def test_prev_track(self):
"""Previous track navigation works."""
wav1 = _write_temp_wav(duration=0.1, frequency=440.0)
wav2 = _write_temp_wav(duration=0.1, frequency=880.0)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav1), title="Song A")
player.playlists.add_track(str(wav2), title="Song B")
player.load_track("bt_000")
player.load_track("bt_001")
# Play track 1
player.select_track(1)
assert player.active_track.id == "bt_001"
# Go back
player.prev_track()
assert player.active_track.id == "bt_000"
finally:
wav1.unlink()
wav2.unlink()
player.shutdown()
def test_select_track(self):
"""Select track by index."""
wav1 = _write_temp_wav(duration=0.1, frequency=440.0)
wav2 = _write_temp_wav(duration=0.1, frequency=880.0)
wav3 = _write_temp_wav(duration=0.1, frequency=1320.0)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav1), title="A")
player.playlists.add_track(str(wav2), title="B")
player.playlists.add_track(str(wav3), title="C")
player.load_playlist_tracks()
player.select_track(2)
assert player.active_track.title == "C"
finally:
for w in [wav1, wav2, wav3]:
w.unlink()
player.shutdown()
def test_load_playlist_tracks(self):
"""Loading all tracks in a playlist."""
wav1 = _write_temp_wav(duration=0.1)
wav2 = _write_temp_wav(duration=0.1)
wav3 = _write_temp_wav(duration=0.1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav1), title="One")
player.playlists.add_track(str(wav2), title="Two")
player.playlists.add_track(str(wav3), title="Three")
count = player.load_playlist_tracks()
assert count == 3
for tid in ["bt_000", "bt_001", "bt_002"]:
assert player.is_loaded(tid)
finally:
for w in [wav1, wav2, wav3]:
w.unlink()
player.shutdown()
def test_unload_track(self):
"""Unloading frees audio data."""
wav = _write_temp_wav(duration=0.1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="One")
player.load_track("bt_000")
assert player.is_loaded("bt_000")
player.unload_track("bt_000")
assert not player.is_loaded("bt_000")
finally:
wav.unlink()
player.shutdown()
def test_get_state(self):
"""get_state returns a meaningful dict."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
state = player.get_state()
assert "playing" in state
assert "active_track" in state
assert "position" in state
assert "transport" in state
assert "metronome_enabled" in state
assert "playlist" in state
assert "loaded_tracks" in state
player.shutdown()
def test_process_buffer_silence_when_stopped(self):
"""process_buffer returns silence when not playing."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
buf = player.process_buffer(256)
assert buf.shape == (256, 2)
assert buf.dtype == np.float32
assert np.all(buf == 0.0)
player.shutdown()
def test_process_buffer_audio(self):
"""process_buffer returns audio when playing."""
wav = _write_temp_wav(duration=1.0, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(
jack_transport_enabled=False,
metronome_enabled=False,
)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.play(count_in=False)
buf = player.process_buffer(256)
assert buf.shape == (256, 2)
# Should contain audio (not silence)
assert np.any(np.abs(buf) > 0.001)
finally:
wav.unlink()
player.shutdown()
def test_process_buffer_muted(self):
"""process_buffer returns silence when track is muted."""
wav = _write_temp_wav(duration=1.0, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(
jack_transport_enabled=False,
metronome_enabled=False,
)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.set_mute("bt_000", True)
player.play(count_in=False)
buf = player.process_buffer(256)
assert np.all(buf == 0.0)
finally:
wav.unlink()
player.shutdown()
def test_loop_mode(self):
"""LOOP mode wraps position back to cue point."""
wav = _write_temp_wav(duration=0.1, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(
jack_transport_enabled=False,
metronome_enabled=False,
)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Loop Song", play_mode=PlayMode.LOOP)
player.load_track("bt_000")
player.play(count_in=False)
# Process enough buffers to exhaust the file
total_frames = 0
max_frames = 48000 * 2 # 2 seconds worth
while total_frames < max_frames and player.is_playing:
buf = player.process_buffer(256)
total_frames += 256
# Track should still be playing (looped)
assert player.is_playing
finally:
wav.unlink()
player.shutdown()
def test_one_shot_mode(self):
"""ONE_SHOT mode stops at end of track."""
wav = _write_temp_wav(duration=0.05, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(
jack_transport_enabled=False,
metronome_enabled=False,
)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="One Shot", play_mode=PlayMode.ONE_SHOT)
player.load_track("bt_000")
player.play(count_in=False)
# Process until done
for _ in range(20):
buf = player.process_buffer(256)
if not player.is_playing:
break
assert not player.is_playing
t = player.playlists.get_track("bt_000")
assert t.state == TrackState.FINISHED
finally:
wav.unlink()
player.shutdown()
def test_segue_mode_advances(self):
"""SEGUE mode advances to next track."""
wav1 = _write_temp_wav(duration=0.05, sample_rate=48000, channels=1, frequency=440.0)
wav2 = _write_temp_wav(duration=0.05, sample_rate=48000, channels=1, frequency=880.0)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(
jack_transport_enabled=False,
metronome_enabled=False,
auto_advance=True,
)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav1), title="A", play_mode=PlayMode.SEGUE)
player.playlists.add_track(str(wav2), title="B", play_mode=PlayMode.ONE_SHOT)
player.load_playlist_tracks()
player.play(count_in=False)
# Process until first track finishes and auto-advances
for _ in range(30):
buf = player.process_buffer(256)
if player.active_track and player.active_track.title == "B":
break
# Should have advanced to track B
assert player.active_track is not None
assert player.active_track.title == "B"
finally:
wav1.unlink()
wav2.unlink()
player.shutdown()
def test_set_loop_region(self):
"""Setting a loop region on a track."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
assert player.set_loop_region("bt_000", 2.0, 8.0)
t = player.playlists.get_track("bt_000")
assert t.loop_start == 2.0
assert t.loop_end == 8.0
player.shutdown()
def test_shutdown_cleanup(self):
"""Shutdown cleans up without errors."""
wav = _write_temp_wav(duration=0.1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav))
player.load_track("bt_000")
player.play(count_in=False)
player.shutdown()
assert not player.is_playing
assert len(player._audio_cache) == 0 # noqa: SLF001
finally:
wav.unlink()
# ═══════════════════════════════════════════════════════════════════════════════
# DSP Integration tests (MIDI parameter handling)
# ═══════════════════════════════════════════════════════════════════════════════
class TestDSPIntegration:
"""Tests for the player's handle_parameter (MIDI → transport/track) dispatch."""
def _param(self, ptype, category="transport", channel=-1, label=""):
"""Helper to create a MixerParameter quickly."""
from src.midi.types import MixerParameter, ParameterType, ParameterCategory
return MixerParameter(
param_type=ParameterType(ptype) if isinstance(ptype, str) else ptype,
category=ParameterCategory(category) if isinstance(category, str) else category,
channel=channel,
label=label or str(ptype),
)
def test_play_stop_parameter(self):
"""PLAY param > 0.5 starts playback, < 0.5 stops it."""
wav = _write_temp_wav(duration=0.5)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
# PLAY=1.0 should start
p = self._param("play")
assert player.handle_parameter(p, 1.0) is True
assert player.is_playing
# PLAY=0.0 should stop
assert player.handle_parameter(p, 0.0) is True
assert not player.is_playing
finally:
wav.unlink()
player.shutdown()
def test_stop_parameter(self):
"""STOP parameter stops playback."""
wav = _write_temp_wav(duration=0.5)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.play(count_in=False)
p = self._param("stop")
assert player.handle_parameter(p, 1.0) is True
assert not player.is_playing
finally:
wav.unlink()
player.shutdown()
def test_tempo_parameter(self):
"""TEMPO parameter sets BPM."""
wav = _write_temp_wav(duration=0.1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav))
p = self._param("tempo")
assert player.handle_parameter(p, 140.0) is True
assert player.transport.state.tempo == 140.0
finally:
wav.unlink()
player.shutdown()
def test_loop_parameter(self):
"""LOOP parameter toggles loop mode on active track."""
wav = _write_temp_wav(duration=0.5)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
player.play(count_in=False)
p = self._param("loop")
assert player.handle_parameter(p, 1.0) is True
assert player.playlists.get_track("bt_000").play_mode == PlayMode.LOOP
assert player.handle_parameter(p, 0.0) is True
assert player.playlists.get_track("bt_000").play_mode == PlayMode.ONE_SHOT
finally:
wav.unlink()
player.shutdown()
def test_channel_volume_maps_to_track(self):
"""Channel 16 volume maps to backing track 0."""
wav = _write_temp_wav(duration=0.1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="BT1")
player.playlists.add_track(str(wav), title="BT2")
p = self._param("volume", category="channel", channel=16)
player.handle_parameter(p, 0.75)
assert player.playlists.get_track("bt_000").volume == 0.75
# Channel 17 = track 1
p2 = self._param("volume", category="channel", channel=17)
player.handle_parameter(p2, 0.5)
assert player.playlists.get_track("bt_001").volume == 0.5
finally:
wav.unlink()
player.shutdown()
def test_channel_mute_maps_to_track(self):
"""Channel mute maps to backing track."""
wav = _write_temp_wav(duration=0.1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track(str(wav))
p = self._param("mute", category="channel", channel=16)
player.handle_parameter(p, 1.0)
assert player.playlists.get_track("bt_000").muted is True
player.handle_parameter(p, 0.0)
assert player.playlists.get_track("bt_000").muted is False
finally:
wav.unlink()
player.shutdown()
def test_non_transport_parameter_ignored(self):
"""Non-transport, non-channel params are ignored."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
p = self._param("eq_low_freq", category="channel", channel=0)
assert player.handle_parameter(p, 100.0) is False
player.shutdown()
def test_tap_tempo_is_noop(self):
"""TAP_TEMPO parameter returns True but is a noop."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
p = self._param("tap_tempo")
assert player.handle_parameter(p, 0.0) is True
player.shutdown()
# ═══════════════════════════════════════════════════════════════════════════════
# Edge cases and error handling
# ═══════════════════════════════════════════════════════════════════════════════
class TestEdgeCases:
def test_stop_when_not_playing(self):
"""Stopping when not playing is a no-op."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.stop() is True
def test_pause_when_not_playing(self):
"""Pausing when not playing returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.pause() is False
def test_resume_when_not_paused(self):
"""Resuming when not paused returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.resume() is False
def test_seek_without_active_track(self):
"""Seeking with no active track returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.seek(1.0) is False
def test_next_track_without_playlist(self):
"""Next track without playlist returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.next_track() is False
def test_prev_track_without_playlist(self):
"""Prev track without playlist returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.prev_track() is False
def test_play_with_no_playlist(self):
"""Playing with no active playlist returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.play() is False
def test_unload_nonexistent_track(self):
"""Unloading a track not in cache returns False."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
assert player.unload_track("nonexistent") is False
def test_volume_clamped_to_range(self):
"""Volume set outside [0,1] is clamped."""
player = BackingTrackPlayer(
BackingPlayerConfig(jack_transport_enabled=False)
)
player.playlists.create("Test")
player.playlists.add_track("/music/song.wav")
player.set_volume("bt_000", 5.0)
assert player.playlists.get_track("bt_000").volume == 1.0
player.set_volume("bt_000", -3.0)
assert player.playlists.get_track("bt_000").volume == 0.0
def test_multiple_playlists_independent(self):
"""Multiple playlists are independent."""
pm = PlaylistManager()
pm.create("Set A")
pm.add_track("/music/a.wav")
pm.create("Set B")
pm.set_active("Set B")
pm.add_track("/music/b.wav")
pm.add_track("/music/c.wav")
assert pm.get("Set A").track_count == 1
assert pm.get("Set B").track_count == 2
assert pm.active.name == "Set B"
# ═══════════════════════════════════════════════════════════════════════════════
# Integration: count-in with audio playback
# ═══════════════════════════════════════════════════════════════════════════════
class TestCountInIntegration:
def test_count_in_before_playback(self):
"""Count-in generates clicks before audio starts."""
wav = _write_temp_wav(duration=0.5, sample_rate=48000, channels=1)
try:
player = BackingTrackPlayer(
BackingPlayerConfig(
jack_transport_enabled=False,
count_in_bars=1, # 4 beats
time_signature_num=4,
metronome_enabled=False, # We want only count-in clicks
)
)
player.metronome.enabled = True # Enable metronome for count-in
player.playlists.create("Test")
player.playlists.add_track(str(wav), title="Song")
player.load_track("bt_000")
# Play WITH count-in (default)
player.play(count_in=True)
# During count-in, is_playing is False
assert not player.is_playing
assert player._count_in_remaining > 0 # noqa: SLF001
# Process until count-in completes
# 1 bar × 4 beats × (60/120)s = 2.0s count-in
# 2.0s / (256 frames / 48000 Hz) ≈ 375 buffers needed
for _ in range(400):
buf = player.process_buffer(256)
if player.is_playing:
break
# After count-in, playback should be active
assert player.is_playing
finally:
wav.unlink()
player.shutdown()