"""Tests for the ODMS client, against a real HTTP server on localhost.

No mocking library and no monkeypatched sockets: the client's whole job is to
speak HTTP correctly, and a fake that intercepts above the socket cannot catch a
malformed query string, a mishandled 429, or a retry that silently repeats a
POST. ``http.server`` is in the standard library and costs milliseconds.
"""

from __future__ import annotations

import json
import sys
import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from odms import NotFound, OdmsClient, RateLimited, Unauthorized  # noqa: E402
from odms.errors import HTTPError, ServiceUnavailable  # noqa: E402


class Recorder:
    """The scripted responses, and what the client actually asked for."""

    def __init__(self):
        self.requests: list[dict] = []
        self.responses: list[tuple] = []

    def reply(self, status: int, body="", headers=None):
        """Queue one response. The last queued one repeats once used up."""
        if isinstance(body, (dict, list)):
            body = json.dumps(body)
            headers = {"Content-Type": "application/json", **(headers or {})}
        self.responses.append((status, body, headers or {}))
        return self

    def next_response(self):
        if len(self.responses) > 1:
            return self.responses.pop(0)
        return self.responses[0] if self.responses else (200, "{}", {})


@pytest.fixture(scope="module")
def _server():
    """One server for the module. Binding a socket and starting a thread costs
    most of a second here, and 22 of them is a suite nobody runs."""
    recorder = Recorder()

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *_args):
            pass  # the test output is not a web server log

        def _handle(self):
            parsed = urlparse(self.path)
            length = int(self.headers.get("Content-Length") or 0)
            recorder.requests.append({
                "method": self.command,
                "path": parsed.path,
                "query": parse_qs(parsed.query),
                "body": self.rfile.read(length).decode() if length else "",
                "headers": dict(self.headers),
            })
            status, body, headers = recorder.next_response()
            payload = body.encode()
            self.send_response(status)
            for key, value in headers.items():
                self.send_header(key, value)
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

        do_GET = do_POST = _handle

    class FastServer(HTTPServer):
        def server_bind(self):
            # HTTPServer.server_bind resolves its own FQDN for the Server header,
            # which costs ~half a second per bind on Windows. Nothing under test
            # reads it.
            super(HTTPServer, self).server_bind()
            self.server_name = "127.0.0.1"
            self.server_port = self.socket.getsockname()[1]

    httpd = FastServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=httpd.serve_forever, kwargs={"poll_interval": 0.02})
    thread.daemon = True
    thread.start()
    recorder.url = f"http://127.0.0.1:{httpd.server_port}"
    try:
        yield recorder
    finally:
        httpd.shutdown()
        httpd.server_close()


@pytest.fixture
def server(_server):
    _server.requests.clear()
    _server.responses.clear()
    return _server


@pytest.fixture
def client(server):
    # Retries with no backoff: the retry *policy* is what is under test, and
    # sleeping through it would only make the suite slow.
    with OdmsClient(server.url, token="t0ken", max_retries=2) as odms:
        odms._transport.backoff_s = 0.0
        # urllib is the floor every consumer gets; httpx, if installed here, is
        # an optimisation the tests should not accidentally become dependent on.
        odms._transport._session = None
        yield odms


# --- request shape ----------------------------------------------------------


def test_the_token_is_sent_as_a_bearer_header(server, client):
    server.reply(200, {"results": []})
    client.satellites()
    assert server.requests[0]["headers"]["Authorization"] == "Bearer t0ken"


def test_unset_filters_are_omitted_rather_than_sent_as_none(server, client):
    """A `?source=None` filter matches no source and returns nothing, silently."""
    server.reply(200, {"count": 0, "results": []})
    client.query(internal_id="iss", source=None, name=None)

    query = server.requests[0]["query"]
    assert query == {"internal_id": ["iss"], "format": ["json"]}


def test_datetimes_are_sent_as_iso_8601(server, client):
    server.reply(200, {"count": 0, "results": []})
    client.at("iss", datetime(2026, 4, 1, tzinfo=timezone.utc))

    assert server.requests[0]["query"]["datetime"] == ["2026-04-01T00:00:00+00:00"]


def test_booleans_are_sent_the_way_django_reads_them(server, client):
    """Python's str(True) is "True"; str(False) is "False" -- but a lowercased
    "false" is truthy to a naive parser, which is the bug this pins."""
    server.reply(201, {"id": 1})
    client.push_message("CCSDS_OPM_VERS = 3.0", key="iss", is_public=False)

    assert json.loads(server.requests[0]["body"])["is_public"] is False


def test_an_unknown_filter_raises_instead_of_being_dropped(client):
    """A typo'd filter that quietly returns the whole catalogue is worse than
    an exception -- the caller gets plausible, wrong data."""
    with pytest.raises(TypeError, match="norrad"):
        client.query(norrad="25544")


def test_an_unknown_format_raises_before_the_request(server, client):
    with pytest.raises(ValueError, match="format"):
        client.query(format="parquet")
    assert server.requests == []


def test_transform_requires_exactly_one_source(client):
    with pytest.raises(TypeError):
        client.transform("ITRF")
    with pytest.raises(TypeError):
        client.transform("ITRF", key="iss", message_id=4)


# --- responses --------------------------------------------------------------


def test_rendered_formats_come_back_as_text_not_json(server, client):
    lines = "ISS (ZARYA)\n1 25544U ...\n2 25544 ..."
    server.reply(200, lines, {"Content-Type": "text/plain"})

    assert client.download(internal_id="iss", format="tle") == lines


def test_latest_returns_none_when_the_satellite_has_no_elements(server, client):
    server.reply(200, {"count": 0, "results": []})
    assert client.latest("iss") is None


def test_decay_latest_returns_none_when_there_is_no_forecast(server, client):
    """No forecast is a normal state: an object with no fittable history is
    skipped deliberately rather than predicted from a coefficient nobody can
    justify. Callers must not have to catch an exception for the normal case."""
    server.reply(404, {"detail": "no forecast"})
    assert client.decay_latest("iss") is None


def test_find_returns_none_where_satellite_raises(server, client):
    server.reply(404, {"detail": "not found"})
    assert client.find("nope") is None
    with pytest.raises(NotFound):
        client.satellite("nope")


# --- errors -----------------------------------------------------------------


@pytest.mark.parametrize("status,expected", [
    (401, Unauthorized),
    (403, Unauthorized),
    (404, NotFound),
    (400, HTTPError),
])
def test_statuses_map_onto_actionable_exceptions(server, client, status, expected):
    server.reply(status, {"detail": "nope"})
    with pytest.raises(expected):
        client.satellite("iss")


def test_the_error_detail_is_lifted_out_of_a_json_body(server, client):
    server.reply(400, {"detail": "end must be >= start"})
    with pytest.raises(HTTPError) as caught:
        client.satellite("iss")
    assert caught.value.detail == "end must be >= start"
    assert caught.value.status == 400


def test_a_non_json_error_body_still_produces_a_usable_message(server, client):
    server.reply(502, "<html>Bad Gateway</html>", {"Content-Type": "text/html"})
    with pytest.raises(ServiceUnavailable) as caught:
        client.satellite("iss")
    assert "Bad Gateway" in str(caught.value)


# --- retries ----------------------------------------------------------------


def test_a_429_is_retried_and_then_succeeds(server, client):
    server.reply(429, "slow down", {"Retry-After": "0"})
    server.reply(200, {"results": [{"internal_id": "iss"}]})

    assert client.satellites() == [{"internal_id": "iss"}]
    assert len(server.requests) == 2


def test_a_persistent_429_raises_with_the_servers_advice(server, client):
    server.reply(429, "slow down", {"Retry-After": "0"})

    with pytest.raises(RateLimited) as caught:
        client.satellites()
    assert caught.value.retry_after == 0
    assert len(server.requests) == 3  # the original plus max_retries


def test_a_500_is_not_retried(server, client):
    """It means ODMS took the request and broke on it. Repeating it breaks it
    again, and turns one alert into three."""
    server.reply(500, "boom")

    with pytest.raises(HTTPError):
        client.satellites()
    assert len(server.requests) == 1


def test_a_503_is_retried_even_for_a_post(server, client):
    """503 proves the request was not acted on, so repeating it cannot
    duplicate anything -- unlike a transport failure."""
    server.reply(503, "busy")
    server.reply(201, {"id": 7})

    assert client.push_message("CCSDS_OPM_VERS = 3.0", key="iss") == {"id": 7}
    assert len(server.requests) == 2


def test_a_transport_failure_never_repeats_a_post(server):
    """A push that died in flight may already have stored the message. Retrying
    would store a second copy, and nothing downstream would notice."""
    from odms.errors import TransportError

    # Nothing is listening on this port, so every attempt fails in transport.
    with OdmsClient("http://127.0.0.1:9", max_retries=3, timeout_s=0.5) as odms:
        odms._transport.backoff_s = 0.0
        odms._transport._session = None
        attempts = []
        original = odms._transport._send_urllib

        def counting(*args, **kwargs):
            attempts.append(1)
            return original(*args, **kwargs)

        odms._transport._send_urllib = counting

        with pytest.raises(TransportError):
            odms.push_message("CCSDS_OPM_VERS = 3.0", key="iss")
        assert len(attempts) == 1

        attempts.clear()
        with pytest.raises(TransportError):
            odms.satellites()
        assert len(attempts) == 4  # a read may be repeated: 1 + max_retries