#!/usr/bin/env python3
# python3-requests-unixsocket probe -- the half of
# 99-python3-requests-unixsocket.test that needs an interpreter.
#
# WHY A PROBE AT ALL. Starting this image's python3 is the expensive part of the
# suite: it is one large static binary and the guest is not fast. Every question
# below needs the module imported AND a live AF_UNIX server, so they are
# answered in ONE start-up and printed as `<case> <value>' lines the .test file
# reads back with sed.
#
# THIS IS THE FIRST SUITE IN THIS TREE THAT HOLDS A FULL HTTP CONVERSATION OVER
# AF_UNIX. `CONFIG_UNIX=y' is in the defconfig for exactly this, and a stream
# socket carrying request, response, headers and body in both directions is what
# exercises it. Nothing here touches the network stack: no AF_INET socket is
# created anywhere, so this suite cannot go red on the virtio-net path.
#
# THE SERVER IS A FORKED CHILD, NOT A THREAD, AND THAT IS A MEASURED CONSTRAINT
# RATHER THAN A PREFERENCE. `Thread.start()' itself RAISES on this arm --
# linux-in-the-browser#85, measured by 99-python3-stopit.test, whose
# `thread-runs' case reads REFUSED -- so the obvious shape (a ThreadingHTTPServer
# in a daemon thread, which is what this probe's x86-64 control ran first) would
# not merely fail, it would fail before serving one byte and leave every
# behavioural case below reading like a broken port. `fork()' DOES work here
# (HAVE_FORK, and 99-python3-filelock.test's `fork_supported' reads yes); it is
# fork+EXEC that does not (brintos/linux#14), and nothing below execs.
#
# NOTHING DEPENDS ON SIGNAL DELIVERY. Signals are unwired on this runtime
# (brintos/linux#21), so the child is never killed: it runs a bounded
# accept loop with its own wall-clock deadline and exits on its own, and the
# parent reaps it with waitpid(WNOHANG) in a bounded poll -- the same shape
# 99-python3-filelock.test uses, for the same reason. A probe that reached for
# SIGTERM would leave a server process behind on every run.
#
# A DEAD PROBE MUST NOT LOOK LIKE A PASSING ONE. Each answer is printed the
# moment it is known, `probe-complete yes' is printed last, and the .test file's
# `field' helper returns empty for anything never reached -- which reds against
# every expectation. The trailing case is what tells a reader why.
import os
import socket
import sys
import time
import traceback


def emit(case, value):
    print("%s %s" % (case, value), flush=True)


def attempt(case, fn):
    """Print fn()'s value, or `EXC:<type>:<msg>' if it raised.

    A raise is a RESULT here, not a crash: several cases below assert that a
    documented exception is raised, and one that swallowed the type would report
    a red as an empty field indistinguishable from a probe that died."""
    try:
        emit(case, fn())
    except Exception as exc:                      # noqa: BLE001 -- see above
        emit(case, "EXC:%s:%s" % (type(exc).__name__, _one(str(exc))))


def _one(s):
    """One line. The runner parses its input line by line, so a value carrying
    an embedded newline lands as one counted result plus N stray lines."""
    return " ".join(str(s).split())


# --------------------------------------------------------------- the import ---
# Measured FIRST and on its own, because everything after it is meaningless if
# it failed, and because it is the case the RDEPENDS correction is about.
try:
    import requests
    import requests_unixsocket
    from requests_unixsocket.adapters import UnixAdapter, UnixHTTPConnection
    emit("bare-import", "ok")
except Exception as exc:                          # noqa: BLE001
    emit("bare-import", "EXC:%s:%s" % (type(exc).__name__, _one(str(exc))))
    traceback.print_exc(file=sys.stderr)
    sys.exit(1)

import urllib3                                    # noqa: E402  (after the gate)

# ------------------------------------------------------------------ identity ---
emit("module-file", requests_unixsocket.__file__)
attempt("version", lambda: __import__("importlib.metadata", fromlist=["x"])
        .version("requests-unixsocket"))
# THERE IS NO `requests_unixsocket.__version__'. Measured on the control against
# the tree unpacked from the built .deb, not read off the recipe: the module
# binds no such name, so the version is discoverable ONLY through the dist-info
# the packaging installed. A suite that asserted the attribute would be asserting
# something upstream never shipped.
emit("has-dunder-version",
     "yes" if hasattr(requests_unixsocket, "__version__") else "no")
emit("default-scheme", requests_unixsocket.DEFAULT_SCHEME)
attempt("public-callables", lambda: ",".join(
    n for n in ("request", "get", "head", "post", "patch", "put", "delete",
                "options")
    if callable(getattr(requests_unixsocket, n, None))))
emit("session-is-requests-session",
     "yes" if issubclass(requests_unixsocket.Session, requests.Session) else "no")
emit("adapter-is-httpadapter",
     "yes" if issubclass(UnixAdapter, requests.adapters.HTTPAdapter) else "no")

# THE TWO PATCHED SEAMS, asserted as facts about the loaded class rather than as
# a grep over the source, so a patch that stopped applying reds here.
#
#   * `UnixAdapter.get_connection_with_tls_context' -- requests 2.32.2 moved the
#     adapter's connection hook to this name and `HTTPAdapter.send' calls ONLY
#     it. Upstream 0.3.0 (2019) overrides `get_connection', which is now a
#     deprecated shim nothing on the send path reaches, so every request went to
#     the stock TCP pool manager and died `InvalidURL: Not supported URL scheme
#     http+unix'. Measured on an x86-64 control at this image's exact versions.
#   * `UnixHTTPConnection' derives from urllib3's HTTPConnection, not from
#     `http.client.HTTPConnection'. urllib3 2.x calls
#     `conn.request(..., chunked=, preload_content=, decode_content=,
#     enforce_content_length=)', which the stdlib class does not accept.
emit("adapter-has-tls-context-hook",
     "yes" if "get_connection_with_tls_context" in vars(UnixAdapter) else "no")
emit("conn-base-is-urllib3",
     "yes" if issubclass(UnixHTTPConnection, urllib3.connection.HTTPConnection)
     else "no")
emit("conn-not-stdlib-direct",
     "yes" if UnixHTTPConnection.__bases__ == (urllib3.connection.HTTPConnection,)
     else "no")
emit("urllib3-version", urllib3.__version__)
emit("requests-version", requests.__version__)

# --------------------------------------------------------------- no threads ---
# The constraint the whole shape of this probe follows from, RE-MEASURED here
# rather than cited: if `Thread.start()' ever starts working on this arm, this
# case is what says so, and the fork below can be reconsidered.
def _thread_start():
    import threading
    try:
        t = threading.Thread(target=lambda: None)
        t.start()
        t.join()
        return "RAN"
    except Exception as exc:                      # noqa: BLE001
        # Normalised to one word: the refusal`s message is not a stable string
        # and the .test cannot assert one. Which exception it was is in
        # `thread-start-detail` beside it.
        emit("thread-start-detail",
             "%s:%s" % (type(exc).__name__, _one(str(exc))))
        return "REFUSED"
emit("thread-start", _thread_start())

# -------------------------------------------------------------- the fixtures ---
# AF_UNIX paths are capped at 108 bytes including the terminator, so the working
# directory is short by construction rather than by luck -- a $TMPDIR-derived
# path plus a descriptive basename has overrun that cap on other images.
W = "/tmp/bst-rus.%d" % os.getpid()
os.makedirs(W, exist_ok=True)
SOCK = os.path.join(W, "s")
DEAD = os.path.join(W, "nx")          # never created: the ENOENT arm
IDLE = os.path.join(W, "id")          # bound and NOT listened on: the refused arm
# The path is under the AF_UNIX cap, asserted as the property rather than as a
# length: the pid in it makes the number vary from run to run.
emit("sockpath-fits", "yes" if len(SOCK) < 108 else "no")

BIG = 64 * 1024                       # spans several reads on any buffer size
POST_BODY = b"".join(bytes([i % 251]) for i in range(4096))


def quoted(path):
    """The library's whole convention: the socket path is the URL's netloc, with
    every `/' percent-encoded. Written out here rather than taken from
    urllib.parse.quote, so the suite asserts the encoding it means."""
    return path.replace("/", "%2F")


def url(path, sockpath=SOCK):
    return "http+unix://" + quoted(sockpath) + path


# ------------------------------------------------------------- the server ---
# A forked child running a FORKING accept loop -- `socketserver.ForkingMixIn'
# over a `UnixStreamServer', with the stdlib's own BaseHTTPRequestHandler on
# top. Two shapes were measured and rejected before this one, and both failures
# are properties of this arm rather than of the code:
#
#   THREADS cannot be used at all. `Thread.start()' raises here
#   (linux-in-the-browser#85); `thread-start' above re-measures it.
#
#   SINGLE-THREADED does not work either, and this is the one that is not
#   obvious. `UnixAdapter' pools connections keyed by URL, so this probe holds
#   SEVERAL keep-alive connections open at once; a server inside
#   `handle_request()' stays in one connection's `handle()' loop until that
#   connection goes idle-timeout, and cannot accept the next one meanwhile.
#   Measured on the x86-64 control: every URL switch cost the connection idle
#   bound (20 s), and the probe went from 6 s to over 120 s without ever
#   deadlocking -- so it would have shipped as an intermittent suite timeout on
#   a slower guest rather than as anything a reader could diagnose.
#
# FORKING is what is left, and it is the shape this arm actually supports:
# `ForkingMixIn' reaps with waitpid(WNOHANG) from `collect_children()' and
# installs no SIGCHLD handler, so it needs nothing from the unwired signal path
# (brintos/linux#21). `collect_children()' is called explicitly below because
# `handle_request()' -- unlike `serve_forever()' -- does not call
# `service_actions()'.
#
# THE CONNECTION COUNTER IS A FILE, NOT AN INTEGER, and that is forced by the
# forking: the parent accepts and the CHILD answers, so a counter in memory
# would be a snapshot frozen at that child's fork and `/conns' answered twice
# over one reused connection would return the same number both times -- which is
# precisely the measurement the reuse cases below depend on. The parent writes
# it at accept; every child reads it fresh.
CONNS = os.path.join(W, "conns")
# ...and for the same reason the STOP flag is a file: `/quit' is answered in a
# forked child, whose `state' the parent`s accept loop can never see. Without
# this the server sat out its full deadline after the probe had finished and the
# reap below reported `not-reaped' -- measured, and the reason this file exists
# rather than an integer.
QUIT = os.path.join(W, "quit")


def child_main():
    import http.server
    import socketserver

    class Handler(http.server.BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"             # keep-alive, and chunked

        def log_message(self, *a):
            pass

        def setup(self):
            http.server.BaseHTTPRequestHandler.setup(self)
            # A client that goes away without closing must not park this child
            # until the server's own deadline.
            self.connection.settimeout(30)

        def _plain(self, body, extra=()):
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("X-Probe", "yes")
            for k, v in extra:
                self.send_header(k, v)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            if self.path == "/big":
                self._plain(b"z" * BIG)
            elif self.path == "/conns":
                try:
                    with open(CONNS) as fh:
                        n = fh.read().strip()
                except OSError:
                    n = "-1"
                self._plain(n.encode())
            elif self.path == "/chunked":
                # Chunked transfer encoding, hand-rolled: BaseHTTPRequestHandler
                # will not frame it, and the point of the case is that urllib3
                # DE-frames what the adapter carried.
                self.send_response(200)
                self.send_header("Content-Type", "text/plain")
                self.send_header("Transfer-Encoding", "chunked")
                self.end_headers()
                for part in (b"alpha", b"beta", b"gamma"):
                    self.wfile.write(b"%x\r\n%s\r\n" % (len(part), part))
                self.wfile.write(b"0\r\n\r\n")
            elif self.path == "/quit":
                # Written from THIS child; the parent polls for the file
                # each time `handle_request()' returns, which its 2 s timeout
                # guarantees happens whether or not a client is connected.
                try:
                    with open(QUIT, "w") as fh:
                        fh.write("1")
                except OSError:
                    pass
                self._plain(b"bye")
                self.close_connection = True
            elif self.path == "/nope":
                self.send_response(404)
                self.send_header("Content-Length", "0")
                self.end_headers()
            else:
                # Every other path is the hello response, so the reuse cases can
                # each take a URL of their own without the handler growing a
                # branch per case.
                self._plain(b"hello-unix",
                            [("X-Peer-Family", str(int(self.connection.family)))])

        def do_POST(self):
            n = int(self.headers.get("Content-Length") or 0)
            body = self.rfile.read(n) if n else b""
            self._plain(body)                     # echoed back byte for byte

    class Server(socketserver.ForkingMixIn, socketserver.UnixStreamServer):
        # BaseHTTPRequestHandler reads these off the server; UnixStreamServer has
        # neither, because its address is a path and not a (host, port).
        server_name = "localhost"
        server_port = 0

        def process_request(self, request, client_address):
            # IN THE PARENT, BEFORE THE FORK -- see the note above on why this
            # count cannot live in memory.
            try:
                with open(CONNS) as fh:
                    n = int(fh.read().strip() or 0)
            except (OSError, ValueError):
                n = 0
            with open(CONNS, "w") as fh:
                fh.write(str(n + 1))
            socketserver.ForkingMixIn.process_request(
                self, request, client_address)

    with open(CONNS, "w") as fh:
        fh.write("0")
    srv = Server(SOCK, Handler)
    srv.timeout = 2                               # handle_request() idle bound
    deadline = time.time() + 180
    while time.time() < deadline and not os.path.exists(QUIT):
        srv.handle_request()
        srv.collect_children()
    srv.server_close()


pid = -1
try:
    pid = os.fork()
except Exception as exc:                          # noqa: BLE001
    emit("fork-supported", "no")
    emit("fork-error", "%s:%s" % (type(exc).__name__, _one(str(exc))))

if pid == 0:
    # ---- child --------------------------------------------------------------
    # THE REDIRECT IS THE FIRST STATEMENT AFTER THE FORK, and nothing above it
    # may emit: everything between fork() and here runs in BOTH processes and
    # would land twice in the parent's transcript.
    try:
        fd = os.open(os.path.join(W, "server.log"),
                     os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        os.dup2(fd, 1)
        os.dup2(fd, 2)
    except Exception:                             # noqa: BLE001
        pass
    try:
        child_main()
    except Exception:                             # noqa: BLE001
        traceback.print_exc()
    os._exit(0)

emit("fork-supported", "yes")

# The socket appearing is the readiness signal. No fixed sleep: on a loaded
# guest a constant is either a stall or a race, and this is the condition the
# client actually needs.
ready = False
for _ in range(600):                              # 60 s, in 100 ms steps
    if os.path.exists(SOCK):
        ready = True
        break
    time.sleep(0.1)
emit("server-ready", "yes" if ready else "no")


def cleanup():
    """Reap the child WITHOUT signalling it: signal delivery is unwired here
    (brintos/linux#21), so `/quit' is what stops it and its own 120 s deadline is
    the backstop."""
    try:
        s = requests_unixsocket.Session()
        s.get(url("/quit"), timeout=10)
        s.close()
    except Exception:                             # noqa: BLE001
        pass
    for _ in range(300):                          # 30 s
        try:
            done, _st = os.waitpid(pid, os.WNOHANG)
        except ChildProcessError:
            return "reaped"
        if done == pid:
            return "reaped"
        time.sleep(0.1)
    return "not-reaped"


try:
    # ------------------------------------------------------- the conversation ---
    sess = requests_unixsocket.Session()

    def _get(path):
        return sess.get(url(path), timeout=30)

    # GET: the status, the headers and the body the handler wrote. All three,
    # because a transport that delivered the status line and lost the body would
    # satisfy a status-only assertion.
    attempt("get-status", lambda: _get("/hello").status_code)
    attempt("get-body", lambda: _get("/hello").text)
    attempt("get-header", lambda: _get("/hello").headers.get("X-Probe"))
    attempt("get-content-type",
            lambda: _get("/hello").headers.get("Content-Type"))

    # THE SERVER'S OWN VIEW OF THE SOCKET FAMILY. socket.AF_UNIX is 1 on Linux;
    # the number is printed rather than the enum's repr, which differs between
    # interpreters.
    attempt("server-peer-family",
            lambda: _get("/hello").headers.get("X-Peer-Family"))
    emit("af-unix-value", str(int(socket.AF_UNIX)))

    # POST with a body echoed back BYTE FOR BYTE. Scored on a digest of the
    # round-trip against a digest of what was sent, so a truncation of any length
    # reds -- and the length is printed beside it so a red says which.
    def _post():
        r = sess.post(url("/echo"), data=POST_BODY, timeout=30)
        return "%d,%d,%s" % (r.status_code, len(r.content),
                             "same" if r.content == POST_BODY else "DIFFERENT")
    attempt("post-echo", _post)

    # A response large enough to span several reads, asserted COMPLETE. A short
    # read is the classic stream-socket defect and a length check is the only
    # thing that catches it.
    def _big():
        r = _get("/big")
        return "%d,%d,%s" % (r.status_code, len(r.content),
                             "z" * 8 == r.text[:8] and r.text[-1] == "z"
                             and "intact" or "CORRUPT")
    attempt("big-response", _big)

    # CHUNKED TRANSFER ENCODING. urllib3 de-frames it, so what `text' holds is
    # the CONCATENATION of the three chunks with none of the framing in it --
    # asserted as a literal, because a transport that passed the raw framing
    # through would produce a plausible-looking body with the sizes embedded.
    # The `Transfer-Encoding: chunked' header SURVIVES on the decoded response
    # and is asserted beside the body: measured, not assumed -- the first draft
    # of this probe expected requests to strip it and it does not, and the two
    # together are what say the framing was understood rather than skipped.
    def _chunked():
        r = _get("/chunked")
        return "%d,%s,%s" % (r.status_code, r.text,
                             r.headers.get("Transfer-Encoding") or "-")
    attempt("chunked", _chunked)

    # THE PERCENT-ENCODING IS THIS LIBRARY'S WHOLE CONVENTION, so it is asserted
    # as an encoding and not only by the requests above happening to work: the
    # URL is shown to carry `%2F' and no bare `/' in its netloc, and the same
    # path spelled with real slashes is shown to FAIL -- one direction alone
    # would pass against a library that ignored the netloc entirely.
    emit("netloc-has-no-bare-slash",
         "yes" if "/" not in quoted(SOCK) else "no")
    emit("netloc-encodes-every-slash",
         "yes" if quoted(SOCK).count("%2F") == SOCK.count("/") else "no")
    # ...and it decodes back to the path the server is actually bound to, which
    # is the half that says the encoding is the library`s and not just a string
    # with no `/' in it.
    from requests.compat import unquote as _unq, urlparse as _urlp
    emit("netloc-unquotes-to-sockpath",
         "yes" if _unq(_urlp(url("/hello")).netloc) == SOCK else "no")

    # THE SAME PATH SPELLED WITH REAL SLASHES MUST FAIL. One direction alone
    # would pass against a library that ignored the netloc entirely. Only the
    # exception TYPE is printed: its message carries this run`s tmpdir.
    def _unencoded():
        try:
            sess.get("http+unix://" + SOCK + "/hello", timeout=10)
            return "NO-RAISE"
        except Exception as exc:                  # noqa: BLE001
            return type(exc).__name__
    attempt("unencoded-path-refused", _unencoded)

    # SEVERAL REQUESTS OVER ONE SESSION, AND CONNECTION REUSE MEASURED AS A
    # DELTA the SERVER counted -- not as anything the client believes about its
    # own pool, and not as a running total, which mixes in every other URL this
    # probe used and cannot be read as a reuse figure at all.
    #
    # THE POOL IS KEYED BY THE WHOLE URL, PATH INCLUDED. `UnixAdapter.pools' is
    # a RecentlyUsedContainer indexed by the request URL, so two paths on the
    # SAME socket get two pools and two connections; only repeats of one URL
    # reuse. That is upstream's design and not a defect of this port, but it is
    # the opposite of what "one connection per socket" would lead a reader to
    # expect, so it is measured in both directions below rather than described.
    def _reuse():
        before = int(_get("/conns").text)
        for _ in range(5):
            sess.get(url("/reuse"), timeout=30)   # ONE url, five requests
        after = int(_get("/conns").text)
        return "%d" % (after - before)
    attempt("reuse-delta-same-url", _reuse)

    def _distinct():
        before = int(_get("/conns").text)
        for path in ("/d1", "/d2", "/d3"):        # three urls, one request each
            sess.get(url(path), timeout=30)
        after = int(_get("/conns").text)
        return "%d" % (after - before)
    attempt("reuse-delta-distinct-urls", _distinct)

    # THE CONNECTED SOCKET IS AF_UNIX, asserted CLIENT-side out of the adapter's
    # own pool. This is the case that fails a suite which silently fell back to
    # TCP: the family is read off the socket the request actually used.
    def _client_family():
        pool = sess.get_adapter(url("/hello")).pools[url("/hello")]
        conn = pool.pool.get()
        try:
            return "%d,%s" % (
                int(conn.sock.family),
                "match" if (conn.sock.getpeername() or SOCK) == SOCK
                else "MISMATCH")
        finally:
            pool.pool.put(conn)
    attempt("client-socket-family", _client_family)

    # The module-level convenience functions build their own Session, so they
    # exercise a DIFFERENT construction path from every case above.
    attempt("module-level-get",
            lambda: requests_unixsocket.get(url("/hello"), timeout=30).text)

    sess.close()

    # ------------------------------------------------------------ the refusals ---
    # A path that DOES NOT EXIST. requests wraps urllib3's NewConnectionError as
    # requests.exceptions.ConnectionError, and the errno text must survive into
    # it -- an exception that reached the caller having lost which errno it was
    # is not a diagnosis.
    def _enoent():
        s2 = requests_unixsocket.Session()
        try:
            s2.get(url("/hello", DEAD), timeout=10)
            return "NO-RAISE"
        except requests.exceptions.ConnectionError as e:
            t = str(e)
            return "ConnectionError,%s,%s" % (
                "ENOENT" if ("No such file" in t or "Errno 2" in t) else "other",
                "chained" if e.__cause__ is not None or e.__context__ is not None
                else "bare")
        finally:
            s2.close()
    attempt("enoent-raises", _enoent)

    # A path that EXISTS and has NO LISTENER -- bound, never listen(2)ed. This is
    # a different kernel refusal from the one above (ECONNREFUSED, not ENOENT)
    # and it is the one that separates "the socket file is missing" from "the
    # server is not there", which is the distinction a caller acts on.
    idle = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    idle.bind(IDLE)
    emit("idle-socket-exists", "yes" if os.path.exists(IDLE) else "no")

    def _refused():
        s3 = requests_unixsocket.Session()
        try:
            s3.get(url("/hello", IDLE), timeout=10)
            return "NO-RAISE"
        except requests.exceptions.ConnectionError as e:
            t = str(e)
            return "ConnectionError,%s" % (
                "ECONNREFUSED" if ("refused" in t or "Errno 111" in t)
                else "other")
        finally:
            s3.close()
    attempt("refused-raises", _refused)
    idle.close()

    # A 404 is a RESPONSE, not a refusal, and the distinction is worth one case:
    # the transport is working in exactly the arm where the application said no.
    def _404():
        s4 = requests_unixsocket.Session()
        try:
            return s4.get(url("/nope"), timeout=30).status_code
        finally:
            s4.close()
    attempt("http-404", _404)

finally:
    emit("cleanup", cleanup())
    for p in (SOCK, IDLE, CONNS, QUIT, os.path.join(W, "server.log")):
        try:
            os.unlink(p)
        except OSError:
            pass
    try:
        os.rmdir(W)
    except OSError:
        pass

emit("probe-complete", "yes")