"""The portal link goes back up on its own, whatever took it down.

A laptop that suspends, a wifi that changes, a portal that restarts: all of
them end the same session, and none of them should need anybody to notice.
"""

from __future__ import annotations

import asyncio
from typing import Any

import pytest

from fluksio.cloud import config as cloud_config
from fluksio.cloud.connector import BASE_BACKOFF_S, MAX_BACKOFF_S, CloudConnector


class _Link:
    """A connector whose sessions do whatever a test says, instantly."""

    def __init__(self, connector: CloudConnector) -> None:
        self.connector = connector
        self.waits: list[float] = []
        self.sessions = 0

    def run(self, outcomes: list[Any]) -> list[float]:
        """Drive `serve_forever` through `outcomes`, then let it return."""

        async def session(_config: Any) -> bool:
            outcome = outcomes[self.sessions]
            self.sessions += 1
            if self.sessions >= len(outcomes):
                # The next iteration reads the config and stops.
                cloud_config.load = lambda: None  # type: ignore[assignment]
            if isinstance(outcome, Exception):
                raise outcome
            return bool(outcome)

        async def sleep(seconds: float) -> None:
            self.waits.append(seconds)

        self.connector._session = session  # type: ignore[assignment]
        with_patched = asyncio.sleep
        asyncio.sleep = sleep  # type: ignore[assignment]
        try:
            asyncio.run(self.connector.serve_forever())
        finally:
            asyncio.sleep = with_patched  # type: ignore[assignment]
        return self.waits


@pytest.fixture
def link(monkeypatch: pytest.MonkeyPatch) -> _Link:
    monkeypatch.setattr(cloud_config, "load", lambda: object())
    return _Link(CloudConnector(app=None))  # type: ignore[arg-type]


def test_a_portal_that_closes_the_socket_is_not_hammered(link: _Link):
    """A clean close used to return normally and reconnect with no wait."""
    waits = link.run([True, True, True])

    # One wait per session. Returning normally from a session used to skip the
    # sleep entirely, so a portal that accepted and closed span the loop.
    assert link.sessions == 3
    assert len(waits) == 3, waits
    assert all(wait > 0 for wait in waits), waits


def test_a_link_that_stood_up_retries_at_once(link: _Link):
    """A dropped connection is a network event, not a refusal."""
    waits = link.run([True, True])

    # Jittered around the base delay rather than backed off.
    assert len(waits) == 2, waits
    assert all(BASE_BACKOFF_S * 0.7 <= w <= BASE_BACKOFF_S * 1.3 for w in waits), waits


def test_a_link_that_never_attaches_backs_off(link: _Link):
    waits = link.run([OSError("no route to host")] * 6)

    assert waits[0] < waits[1] < waits[2], waits
    assert max(waits) <= MAX_BACKOFF_S * 1.3


def test_backing_off_stops_at_the_ceiling(link: _Link):
    waits = link.run([OSError("refused")] * 12)

    assert max(waits) <= MAX_BACKOFF_S * 1.3, waits


def test_a_failure_after_a_good_session_starts_over_from_the_base(link: _Link):
    """Ten minutes of uptime should not inherit the last outage's delay."""
    waits = link.run([OSError("x"), OSError("x"), OSError("x"), True, True])

    assert waits[-1] <= BASE_BACKOFF_S * 1.3, waits