meeting-room/tests/test_async_bridge.py

"""Tests for AsyncEventBridge — sync EventEmitter → asyncio.Queue bridge."""

from __future__ import annotations

import asyncio

import pytest

from meeting_room.events import (
    AGENT_MESSAGE,
    AGENT_TURN_START,
    DISCUSSION_END,
    DISCUSSION_START,
    ROUND_END,
    ROUND_START,
    TOOL_CALL,
    AsyncEventBridge,
    EventEmitter,
)


class TestAsyncEventBridgeSubscribe:
    """subscribe() creates a queue that receives events."""

    @pytest.mark.asyncio
    async def test_subscribe_returns_queue(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        q = bridge.subscribe()
        assert isinstance(q, asyncio.Queue)

    @pytest.mark.asyncio
    async def test_multiple_subscribers(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        q1 = bridge.subscribe()
        q2 = bridge.subscribe()
        assert q1 is not q2


class TestAsyncEventBridgeStartStop:
    """start()/stop() wire the bridge to the sync bus."""

    @pytest.mark.asyncio
    async def test_start_subscribes_to_all_event_types(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        for event_type in [
            DISCUSSION_START, ROUND_START, AGENT_TURN_START,
            AGENT_MESSAGE, TOOL_CALL, ROUND_END, DISCUSSION_END,
        ]:
            assert len(bus.handlers(event_type)) == 1

        bridge.stop()

    @pytest.mark.asyncio
    async def test_stop_unsubscribes(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)
        bridge.stop()

        for event_type in [
            DISCUSSION_START, ROUND_START, AGENT_TURN_START,
            AGENT_MESSAGE, TOOL_CALL, ROUND_END, DISCUSSION_END,
        ]:
            assert len(bus.handlers(event_type)) == 0


class TestAsyncEventBridgeForwarding:
    """Events emitted on the sync bus arrive in async queues."""

    @pytest.mark.asyncio
    async def test_forward_event_to_queue(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        q = bridge.subscribe()

        # Emit a sync event
        bus.emit(AGENT_MESSAGE, role_id="moderator", name="Moderator", content="Hello")

        # Give the event loop a chance to process call_soon_threadsafe
        await asyncio.sleep(0.05)

        payload = q.get_nowait()
        assert payload["type"] == AGENT_MESSAGE
        assert payload["data"]["role_id"] == "moderator"
        assert payload["data"]["content"] == "Hello"
        assert "timestamp" in payload

        bridge.stop()

    @pytest.mark.asyncio
    async def test_multiple_queues_receive_same_event(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        q1 = bridge.subscribe()
        q2 = bridge.subscribe()

        bus.emit(DISCUSSION_START, name="test")
        await asyncio.sleep(0.05)

        p1 = q1.get_nowait()
        p2 = q2.get_nowait()
        assert p1["type"] == DISCUSSION_START
        assert p2["type"] == DISCUSSION_START
        assert p1["data"]["name"] == "test"
        assert p2["data"]["name"] == "test"

        bridge.stop()

    @pytest.mark.asyncio
    async def test_unsubscribe_stops_delivery(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        q = bridge.subscribe()
        bridge.unsubscribe(q)

        bus.emit(AGENT_MESSAGE, role_id="x", name="X", content="Y")
        await asyncio.sleep(0.05)

        assert q.empty()

        bridge.stop()


class TestAsyncEventBridgeThreadedForwarding:
    """Events from a worker thread reach async consumers."""

    @pytest.mark.asyncio
    async def test_forward_from_thread(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        q = bridge.subscribe()

        # Run engine in a thread (simplified)
        def sync_work() -> None:
            bus.emit(DISCUSSION_START, name="threaded")
            bus.emit(AGENT_MESSAGE, role_id="bot", name="Bot", content="Hi")
            bus.emit(DISCUSSION_END, total_messages=1)

        await asyncio.to_thread(sync_work)
        await asyncio.sleep(0.1)

        events = []
        while not q.empty():
            events.append(q.get_nowait())

        types = [e["type"] for e in events]
        assert "discussion_start" in types
        assert "agent_message" in types
        assert "discussion_end" in types

        bridge.stop()


class TestAsyncEventBridgePayloadFormat:
    """Serialized events contain type, data, and ISO timestamp."""

    @pytest.mark.asyncio
    async def test_payload_structure(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        q = bridge.subscribe()
        bus.emit(TOOL_CALL, role_id="analyst", tool_name="search", arguments='{"q":"test"}')
        await asyncio.sleep(0.05)

        payload = q.get_nowait()
        assert "type" in payload
        assert "data" in payload
        assert "timestamp" in payload
        # timestamp is ISO format string
        assert isinstance(payload["timestamp"], str)
        assert payload["data"]["tool_name"] == "search"

        bridge.stop()


class TestAsyncEventBridgeDropPolicy:
    """When a queue is full, events are dropped without crashing."""

    @pytest.mark.asyncio
    async def test_full_queue_drops_events(self) -> None:
        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        loop = asyncio.get_event_loop()
        bridge.start(loop)

        # maxsize=2 to make it fill quickly
        q: asyncio.Queue[dict] = asyncio.Queue(maxsize=2)
        with bridge._lock:
            bridge._queues.append(q)

        # Emit 5 events — queue can hold 2
        for i in range(5):
            bus.emit(AGENT_MESSAGE, role_id="x", name="X", content=f"msg{i}")
        await asyncio.sleep(0.1)

        # Queue should have at most 2 items (rest dropped)
        received = []
        while not q.empty():
            received.append(q.get_nowait())

        assert len(received) <= 2

        bridge.stop()