"""Outlier-hardening test battery for the deterministic pipeline.
These run locally (no GPU/Modal). They prove the feature extractor, rule
engine, and output validator survive every degenerate input we could think of
and always emit finite, sane, UI-safe values.
Run: python -m pytest tests/ -q
"""
import math
import os
import sys
import numpy as np
import pytest
import soundfile as sf
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from audio_analyzer import extract_features, AudioFeatures, SR # noqa: E402
from fault_rules import rank_candidates, RULES # noqa: E402
from json_guard import validate # noqa: E402
ASSETS = os.path.join(os.path.dirname(os.path.dirname(__file__)), "assets")
APPLIANCES = list(RULES.keys()) + ["Unknown gadget", "", None]
# --- helpers ---------------------------------------------------------------
def _write(tmp_path, y, sr=SR, name="t.wav"):
p = tmp_path / name
sf.write(p, np.asarray(y, dtype=np.float32), sr)
return str(p)
def _assert_finite(f: AudioFeatures):
for k, v in f.to_dict().items():
if isinstance(v, bool):
continue
if isinstance(v, (int, float)):
assert math.isfinite(v), f"{k} is not finite: {v}"
# --- real samples ----------------------------------------------------------
@pytest.mark.skipif(not os.path.exists(os.path.join(ASSETS, "sample_washer_bearing.wav")),
reason="run assets/generate_samples.py first")
def test_bearing_detected_and_finite():
f = extract_features(os.path.join(ASSETS, "sample_washer_bearing.wav"))
_assert_finite(f)
assert f.signal_present and f.has_regular_pattern
assert any("bearing" in c.name.lower() for c in rank_candidates(f, "Washing machine"))
@pytest.mark.skipif(not os.path.exists(os.path.join(ASSETS, "sample_washer_good.wav")),
reason="run assets/generate_samples.py first")
def test_good_sample_calm():
f = extract_features(os.path.join(ASSETS, "sample_washer_good.wav"))
_assert_finite(f)
assert not f.has_regular_pattern and f.anomaly_score < 0.6
# --- audio outliers --------------------------------------------------------
def test_missing_file():
f = extract_features("does_not_exist.wav")
assert not f.signal_present
_assert_finite(f)
def test_none_and_empty_path():
for bad in (None, "", 123):
f = extract_features(bad) # type: ignore[arg-type]
assert not f.signal_present
def test_non_audio_file(tmp_path):
p = tmp_path / "notaudio.wav"
p.write_bytes(b"this is not audio data at all")
f = extract_features(str(p))
_assert_finite(f) # must not raise, must be finite
def test_silence(tmp_path):
f = extract_features(_write(tmp_path, np.zeros(SR)))
assert not f.signal_present
_assert_finite(f)
def test_all_nan(tmp_path):
p = tmp_path / "nan.wav"
sf.write(p, np.full(SR, np.nan, dtype=np.float32), SR)
f = extract_features(str(p))
_assert_finite(f)
def test_single_sample(tmp_path):
f = extract_features(_write(tmp_path, np.array([0.5])))
_assert_finite(f)
def test_very_short(tmp_path):
f = extract_features(_write(tmp_path, np.random.randn(200) * 0.5))
_assert_finite(f)
def test_clipping(tmp_path):
y = np.sign(np.sin(2 * np.pi * 200 * np.linspace(0, 3, SR * 3))) # full-scale square
f = extract_features(_write(tmp_path, y))
_assert_finite(f)
assert f.peak_db <= 6.0
def test_dc_offset(tmp_path):
f = extract_features(_write(tmp_path, np.full(SR * 2, 0.8, dtype=np.float32)))
_assert_finite(f)
def test_pure_tone(tmp_path):
y = 0.6 * np.sin(2 * np.pi * 440 * np.linspace(0, 4, SR * 4))
f = extract_features(_write(tmp_path, y))
_assert_finite(f)
assert 0.0 <= f.harmonic_ratio <= 1.0
def test_white_noise(tmp_path):
f = extract_features(_write(tmp_path, np.random.uniform(-0.7, 0.7, SR * 4)))
_assert_finite(f)
def test_stereo_input(tmp_path):
stereo = np.random.randn(SR * 2, 2).astype(np.float32) * 0.3
f = extract_features(_write(tmp_path, stereo))
_assert_finite(f)
def test_odd_sample_rate(tmp_path):
y = 0.5 * np.sin(2 * np.pi * 300 * np.linspace(0, 3, 8000 * 3))
f = extract_features(_write(tmp_path, y, sr=8000))
_assert_finite(f)
def test_overlong_is_capped(tmp_path):
y = 0.4 * np.sin(2 * np.pi * 120 * np.linspace(0, 30, SR * 30))
f = extract_features(_write(tmp_path, y))
assert f.duration_s <= 10.0 + 1e-3
# --- rule engine outliers --------------------------------------------------
def test_rules_never_crash_on_extreme_features():
extremes = AudioFeatures(
duration_s=10.0, rms_db=20.0, rms_variance=1e6, zero_crossing_rate=1.0,
spectral_centroid_hz=SR / 2, spectral_bandwidth_hz=SR, spectral_rolloff_hz=SR / 2,
dominant_frequency_hz=SR / 2, harmonic_ratio=1.0, onset_rate_per_sec=1000.0,
has_regular_pattern=True, pattern_interval_ms=1.0, peak_db=6.0,
anomaly_score=1.0, signal_present=True,
)
zeros = AudioFeatures(*([0.0] * 10), False, 0.0, 0.0, 0.0, False)
for f in (extremes, zeros):
for ap in APPLIANCES:
cands = rank_candidates(f, ap)
assert cands and all(0.0 <= c.weight <= 1.0 for c in cands)
def test_every_appliance_returns_candidate():
f = extract_features(os.path.join(ASSETS, "sample_washer_bearing.wav")) \
if os.path.exists(os.path.join(ASSETS, "sample_washer_bearing.wav")) \
else AudioFeatures(*([0.0] * 10), False, 0.0, 0.0, 0.0, True)
for ap in RULES:
assert rank_candidates(f, ap)
# --- validator outliers ----------------------------------------------------
def _cands():
return rank_candidates(
AudioFeatures(8, -18, 0.02, 0.11, 2400, 2500, 4000, 1800, 0.6, 4,
True, 250, -1, 0.5, True), "Washing machine")
def test_validate_garbage_json():
r = validate("the model rambled with no json", _cands())
assert r.fault and r.urgency in {"CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"}
def test_validate_empty_and_none():
for raw in ("", None):
r = validate(raw, _cands()) # type: ignore[arg-type]
assert r.fault
def test_validate_html_injection_is_contained():
raw = '{"fault": "", "urgency": "HIGH", "checks": ["x"], "confidence": 50}'
r = validate(raw, _cands())
# ungrounded fault snaps back to a real candidate (no script survives)
assert "