95029c6c8c
CI / test (push) Has been cancelled
- Frontend now reads auth_pin from /api/state and includes X-Pedal-Auth on all non-GET requests - Fixed test imports after removing module-level BLOCK_SIZE/SAMPLE_RATE constants - Removed rogue test_issue_2_3.py from tracking
1111 lines
53 KiB
Python
1111 lines
53 KiB
Python
"""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,
|
|
_CombFilter,
|
|
_AllpassFilter,
|
|
_DelayLine,
|
|
_compute_lowshelf_coeffs,
|
|
_compute_highshelf_coeffs,
|
|
_compute_peaking_coeffs,
|
|
)
|
|
|
|
# Local test constants (no longer exported by src.dsp.pipeline)
|
|
BLOCK_SIZE = 256
|
|
SAMPLE_RATE = 48000
|
|
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)
|