festifun/backend/tests/test_events_ai.py

"""Tests for events_ai.py (lane D / spec M7).

No test hits a real API or ffmpeg: candidate detection runs on the synthetic fixture's
in-memory audio, and the classifier orchestration is exercised with stub classifiers and a
monkeypatched input-preparer. This mirrors spec M7's acceptance:
  - candidates land within ±0.5 s of the ground-truth pulse times, and
  - the two-stage loop (detect -> classify) works independent of any provider.
"""

from __future__ import annotations

from pathlib import Path

import pytest
from pydantic import ValidationError

from festival4d import config, db, events_ai, synthetic
from festival4d.events_ai import (
    ClaudeClassifier,
    GeminiClassifier,
    LocalClassifier,
    MomentClassification,
    MomentClassifier,
    detect_candidates,
    get_classifier,
    run_events,
)

SR = config.SYNTH_AUDIO_SR


# ---------------------------------------------------------------------------
# Helpers / stub classifiers (no network)
# ---------------------------------------------------------------------------
def _reference_audio():
    """cam0's audio: the master signal sliced at offset 0 (local == global time)."""
    master, gt = synthetic.synth_master_audio(SR)
    audio = synthetic.slice_for_offset(master, SR, 0.0, config.SYNTH_DURATION_S)
    return audio, gt["pulse_times_global_s"]


class StubClassifier:
    """Always returns the same valid classification; counts calls."""

    def __init__(self, event_type: str = "bass_drop") -> None:
        self.calls = 0
        self.event_type = event_type

    def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification:
        self.calls += 1
        return MomentClassification(
            event_type=self.event_type, confidence=0.9, description="stub classification"
        )


class FlakyClassifier:
    """Raises on a chosen call to exercise per-candidate exception isolation."""

    def __init__(self, fail_on_call: int = 2) -> None:
        self.calls = 0
        self.fail_on_call = fail_on_call

    def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification:
        self.calls += 1
        if self.calls == self.fail_on_call:
            raise RuntimeError("simulated provider failure")
        return MomentClassification(event_type="pyro", confidence=0.6, description="ok")


def _fresh_reference_db(offset_ms: float = 0.0) -> None:
    """A clean DB with a single reference video (matching cam0)."""
    db.init_engine()
    db.reset_db()
    db.add_video(
        filename="cam0.mp4", duration_s=config.SYNTH_DURATION_S, fps=config.SYNTH_FPS,
        width=config.SYNTH_VIDEO_W, height=config.SYNTH_VIDEO_H,
        offset_ms=offset_ms, drift_ppm=0.0, sync_confidence=1.0,
    )


def _stub_run(monkeypatch, classifier):
    """Wire run_events to run real detection on fixture audio + a stub classifier."""
    audio, _ = _reference_audio()
    monkeypatch.setattr(events_ai, "_load_reference_audio", lambda video: (audio, SR))
    monkeypatch.setattr(events_ai, "get_classifier", lambda name=None: classifier)
    monkeypatch.setattr(
        events_ai, "prepare_inputs",
        lambda t: (Path("/nonexistent/clip.mp4"), [Path("/nonexistent/frame_00.jpg")]),
    )


# ---------------------------------------------------------------------------
# Candidate detection (the acceptance-critical numeric test)
# ---------------------------------------------------------------------------
def test_detect_candidates_matches_ground_truth():
    audio, pulses = _reference_audio()
    cands = detect_candidates(audio, SR)

    # Recall: every ground-truth pulse has a candidate within ±0.5 s.
    for pulse in pulses:
        nearest = min(abs(c - pulse) for c in cands)
        assert nearest < 0.5, f"pulse {pulse}s unmatched; nearest candidate {nearest:.3f}s away"

    # Precision: no spurious candidates — each maps to a distinct ground-truth pulse.
    for c in cands:
        nearest = min(abs(c - pulse) for pulse in pulses)
        assert nearest < 0.5, f"spurious candidate at {c:.3f}s ({nearest:.3f}s from any pulse)"
    assert len(cands) == len(pulses), (cands, pulses)


def test_detect_candidates_short_audio_is_empty():
    import numpy as np

    assert detect_candidates(np.zeros(16, dtype=np.float32), SR) == []


def test_detect_candidates_silent_or_flat_audio_is_empty():
    """Silence / DC / a bare noise floor must yield NO candidates, never fabricate events."""
    import numpy as np

    n = SR * 6
    assert detect_candidates(np.zeros(n, dtype=np.float32), SR) == []          # digital silence
    assert detect_candidates(np.full(n, 0.5, dtype=np.float32), SR) == []      # constant / DC
    rng = np.random.default_rng(0)
    assert detect_candidates((1e-4 * rng.standard_normal(n)).astype(np.float32), SR) == []  # noise floor


# ---------------------------------------------------------------------------
# Frozen contract sanity (MomentClassification / MomentClassifier)
# ---------------------------------------------------------------------------
def test_moment_classification_validates_bounds_and_enum():
    ok = MomentClassification(event_type="pyro", confidence=0.5, description="x")
    assert ok.event_type == "pyro"
    with pytest.raises(ValidationError):
        MomentClassification(event_type="pyro", confidence=1.5, description="x")  # >1
    with pytest.raises(ValidationError):
        MomentClassification(event_type="not_a_type", confidence=0.5, description="x")


def test_stub_satisfies_classifier_protocol():
    assert isinstance(StubClassifier(), MomentClassifier)


# ---------------------------------------------------------------------------
# Provider selection + degradation (no network — constructors don't call out)
# ---------------------------------------------------------------------------
def test_get_classifier_unconfigured_returns_none(monkeypatch):
    for var in ("GEMINI_API_KEY", "ANTHROPIC_API_KEY",
                "FESTIVAL4D_OPENAI_BASE_URL", "FESTIVAL4D_OPENAI_MODEL", "FESTIVAL4D_OPENAI_KEY"):
        monkeypatch.delenv(var, raising=False)
    assert get_classifier("gemini") is None
    assert get_classifier("claude") is None
    assert get_classifier("local") is None


def test_get_classifier_unknown_provider_returns_none():
    assert get_classifier("does-not-exist") is None


def test_get_classifier_defaults_to_gemini(monkeypatch):
    monkeypatch.delenv("FESTIVAL4D_CLASSIFIER", raising=False)
    monkeypatch.setenv("GEMINI_API_KEY", "test-key")
    assert isinstance(get_classifier(), GeminiClassifier)


def test_get_classifier_local_when_configured(monkeypatch):
    monkeypatch.setenv("FESTIVAL4D_OPENAI_BASE_URL", "http://localhost:11434/v1")
    monkeypatch.setenv("FESTIVAL4D_OPENAI_MODEL", "qwen2.5-vl")
    monkeypatch.delenv("FESTIVAL4D_OPENAI_KEY", raising=False)
    clf = get_classifier("local")
    assert isinstance(clf, LocalClassifier)
    assert isinstance(clf, MomentClassifier)


def test_get_classifier_claude_when_configured(monkeypatch):
    monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key")
    assert isinstance(get_classifier("claude"), ClaudeClassifier)


# ---------------------------------------------------------------------------
# Orchestration loop (stub classifier — no API, no ffmpeg)
# ---------------------------------------------------------------------------
def test_run_events_classifies_all_candidates(monkeypatch):
    _fresh_reference_db()
    stub = StubClassifier(event_type="bass_drop")
    _stub_run(monkeypatch, stub)

    result = run_events()

    assert result["candidates"] == 3
    assert result["classified"] == 3
    assert result["candidates_only"] == 0
    assert result["classifier_configured"] is True
    assert stub.calls == 3

    events = db.get_events()
    assert len(events) == 3
    assert all(e.source == "ai" for e in events)
    assert all(e.event_type == "bass_drop" for e in events)
    # Candidate times map to t_global (reference offset 0) near the fixture pulses.
    times = sorted(e.t_global_s for e in events)
    for got, pulse in zip(times, (3.0, 10.0, 17.0)):
        assert abs(got - pulse) < 0.5


def test_run_events_unconfigured_stores_candidates_only(monkeypatch):
    _fresh_reference_db()
    audio, _ = _reference_audio()
    monkeypatch.setattr(events_ai, "_load_reference_audio", lambda video: (audio, SR))
    monkeypatch.setattr(events_ai, "get_classifier", lambda name=None: None)
    # Degraded path must never prepare inputs. A *raised* tripwire would be swallowed by
    # _classify_one's `except Exception`; record calls with a side effect instead.
    prepared: list = []
    monkeypatch.setattr(events_ai, "prepare_inputs", lambda t: prepared.append(t))

    result = run_events()

    assert prepared == []  # prepare_inputs was never called when unconfigured

    assert result["candidates"] == 3
    assert result["classified"] == 0
    assert result["candidates_only"] == 3
    assert result["classifier_configured"] is False

    events = db.get_events()
    assert len(events) == 3
    assert all(e.source == "audio_auto" for e in events)
    assert all(e.event_type == "candidate" for e in events)


def test_run_events_isolates_per_candidate_failure(monkeypatch):
    _fresh_reference_db()
    flaky = FlakyClassifier(fail_on_call=2)  # 2nd candidate fails
    _stub_run(monkeypatch, flaky)

    result = run_events()

    assert result["candidates"] == 3
    assert result["classified"] == 2       # two succeeded
    assert result["candidates_only"] == 1  # the failed one fell back to a candidate
    assert flaky.calls == 3                # the batch was not aborted

    events = db.get_events()
    assert sum(e.source == "ai" for e in events) == 2
    assert sum(e.source == "audio_auto" for e in events) == 1


def test_run_events_whole_track_replaces_machine_events_keeps_user(monkeypatch):
    _fresh_reference_db()
    # Pre-existing events: a stale AI one and a user correction.
    db.add_event(t_global_s=99.0, event_type="other", source="ai", description="stale")
    db.add_event(t_global_s=5.0, event_type="crowd_wave", source="user", description="mine")
    _stub_run(monkeypatch, StubClassifier())

    run_events()

    events = db.get_events()
    user = [e for e in events if e.source == "user"]
    ai = [e for e in events if e.source == "ai"]
    assert len(user) == 1 and user[0].description == "mine"   # user edit preserved
    assert len(ai) == 3                                       # fresh detection, stale AI gone
    assert not any(e.description == "stale" for e in events)


def test_run_events_windowed_adds_in_window_and_clears_nothing(monkeypatch):
    _fresh_reference_db()
    # A machine event OUTSIDE the window must survive — windowed detection is purely additive.
    db.add_event(t_global_s=3.0, event_type="bass_drop", source="ai", description="pre-existing")
    _stub_run(monkeypatch, StubClassifier())

    result = run_events(t_global_s=10.0, window_s=4.0)  # window [8, 12] -> only the ~10 s pulse

    assert result["candidates"] == 1
    assert result["window"] == {"t_global_s": 10.0, "window_s": 4.0}
    events = db.get_events()
    assert any(e.description == "pre-existing" for e in events)  # out-of-window event preserved
    assert len(events) == 2
    new = [e for e in events if e.description != "pre-existing"]
    assert len(new) == 1 and abs(new[0].t_global_s - 10.0) < 0.5


def test_run_events_no_videos_degrades(monkeypatch):
    db.init_engine()
    db.reset_db()
    result = run_events()
    assert result["candidates"] == 0
    assert "note" in result
    assert db.get_events() == []