"""
Tests for BeatDetector — AUD-03 (system audio capture) and AUD-04 (beat detection).
TDD RED phase: these tests define expected behavior before implementation.

Threading safety rules (D-04, D-05):
- _audio_callback MUST NOT call asyncio APIs directly
- Beat events MUST travel through call_soon_threadsafe
"""
import asyncio
import inspect
import threading
import time
import pytest


# ─── Structural tests (no audio device required) ────────────────────────────

def test_beat_detector_importable():
    """BeatDetector must be importable from led_sync.audio."""
    from led_sync.audio import BeatDetector
    assert BeatDetector is not None


def test_beat_detector_importable_from_module():
    """BeatDetector must be importable from led_sync.audio.beat_detector."""
    from led_sync.audio.beat_detector import BeatDetector
    assert BeatDetector is not None


def test_constants_present():
    """Module constants must match latency specification."""
    from led_sync.audio.beat_detector import SAMPLE_RATE, HOP_SIZE, BUF_SIZE
    assert SAMPLE_RATE == 44100
    assert HOP_SIZE == 512
    assert BUF_SIZE == 1024


def test_latency_spec():
    """HOP_SIZE / SAMPLE_RATE must yield < 12ms latency (target ~11ms)."""
    from led_sync.audio.beat_detector import SAMPLE_RATE, HOP_SIZE
    latency_ms = HOP_SIZE / SAMPLE_RATE * 1000
    assert latency_ms < 12.0, f"Latency {latency_ms:.2f}ms exceeds 11ms spec"


@pytest.mark.asyncio
async def test_initial_state():
    """beat_count starts at 0, bpm starts at 0.0."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)
    assert detector.beat_count == 0
    assert detector.bpm == 0.0


@pytest.mark.asyncio
async def test_stop_before_start_is_noop():
    """Calling stop() before start() must not raise any exception."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)
    # Must not raise
    detector.stop()
    detector.stop()  # second call also noop


@pytest.mark.asyncio
async def test_double_stop_is_noop():
    """Calling stop() twice must not raise even if stream was opened."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)
    detector.stop()
    # Shouldn't raise
    detector.stop()


def test_beat_count_property_is_thread_safe():
    """beat_count property must be accessible from multiple threads (uses lock)."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.new_event_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)

    errors = []
    results = []

    def read_count():
        try:
            results.append(detector.beat_count)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=read_count) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not errors, f"Thread safety errors: {errors}"
    assert all(r == 0 for r in results)
    loop.close()


def test_bpm_property_is_thread_safe():
    """bpm property must be accessible from multiple threads (uses lock)."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.new_event_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)

    errors = []
    results = []

    def read_bpm():
        try:
            results.append(detector.bpm)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=read_bpm) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not errors, f"Thread safety errors: {errors}"
    assert all(r == 0.0 for r in results)
    loop.close()


# ─── Audio callback thread-safety check ─────────────────────────────────────

def test_audio_callback_uses_call_soon_threadsafe():
    """
    _audio_callback source code must use call_soon_threadsafe (not asyncio APIs directly).
    This is a static analysis check enforcing D-05.
    """
    import inspect
    from led_sync.audio import beat_detector as mod

    source = inspect.getsource(mod)
    # Must contain the thread-safe bridge
    assert "call_soon_threadsafe" in source, (
        "_audio_callback must use call_soon_threadsafe to bridge audio thread -> asyncio"
    )
    # Must NOT call asyncio.create_task or asyncio.ensure_future directly in callback
    # (these would be wrong — asyncio is not thread-safe to call directly)
    # We check by ensuring there's no bare asyncio. call in the source
    # (call_soon_threadsafe is the correct pattern)
    lines = source.splitlines()
    callback_lines = []
    in_callback = False
    brace_depth = 0
    for line in lines:
        if "def _audio_callback" in line:
            in_callback = True
        if in_callback:
            callback_lines.append(line)
            # Stop at next non-indented def (crude but sufficient)
            if len(callback_lines) > 1 and line.strip().startswith("def "):
                break

    callback_text = "\n".join(callback_lines)
    # Must not directly call asyncio.run() or asyncio.get_event_loop() in callback
    assert "asyncio.run(" not in callback_text
    assert "asyncio.get_event_loop()" not in callback_text


# ─── Simulated beat injection (no real audio device) ────────────────────────

@pytest.mark.asyncio
async def test_beat_callback_called_via_asyncio_bridge():
    """
    Simulate what the audio thread does: call_soon_threadsafe with on_beat coroutine.
    Verify beat_count increments and on_beat is called in asyncio context.
    """
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()
    received_beats = []
    received_threads = []

    async def on_beat(beat_time: float):
        received_beats.append(beat_time)
        received_threads.append(threading.current_thread())

    detector = BeatDetector(loop=loop, on_beat=on_beat)

    # Simulate what _audio_callback does internally when a beat is detected
    # We manually trigger the bridge to test it works correctly
    with detector._lock:
        detector._beat_count += 1
        detector._bpm = 120.0

    loop.call_soon_threadsafe(
        loop.create_task,
        on_beat(1.0),
    )

    # Give asyncio a chance to process
    await asyncio.sleep(0.05)

    assert len(received_beats) == 1
    assert received_beats[0] == 1.0
    # on_beat should run in the asyncio thread (main thread in test)
    assert received_threads[0] == threading.main_thread()


@pytest.mark.asyncio
async def test_beat_count_increments_correctly():
    """beat_count reflects simulated beats correctly."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)
    assert detector.beat_count == 0

    # Simulate 3 beats
    for i in range(3):
        with detector._lock:
            detector._beat_count += 1

    assert detector.beat_count == 3


@pytest.mark.asyncio
async def test_bpm_updated():
    """bpm reflects simulated BPM correctly."""
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)
    assert detector.bpm == 0.0

    with detector._lock:
        detector._bpm = 128.5

    assert detector.bpm == 128.5


# ─── ImportError handling ────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_start_raises_runtime_error_on_missing_deps(monkeypatch):
    """start() must raise RuntimeError if sounddevice or aubio cannot be imported."""
    import builtins
    from led_sync.audio.beat_detector import BeatDetector

    loop = asyncio.get_running_loop()

    async def on_beat(beat_time: float):
        pass

    detector = BeatDetector(loop=loop, on_beat=on_beat)

    original_import = builtins.__import__

    def mock_import(name, *args, **kwargs):
        if name == "sounddevice":
            raise ImportError("mocked missing sounddevice")
        return original_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, "__import__", mock_import)

    with pytest.raises(RuntimeError, match="sounddevice"):
        detector.start()