tests/test_pipeline.py

"""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": "<script>alert(1)</script>", "urgency": "HIGH", "checks": ["x"], "confidence": 50}'
    r = validate(raw, _cands())
    # ungrounded fault snaps back to a real candidate (no script survives)
    assert "<script>" not in r.fault


def test_validate_oversized_strings_truncated():
    raw = ('{"fault": "' + "A" * 5000 + '", "urgency": "HIGH", "checks": ["'
           + "B" * 5000 + '"], "safety": "' + "C" * 5000 + '", "confidence": 999}')
    r = validate(raw, _cands())
    assert len(r.fault) <= 80
    assert all(len(c) <= 240 for c in r.checks)
    assert len(r.safety) <= 240
    assert 0 <= r.confidence <= 100


def test_validate_bad_types():
    raw = '{"fault": 123, "urgency": "purple", "checks": "not a list", "confidence": "lots"}'
    r = validate(raw, _cands())
    assert r.urgency in {"CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"}
    assert isinstance(r.checks, list) and r.checks
    assert 0 <= r.confidence <= 100


def test_validate_empty_candidates():
    r = validate("{}", [])
    assert r.fault  # must not IndexError