from __future__ import annotations

import collections.abc as cabc
import http.client
import json
import os
import socket
import ssl
import subprocess
import sys
import time
import typing as t
from contextlib import closing
from contextlib import ExitStack
from pathlib import Path
from types import TracebackType

import ephemeral_port_reserve
import pytest

if t.TYPE_CHECKING:
    import typing_extensions as te


class UnixSocketHTTPConnection(http.client.HTTPConnection):
    def connect(self) -> None:
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        # Raises FileNotFoundError if the server hasn't started yet.
        self.sock.connect(self.host)


# Used to annotate the ``DevServerClient.request`` return value.
class DataHTTPResponse(http.client.HTTPResponse):
    data: bytes
    json: t.Any


class DevServerClient:
    """Manage a live dev server process and make requests to it. Must be used
    as a context manager.

    If ``hostname`` starts with ``unix://``, the server listens to a unix socket
    file instead of a TCP socket.

    If ``port`` is not given, a random port is reserved for use by the server,
    to allow multiple servers to run simultaneously.

    If ``ssl_context`` is given, the server listens with TLS enabled. It can be
    the special value ``custom`` to generate and pass a context to
    ``run_simple``, as opposed to ``adhoc`` which tells ``run_simple`` to
    generate the context.

    :param app_name: The name of the app from the ``live_apps`` folder to load.
    :param tmp_path: The current test's temporary directory. The server process
        sets the working dir here, it is added to the Python path, the log file
        is written here, and for unix connections the socket is opened here.
    :param server_kwargs: Arguments to pass to ``live_apps/run.py`` to control
        how ``run_simple`` is called in the subprocess.
    """

    scheme: str
    """One of ``http``, ``https``, or ``unix``. Set based on ``ssl_context`` or
    ``hostname``.
    """
    addr: str
    """The host and port."""
    url: str
    """The scheme, host, and port."""

    def __init__(
        self, app_name: str = "standard", *, tmp_path: Path, **server_kwargs: t.Any
    ) -> None:
        host = server_kwargs.get("hostname", "127.0.0.1")

        if not host.startswith("unix://"):
            port = server_kwargs.get("port")

            if port is None:
                server_kwargs["port"] = port = ephemeral_port_reserve.reserve(host)

            self.scheme = "https" if "ssl_context" in server_kwargs else "http"
            self.addr = f"{host}:{port}"
            self.url = f"{self.scheme}://{self.addr}"
        else:
            self.scheme = "unix"
            self.addr = host[7:]  # strip "unix://"
            self.url = host

        self._app_name = app_name
        self._server_kwargs = server_kwargs
        self._tmp_path = tmp_path
        self._log_write: t.IO[bytes] | None = None
        self._log_read: t.IO[str] | None = None
        self._proc: subprocess.Popen[bytes] | None = None

    def __enter__(self) -> te.Self:
        """Start the server process and wait for it to be ready."""
        log_path = self._tmp_path / "log.txt"
        self._log_write = open(log_path, "wb")
        self._log_read = open(log_path, encoding="utf8", errors="surrogateescape")
        tmp_dir = os.fspath(self._tmp_path)
        self._proc = subprocess.Popen(
            [
                sys.executable,
                os.fspath(Path(__file__).parent / "live_apps/run.py"),
                self._app_name,
                json.dumps(self._server_kwargs),
            ],
            env={**os.environ, "PYTHONUNBUFFERED": "1", "PYTHONPATH": tmp_dir},
            cwd=tmp_dir,
            close_fds=True,
            stdout=self._log_write,
            stderr=subprocess.STDOUT,
        )
        self.wait_ready()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException],
        exc_val: BaseException,
        exc_tb: TracebackType,
    ) -> None:
        """Clean up the server process."""
        assert self._proc is not None
        self._proc.terminate()
        self._proc.wait()
        self._proc = None
        assert self._log_read is not None
        self._log_read.close()
        self._log_read = None
        assert self._log_write is not None
        self._log_write.close()
        self._log_write = None

    def connect(self, **kwargs: t.Any) -> http.client.HTTPConnection:
        """Create a connection to the server, without sending a request.
        Useful if a test requires lower level methods to try something that
        ``HTTPClient.request`` will not do.

        If the server's scheme is HTTPS and the TLS ``context`` argument is not
        given, a default permissive context is used.

        :param kwargs: Arguments to :class:`http.client.HTTPConnection`.
        """
        if self.scheme == "https":
            if "context" not in kwargs:
                context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                context.check_hostname = False
                context.verify_mode = ssl.CERT_NONE
                kwargs["context"] = context

            return http.client.HTTPSConnection(self.addr, **kwargs)

        if self.scheme == "unix":
            return UnixSocketHTTPConnection(self.addr, **kwargs)

        return http.client.HTTPConnection(self.addr, **kwargs)

    def request(self, url: str = "", **kwargs: t.Any) -> DataHTTPResponse:
        """Open a connection and make a request to the server, returning the
        response.

        The response object ``data`` parameter has the result of
        ``response.read()``. If the response has a ``application/json`` content
         type, the ``json`` parameter is populated with ``json.loads(data)``.

        :param url: URL to put in the request line.
        :param kwargs: Arguments to :meth:`http.client.HTTPConnection.request`.
        """
        kwargs.setdefault("method", "GET")
        kwargs["url"] = url
        response: DataHTTPResponse

        with closing(self.connect()) as conn:
            conn.request(**kwargs)

            with conn.getresponse() as response:  # type: ignore[assignment]
                response.data = response.read()

        if response.headers.get("Content-Type", "").startswith("application/json"):
            response.json = json.loads(response.data)
        else:
            response.json = None

        return response

    def wait_ready(self) -> None:
        """Wait until a request to ``/ensure`` is successful, indicating the
        server has started and is listening.
        """
        while True:
            try:
                self.request("/ensure")
                return
            # ConnectionRefusedError for http, FileNotFoundError for unix
            except (ConnectionRefusedError, FileNotFoundError):
                time.sleep(0.1)

    def read_log(self) -> str:
        """Read from the current position to the current end of the log."""
        assert self._log_read is not None
        return self._log_read.read()

    def wait_for_log(self, value: str) -> None:
        """Wait until a line in the log contains the given string.

        :param value: The string to search for.
        """
        assert self._log_read is not None

        while True:
            for line in self._log_read:
                if value in line:
                    return

            time.sleep(0.1)

    def wait_for_reload(self) -> None:
        """Wait until the server logs that it is restarting, then wait for it to
        be ready.
        """
        self.wait_for_log("Restarting with")
        self.wait_ready()


class StartDevServer(t.Protocol):
    def __call__(self, name: str = "standard", **kwargs: t.Any) -> DevServerClient: ...


@pytest.fixture()
def dev_server(tmp_path: Path) -> cabc.Iterator[StartDevServer]:
    """A function that will start a dev server in a subprocess and return a
    client for interacting with the server.
    """
    exit_stack = ExitStack()

    def start_dev_server(name: str = "standard", **kwargs: t.Any) -> DevServerClient:
        client = DevServerClient(name, tmp_path=tmp_path, **kwargs)
        exit_stack.enter_context(client)  # type: ignore[arg-type]
        return client

    with exit_stack:
        yield start_dev_server


@pytest.fixture()
def standard_app(dev_server: t.Callable[..., DevServerClient]) -> DevServerClient:
    """Equivalent to ``dev_server("standard")``."""
    return dev_server()