"""Tests for the in-process EventBus.

These exercise the drop-oldest overflow strategy, per-kind fan-out, and
the non-blocking publish contract. They are async because the bus
exposes an async iterator for subscribers.
"""

from __future__ import annotations

import asyncio
import time

import pytest

from cyclone.pubsub import EventBus, get_event_bus


@pytest.mark.asyncio
async def test_publish_no_subscribers_is_noop():
    bus = EventBus()
    # No subscribers at all — must not raise.
    await bus.publish("x", {"a": 1})


@pytest.mark.asyncio
async def test_publish_to_n_subscribers_delivers_to_all():
    bus = EventBus()
    queues = [bus.subscribe(["claim"]) for _ in range(3)]

    await bus.publish("claim", {"id": "C1"})

    received = [await asyncio.wait_for(q.__anext__(), timeout=0.5) for q in queues]
    for event in received:
        assert event == {"id": "C1", "_kind": "claim"}


@pytest.mark.asyncio
async def test_subscriber_only_receives_matching_kinds():
    bus = EventBus()
    queue = bus.subscribe(["a"])

    await bus.publish("b", {"n": 1})
    await bus.publish("a", {"n": 2})

    event = await asyncio.wait_for(queue.__anext__(), timeout=0.5)
    assert event == {"n": 2, "_kind": "a"}


@pytest.mark.asyncio
async def test_slow_subscriber_does_not_block_publish():
    bus = EventBus(max_queue_size=2)
    queue = bus.subscribe(["e"])

    async def emit():
        for i in range(3):
            await bus.publish("e", {"i": i})

    start = time.monotonic()
    await emit()
    elapsed = time.monotonic() - start

    # Publishing must not block even though the subscriber queue is full
    # after the second event; the drop-oldest policy keeps publishes non-blocking.
    assert elapsed < 0.05, f"publish blocked for {elapsed:.3f}s"

    # The slow subscriber retains only the last two events.
    seen = []
    for _ in range(2):
        seen.append(await asyncio.wait_for(queue.__anext__(), timeout=0.5))
    assert [e["i"] for e in seen] == [1, 2]


@pytest.mark.asyncio
async def test_queue_overflow_drops_oldest():
    bus = EventBus(max_queue_size=2)
    queue = bus.subscribe(["e"])

    for i in range(3):
        await bus.publish("e", {"i": i})

    seen = []
    for _ in range(2):
        seen.append(await asyncio.wait_for(queue.__anext__(), timeout=0.5))
    assert [e["i"] for e in seen] == [1, 2]


@pytest.mark.asyncio
async def test_multiple_concurrent_subscribers_each_get_all_events():
    bus = EventBus()
    queues = [bus.subscribe(["claim"]) for _ in range(2)]

    for i in range(5):
        await bus.publish("claim", {"i": i})

    for q in queues:
        seen = []
        for _ in range(5):
            seen.append(await asyncio.wait_for(q.__anext__(), timeout=0.5))
        assert [e["i"] for e in seen] == [0, 1, 2, 3, 4]


def test_get_event_bus_raises_when_app_state_uninitialized():
    """get_event_bus() late-imports cyclone.api; if the lifespan handler
    hasn't set ``app.state.event_bus`` (or it's been cleared), the
    AttributeError must surface so callers can fail loudly."""
    from cyclone.api import app

    # The autouse conftest fixture sets app.state.event_bus for every test
    # so that endpoint code can call request.app.state.event_bus. We pop
    # the key entirely here so __getattr__ raises AttributeError rather
    # than returning None for an explicitly-None slot.
    state_dict = app.state._state
    saved = state_dict.pop("event_bus", None)
    try:
        with pytest.raises((AttributeError, RuntimeError)):
            get_event_bus()
    finally:
        if saved is not None:
            state_dict["event_bus"] = saved


def test_get_event_bus_returns_app_state_bus():
    """When the lifespan handler has initialised the bus, get_event_bus()
    returns it. Validates the happy path under the same late-import."""
    from cyclone.api import app
    from cyclone.pubsub import EventBus

    saved = getattr(app.state, "event_bus", None)
    bus = EventBus()
    app.state.event_bus = bus
    try:
        assert get_event_bus() is bus
    finally:
        app.state.event_bus = saved


def test_unsubscribe_removes_queue_from_each_kind():
    bus = EventBus()
    q1, _ = bus.subscribe_raw(["a", "b"])
    q2, _ = bus.subscribe_raw(["a"])

    bus.unsubscribe(q1, ["a", "b"])

    assert q1 not in bus._subscribers.get("a", [])
    assert q1 not in bus._subscribers.get("b", [])
    assert q2 in bus._subscribers["a"]


def test_unsubscribe_is_idempotent():
    bus = EventBus()
    q, _ = bus.subscribe_raw(["a"])
    bus.unsubscribe(q, ["a"])
    # Second call must not raise.
    bus.unsubscribe(q, ["a"])
    bus.unsubscribe(q, ["b"])  # wrong kind — also fine
    assert "a" not in bus._subscribers