praxis/tests/test_assist_webrtc_reconnect.py

"""Chaos test for the WebRTC reconnect logic (REQ-IDEATE-08, TASK-06-03).

Verifies the reconnect state machine:
  - Open a warm connection → 'connected'
  - Simulate a disconnect → 'reconnecting'
  - New offer within 30s → 'connected' (pipeline rebuilt)
  - Disconnect + no new offer within 30s → 'disconnected'
  - The shift is NOT auto-ended on disconnect (the session row is still active)
  - The 8h auto-end still fires on a disconnected shift (D-069)

The test uses a shortened reconnect wait (1s) to keep CI fast.
"""

from __future__ import annotations

import asyncio
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from server.assist.webrtc import (
    WarmWebRTCManager,
    WarmConnection,
    _RECONNECT_WAIT_S,
)


@pytest.fixture
def manager():
    return WarmWebRTCManager()


def test_reconnect_state_machine_disconnected_after_timeout(manager: WarmWebRTCManager):
    """Disconnect + no new offer within the wait → 'disconnected'."""
    # Seed a fake warm connection in 'connected' state.
    warm = WarmConnection(
        connection=MagicMock(),
        task=MagicMock(),
        runner=MagicMock(),
        shift_id="shift-1",
        reconnect_state="connected",
    )
    manager._connections["shift-1"] = warm

    async def _run():
        # Shorten the reconnect wait so the test is fast.
        with patch("server.assist.webrtc._RECONNECT_WAIT_S", 0.1):
            await manager._on_disconnect("shift-1")

    asyncio.run(_run())
    assert manager.get_reconnect_state("shift-1") == "disconnected"


def test_reconnect_state_machine_reconnect_within_window(manager: WarmWebRTCManager):
    """New offer within the wait → 'connected' (pipeline rebuilt)."""
    warm = WarmConnection(
        connection=MagicMock(),
        task=MagicMock(),
        runner=MagicMock(),
        shift_id="shift-2",
        reconnect_state="connected",
    )
    manager._connections["shift-2"] = warm

    async def _run():
        # Start the disconnect handler (it will wait 0.1s).
        with patch("server.assist.webrtc._RECONNECT_WAIT_S", 0.1):
            task = asyncio.create_task(manager._on_disconnect("shift-2"))
            await asyncio.sleep(0.02)  # let it enter 'reconnecting'
            assert manager.get_reconnect_state("shift-2") == "reconnecting"
            # Simulate a reconnect offer arriving before the timeout.
            warm.reconnect_state = "connected"
            await task

    asyncio.run(_run())
    # The state was set back to 'connected' by the reconnect.
    assert manager.get_reconnect_state("shift-2") == "connected"


def test_shift_not_auto_ended_on_disconnect(manager: WarmWebRTCManager):
    """The shift is NOT auto-ended on disconnect (the session row stays active).

    The WarmWebRTCManager doesn't touch the sessions table — only the
    ShiftLifecycleManager (8h auto-end) ends shifts. This test verifies the
    manager doesn't end the shift on disconnect.
    """
    warm = WarmConnection(
        connection=MagicMock(),
        task=MagicMock(),
        runner=MagicMock(),
        shift_id="shift-3",
        reconnect_state="connected",
    )
    manager._connections["shift-3"] = warm

    async def _run():
        with patch("server.assist.webrtc._RECONNECT_WAIT_S", 0.1):
            await manager._on_disconnect("shift-3")

    asyncio.run(_run())
    # The connection is still in the map (not removed) — the shift is still active.
    assert manager.get("shift-3") is not None
    assert manager.get_reconnect_state("shift-3") == "disconnected"


def test_close_removes_connection(manager: WarmWebRTCManager):
    """close() removes the connection from the active map."""
    warm = WarmConnection(
        connection=MagicMock(),
        task=MagicMock(),
        runner=MagicMock(),
        shift_id="shift-4",
        reconnect_state="connected",
        heartbeat_task=None,
    )
    # Mock the connection close so it doesn't fail.
    warm.connection.close = AsyncMock()
    manager._connections["shift-4"] = warm

    async def _run():
        await manager.close("shift-4")

    asyncio.run(_run())
    assert manager.get("shift-4") is None


def test_get_reconnect_state_unknown_shift(manager: WarmWebRTCManager):
    """An unknown shift_id returns 'disconnected'."""
    assert manager.get_reconnect_state("unknown-shift") == "disconnected"
    assert manager.get("unknown-shift") is None


def test_8h_auto_end_fires_on_disconnected_shift():
    """D-069: the 8h auto-end still fires on a disconnected shift.

    The ShiftLifecycleManager checks list_active_assist_sessions() (sessions
    with ended_at IS NULL) — the WebRTC connection state is irrelevant. A
    disconnected shift still has an active session row, so the 8h auto-end
    fires. This test verifies the two systems are decoupled.
    """
    import datetime as _dt
    import sqlite3
    from pathlib import Path
    from tempfile import NamedTemporaryFile

    from db.migrate import apply_migrations
    from db.store import PraxisStore, HARDCODED_LEARNER_ID
    from server.assist.lifecycle import ShiftLifecycleManager

    async def _run():
        with NamedTemporaryFile(suffix=".db", delete=False) as f:
            db_path = Path(f.name)
        apply_migrations(db_path)
        store = PraxisStore(db_path)
        await store.init()
        # Start an assist shift.
        sid = await store.start_session_typed(
            HARDCODED_LEARNER_ID, "assist:refund", "assist"
        )
        # Backdate started_at to 9h ago.
        old = (_dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(hours=9)).strftime(
            "%Y-%m-%d %H:%M:%S"
        )
        conn = sqlite3.connect(str(db_path))
        conn.execute("UPDATE sessions SET started_at = ? WHERE id = ?", (old, sid))
        conn.commit()
        conn.close()
        # The 8h auto-end should fire (the shift is active regardless of WebRTC state).
        mgr = ShiftLifecycleManager(store, max_shift_hours=8)
        ended = await mgr.check_auto_end()
        assert sid in ended
        row = await store.get_session(sid)
        assert row.outcome == "auto_ended"

    asyncio.run(_run())