"""Unit tests for individual FX blocks in the audio pipeline. Each test validates a specific effect against known output shapes (silence clipping, envelope tracking, modulation range, etc.). All tests use 256-sample blocks at 48kHz to match real-time operation. """ from __future__ import annotations import numpy as np import pytest from src.dsp.pipeline import ( AudioPipeline, BLOCK_SIZE, SAMPLE_RATE, _DelayLine, _CombFilter, _AllpassFilter, _compute_lowshelf_coeffs, _compute_highshelf_coeffs, _compute_peaking_coeffs, ) from src.presets.types import FXBlock, FXType, Preset # ── Fixtures ─────────────────────────────────────────────────────── SILENCE = np.zeros(BLOCK_SIZE, dtype=np.float32) SINE_TONE = (np.sin(2 * np.pi * 440.0 * np.arange(BLOCK_SIZE) / SAMPLE_RATE) .astype(np.float32)) HALF_SCALE = np.full(BLOCK_SIZE, 0.5, dtype=np.float32) FULL_SCALE = np.full(BLOCK_SIZE, 0.99, dtype=np.float32) @pytest.fixture def pipeline(): p = AudioPipeline() yield p def _load_fx(pipeline: AudioPipeline, fx_type: FXType, params: dict[str, float] | None = None) -> None: """Quick-apply a single FX block to the pipeline for testing.""" block = FXBlock( fx_type=fx_type, enabled=True, bypass=False, params=params or {}, ) preset = Preset( name="test", chain=[block], master_volume=1.0, ) pipeline.load_preset(preset) # ═══════════════════════════════════════════════════════════════════ # 1. Noise Gate # ═══════════════════════════════════════════════════════════════════ class TestNoiseGate: def test_silence_muted(self, pipeline): """Gate at default threshold (0.01) should silence silence.""" _load_fx(pipeline, FXType.NOISE_GATE, {"threshold": 0.01}) out = pipeline.process(SILENCE) assert np.max(np.abs(out)) == 0.0, "Silence should be muted" def test_strong_signal_passes(self, pipeline): """Gate should pass full-scale tone.""" _load_fx(pipeline, FXType.NOISE_GATE, {"threshold": 0.01}) out = pipeline.process(FULL_SCALE) assert np.max(np.abs(out)) > 0.8, "Strong signal should pass" def test_release_tail(self, pipeline): """Gate should have exponential release, not instant cut.""" _load_fx(pipeline, FXType.NOISE_GATE, {"threshold": 0.1, "release": 50.0}) # Transition: silence -> tone -> silence out1 = pipeline.process(SILENCE) # below threshold — muted assert np.max(np.abs(out1)) == 0.0, "Initial silence muted" out2 = pipeline.process(SINE_TONE * 0.5) # strong — opens assert np.max(np.abs(out2)) > 0.1, "Gate opened for tone" out3 = pipeline.process(SILENCE) # release tail out4 = pipeline.process(SILENCE) # should fully decay out5 = pipeline.process(SILENCE) # After enough silence blocks, output should be zero assert np.max(np.abs(out5)) < 0.001, "Release should fully decay" def test_bypass(self, pipeline): """Bypassed gate passes audio unchanged.""" block = FXBlock( fx_type=FXType.NOISE_GATE, enabled=True, bypass=True, params={"threshold": 1.0}, ) preset = Preset(name="test", chain=[block], master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE), "Bypassed gate should passthrough" # ═══════════════════════════════════════════════════════════════════ # 2. Compressor # ═══════════════════════════════════════════════════════════════════ class TestCompressor: def test_below_threshold_no_change(self, pipeline): """Signal below threshold should pass at unity (after makeup).""" _load_fx(pipeline, FXType.COMPRESSOR, {"threshold": 0.0, "ratio": 4.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.01) # very quiet assert np.max(np.abs(out)) > 0, "Quiet signal passes" def test_limits_above_threshold(self, pipeline): """Signal well above threshold gets compressed.""" _load_fx(pipeline, FXType.COMPRESSOR, {"threshold": -10.0, "ratio": 8.0, "gain": 1.0}) out = pipeline.process(FULL_SCALE) rms_out = np.sqrt(np.mean(out ** 2)) # High ratio should substantially reduce dynamics assert rms_out < 0.8, "Compressor should reduce level above threshold" def test_makeup_gain(self, pipeline): """Makeup gain should be applied after compression.""" _load_fx(pipeline, FXType.COMPRESSOR, {"threshold": -20.0, "ratio": 4.0, "gain": 1.5}) out = pipeline.process(SINE_TONE * 0.3) assert np.max(np.abs(out)) <= 1.0, "Output must be in [-1, 1]" def test_output_clipped(self, pipeline): """Compressor output never exceeds [-1, 1].""" _load_fx(pipeline, FXType.COMPRESSOR, {"threshold": -50.0, "ratio": 2.0, "gain": 20.0}) out = pipeline.process(HALF_SCALE) assert np.all(out >= -1.0) and np.all(out <= 1.0) # ═══════════════════════════════════════════════════════════════════ # 3. Boost / Overdrive / Distortion / Fuzz # ═══════════════════════════════════════════════════════════════════ class TestBoost: def test_linear_gain(self, pipeline): """Boost applies linear gain without clipping (small signal).""" _load_fx(pipeline, FXType.BOOST, {"gain_db": 6.0}) quiet = SINE_TONE * 0.1 out = pipeline.process(quiet) expected = np.clip(quiet * 10 ** (6.0 / 20.0), -1.0, 1.0) assert np.allclose(out, expected, atol=1e-6), \ "6dB boost should double amplitude" def test_clips_at_unity(self, pipeline): """Boost clips to [-1, 1].""" _load_fx(pipeline, FXType.BOOST, {"gain_db": 60.0}) out = pipeline.process(HALF_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 class TestOverdrive: def test_asymmetric_clipping(self, pipeline): """Overdrive clips asymmetrically (tube-like).""" _load_fx(pipeline, FXType.OVERDRIVE, {"drive": 0.8, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 # Should have harmonics — check shape differs from sine input assert not np.allclose(out, SINE_TONE * 0.5, atol=0.05) def test_low_drive_passthrough(self, pipeline): """Low drive should pass nearly clean.""" _load_fx(pipeline, FXType.OVERDRIVE, {"drive": 0.0, "gain": 1.0}) quiet = SINE_TONE * 0.05 out = pipeline.process(quiet) assert np.max(np.abs(out)) > 0.0 class TestDistortion: def test_harder_clipping(self, pipeline): """Distortion clips harder than overdrive.""" _load_fx(pipeline, FXType.DISTORTION, {"drive": 0.8, "gain": 1.0}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_distortion_changes_waveform(self, pipeline): """Distortion should significantly reshape the waveform.""" _load_fx(pipeline, FXType.DISTORTION, {"drive": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) # Should have square-ish shape (overtones) assert np.std(out) > 0, "Distortion produces output" class TestFuzz: def test_hard_clip_shape(self, pipeline): """Fuzz creates near-square-wave shape.""" _load_fx(pipeline, FXType.FUZZ, {"drive": 1.0, "gain": 0.5}) out = pipeline.process(SINE_TONE * 0.8) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 # Fuzz should be very clipped rms_out = np.sqrt(np.mean(out ** 2)) assert rms_out > 0.2, "Fuzz should produce significant output" # ═══════════════════════════════════════════════════════════════════ # 4. EQ # ═══════════════════════════════════════════════════════════════════ class TestEQ: def test_flat_chain(self, pipeline): """EQ with 0dB on all bands passes signal unchanged.""" _load_fx(pipeline, FXType.EQ, {"bass": 0.0, "mid": 0.0, "treble": 0.0}) out = pipeline.process(SINE_TONE) assert np.allclose(out, SINE_TONE, atol=1e-4), \ "Flat EQ should pass through" def test_bass_boost(self, pipeline): """Bass boost amplifies low frequencies.""" _load_fx(pipeline, FXType.EQ, {"bass": 12.0, "mid": 0.0, "treble": 0.0, "bass_freq": 200.0}) # Low frequency test tone low_tone = (np.sin(2 * np.pi * 80.0 * np.arange(BLOCK_SIZE) / SAMPLE_RATE) .astype(np.float32)) out = pipeline.process(low_tone * 0.3) rms_out = np.sqrt(np.mean(out ** 2)) rms_in = np.sqrt(np.mean((low_tone * 0.3) ** 2)) assert rms_out > rms_in * 1.5, \ "Bass boost should amplify low tones" def test_output_range(self, pipeline): """EQ stays in [-3, 3] before final clip in process().""" _load_fx(pipeline, FXType.EQ, {"bass": 15.0, "mid": 15.0, "treble": 15.0}) out = pipeline.process(SINE_TONE * 0.3) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_biquad_coeffs_stable(self): """Biquad coefficient generators produce stable filters.""" coeffs = _compute_lowshelf_coeffs(200, 6.0, 0.707, SAMPLE_RATE) assert all(np.isfinite(c) for c in coeffs) coeffs = _compute_highshelf_coeffs(3500, -6.0, 0.707, SAMPLE_RATE) assert all(np.isfinite(c) for c in coeffs) coeffs = _compute_peaking_coeffs(1000, 6.0, 0.707, SAMPLE_RATE) assert all(np.isfinite(c) for c in coeffs) # ═══════════════════════════════════════════════════════════════════ # 5. Chorus # ═══════════════════════════════════════════════════════════════════ class TestChorus: def test_output_range(self, pipeline): """Chorus output stays in [-1, 1].""" _load_fx(pipeline, FXType.CHORUS, {"rate": 0.5, "depth": 0.5, "mix": 0.5}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_dry_only_at_zero_mix(self, pipeline): """0% mix = pass-through.""" _load_fx(pipeline, FXType.CHORUS, {"rate": 0.5, "depth": 0.5, "mix": 0.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.allclose(out, SINE_TONE * 0.5, atol=1e-4) def test_wet_at_full_mix(self, pipeline): """100% mix = modulated signal only.""" _load_fx(pipeline, FXType.CHORUS, {"rate": 0.5, "depth": 0.5, "mix": 1.0}) # Warm up delay line so first reads are meaningful for _ in range(5): pipeline.process(SINE_TONE * 0.5) out = pipeline.process(SINE_TONE * 0.5) out2 = pipeline.process(SINE_TONE * 0.5) # Chorus should produce varied output (LFO modulation) assert not np.allclose(out, out2, atol=0.01), \ "Chorus LFO should produce different samples each block" # ═══════════════════════════════════════════════════════════════════ # 6. Flanger # ═══════════════════════════════════════════════════════════════════ class TestFlanger: def test_output_range(self, pipeline): _load_fx(pipeline, FXType.FLANGER, {"rate": 0.25, "depth": 0.7, "feedback": 0.3, "mix": 0.5}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_zero_mix_pass(self, pipeline): """0% mix = pass-through.""" _load_fx(pipeline, FXType.FLANGER, {"rate": 0.25, "depth": 0.7, "feedback": 0.3, "mix": 0.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.allclose(out, SINE_TONE * 0.5, atol=1e-4) def test_feedback_accumulates(self, pipeline): """Flanger with high feedback should produce different output over successive blocks than without feedback.""" _load_fx(pipeline, FXType.FLANGER, {"rate": 0.25, "depth": 0.7, "feedback": 0.8, "mix": 1.0}) # Warm up delay line for _ in range(10): pipeline.process(SINE_TONE * 0.3) out1 = pipeline.process(SINE_TONE * 0.3) # Advance many blocks to get different LFO phase for _ in range(20): pipeline.process(SINE_TONE * 0.3) out_far = pipeline.process(SINE_TONE * 0.3) # With high feedback, widely-spaced blocks should differ # due to feedback accumulation + different LFO phase assert not np.allclose(out1, out_far, atol=0.05), \ "High feedback should accumulate in flanger across LFO phases" # ═══════════════════════════════════════════════════════════════════ # 7. Phaser # ═══════════════════════════════════════════════════════════════════ class TestPhaser: def test_output_range(self, pipeline): _load_fx(pipeline, FXType.PHASER, {"rate": 0.4, "depth": 0.5, "feedback": 0.3, "mix": 0.5}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_zero_mix_pass(self, pipeline): _load_fx(pipeline, FXType.PHASER, {"rate": 0.4, "depth": 0.5, "feedback": 0.3, "mix": 0.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.allclose(out, SINE_TONE * 0.5, atol=1e-4) def test_phaser_modulates(self, pipeline): """Phaser produces different output across successive blocks (LFO sweep).""" _load_fx(pipeline, FXType.PHASER, {"rate": 0.4, "depth": 0.5, "feedback": 0.3, "mix": 1.0}) out1 = pipeline.process(SINE_TONE * 0.3) out2 = pipeline.process(SINE_TONE * 0.3) assert not np.allclose(out1, out2, atol=0.01), \ "Phaser LFO should modulate across blocks" # ═══════════════════════════════════════════════════════════════════ # 8. Tremolo # ═══════════════════════════════════════════════════════════════════ class TestTremolo: def test_output_range(self, pipeline): _load_fx(pipeline, FXType.TREMOLO, {"rate": 4.0, "depth": 0.7, "shape": "sine"}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_silence_stays_silent(self, pipeline): _load_fx(pipeline, FXType.TREMOLO, {"rate": 4.0, "depth": 1.0, "shape": "sine"}) out = pipeline.process(SILENCE) assert np.max(np.abs(out)) == 0.0 def test_square_lfo_shape(self, pipeline): """Square wave LFO with depth=1 cuts out half the signal.""" _load_fx(pipeline, FXType.TREMOLO, {"rate": 187.5, "depth": 1.0, "shape": "square"}) # At 187.5 Hz, one full period = 256 samples = exactly 1 block. # Phase: first half of block < 0.5 (LFO=1.0, passes signal), # second half >= 0.5 (LFO=0.0, cuts out). Depth=1.0 means full cut. out = pipeline.process(HALF_SCALE) # First half of output should equal HALF_SCALE, second half = 0 assert np.allclose(out[:128], HALF_SCALE[:128], atol=1e-4), \ "First half should pass when LFO=1" assert np.max(np.abs(out[128:])) < 1e-4, \ "Second half should be silent when LFO=0" def test_zero_depth_no_effect(self, pipeline): """0% depth = no modulation.""" _load_fx(pipeline, FXType.TREMOLO, {"rate": 4.0, "depth": 0.0, "shape": "sine"}) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE, atol=1e-4) def test_modulation_changes_each_block(self, pipeline): """LFO phase advances across blocks.""" _load_fx(pipeline, FXType.TREMOLO, {"rate": 4.0, "depth": 1.0, "shape": "sine"}) out1 = pipeline.process(HALF_SCALE) out2 = pipeline.process(HALF_SCALE) assert not np.allclose(out1, out2, atol=0.001), \ "LFO should produce different modulation each block" # ═══════════════════════════════════════════════════════════════════ # 9. Vibrato # ═══════════════════════════════════════════════════════════════════ class TestVibrato: def test_output_range(self, pipeline): _load_fx(pipeline, FXType.VIBRATO, {"rate": 3.0, "depth": 0.5}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_pitch_modulation(self, pipeline): """Vibrato produces time-varying delay (pitch warble).""" _load_fx(pipeline, FXType.VIBRATO, {"rate": 3.0, "depth": 0.5}) out1 = pipeline.process(SINE_TONE * 0.3) out2 = pipeline.process(SINE_TONE * 0.3) # Identical input blocks produce different output due to LFO assert not np.allclose(out1, out2, atol=0.001), \ "Vibrato LFO should modulate pitch across blocks" # ═══════════════════════════════════════════════════════════════════ # 10. Delay # ═══════════════════════════════════════════════════════════════════ class TestDelay: def test_output_range(self, pipeline): _load_fx(pipeline, FXType.DELAY, {"time": 400.0, "feedback": 0.3, "mix": 0.4}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_dry_only_at_zero_mix(self, pipeline): _load_fx(pipeline, FXType.DELAY, {"time": 400.0, "feedback": 0.3, "mix": 0.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.allclose(out, SINE_TONE * 0.5, atol=1e-4) def test_no_silence_output_on_silence(self, pipeline): """Silence in with delay should produce echo decay tail.""" _load_fx(pipeline, FXType.DELAY, {"time": 50.0, "feedback": 0.5, "mix": 1.0}) # Fill delay line with enough blocks to cover 50ms delay for _ in range(20): pipeline.process(SINE_TONE * 0.5) out1 = pipeline.process(SILENCE) out2 = pipeline.process(SILENCE) # Should have decaying echo assert np.max(np.abs(out1)) > 0, "Delay tail should still be present" # Echo should decay toward zero assert np.max(np.abs(out2)) <= np.max(np.abs(out1)) + 0.001, \ "Echo should decay" def test_tap_tempo_callback(self, pipeline): """Tap tempo overrides time_ms when set.""" _load_fx(pipeline, FXType.DELAY, {"time": 400.0, "tap_tempo": 200.0, "feedback": 0.3, "mix": 0.5}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) # ═══════════════════════════════════════════════════════════════════ # 11. Reverb (Schroeder) # ═══════════════════════════════════════════════════════════════════ class TestReverb: def test_output_range(self, pipeline): _load_fx(pipeline, FXType.REVERB, {"decay": 0.5, "damping": 0.4, "mix": 0.3}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_zero_mix_passthrough(self, pipeline): _load_fx(pipeline, FXType.REVERB, {"decay": 0.5, "damping": 0.4, "mix": 0.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.allclose(out, SINE_TONE * 0.5, atol=1e-4) def test_decay_tail(self, pipeline): """Reverb produces decaying tail after input stops.""" _load_fx(pipeline, FXType.REVERB, {"decay": 0.8, "damping": 0.4, "mix": 1.0}) # Fill reverb — long comb delays need ~50 blocks to energise for _ in range(50): pipeline.process(FULL_SCALE) tail1 = pipeline.process(SILENCE) tail2 = pipeline.process(SILENCE) # Should have a decay tail assert np.max(np.abs(tail1)) > 0.001, "Reverb tail should be audible" # Should decay (not necessarily monotonic but trend downward) tail_energy = [np.sqrt(np.mean(t ** 2)) for t in [tail1, tail2]] assert sum(tail_energy) > 0, "Tail must have energy" def test_different_decay_values(self, pipeline): """Higher decay = longer tail.""" _load_fx(pipeline, FXType.REVERB, {"decay": 0.9, "damping": 0.2, "mix": 1.0}) pipeline.process(FULL_SCALE) high_tail = pipeline.process(SILENCE) _load_fx(pipeline, FXType.REVERB, {"decay": 0.3, "damping": 0.2, "mix": 1.0}) pipeline.process(FULL_SCALE) low_tail = pipeline.process(SILENCE) energy_high = np.sqrt(np.mean(high_tail ** 2)) energy_low = np.sqrt(np.mean(low_tail ** 2)) # Higher decay should generally produce more tail energy # (not assertion — informational only; structural differences matter) assert energy_high >= 0 and energy_low >= 0, \ "Both decay values should produce non-negative energy" # ═══════════════════════════════════════════════════════════════════ # 12. Volume # ═══════════════════════════════════════════════════════════════════ class TestVolume: def test_linear_scaling(self, pipeline): _load_fx(pipeline, FXType.VOLUME, {"level": 0.5}) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE * 0.5, atol=1e-6) def test_unity_gain(self, pipeline): _load_fx(pipeline, FXType.VOLUME, {"level": 1.0}) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE) # ═══════════════════════════════════════════════════════════════════ # 13. Integrated chain tests # ═══════════════════════════════════════════════════════════════════ class TestChain: def test_multiple_fx_chain(self, pipeline): """Chain multiple effects without error.""" chain = [ FXBlock(FXType.NOISE_GATE, params={"threshold": 0.001}), FXBlock(FXType.COMPRESSOR, params={"threshold": -20.0, "ratio": 4.0}), FXBlock(FXType.OVERDRIVE, params={"drive": 0.4}), FXBlock(FXType.CHORUS, params={"rate": 0.5, "mix": 0.3}), FXBlock(FXType.DELAY, params={"time": 300.0, "mix": 0.3}), FXBlock(FXType.REVERB, params={"decay": 0.4, "mix": 0.2}), FXBlock(FXType.VOLUME, params={"level": 0.8}), ] preset = Preset(name="chain_test", chain=chain, master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(SINE_TONE * 0.5) assert np.all(np.isfinite(out)), "Chain should produce finite output" assert np.all(out >= -1.0) and np.all(out <= 1.0), \ "Chain output must be in [-1, 1]" def test_all_bypass_passthrough(self, pipeline): """All blocks bypassed = pass-through.""" chain = [ FXBlock(FXType.NOISE_GATE, enabled=False), FXBlock(FXType.COMPRESSOR, bypass=True), FXBlock(FXType.OVERDRIVE, bypass=True), FXBlock(FXType.EQ, bypass=True), FXBlock(FXType.DELAY, bypass=True), FXBlock(FXType.REVERB, bypass=True), ] preset = Preset(name="bypass_test", chain=chain, master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE, atol=1e-6) def test_global_bypass(self, pipeline): """Global bypass = dry output at master volume.""" _load_fx(pipeline, FXType.OVERDRIVE, {"drive": 1.0}) pipeline.bypassed = True out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE, atol=1e-6) def test_master_volume(self, pipeline): _load_fx(pipeline, FXType.VOLUME, {"level": 1.0}) pipeline.master_volume = 0.5 out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE * 0.5, atol=1e-6) # ═══════════════════════════════════════════════════════════════════ # 14. DelayLine unit tests # ═══════════════════════════════════════════════════════════════════ class TestDelayLine: def test_write_read_identity(self): dl = _DelayLine(1024) block = np.arange(BLOCK_SIZE, dtype=np.float32) * 0.001 dl.write_block(block) out = dl.read_block(float(BLOCK_SIZE), BLOCK_SIZE) assert np.allclose(out, block, atol=1e-5), \ "Read after write at exact delay should return original" def test_fractional_interpolation(self): dl = _DelayLine(128) dl.write_block(np.array([1.0, 0.0, 0.0, 0.0], dtype=np.float32)) # Read at delay 2.5 — should interpolate between 0 and 0 dl.write_block(np.zeros(120, dtype=np.float32)) out = dl.read_block(2.5, 1) assert 0.0 <= out[0] <= 0.6, \ f"Interpolation at 2.5 should be ~0.5 (±error), got {out[0]:.4f}" def test_circular_wraparound(self): dl = _DelayLine(4) dl.write_block(np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32)) dl.write_block(np.array([5.0, 6.0, 0.0, 0.0], dtype=np.float32)) out = dl.read_block(2.0, 2) # Two full writes to a 4-element buffer: # After first write: buf=[1,2,3,4], write_idx=4 (wraps to 0 internally) # After second write: buf=[5,6,0,0], write_idx=4 # Read at delay 2: read_start=(4-2)%4=2 -> buf[2]=0, buf[3]=0 assert len(out) == 2, f"Should read 2 samples, got {len(out)}" def test_wraparound_multi_block(self): """Writing a block larger than buffer wraps correctly.""" dl = _DelayLine(4) dl.write_block(np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0], dtype=np.float32)) # After writing 6 values to 4-element buffer: # First 4: [1,2,3,4], write_idx wraps to 0 # Next 2: [5,6] written at idx 0-1 -> buf=[5,6,3,4], write_idx=2 # Read at delay 2: (2-2)%4=0 -> buf[0]=5, buf[1]=6 out = dl.read_block(2.0, 2) assert np.allclose(out, [5.0, 6.0], atol=1e-5), \ f"Wraparound should get [5, 6], got {out}" class TestCombFilter: def test_output_finite(self): cf = _CombFilter(7) out = cf.process(SINE_TONE * 0.5) assert np.all(np.isfinite(out)) assert np.all(np.abs(out) < 10.0) # Should be well-behaved class TestAllpassFilter: def test_output_finite(self): ap = _AllpassFilter(5) out = ap.process(SINE_TONE * 0.5) assert np.all(np.isfinite(out)) # ═══════════════════════════════════════════════════════════════════ # 15. State isolation between blocks # ═══════════════════════════════════════════════════════════════════ class TestStateIsolation: def test_two_reverbs_different(self, pipeline): """Two reverb blocks in chain should have separate state.""" chain = [ FXBlock(FXType.REVERB, params={"decay": 0.7, "mix": 0.5}), FXBlock(FXType.REVERB, params={"decay": 0.7, "mix": 0.5}), ] preset = Preset(name="dual_reverb", chain=chain, master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(SINE_TONE * 0.3) assert np.all(np.isfinite(out)) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_delay_and_reverb_state_independent(self, pipeline): """Delay and reverb should not share state.""" _load_fx(pipeline, FXType.DELAY, {"time": 100.0, "mix": 0.5}) pipeline.process(SINE_TONE * 0.3) out1 = pipeline._state.get("fx_0", {}).get("delay", None) assert out1 is not None, "Delay block should have 'delay' in state" # ═══════════════════════════════════════════════════════════════════ # 16-21. New effects: Octaver, Ring Mod, Bitcrusher, Parametric EQ, # Ping-pong Delay, Multi-tap Delay # ═══════════════════════════════════════════════════════════════════ class TestOctaver: def test_output_shape_and_range(self, pipeline): """Octaver should produce finite output in [-1, 1].""" _load_fx(pipeline, FXType.OCTAVER, {"mix": 0.5}) out = pipeline.process(FULL_SCALE) assert np.all(np.isfinite(out)) assert np.all(out >= -1.0) and np.all(out <= 1.0) assert np.max(np.abs(out)) > 0.5, "Octaver should pass signal" def test_dry_only(self, pipeline): """Mix=0 should pass dry signal unchanged (minus clip).""" _load_fx(pipeline, FXType.OCTAVER, {"mix": 0.0}) out = pipeline.process(FULL_SCALE) assert np.allclose(out, np.clip(FULL_SCALE, -1.0, 1.0), atol=0.01) def test_silence_muted(self, pipeline): """Silence in should produce silence out.""" _load_fx(pipeline, FXType.OCTAVER, {"mix": 1.0}) out = pipeline.process(SILENCE) assert np.max(np.abs(out)) < 0.001 class TestRingModulator: def test_output_finite(self, pipeline): """Ring mod should produce finite output.""" _load_fx(pipeline, FXType.RING_MODULATOR, {"rate": 100.0, "depth": 0.5}) out = pipeline.process(SINE_TONE * 0.3) assert np.all(np.isfinite(out)) def test_zero_depth_is_dry(self, pipeline): """Depth=0 should pass signal unchanged.""" _load_fx(pipeline, FXType.RING_MODULATOR, {"depth": 0.0, "mix": 1.0}) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE, atol=0.01) def test_output_clamped(self, pipeline): """Output should stay in [-1, 1].""" _load_fx(pipeline, FXType.RING_MODULATOR, {"depth": 1.0, "mix": 1.0}) out = pipeline.process(FULL_SCALE) assert np.all(out >= -1.0) and np.all(out <= 1.0) class TestBitcrusher: def test_output_finite(self, pipeline): """Bitcrusher should produce finite output.""" _load_fx(pipeline, FXType.BITCRUSHER, {"bits": 4, "rate": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(np.isfinite(out)) def test_16bit_no_crush(self, pipeline): """16-bit at rate=1 should be near-transparent.""" _load_fx(pipeline, FXType.BITCRUSHER, {"bits": 16, "rate": 1.0, "mix": 1.0}) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE, atol=0.01) def test_rate_reduction_holds_samples(self, pipeline): """Rate reduction >1 should create stepped output.""" _load_fx(pipeline, FXType.BITCRUSHER, {"bits": 8, "rate": 8.0, "mix": 1.0}) out = pipeline.process(SINE_TONE * 0.5) # With rate=8, every 8th sample should repeat 8 times diffs = np.diff(out) # At least some consecutive samples should be zero-difference (held) assert np.any(np.abs(diffs) < 1e-6), "Rate reduction should create held samples" def test_silence_muted(self, pipeline): """Silence in should produce silence out.""" _load_fx(pipeline, FXType.BITCRUSHER, {"bits": 4}) out = pipeline.process(SILENCE) assert np.max(np.abs(out)) < 0.001 class TestParametricEQ: def test_output_finite(self, pipeline): """PEQ should produce finite output.""" _load_fx(pipeline, FXType.PARAMETRIC_EQ, {}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(np.isfinite(out)) def test_zero_gain_passthrough(self, pipeline): """All gains 0 should pass signal (some phase shift OK).""" _load_fx(pipeline, FXType.PARAMETRIC_EQ, { "freq_0": 500.0, "gain_0": 0.0, "q_0": 0.707, }) out = pipeline.process(HALF_SCALE) assert np.max(np.abs(out)) > 0.4 # Signal should pass def test_boost_band(self, pipeline): """Positive gain should boost signal in that band.""" _load_fx(pipeline, FXType.PARAMETRIC_EQ, { "freq_0": 440.0, "gain_0": 6.0, "q_0": 1.0, }) out = pipeline.process(SINE_TONE * 0.3) assert np.max(np.abs(out)) > 0.3, "Boosted signal should be louder" def test_output_clamped(self, pipeline): """Output should stay in [-1, 1].""" _load_fx(pipeline, FXType.PARAMETRIC_EQ, { "freq_0": 440.0, "gain_0": 12.0, "q_0": 1.0, }) out = pipeline.process(FULL_SCALE) assert np.all(out >= -1.0) and np.all(out <= 1.0) class TestPingPongDelay: def test_output_finite(self, pipeline): """Ping-pong delay should produce finite output.""" _load_fx(pipeline, FXType.PING_PONG_DELAY, {"time": 5.0, "feedback": 0.0, "mix": 0.5}) out = pipeline.process(FULL_SCALE) assert np.all(np.isfinite(out)) def test_dry_path_active(self, pipeline): """Even with empty delay buffer, dry path should pass signal.""" _load_fx(pipeline, FXType.PING_PONG_DELAY, {"time": 3.0, "feedback": 0.0, "mix": 0.3}) out1 = pipeline.process(HALF_SCALE) assert np.max(np.abs(out1)) > 0.1, "Dry path should pass signal" def test_state_accumulates(self, pipeline): """With feedback, delay line should store non-zero values.""" _load_fx(pipeline, FXType.PING_PONG_DELAY, {"time": 3.0, "feedback": 0.5, "mix": 0.5}) _ = pipeline.process(HALF_SCALE) fx_state = pipeline._state.get("fx_0", {}) delay_line = fx_state.get("delay") assert delay_line is not None, "Ping-pong delay should have delay line" # Delay line buffer should have non-zero content after processing buf_max = np.max(np.abs(delay_line.buf)) assert buf_max > 0.0, f"Delay buffer should have content, got max={buf_max}" class TestMultiTapDelay: def test_output_finite(self, pipeline): """Multi-tap delay should produce finite output.""" _load_fx(pipeline, FXType.MULTI_TAP_DELAY, {"pattern": "quarter", "mix": 1.0}) out = pipeline.process(HALF_SCALE) assert np.all(np.isfinite(out)) def test_state_initialized(self, pipeline): """Multi-tap delay should initialize delay state on first call.""" _load_fx(pipeline, FXType.MULTI_TAP_DELAY, {"pattern": "quarter", "feedback": 0.0, "mix": 1.0}) out = pipeline.process(HALF_SCALE) assert np.all(np.isfinite(out)) # State should have a delay line after first call fx_state = pipeline._state.get("fx_0", {}) assert "delay" in fx_state, "Multi-tap should initialize delay line" class TestBypassNew: def test_octaver_bypass(self, pipeline): """Bypassed octaver passes audio unchanged.""" block = FXBlock(FXType.OCTAVER, enabled=True, bypass=True, params={"mix": 1.0}) preset = Preset(name="test", chain=[block], master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE) def test_bitcrusher_bypass(self, pipeline): """Bypassed bitcrusher passes audio unchanged.""" block = FXBlock(FXType.BITCRUSHER, enabled=True, bypass=True, params={"bits": 4}) preset = Preset(name="test", chain=[block], master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE) def test_ring_mod_bypass(self, pipeline): """Bypassed ring modulator passes audio unchanged.""" block = FXBlock(FXType.RING_MODULATOR, enabled=True, bypass=True, params={"rate": 100.0}) preset = Preset(name="test", chain=[block], master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE) # ═══════════════════════════════════════════════════════════════════ # Analog delay (subtype of DELAY) # ═══════════════════════════════════════════════════════════════════ class TestAnalogDelay: def test_output_range(self, pipeline): """Analog delay output must be in [-1, 1].""" _load_fx(pipeline, FXType.DELAY, {"subtype": "analog", "time": 400.0, "feedback": 0.3, "mix": 0.5}) out = pipeline.process(SINE_TONE * 0.5) assert np.all(out >= -1.0) and np.all(out <= 1.0) def test_dry_only_at_zero_mix(self, pipeline): """Zero mix = dry passthrough.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "analog", "time": 400.0, "feedback": 0.3, "mix": 0.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.allclose(out, SINE_TONE * 0.5, atol=1e-4) def test_feedback_tail_decays(self, pipeline): """Analog delay produces decaying echo tail after input stops.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "analog", "time": 50.0, "feedback": 0.6, "mix": 1.0}) for _ in range(20): pipeline.process(SINE_TONE * 0.5) tail1 = pipeline.process(SILENCE) tail2 = pipeline.process(SILENCE) assert np.max(np.abs(tail1)) > 0, "Delay tail should be present" assert np.max(np.abs(tail2)) <= np.max(np.abs(tail1)) + 0.001, \ "Echo should decay" def test_tone_affects_spectrum(self, pipeline): """Tone=0 (dark) should have less high-frequency energy than tone=1 (bright).""" _load_fx(pipeline, FXType.DELAY, {"subtype": "analog", "time": 100.0, "feedback": 0.7, "mix": 1.0, "tone": 0.0}) for _ in range(10): pipeline.process(SINE_TONE * 0.8) dark_tail = pipeline.process(SILENCE) _load_fx(pipeline, FXType.DELAY, {"subtype": "analog", "time": 100.0, "feedback": 0.7, "mix": 1.0, "tone": 1.0}) for _ in range(10): pipeline.process(SINE_TONE * 0.8) bright_tail = pipeline.process(SILENCE) # Darker tone should have less total energy (highs removed) dark_energy = np.sqrt(np.mean(dark_tail ** 2)) bright_energy = np.sqrt(np.mean(bright_tail ** 2)) # Bright should have equal or more energy than dark assert bright_energy >= dark_energy * 0.9, \ f"Bright tail ({bright_energy:.6f}) should not be much lower than dark ({dark_energy:.6f})" def test_state_initialized(self, pipeline): """Analog delay should initialize delay line on first call.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "analog", "time": 200.0, "feedback": 0.3, "mix": 0.5}) out = pipeline.process(HALF_SCALE) assert np.all(np.isfinite(out)) fx_state = pipeline._state.get("fx_0", {}) assert "delay" in fx_state, "Analog delay should initialize delay line" assert "lp_z" in fx_state, "Analog delay should initialize LP filter state" def test_bypass(self, pipeline): """Bypassed analog delay passes audio unchanged.""" block = FXBlock(FXType.DELAY, enabled=True, bypass=True, params={"subtype": "analog", "time": 400.0, "mix": 1.0}) preset = Preset(name="test", chain=[block], master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE) # ═══════════════════════════════════════════════════════════════════ # Drive Subtype: Klon Centaur # ═══════════════════════════════════════════════════════════════════ class TestKlon: """Klon Centaur-style transparent overdrive with clean blend.""" def test_output_clamped(self, pipeline): """Klon output should stay within [-1, 1].""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 0.8, "gain": 1.0}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_shapes_waveform(self, pipeline): """Klon should change waveform shape at high drive.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert not np.allclose(out, SINE_TONE * 0.5, atol=0.05) def test_clean_blend_dry(self, pipeline): """Blend=0 should pass clean signal (mostly dry).""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 1.0, "blend": 0.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.3) assert np.allclose(out, SINE_TONE * 0.3, atol=0.02) def test_clean_blend_wet(self, pipeline): """Blend=1 should pass only overdriven signal.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 1.0, "blend": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.3) assert np.max(out) > 0.0 assert np.max(np.abs(out - SINE_TONE * 0.3)) > 0.02 def test_low_drive_passthrough(self, pipeline): """Low drive should produce output.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 0.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.05) assert np.max(np.abs(out)) > 0.0 def test_default_subtype_ts808(self, pipeline): """No subtype=ts808 should use original overdrive.""" _load_fx(pipeline, FXType.OVERDRIVE, {"drive": 0.5, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 # ═══════════════════════════════════════════════════════════════════ # Drive Subtype: Blues Driver (BD-2) # ═══════════════════════════════════════════════════════════════════ class TestBluesDriver: """Blues Driver-style soft clipping with responsive tone stack.""" def test_output_clamped(self, pipeline): """BD-2 output should stay within [-1, 1].""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 0.8, "gain": 1.0}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_shapes_waveform(self, pipeline): """BD-2 should change waveform shape at high drive.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert not np.allclose(out, SINE_TONE * 0.5, atol=0.05) def test_tone_affects_output(self, pipeline): """Different tone values should produce different output.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 0.5, "tone": 0.0, "gain": 1.0}) out_a = pipeline.process(SINE_TONE * 0.5) pipeline.load_preset(Preset( name="test", chain=[FXBlock(FXType.OVERDRIVE, enabled=True, params={"subtype": "bd2", "drive": 0.5, "tone": 1.0, "gain": 1.0})], master_volume=1.0, )) out_b = pipeline.process(SINE_TONE * 0.5) assert not np.allclose(out_a, out_b, atol=0.01) def test_low_drive_passthrough(self, pipeline): """Low drive should pass signal.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 0.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.05) assert np.max(np.abs(out)) > 0.0 # ═══════════════════════════════════════════════════════════════════ # Drive Subtype: Big Muff Pi # ═══════════════════════════════════════════════════════════════════ class TestBigMuff: """Big Muff Pi-style fuzz with tone stack.""" def test_output_clamped(self, pipeline): """Muff output should stay within [-1, 1].""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 0.8, "volume": 0.5}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_fuzz_shapes_waveform(self, pipeline): """Muff should significantly reshape waveform (fuzz).""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 1.0, "volume": 0.5}) out = pipeline.process(SINE_TONE * 0.8) assert not np.allclose(out, SINE_TONE * 0.8, atol=0.05) rms_out = np.sqrt(np.mean(out ** 2)) assert rms_out > 0.2, "Muff should produce significant output" def test_tone_sweep(self, pipeline): """Different tone positions should produce different EQ.""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 0.5, "tone": 0.0, "volume": 0.5}) out_a = pipeline.process(SINE_TONE) pipeline.load_preset(Preset( name="test", chain=[FXBlock(FXType.FUZZ, enabled=True, params={"subtype": "muff", "sustain": 0.5, "tone": 1.0, "volume": 0.5})], master_volume=1.0, )) out_b = pipeline.process(SINE_TONE) assert not np.allclose(out_a, out_b, atol=0.01) def test_silence_muted(self, pipeline): """Silence in should give silence out.""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 0.5, "volume": 0.5}) out = pipeline.process(SILENCE) assert np.max(np.abs(out)) == 0.0 def test_default_subtype_fuzz(self, pipeline): """No subtype on FUZZ should use original fuzz algorithm.""" _load_fx(pipeline, FXType.FUZZ, {"drive": 0.5, "gain": 0.5}) out = pipeline.process(SINE_TONE * 0.8) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 # ═══════════════════════════════════════════════════════════════════ # Ping-pong delay (subtype of DELAY) # ═══════════════════════════════════════════════════════════════════ class TestDelaySubtypePingPong: def test_output_finite(self, pipeline): """Ping-pong as delay subtype should produce finite output.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "ping_pong", "time": 5.0, "feedback": 0.0, "mix": 0.5}) out = pipeline.process(FULL_SCALE) assert np.all(np.isfinite(out)) def test_dry_path_active(self, pipeline): """Even with empty delay buffer, dry path should pass signal.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "ping_pong", "time": 3.0, "feedback": 0.0, "mix": 0.3}) out1 = pipeline.process(HALF_SCALE) assert np.max(np.abs(out1)) > 0.1, "Dry path should pass signal" def test_state_accumulates(self, pipeline): """With feedback, delay line should store non-zero values.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "ping_pong", "time": 3.0, "feedback": 0.5, "mix": 0.5}) _ = pipeline.process(HALF_SCALE) fx_state = pipeline._state.get("fx_0", {}) delay_line = fx_state.get("delay") assert delay_line is not None, "Ping-pong delay should have delay line" buf_max = np.max(np.abs(delay_line.buf)) assert buf_max > 0.0, f"Delay buffer should have content, got max={buf_max}" def test_ping_alternates(self, pipeline): """Ping-pong pan should alternate each block.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "ping_pong", "time": 2.0, "feedback": 0.0, "mix": 1.0}) for _ in range(5): pipeline.process(FULL_SCALE) fx_state = pipeline._state.get("fx_0", {}) # After 5 blocks starting at 1, pattern is: 1, -1, 1, -1, 1 # ping should be -1 after 5 alternations (odd count) assert fx_state.get("ping") == -1, \ f"Ping should alternate to -1 after 5 blocks, got {fx_state.get('ping')}" def test_zero_mix_passthrough(self, pipeline): """Zero mix should pass dry signal through.""" _load_fx(pipeline, FXType.DELAY, {"subtype": "ping_pong", "time": 10.0, "feedback": 0.0, "mix": 0.0}) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE, atol=1e-4) def test_bypass(self, pipeline): """Bypassed ping-pong delay passes audio unchanged.""" block = FXBlock(FXType.DELAY, enabled=True, bypass=True, params={"subtype": "ping_pong", "time": 400.0, "mix": 1.0}) preset = Preset(name="test", chain=[block], master_volume=1.0) pipeline.load_preset(preset) out = pipeline.process(HALF_SCALE) assert np.allclose(out, HALF_SCALE) # ═══════════════════════════════════════════════════════════════════ # Drive Subtype: Klon Centaur # ═══════════════════════════════════════════════════════════════════ class TestKlon: """Klon Centaur-style transparent overdrive with clean blend.""" def test_output_clamped(self, pipeline): """Klon output should stay within [-1, 1].""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 0.8, "gain": 1.0}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_shapes_waveform(self, pipeline): """Klon should change waveform shape at high drive.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert not np.allclose(out, SINE_TONE * 0.5, atol=0.05) def test_clean_blend_dry(self, pipeline): """Blend=0 should pass clean signal (mostly dry).""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 1.0, "blend": 0.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.3) assert np.allclose(out, SINE_TONE * 0.3, atol=0.02) def test_clean_blend_wet(self, pipeline): """Blend=1 should pass only overdriven signal.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 1.0, "blend": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.3) assert np.max(out) > 0.0 assert np.max(np.abs(out - SINE_TONE * 0.3)) > 0.02 def test_low_drive_passthrough(self, pipeline): """Low drive should produce output.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "klon", "drive": 0.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.05) assert np.max(np.abs(out)) > 0.0 def test_default_subtype_ts808(self, pipeline): """No subtype=ts808 should use original overdrive.""" _load_fx(pipeline, FXType.OVERDRIVE, {"drive": 0.5, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 # ═══════════════════════════════════════════════════════════════════ # Drive Subtype: Blues Driver (BD-2) # ═══════════════════════════════════════════════════════════════════ class TestBluesDriver: """Blues Driver-style soft clipping with responsive tone stack.""" def test_output_clamped(self, pipeline): """BD-2 output should stay within [-1, 1].""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 0.8, "gain": 1.0}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_shapes_waveform(self, pipeline): """BD-2 should change waveform shape at high drive.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 1.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.5) assert not np.allclose(out, SINE_TONE * 0.5, atol=0.05) def test_tone_affects_output(self, pipeline): """Different tone values should produce different output.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 0.5, "tone": 0.0, "gain": 1.0}) out_a = pipeline.process(SINE_TONE * 0.5) pipeline.load_preset(Preset( name="test", chain=[FXBlock(FXType.OVERDRIVE, enabled=True, params={"subtype": "bd2", "drive": 0.5, "tone": 1.0, "gain": 1.0})], master_volume=1.0, )) out_b = pipeline.process(SINE_TONE * 0.5) assert not np.allclose(out_a, out_b, atol=0.01) def test_low_drive_passthrough(self, pipeline): """Low drive should pass signal.""" _load_fx(pipeline, FXType.OVERDRIVE, {"subtype": "bd2", "drive": 0.0, "gain": 1.0}) out = pipeline.process(SINE_TONE * 0.05) assert np.max(np.abs(out)) > 0.0 # ═══════════════════════════════════════════════════════════════════ # Drive Subtype: Big Muff Pi # ═══════════════════════════════════════════════════════════════════ class TestBigMuff: """Big Muff Pi-style fuzz with tone stack.""" def test_output_clamped(self, pipeline): """Muff output should stay within [-1, 1].""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 0.8, "volume": 0.5}) out = pipeline.process(FULL_SCALE) assert np.max(out) <= 1.0 and np.min(out) >= -1.0 def test_fuzz_shapes_waveform(self, pipeline): """Muff should significantly reshape waveform (fuzz).""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 1.0, "volume": 0.5}) out = pipeline.process(SINE_TONE * 0.8) assert not np.allclose(out, SINE_TONE * 0.8, atol=0.05) rms_out = np.sqrt(np.mean(out ** 2)) assert rms_out > 0.2, "Muff should produce significant output" def test_tone_sweep(self, pipeline): """Different tone positions should produce different EQ.""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 0.5, "tone": 0.0, "volume": 0.5}) out_a = pipeline.process(SINE_TONE) pipeline.load_preset(Preset( name="test", chain=[FXBlock(FXType.FUZZ, enabled=True, params={"subtype": "muff", "sustain": 0.5, "tone": 1.0, "volume": 0.5})], master_volume=1.0, )) out_b = pipeline.process(SINE_TONE) assert not np.allclose(out_a, out_b, atol=0.01) def test_silence_muted(self, pipeline): """Silence in should give silence out.""" _load_fx(pipeline, FXType.FUZZ, {"subtype": "muff", "sustain": 0.5, "volume": 0.5}) out = pipeline.process(SILENCE) assert np.max(np.abs(out)) == 0.0 def test_default_subtype_fuzz(self, pipeline): """No subtype on FUZZ should use original fuzz algorithm.""" _load_fx(pipeline, FXType.FUZZ, {"drive": 0.5, "gain": 0.5}) out = pipeline.process(SINE_TONE * 0.8) assert np.max(out) <= 1.0 and np.min(out) >= -1.0