"""Proxy recovery tests.

htmx reconnects a browser WebSocket only after an abnormal close, so a btcdata
restart must either be invisible to the browser or arrive as a retryable close
code. These tests run a real loopback WebSocket server as the stand-in btcdata:
the recovery being tested is connection lifecycle behaviour, which a stubbed
client would not exercise.
"""
import asyncio
import json
import socket
import threading

import pytest
import websockets
from fastapi.testclient import TestClient

from app import main

PRICE = json.dumps({"price": 67432.1, "prev_price": 67400, "recorded_at": ""})
LATER_PRICE = json.dumps({"price": 67500.0, "prev_price": 67432.1, "recorded_at": ""})


class FakeBtcdata:
    """Serves one list of messages per incoming connection.

    Every session but the last closes as soon as its messages are sent, which is
    what a btcdata restart looks like from the proxy's side. The last session
    stays open so the test can assert on a live stream.
    """

    def __init__(self, sessions):
        self._sessions = [list(session) for session in sessions]
        self._ready = threading.Event()
        self.connections = 0
        self.port = None

    async def _handler(self, connection):
        self.connections += 1
        session = self._sessions.pop(0) if self._sessions else []
        for message in session:
            await connection.send(message)
        if not self._sessions:
            await connection.wait_closed()

    async def _start(self):
        # websockets.serve needs a running loop, so it cannot be called from
        # run_until_complete's argument list.
        self._server = await websockets.serve(self._handler, "127.0.0.1", 0)
        self.port = self._server.sockets[0].getsockname()[1]

    async def _stop(self):
        self._server.close()
        await self._server.wait_closed()

    def _run(self):
        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._loop)
        self._loop.run_until_complete(self._start())
        self._ready.set()
        self._loop.run_forever()
        self._loop.run_until_complete(self._stop())
        self._loop.close()

    def __enter__(self):
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()
        assert self._ready.wait(10), "fake btcdata did not start"
        return self

    def stop(self):
        """Stop listening. Idempotent, so a test may stop the upstream early."""
        if self._thread.is_alive():
            self._loop.call_soon_threadsafe(self._loop.stop)
            self._thread.join(10)

    def __exit__(self, *exception):
        self.stop()
        return False


@pytest.fixture(autouse=True)
def quick_retries(monkeypatch):
    """Keep the real schedule's shape without waiting out its real delays."""
    monkeypatch.setattr(main, "UPSTREAM_RETRY_DELAYS_SECONDS", (0.01, 0.01, 0.01))


def closed_port():
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


def test_upstream_restart_recovers_on_the_same_browser_socket(monkeypatch):
    with FakeBtcdata([[PRICE], [LATER_PRICE]]) as upstream:
        monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{upstream.port}")
        with TestClient(main.app).websocket_connect("/ws/price") as browser:
            assert "$67.432,10" in browser.receive_text()
            # Same browser socket, no reload: the proxy reconnected upstream.
            assert "$67.500,00" in browser.receive_text()
        assert upstream.connections == 2


def test_browser_is_asked_to_reconnect_when_upstream_stays_down(monkeypatch):
    monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{closed_port()}")
    with TestClient(main.app).websocket_connect("/ws/price") as browser:
        with pytest.raises(main.WebSocketDisconnect) as disconnect:
            browser.receive_text()
    # 1000 is the regression: htmx treats it as "done" and never reconnects.
    assert disconnect.value.code == main.BROWSER_RETRY_CLOSE_CODE
    assert disconnect.value.code in (1006, 1011, 1012, 1013)


def test_browser_is_asked_to_reconnect_after_a_lost_upstream(monkeypatch):
    with FakeBtcdata([[PRICE]]) as upstream:
        port = upstream.port
        monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{port}")
        with TestClient(main.app).websocket_connect("/ws/price") as browser:
            assert "$67.432,10" in browser.receive_text()
            # The fake stops listening entirely, so in-proxy retries cannot win.
            upstream.stop()
            with pytest.raises(main.WebSocketDisconnect) as disconnect:
                browser.receive_text()
    assert disconnect.value.code == main.BROWSER_RETRY_CLOSE_CODE


def test_unrenderable_message_does_not_end_the_stream(monkeypatch):
    with FakeBtcdata([["not json", "", PRICE]]) as upstream:
        monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{upstream.port}")
        with TestClient(main.app).websocket_connect("/ws/price") as browser:
            assert "$67.432,10" in browser.receive_text()
        assert upstream.connections == 1


def test_strike_stream_recovers_on_the_same_browser_socket(monkeypatch):
    rates = json.dumps({"BTC/USDT": 67432.1, "BTC/EUR": 61000.5, "quote-sats": 1476})
    with FakeBtcdata([[rates], [rates]]) as upstream:
        monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{upstream.port}")
        with TestClient(main.app).websocket_connect("/ws/strike") as browser:
            assert "1.476 <span" in browser.receive_text()
            assert 'id="strike-display"' in browser.receive_text()
        assert upstream.connections == 2