"""Tests for the recording engine — WAV writer, recorder, session, disk monitor, bounce. Tests are designed to run without JACK — they validate the logic, state management, file I/O, and audio processing without requiring actual audio hardware. """ from __future__ import annotations import json import os import queue import struct import tempfile import threading import time from pathlib import Path import numpy as np import pytest from src.recording.wav_writer import ( WAVWriter, WAVWriterConfig, WAVBitDepth, write_wav, _float32_to_int16, _float32_to_int24_fast, _float32_to_int32, _float32_to_float32, ) from src.recording.recorder import ( MultiTrackRecorder, RecorderConfig, RecorderState, ChannelState as RecChanState, ChannelRecordStatus, PunchRegion, ) from src.recording.session import ( Session, Take, TrackInfo, PunchInfo, _now_iso, ) from src.recording.disk_monitor import ( DiskMonitor, DiskMonitorConfig, DiskSpaceCallback, ) from src.recording.bounce import ( BounceEngine, BounceConfig, BounceTrack, BounceResult, db_to_linear, linear_to_db, _pan_to_gains, ) # ═════════════════════════════════════════════════════════════════════════════ # Helpers # ═════════════════════════════════════════════════════════════════════════════ def _make_sine(freq: float, sample_rate: int, duration: float, amplitude: float = 0.5) -> np.ndarray: """Generate a sine wave as float32 numpy array.""" t = np.arange(int(sample_rate * duration), dtype=np.float32) / sample_rate return (amplitude * np.sin(2.0 * np.pi * freq * t)).astype(np.float32) def _read_wav_header(file_path: Path) -> dict: """Read RIFF/WAVE header fields for validation.""" with open(file_path, "rb") as f: riff = f.read(4) file_size = struct.unpack(" np.ndarray: """Read WAV file back as float32 using soundfile or custom code.""" try: import soundfile as sf samples, _ = sf.read(str(file_path), dtype="float32") return samples except ImportError: raise RuntimeError("soundfile required for test") # ═════════════════════════════════════════════════════════════════════════════ # WAVWriter tests # ═════════════════════════════════════════════════════════════════════════════ class TestSampleConversion: """Test float32 → integer sample conversion.""" def test_int16_zero(self): out = _float32_to_int16(np.array([0.0], dtype=np.float32)) assert out[0] == 0 def test_int16_full_scale(self): out = _float32_to_int16(np.array([1.0, -1.0], dtype=np.float32)) assert out[0] == 32767 assert out[1] == -32767 def test_int16_clipping(self): out = _float32_to_int16(np.array([2.0, -2.0], dtype=np.float32)) assert out[0] == 32767 assert out[1] == -32767 def test_int24_fast_roundtrip_range(self): """Int24 should produce correct byte count and values within range.""" sine = _make_sine(440, 48000, 0.1, 0.9) packed = _float32_to_int24_fast(sine) assert len(packed) == len(sine) * 3 # 3 bytes per sample assert packed.dtype == np.uint8 def test_int24_fast_zero(self): packed = _float32_to_int24_fast(np.array([0.0], dtype=np.float32)) assert len(packed) == 3 assert packed[0] == 0 and packed[1] == 0 and packed[2] == 0 def test_int24_fast_positive(self): packed = _float32_to_int24_fast(np.array([1.0], dtype=np.float32)) assert len(packed) == 3 # Reconstruct 24-bit value from 3 uint8 bytes b0 = int(packed[0]) b1 = int(packed[1]) b2 = int(packed[2]) val = b0 | (b1 << 8) | (b2 << 16) # Should be close to 8388607 (max positive 24-bit) assert val in (8388607, 8388606) # rounding variation def test_int24_fast_negative(self): packed = _float32_to_int24_fast(np.array([-1.0], dtype=np.float32)) assert len(packed) == 3 # Read as signed 24-bit b0 = int(packed[0]) b1 = int(packed[1]) b2 = int(packed[2]) val = b0 | (b1 << 8) | (b2 << 16) if val >= 0x800000: val -= 0x1000000 assert val in (-8388608, -8388607) def test_int32_range(self): out = _float32_to_int32(np.array([1.0, -1.0], dtype=np.float32)) assert out[0] == 2147483647 assert out[1] == -2147483647 def test_float32_passthrough(self): vals = np.array([0.5, -0.5], dtype=np.float32) out = _float32_to_float32(vals) np.testing.assert_array_equal(out, vals) class TestWAVWriterBasic: """Test WAV file creation and header correctness.""" def test_write_mono_int16(self, tmp_path): file_path = tmp_path / "test_int16.wav" sine = _make_sine(440, 48000, 0.5) with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT16) as w: w.write_frames(sine) assert file_path.exists() assert file_path.stat().st_size > 44 # header + data header = _read_wav_header(file_path) assert header["riff"] == b"RIFF" assert header["wave"] == b"WAVE" assert header["audio_format"] == 1 # PCM assert header["channels"] == 1 assert header["sample_rate"] == 48000 assert header["bits_per_sample"] == 16 def test_write_mono_int24(self, tmp_path): file_path = tmp_path / "test_int24.wav" sine = _make_sine(440, 48000, 0.5) with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT24) as w: w.write_frames(sine) header = _read_wav_header(file_path) assert header["bits_per_sample"] == 24 # Read back and verify it's valid audio samples = _read_wav_samples(file_path) assert len(samples) == len(sine) # Check correlation (should be very high) corr = np.corrcoef(samples, sine)[0, 1] assert corr > 0.95 def test_write_mono_int32(self, tmp_path): file_path = tmp_path / "test_int32.wav" sine = _make_sine(440, 48000, 0.5) with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT32) as w: w.write_frames(sine) header = _read_wav_header(file_path) assert header["bits_per_sample"] == 32 def test_write_mono_float32(self, tmp_path): file_path = tmp_path / "test_float32.wav" sine = _make_sine(440, 48000, 0.5) with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.FLOAT32) as w: w.write_frames(sine) header = _read_wav_header(file_path) assert header["audio_format"] == 3 # IEEE float assert header["bits_per_sample"] == 32 def test_write_stereo(self, tmp_path): file_path = tmp_path / "test_stereo.wav" left = _make_sine(440, 48000, 0.5, 0.5) right = _make_sine(880, 48000, 0.5, 0.3) stereo = np.column_stack([left, right]) with WAVWriter(file_path, sample_rate=48000, channels=2, bit_depth=WAVBitDepth.INT16) as w: w.write_frames(stereo) header = _read_wav_header(file_path) assert header["channels"] == 2 assert header["block_align"] == 4 # 2ch * 2 bytes def test_multi_chunk_write(self, tmp_path): """Test writing in multiple chunks produces valid output.""" file_path = tmp_path / "test_chunks.wav" total_frames = 4800 # 100ms at 48kHz chunk_size = 256 with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT24) as w: for start in range(0, total_frames, chunk_size): end = min(start + chunk_size, total_frames) chunk = _make_sine(440, 48000, (end - start) / 48000) w.write_frames(chunk) assert w.frames_written == total_frames samples = _read_wav_samples(file_path) assert len(samples) == total_frames def test_context_manager(self, tmp_path): """Test __enter__/__exit__.""" file_path = tmp_path / "test_ctx.wav" with WAVWriter(file_path, sample_rate=48000, channels=1) as w: w.write_frames(_make_sine(440, 48000, 0.1)) assert w.closed assert file_path.exists() def test_close_idempotent(self, tmp_path): """Calling close() multiple times is safe.""" file_path = tmp_path / "test_close.wav" w = WAVWriter(file_path) w.write_frames(_make_sine(440, 48000, 0.1)) w.close() w.close() # should not raise assert w.closed def test_write_after_close_raises(self, tmp_path): file_path = tmp_path / "test_err.wav" w = WAVWriter(file_path) w.close() with pytest.raises(ValueError): w.write_frames(_make_sine(440, 48000, 0.1)) def test_duration_seconds(self, tmp_path): file_path = tmp_path / "test_dur.wav" duration = 0.25 with WAVWriter(file_path, sample_rate=48000) as w: w.write_frames(_make_sine(440, 48000, duration)) assert w.duration_seconds == pytest.approx(duration, abs=0.001) def test_variable_sample_rate(self, tmp_path): """Test 44100 Hz and 96000 Hz sample rates.""" for sr in [44100, 96000]: file_path = tmp_path / f"test_{sr}.wav" with WAVWriter(file_path, sample_rate=sr) as w: w.write_frames(_make_sine(440, sr, 0.1)) header = _read_wav_header(file_path) assert header["sample_rate"] == sr def test_1d_input_broadcast_to_stereo(self, tmp_path): """Mono input to stereo writer should duplicate.""" file_path = tmp_path / "test_broadcast.wav" mono = _make_sine(440, 48000, 0.1) with WAVWriter(file_path, sample_rate=48000, channels=2) as w: w.write_frames(mono) samples = _read_wav_samples(file_path) assert samples.ndim == 2 assert samples.shape[1] == 2 np.testing.assert_array_almost_equal(samples[:, 0], samples[:, 1]) def test_flush(self, tmp_path): """Flush should write data to disk without closing.""" file_path = tmp_path / "test_flush.wav" w = WAVWriter(file_path) w.write_frames(_make_sine(440, 48000, 0.1)) w.flush() assert file_path.stat().st_size > 44 # data should be on disk w.close() def test_empty_write(self, tmp_path): """Writing zero frames should not fail.""" file_path = tmp_path / "test_empty.wav" with WAVWriter(file_path, sample_rate=48000) as w: w.write_frames(np.array([], dtype=np.float32)) assert w.frames_written == 0 def test_mismatched_channels_auto_adapt(self, tmp_path): """Writer with 1 channel should take first channel from multi-channel input.""" file_path = tmp_path / "test_adapt.wav" stereo = np.column_stack([ _make_sine(440, 48000, 0.1, 0.5), _make_sine(880, 48000, 0.1, 0.3), ]) with WAVWriter(file_path, sample_rate=48000, channels=1) as w: w.write_frames(stereo) samples = _read_wav_samples(file_path) assert samples.ndim == 1 class TestWAVWriterConfig: """Test WAVWriterConfig behaviour.""" def test_default_flush_interval(self, tmp_path): config = WAVWriterConfig(flush_interval_bytes=100) file_path = tmp_path / "test_config.wav" with WAVWriter(file_path, config=config) as w: w.write_frames(_make_sine(440, 48000, 0.01)) # ~480 samples assert file_path.exists() class TestWriteWAVConvenience: """Test the write_wav convenience function.""" def test_write_wav_mono(self, tmp_path): file_path = tmp_path / "quick.wav" sine = _make_sine(440, 48000, 0.5) result = write_wav(file_path, sine, sample_rate=48000, bit_depth=WAVBitDepth.INT16) assert result.exists() assert result.suffix == ".wav" def test_write_wav_stereo(self, tmp_path): file_path = tmp_path / "quick_stereo.wav" left = _make_sine(440, 48000, 0.2) right = _make_sine(880, 48000, 0.2) stereo = np.column_stack([left, right]) write_wav(file_path, stereo, sample_rate=48000) assert file_path.exists() # ═════════════════════════════════════════════════════════════════════════════ # MultiTrackRecorder tests # ═════════════════════════════════════════════════════════════════════════════ class TestRecorderConfig: """Test RecorderConfig defaults.""" def test_defaults(self): cfg = RecorderConfig() assert cfg.sample_rate == 48000 assert cfg.channels == 16 assert cfg.bit_depth == WAVBitDepth.INT24 def test_custom_session_dir(self): cfg = RecorderConfig(session_dir="/tmp/test") assert cfg.session_dir == "/tmp/test" class TestMultiTrackRecorderBasic: """Test recorder creation and state management.""" def test_initial_state(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert rec.state == RecorderState.IDLE assert not rec.is_recording assert rec.global_frame == 0 def test_arm_disarm(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert rec.arm_channel(0) assert rec.is_armed(0) assert rec.get_channel_state(0) == RecChanState.ARMED rec.disarm_channel(0) assert not rec.is_armed(0) assert rec.get_channel_state(0) == RecChanState.DISARMED def test_arm_invalid_channel(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert not rec.arm_channel(99) # out of range def test_get_armed_channels(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.arm_channel(3) rec.arm_channel(7) armed = rec.get_armed_channels() assert armed == [0, 3, 7] def test_start_with_no_armed_channels(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert rec.start() # starts but warns # Should still transition to RECORDING assert rec.state == RecorderState.RECORDING rec.stop() def test_start_stop_cycle(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() assert rec.state == RecorderState.RECORDING assert rec.get_channel_state(0) == RecChanState.RECORDING rec.stop() assert rec.state == RecorderState.IDLE # After stop, channel should be ARMED (not recording), since we disarm on stop # Actually stop doesn't disarm — let me check... stop punts out then closes writers. # After stop, state goes to IDLE but channel state is updated by _punch_out_channel # which sets it to ARMED. assert rec.get_channel_state(0) == RecChanState.ARMED def test_double_start(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) assert rec.start() assert not rec.start() # already recording def test_pause_resume(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() assert rec.state == RecorderState.RECORDING rec.pause() assert rec.state == RecorderState.PAUSED rec.resume() assert rec.state == RecorderState.RECORDING rec.stop() class TestMultiTrackRecorderWriting: """Test writing audio through the recorder.""" def test_write_queues_samples(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() chunk = _make_sine(440, 48000, 0.01) result = rec.write(0, chunk) assert result # should be queued rec.stop() def test_write_to_unarmed_channel(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.start() # Channel 0 not armed — write should return False chunk = _make_sine(440, 48000, 0.01) result = rec.write(0, chunk) assert not result rec.stop() def test_write_while_paused(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() rec.pause() chunk = _make_sine(440, 48000, 0.01) result = rec.write(0, chunk) assert not result # paused → no recording rec.stop() def test_writer_thread_starts_and_stops(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() assert rec._writer_thread is not None assert rec._writer_thread.is_alive() rec.stop() # Writer thread should have stopped assert not rec._writer_thread.is_alive() def test_write_creates_wav_file(self, tmp_path): session_dir = tmp_path / "session" rec = MultiTrackRecorder(session_dir=str(session_dir)) rec.arm_channel(0) rec.start() # Write enough frames to get a real file chunk = _make_sine(440, 48000, 0.1, 0.5) rec.write(0, chunk) rec.advance_frame(len(chunk)) rec.stop() # Check that a WAV file was created wav_files = list(session_dir.glob("ch00*.wav")) assert len(wav_files) >= 1 assert wav_files[0].stat().st_size > 0 def test_multi_channel_recording(self, tmp_path): session_dir = tmp_path / "multi" rec = MultiTrackRecorder(session_dir=str(session_dir)) rec.arm_channel(0) rec.arm_channel(1) rec.arm_channel(2) rec.start() chunk = _make_sine(440, 48000, 0.1, 0.5) rec.write(0, chunk) rec.write(1, chunk * 0.8) rec.write(2, chunk * 0.6) rec.advance_frame(len(chunk)) rec.stop() # All three channels should have files for ch in range(3): wav_files = list(session_dir.glob(f"ch{ch:02d}*.wav")) assert len(wav_files) >= 1, f"Channel {ch} has no WAV file" class TestPunchInOut: """Test punch in/out functionality.""" def test_punch_in_while_recording(self, tmp_path): session_dir = tmp_path / "punch" rec = MultiTrackRecorder(session_dir=str(session_dir)) rec.arm_channel(0) rec.start() # Record some data on channel 0 chunk = _make_sine(440, 48000, 0.05) rec.write(0, chunk) rec.advance_frame(len(chunk)) # Now punch in channel 1 (which wasn't armed) rec.arm_channel(1) assert rec.punch_in(1) chunk2 = _make_sine(880, 48000, 0.05) rec.write(1, chunk2) rec.advance_frame(len(chunk2)) # Punch out channel 1 assert rec.punch_out(1) # Punch out channel 0 assert rec.punch_out(0) rec.stop() # Both channels should have files assert len(list(session_dir.glob("ch00*.wav"))) >= 1 assert len(list(session_dir.glob("ch01*.wav"))) >= 1 def test_punch_creates_punch_regions(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() chunk = _make_sine(440, 48000, 0.05) rec.write(0, chunk) rec.advance_frame(len(chunk)) rec.punch_out(0) rec.advance_frame(100) rec.punch_in(0) chunk2 = _make_sine(440, 48000, 0.05) rec.write(0, chunk2) rec.advance_frame(len(chunk2)) rec.stop() status = rec.get_channel_record_status(0) assert status is not None # Should have at least 2 punch regions (initial + re-punch) assert len(status.punch_regions) >= 2 def test_punch_in_without_recording(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) # Not started yet assert not rec.punch_in(0) def test_punch_out_without_recording(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert not rec.punch_out(0) class TestRecorderStats: """Test recording statistics.""" def test_stats_during_recording(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() chunk = _make_sine(440, 48000, 0.1) rec.write(0, chunk) rec.advance_frame(len(chunk)) stats = rec.get_recording_stats() assert stats["state"] == "RECORDING" assert stats["global_frame"] == len(chunk) assert RecChanState.RECORDING.name in str(stats["channels"]) rec.stop() def test_writer_error_propagation(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert rec.writer_error is None class TestRecorderChannelState: """Test per-channel state transitions.""" def test_disarmed_to_armed(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert rec.get_channel_state(0) == RecChanState.DISARMED rec.arm_channel(0) assert rec.get_channel_state(0) == RecChanState.ARMED def test_armed_to_recording_on_start(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) rec.arm_channel(0) rec.start() assert rec.get_channel_state(0) == RecChanState.RECORDING rec.stop() def test_invalid_channel(self, tmp_path): rec = MultiTrackRecorder(session_dir=str(tmp_path)) assert rec.get_channel_state(-1) == RecChanState.DISARMED assert rec.get_channel_state(999) == RecChanState.DISARMED assert rec.get_channel_record_status(-1) is None # ═════════════════════════════════════════════════════════════════════════════ # Session tests # ═════════════════════════════════════════════════════════════════════════════ class TestSessionCreate: """Test session creation.""" def test_create_defaults(self): session = Session.create("Test Session") assert session.name == "Test Session" assert session.version == 1 assert session.sample_rate == 48000 assert session.bit_depth == "int24" assert session.created_at != "" assert session.takes == [] def test_create_with_options(self): session = Session.create("Session", sample_rate=96000) assert session.sample_rate == 96000 def test_take_count_zero_initially(self): session = Session.create("Empty") assert session.take_count == 0 class TestTakeManagement: """Test take creation, retrieval, deletion.""" def test_new_take(self): session = Session.create("Test") take = session.new_take("Take 1") assert take.name == "Take 1" assert take.id.startswith("take_") assert session.take_count == 1 def test_auto_naming(self): session = Session.create("Test") t1 = session.new_take() t2 = session.new_take() t3 = session.new_take() assert t1.name == "Take 1" assert t2.name == "Take 2" assert t3.name == "Take 3" def test_get_take(self): session = Session.create("Test") take = session.new_take("My Take") found = session.get_take(take.id) assert found is not None assert found.name == "My Take" def test_get_take_not_found(self): session = Session.create("Test") assert session.get_take("nonexistent") is None def test_delete_take(self): session = Session.create("Test") take = session.new_take("Delete Me") assert session.delete_take(take.id) assert session.take_count == 0 def test_delete_nonexistent(self): session = Session.create("Test") assert not session.delete_take("nope") def test_rename_take(self): session = Session.create("Test") take = session.new_take("Original") assert session.rename_take(take.id, "Renamed") renamed = session.get_take(take.id) assert renamed is not None assert renamed.name == "Renamed" def test_list_takes(self): session = Session.create("Test") session.new_take("A") session.new_take("B") takes = session.list_takes() assert len(takes) == 2 assert takes[0].name == "A" assert takes[1].name == "B" def test_get_last_take(self): session = Session.create("Test") assert session.get_last_take() is None t1 = session.new_take("First") t2 = session.new_take("Second") last = session.get_last_take() assert last is not None assert last.id == t2.id class TestTrackManagement: """Test track addition within takes.""" def test_add_track(self): session = Session.create("Test") take = session.new_take("Take") track = session.add_track(take.id, channel=0, file_path="tracks/ch00.wav", frames=48000) assert track is not None assert track.channel == 0 assert track.file_path == "tracks/ch00.wav" assert track.frames == 48000 def test_add_track_nonexistent_take(self): session = Session.create("Test") track = session.add_track("bad_id", channel=0, file_path="x.wav") assert track is None def test_track_lookup(self): session = Session.create("Test") take = session.new_take("Take") session.add_track(take.id, channel=3, file_path="ch03.wav") session.add_track(take.id, channel=7, file_path="ch07.wav") t = take.get_track(3) assert t is not None assert t.file_path == "ch03.wav" assert take.get_track(99) is None def test_get_all_track_files(self): session = Session.create("Test") take1 = session.new_take("A") session.add_track(take1.id, 0, "a_ch0.wav") session.add_track(take1.id, 1, "a_ch1.wav") take2 = session.new_take("B") session.add_track(take2.id, 0, "b_ch0.wav") files = session.get_all_track_files() assert len(files) == 3 assert "a_ch0.wav" in files assert "a_ch1.wav" in files assert "b_ch0.wav" in files class TestSessionSerialization: """Test session save/load and dict conversion.""" def test_to_dict_and_back(self): session = Session.create("Test") take = session.new_take("Take 1") session.add_track(take.id, 0, "ch00.wav", frames=1000) session.mixer_state = {"master_volume": -6.0} session.notes = "Test session" data = session.to_dict() assert data["version"] == 1 assert data["name"] == "Test" assert len(data["takes"]) == 1 assert data["mixer_state"] == {"master_volume": -6.0} assert data["notes"] == "Test session" restored = Session.from_dict(data) assert restored.name == "Test" assert restored.take_count == 1 assert restored.takes[0].name == "Take 1" assert restored.mixer_state == {"master_volume": -6.0} def test_save_load_roundtrip(self, tmp_path): session_file = tmp_path / "test.session.json" session = Session.create("Roundtrip Test", sample_rate=44100, bit_depth="int16") take = session.new_take("Main Take") session.add_track(take.id, 0, "tracks/ch00.wav", frames=44100) session.add_track(take.id, 1, "tracks/ch01.wav", frames=44100) session.mixer_state = {"ch0_volume": 0.0, "ch1_volume": -3.0} session.notes = "Roundtrip test" session.save(session_file) assert session_file.exists() loaded = Session.load(session_file) assert loaded.name == "Roundtrip Test" assert loaded.sample_rate == 44100 assert loaded.bit_depth == "int16" assert loaded.take_count == 1 assert loaded.takes[0].name == "Main Take" assert loaded.mixer_state == session.mixer_state assert loaded.notes == session.notes assert loaded._file_path == str(session_file) def test_save_without_path(self): session = Session.create("Test") with pytest.raises(ValueError): session.save() # no path set def test_save_then_resave(self, tmp_path): session_file = tmp_path / "session.json" session = Session.create("Test") session.save(session_file) session.name = "Updated" session.save() # uses _file_path from last save loaded = Session.load(session_file) assert loaded.name == "Updated" def test_directory_property(self, tmp_path): session_file = tmp_path / "dir" / "session.json" session = Session.create("Test") session.save(session_file) assert session.directory == tmp_path / "dir" def test_resolve_track_path(self, tmp_path): session_file = tmp_path / "sessions" / "test.session.json" session = Session.create("Test") session.save(session_file) take = session.new_take("T") track = session.add_track(take.id, 0, "tracks/ch00.wav") assert track is not None resolved = session.resolve_track_path(track) assert resolved == tmp_path / "sessions" / "tracks" / "ch00.wav" def test_load_nonexistent(self): with pytest.raises(FileNotFoundError): Session.load("/tmp/nonexistent_session.json") def test_update_mixer_state(self): session = Session.create("Test") session.update_mixer_state({"master_vol": -10.0}) assert session.mixer_state["master_vol"] == -10.0 class TestTakeSerialization: """Test Take to_dict / from_dict.""" def test_take_roundtrip(self): take = Take( id="take_abc", name="My Take", created_at="2024-01-01T00:00:00Z", sample_rate=48000, bit_depth="int24", start_frame=0, end_frame=48000, duration_seconds=1.0, ) take.add_track(TrackInfo(channel=0, file_path="ch00.wav", frames=48000)) take.add_track(TrackInfo(channel=1, file_path="ch01.wav", frames=48000)) data = take.to_dict() restored = Take.from_dict(data) assert restored.id == take.id assert restored.name == take.name assert len(restored.tracks) == 2 assert restored.tracks[0].channel == 0 def test_take_total_frames(self): take = Take(id="t", name="T", end_frame=48000) assert take.total_frames() == 48000 class TestPunchInfo: """Test punch info serialization in tracks.""" def test_punch_info_in_track(self): take = Take(id="t1", name="T") track = TrackInfo( channel=0, file_path="ch00.wav", punch_regions=[ PunchInfo(start_frame=0, end_frame=24000), PunchInfo(start_frame=30000, end_frame=48000), ], ) take.add_track(track) data = take.to_dict() restored = Take.from_dict(data) t = restored.tracks[0] assert len(t.punch_regions) == 2 assert t.punch_regions[0].start_frame == 0 assert t.punch_regions[0].end_frame == 24000 # ═════════════════════════════════════════════════════════════════════════════ # DiskMonitor tests # ═════════════════════════════════════════════════════════════════════════════ class TestDiskMonitor: """Test disk space monitoring.""" def test_initial_check(self): monitor = DiskMonitor("/tmp") assert monitor.free_bytes > 0 assert monitor.total_bytes > 0 def test_free_mb(self): monitor = DiskMonitor("/tmp") mb = monitor.free_mb assert mb > 0 def test_usage_percent(self): monitor = DiskMonitor("/tmp") pct = monitor.usage_percent assert 0.0 <= pct <= 100.0 def test_level_ok_when_plenty_of_space(self): cfg = DiskMonitorConfig( warning_threshold_bytes=500, critical_threshold_bytes=100, ) monitor = DiskMonitor("/tmp", config=cfg) monitor.check_now() # Unless we're on a very full disk, this should be ok assert monitor.level in (0, 1, 2) # just check it doesn't crash def test_is_low_space_with_high_threshold(self, tmp_path): cfg = DiskMonitorConfig( warning_threshold_bytes=10 * 1024 * 1024 * 1024 * 1024, # 10 TB — impossible critical_threshold_bytes=5 * 1024 * 1024 * 1024 * 1024, ) monitor = DiskMonitor(str(tmp_path), config=cfg) assert monitor.is_low_space() # will trigger warning assert monitor.is_critical() # will trigger critical def test_has_minimum(self, tmp_path): cfg = DiskMonitorConfig(minimum_required_bytes=0) monitor = DiskMonitor(str(tmp_path), config=cfg) assert monitor.has_minimum_for_recording() def test_estimated_time_positive(self, tmp_path): monitor = DiskMonitor(str(tmp_path)) t = monitor.estimated_recording_time_seconds() assert t > 0 def test_format_bytes(self): assert DiskMonitor.format_bytes(500) == "500 B" assert "KB" in DiskMonitor.format_bytes(2048) assert "MB" in DiskMonitor.format_bytes(10 * 1024 * 1024) assert "GB" in DiskMonitor.format_bytes(2 * 1024 * 1024 * 1024) def test_static_helpers(self): free = DiskMonitor.get_free_bytes("/tmp") total = DiskMonitor.get_total_bytes("/tmp") assert free > 0 assert total >= free def test_callback_fires_on_level_change(self, tmp_path): callback_calls = [] def cb(free, total, level): callback_calls.append(level) cfg = DiskMonitorConfig( warning_threshold_bytes=10 * 1024**4, # enormous critical_threshold_bytes=5 * 1024**4, poll_interval_seconds=0.01, ) monitor = DiskMonitor(str(tmp_path), config=cfg) monitor.set_callback(cb) # Reset internal level so next check triggers callback monitor._current_level = 0 monitor.check_now() # force check # Should have been called at least once with level >= 1 assert len(callback_calls) > 0 assert callback_calls[-1] >= 1 def test_start_stop(self): monitor = DiskMonitor("/tmp") monitor.start() assert monitor._running monitor.stop() assert not monitor._running def test_level_label(self): monitor = DiskMonitor("/tmp") assert monitor.level_label in ("ok", "warning", "critical") # ═════════════════════════════════════════════════════════════════════════════ # BounceEngine tests # ═════════════════════════════════════════════════════════════════════════════ class TestDBAConversion: """Test dB ↔ linear conversion utilities.""" def test_db_to_linear_zero(self): assert db_to_linear(0.0) == pytest.approx(1.0) def test_db_to_linear_minus_6(self): assert db_to_linear(-6.0) == pytest.approx(0.5012, abs=0.001) def test_db_to_linear_plus_6(self): assert db_to_linear(6.0) == pytest.approx(1.995, abs=0.01) def test_linear_to_db_one(self): assert linear_to_db(1.0) == pytest.approx(0.0, abs=0.01) def test_linear_to_db_half(self): assert linear_to_db(0.5) == pytest.approx(-6.02, abs=0.1) def test_roundtrip(self): for db in [-12.0, -6.0, -3.0, 0.0, 3.0, 6.0, 12.0]: assert linear_to_db(db_to_linear(db)) == pytest.approx(db, abs=0.01) class TestPanLaw: """Test pan law conversion.""" def test_center(self): l, r = _pan_to_gains(0.0) assert l == pytest.approx(r) assert l > 0.7 # -3dB center def test_hard_left(self): l, r = _pan_to_gains(-1.0) assert l == pytest.approx(1.0, abs=0.001) assert r == pytest.approx(0.0, abs=0.001) def test_hard_right(self): l, r = _pan_to_gains(1.0) assert l == pytest.approx(0.0, abs=0.001) assert r == pytest.approx(1.0, abs=0.001) def test_power_sum(self): """Pan law should maintain constant power (l^2 + r^2 ≈ 1).""" for pan in np.linspace(-1.0, 1.0, 21): l, r = _pan_to_gains(pan) power = l ** 2 + r ** 2 assert power == pytest.approx(1.0, abs=0.01), f"pan={pan}: power={power}" class TestBounceEngine: """Test bounce engine functionality.""" def _create_mono_wav(self, path: Path, freq: float, duration: float, amplitude: float = 0.5) -> None: """Create a test mono WAV file.""" sine = _make_sine(freq, 48000, duration, amplitude) write_wav(path, sine, sample_rate=48000, bit_depth=WAVBitDepth.INT16) def _create_stereo_wav(self, path: Path, freq_l: float, freq_r: float, duration: float) -> None: """Create a test stereo WAV file.""" left = _make_sine(freq_l, 48000, duration, 0.5) right = _make_sine(freq_r, 48000, duration, 0.3) stereo = np.column_stack([left, right]) write_wav(path, stereo, sample_rate=48000, bit_depth=WAVBitDepth.INT16) def test_bounce_single_mono_track(self, tmp_path): input_path = tmp_path / "input.wav" output_path = tmp_path / "output.wav" self._create_mono_wav(input_path, 440, 0.5) engine = BounceEngine(sample_rate=48000) engine.add_track(input_path, channel=0) result = engine.bounce(output_path) assert result.output_path.exists() assert result.frames_written == int(48000 * 0.5) assert result.duration_seconds == pytest.approx(0.5, abs=0.01) assert result.files_processed == 1 # Should be stereo output samples = _read_wav_samples(output_path) assert samples.ndim == 2 assert samples.shape[1] == 2 def test_bounce_two_tracks_stereo(self, tmp_path): input1 = tmp_path / "track1.wav" input2 = tmp_path / "track2.wav" output_path = tmp_path / "mix.wav" self._create_mono_wav(input1, 440, 0.5, 0.5) self._create_mono_wav(input2, 880, 0.5, 0.3) engine = BounceEngine(sample_rate=48000) engine.add_track(input1, channel=0, volume_db=0.0) engine.add_track(input2, channel=1, volume_db=-6.0) result = engine.bounce(output_path) assert result.output_path.exists() assert result.files_processed == 2 def test_bounce_panning(self, tmp_path): input_path = tmp_path / "mono.wav" output_path = tmp_path / "panned.wav" self._create_mono_wav(input_path, 440, 0.2, 1.0) engine = BounceEngine(sample_rate=48000) # Hard left engine.add_track(input_path, channel=0, pan=-1.0) result = engine.bounce(output_path) samples = _read_wav_samples(output_path) # Right channel should be silent right_rms = float(np.sqrt(np.mean(samples[:, 1] ** 2))) left_rms = float(np.sqrt(np.mean(samples[:, 0] ** 2))) assert left_rms > right_rms * 5 # right should be much quieter def test_bounce_muted_track(self, tmp_path): input1 = tmp_path / "loud.wav" input2 = tmp_path / "muted.wav" output_path = tmp_path / "mix.wav" self._create_mono_wav(input1, 440, 0.3, 0.5) self._create_mono_wav(input2, 880, 0.3, 1.0) engine = BounceEngine(sample_rate=48000) engine.add_track(input1, channel=0) engine.add_track(input2, channel=1, muted=True) result = engine.bounce(output_path) # The output should only contain track 1 samples = _read_wav_samples(output_path) full_mix = _make_sine(440, 48000, 0.3, 0.5) mix_stereo = np.column_stack([full_mix, full_mix]) # Correlation should be close to 1.0 with track 1 corr = np.corrcoef(samples[:, 0], mix_stereo[:, 0])[0, 1] assert corr > 0.99 def test_bounce_normalisation(self, tmp_path): input_path = tmp_path / "quiet.wav" output_path = tmp_path / "norm.wav" # Very quiet signal self._create_mono_wav(input_path, 440, 0.3, 0.1) engine = BounceEngine(sample_rate=48000) engine.add_track(input_path, channel=0) result = engine.bounce(output_path, normalise=True) samples = _read_wav_samples(output_path) peak = float(np.max(np.abs(samples))) # Should be close to target (0 dB = 1.0, with -0.3 dBFS ≈ 0.966) assert peak > 0.9 def test_bounce_no_normalise(self, tmp_path): input_path = tmp_path / "quiet.wav" output_path = tmp_path / "nonorm.wav" self._create_mono_wav(input_path, 440, 0.3, 0.1) config = BounceConfig(normalise=False) engine = BounceEngine(sample_rate=48000, config=config) engine.add_track(input_path, channel=0) result = engine.bounce(output_path) samples = _read_wav_samples(output_path) peak = float(np.max(np.abs(samples))) # Should be low (near 0.1) assert peak < 0.3 def test_bounce_master_volume(self, tmp_path): input_path = tmp_path / "input.wav" output_path = tmp_path / "mastervol.wav" self._create_mono_wav(input_path, 440, 0.3, 0.5) config = BounceConfig(master_volume_db=-20.0, normalise=False) engine = BounceEngine(sample_rate=48000, config=config) engine.add_track(input_path, channel=0) result = engine.bounce(output_path) samples = _read_wav_samples(output_path) peak = float(np.max(np.abs(samples))) # -20 dB = 0.1 linear, so peak should be ~0.5 * 0.1 = 0.05 assert peak < 0.1 def test_bounce_master_muted(self, tmp_path): input_path = tmp_path / "input.wav" output_path = tmp_path / "muted_master.wav" self._create_mono_wav(input_path, 440, 0.3, 0.5) config = BounceConfig(master_muted=True, normalise=False) engine = BounceEngine(sample_rate=48000, config=config) engine.add_track(input_path, channel=0) result = engine.bounce(output_path) samples = _read_wav_samples(output_path) # Should be silent assert float(np.max(np.abs(samples))) < 1e-6 def test_bounce_master_dim(self, tmp_path): input_path = tmp_path / "input.wav" output_path = tmp_path / "dimmed.wav" self._create_mono_wav(input_path, 440, 0.3, 0.5) config = BounceConfig(master_dim_active=True, master_dim_db=-40.0, normalise=False) engine = BounceEngine(sample_rate=48000, config=config) engine.add_track(input_path, channel=0) result = engine.bounce(output_path) samples = _read_wav_samples(output_path) peak = float(np.max(np.abs(samples))) # -40 dB dim = 0.01 linear assert peak < 0.05 def test_preview(self, tmp_path): input_path = tmp_path / "preview.wav" self._create_mono_wav(input_path, 440, 0.5) engine = BounceEngine(sample_rate=48000) engine.add_track(input_path, channel=0) preview = engine.preview() assert preview["total_frames"] == int(48000 * 0.5) assert preview["duration_seconds"] == pytest.approx(0.5, abs=0.01) assert preview["estimated_file_size_bytes"] > 0 assert preview["track_count"] == 1 def test_bounce_no_tracks_raises(self, tmp_path): engine = BounceEngine(sample_rate=48000) with pytest.raises(ValueError): engine.bounce(tmp_path / "output.wav") def test_bounce_clear_tracks(self, tmp_path): input_path = tmp_path / "input.wav" self._create_mono_wav(input_path, 440, 0.1) engine = BounceEngine(sample_rate=48000) engine.add_track(input_path, channel=0) engine.add_track(input_path, channel=1) engine.remove_track(0) assert len(engine._tracks) == 1 engine.clear_tracks() assert len(engine._tracks) == 0 def test_bounce_stereo_input(self, tmp_path): """Bouncing a stereo input should preserve left/right.""" input_path = tmp_path / "stereo.wav" output_path = tmp_path / "stereo_out.wav" self._create_stereo_wav(input_path, 440, 880, 0.3) engine = BounceEngine(sample_rate=48000) engine.add_track(input_path, channel=0, pan=0.0) result = engine.bounce(output_path) samples = _read_wav_samples(output_path) assert samples.shape[1] == 2 def test_bounce_result_peak_rms(self, tmp_path): input_path = tmp_path / "input.wav" output_path = tmp_path / "result.wav" self._create_mono_wav(input_path, 440, 0.3, 0.7) engine = BounceEngine(sample_rate=48000) config = BounceConfig(normalise=False) engine.config = config engine.add_track(input_path, channel=0) result = engine.bounce(output_path) # Peak should be close to 0.7 assert result.peak_db > -10 assert result.rms_db > -20 # ═════════════════════════════════════════════════════════════════════════════ # Integration / edge-case tests # ═════════════════════════════════════════════════════════════════════════════ class TestRecordingIntegration: """Tests that combine multiple recording components.""" def test_record_then_bounce(self, tmp_path): """Record audio, save session, bounce to stereo.""" session_dir = tmp_path / "project" session_dir.mkdir() # 1. Create session session = Session.create("Integration Test", sample_rate=48000) session_file = session_dir / "session.json" # 2. Record rec = MultiTrackRecorder(session_dir=str(session_dir)) rec.arm_channel(0) rec.arm_channel(1) rec.start() chunk = _make_sine(440, 48000, 0.2, 0.5) rec.write(0, chunk) rec.write(1, chunk * 0.8) rec.advance_frame(len(chunk)) rec.stop() # 3. Find recorded files wav_files_0 = sorted(session_dir.glob("ch00*.wav")) wav_files_1 = sorted(session_dir.glob("ch01*.wav")) # 4. Create take from recording take = session.new_take("Take 1") if wav_files_0: session.add_track(take.id, 0, str(wav_files_0[0].relative_to(session_dir)), frames=len(chunk)) if wav_files_1: session.add_track(take.id, 1, str(wav_files_1[0].relative_to(session_dir)), frames=len(chunk)) session.save(session_file) assert session_file.exists() # 5. Bounce output_path = session_dir / "master.wav" engine = BounceEngine(sample_rate=48000) for wav_file in wav_files_0: engine.add_track(wav_file, channel=0, volume_db=0.0, pan=-0.5) for wav_file in wav_files_1: engine.add_track(wav_file, channel=1, volume_db=-3.0, pan=0.5) result = engine.bounce(output_path, normalise=True) assert result.output_path.exists() # 6. Reload session loaded = Session.load(session_file) assert loaded.take_count == 1 assert loaded.takes[0].name == "Take 1" def test_bounce_with_session_mixer_state(self, tmp_path): """Bounce using mixer state from a session.""" session_dir = tmp_path / "mixer_session" session_dir.mkdir() # Create test audio input_path = session_dir / "ch00.wav" _create_mono_wav(input_path, 440, 0.3, 0.5) input_path2 = session_dir / "ch01.wav" _create_mono_wav(input_path2, 880, 0.3, 0.3) session = Session.create("Test", sample_rate=48000) session.mixer_state = { "ch0_volume": -3.0, "ch0_pan": -0.5, "ch1_volume": -6.0, "ch1_pan": 0.5, "master_volume": -2.0, } session.save(session_dir / "session.json") # Load mixer state into bounce config ms = session.mixer_state engine = BounceEngine( sample_rate=48000, config=BounceConfig( master_volume_db=ms.get("master_volume", 0.0), normalise=False, ), ) engine.add_track(input_path, channel=0, volume_db=ms.get("ch0_volume", 0.0), pan=ms.get("ch0_pan", 0.0)) engine.add_track(input_path2, channel=1, volume_db=ms.get("ch1_volume", 0.0), pan=ms.get("ch1_pan", 0.0)) output_path = session_dir / "master.wav" result = engine.bounce(output_path) assert result.output_path.exists() # Helper for integration test def _create_mono_wav(path: Path, freq: float, duration: float, amplitude: float = 0.5) -> None: """Create a test mono WAV file (function version for use outside class).""" sine = _make_sine(freq, 48000, duration, amplitude) write_wav(path, sine, sample_rate=48000, bit_depth=WAVBitDepth.INT16)