# Copyright (C) 2026 half144 and PS5 React contributors
# SPDX-License-Identifier: GPL-3.0-or-later
# Additional attribution term: see LICENSE-ATTRIBUTION.
"""Local native networking integration tests; never access a console or catalog."""
import hashlib
import http.server
import json
import os
from pathlib import Path
import re
import signal
import socket
import ssl
import struct
import subprocess
import tempfile
import threading
import time

ROOT = Path(__file__).resolve().parents[1]
DATA = bytes(range(256)) * (16 * 1024 * 1024 // 256)
# /paced serves this at about 2 MB/s per connection, MediaFire's pace, so refusal bursts land mid-download.
LARGE = DATA * 8
PIECES = [bytes([19+i]) * size for i, size in enumerate((3*1024*1024+17, 5*1024*1024+31, 7*1024*1024+9))]


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    requests = []
    guard = threading.Lock()
    retried = set()
    # Signatures that now answer 403, like an expired signed CDN link.
    expired = set()
    modified = "Wed, 01 Jan 2025 00:00:00 GMT"
    mutated = False
    # /throttled answers like an archive.org storage node: past a few concurrent ranges it refuses
    # with 500/503/429, and it resets some ranges halfway once.
    busy = 0
    refused = []
    # /hop sends its first request (the probe) to /node-a, whose ranges then fail with HTTP 500, and
    # every later one to /node-b, a healthy node with the same file.
    hops = 0
    # Connections serving /suspend ranges, which the test cuts while the client is stopped, and how many
    # of the ranges asked for next answer 503.
    streams = set()
    suspend_refusals = 0

    def respond(self, head=False):
        if self.path == "/hop":
            with self.guard:
                Handler.hops += 1
                target = "/node-a" if Handler.hops == 1 else "/node-b"
            self.send_response(302)
            self.send_header("Location", target)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        if self.path == "/node-a" and self.headers.get("Range") != "bytes=0-0":
            self.send_response(500)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        if not self.path.startswith("/throttled") or self.headers.get("Range") in (None, "bytes=0-0"):
            return self.serve(head)
        with self.guard:
            Handler.busy += 1
            busy = Handler.busy
        try:
            with self.guard:
                first = ("refuse", self.headers["Range"]) not in self.retried
                self.retried.add(("refuse", self.headers["Range"]))
            # The first range's first answer is archive.org's HTML error page; it must not read as a web page.
            if busy > 6 or (first and self.headers["Range"].startswith("bytes=0-")):
                status = (500, 503, 429)[len(self.refused) % 3]
                self.refused.append(status)
                body = b"<html><body>Internal Server Error</body></html>" if status == 500 else b""
                self.send_response(status)
                if status == 503: self.send_header("Retry-After", "0")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
                return
            self.serve(head)
        finally:
            with self.guard:
                Handler.busy -= 1

    def handle(self):
        try:
            super().handle()
        except (BrokenPipeError, ConnectionResetError):
            pass  # Expected when the client cancels or rejects a response.

    def log_message(self, *_):
        pass

    def do_HEAD(self):
        self.respond(head=True)

    def do_GET(self):
        self.respond()

    def serve(self, head=False):
        path, _, query = self.path.partition("?")
        selected = self.headers.get("Range")
        with self.guard:
            self.requests.append((path, selected, self.headers.get("If-Range")))
        if query in self.expired:
            self.send_response(403)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        if path == "/redirect":
            self.send_response(302)
            self.send_header("Location", "/file")
            self.send_header("HX-Redirect", "/old-page")
            self.send_header("Set-Cookie", "secret=never-expose")
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        if path in ("/html", "/html-no-type", "/html-fragmented"):
            body = b"<!DOCTYPE html><html>Verify</html>"
            self.send_response(200)
            if path == "/html": self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            if not head:
                if path == "/html-fragmented":
                    self.wfile.write(body[:1]); self.wfile.flush(); time.sleep(0.03)
                    self.wfile.write(body[1:])
                else: self.wfile.write(body)
            return
        if path in ("/json", "/not-found", "/large-json"):
            body = b'{"value":42}' if path != "/large-json" else b"x" * 4096
            self.send_response(404 if path == "/not-found" else 200)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            if not head:
                self.wfile.write(body)
            return
        piece = re.search(r"/piece([0-2])$", path)
        payload = PIECES[int(piece[1])] if piece else LARGE if path == "/paced" else DATA
        if path.endswith("-mutable") and self.mutated:
            payload = bytes(reversed(DATA))
        total = 4*1024**3 + 1024**2 if path == "/large-offset" else len(payload)
        start, end, status = 0, total-1, 200
        if selected and path not in ("/no-range", "/slow-no-range", "/multipart-no-range/piece1"):
            match = re.fullmatch(r"bytes=(\d+)-(\d+)", selected)
            if not match:
                self.send_error(416)
                return
            start, end = map(int, match.groups())
            status = 206
        if (path == "/flaky" and selected != "bytes=0-0") or path == "/flaky-probe":
            with self.guard:
                retry_key = (path, selected)
                first = retry_key not in self.retried
                self.retried.add(retry_key)
            if first:
                self.send_response(503)
                self.send_header("Retry-After", "0")
                self.send_header("Content-Length", "0")
                self.end_headers()
                return
        # Every range, and every part of a split one, answers 503 twice before it serves.
        if path == "/slow-twice" and selected != "bytes=0-0":
            with self.guard:
                refused = sum(r[:2] == (path, selected) for r in self.requests) <= 2
            if refused:
                self.send_response(503)
                self.send_header("Retry-After", "0")
                self.send_header("Content-Length", "0")
                self.end_headers()
                return
        if path == "/suspend" and selected != "bytes=0-0":
            with self.guard:
                refuse = Handler.suspend_refusals > 0
                Handler.suspend_refusals -= refuse
            if refuse:
                self.send_response(503)
                self.send_header("Retry-After", "0")
                self.send_header("Content-Length", "0")
                self.end_headers()
                return
        if path == "/changed" and selected != "bytes=0-0":
            status = 200
        changed_etag = path == "/changed-etag" and selected != "bytes=0-0"
        self.send_response(status)
        self.send_header("Content-Length", str(end-start+1))
        if "modified" in path:
            self.send_header("Last-Modified", self.modified)
            self.send_header("ETag", 'W/"weak"')
        elif "no-validator" not in path:
            self.send_header("ETag", '"fixture-v2"' if path == "/changed" or changed_etag else '"fixture-v1"')
        if status == 206:
            reported = start+1 if path == "/bad-range" and selected != "bytes=0-0" else start
            self.send_header("Content-Range", f"bytes {reported}-{end}/{total}")
        self.end_headers()
        if head:
            return
        if path == "/suspend" and selected != "bytes=0-0":
            with self.guard:
                Handler.streams.add(self.connection)
        try:
            for offset in range(start, end+1, 65536):
                index = offset % len(payload)
                self.wfile.write(payload[index:index+min(65536, end-offset+1)])
                if path.startswith("/slow") or path.startswith("/multipart-slow"):
                    time.sleep(0.006)
                if path == "/suspend":
                    time.sleep(0.02)
                if path == "/paced":
                    time.sleep(0.03)
                if path == "/truncated" and selected != "bytes=0-0":
                    self.close_connection = True
                    self.connection.shutdown(socket.SHUT_RDWR)
                    return
                if path == "/throttled" and offset == start and start % (3*1024*1024) == 0:
                    with self.guard:
                        first = ("reset", selected) not in self.retried
                        self.retried.add(("reset", selected))
                    if first:
                        self.close_connection = True
                        self.connection.shutdown(socket.SHUT_RDWR)
                        return
                if path == "/throttled":
                    time.sleep(0.004)
        except (BrokenPipeError, ConnectionResetError, OSError):
            pass


class Refusing(http.server.ThreadingHTTPServer):
    """A TLS origin that turns new handshakes away like MediaFire: for eight seconds from 1.5 s after
    `started` while `burst` is set, and every one after the first while `dead`."""

    def __init__(self, context):
        super().__init__(("127.0.0.1", 0), Handler)
        self.context = context
        self.started, self.burst, self.dead = 0.0, False, False
        self.handshakes, self.refused = 0, []

    def refusing(self):
        if self.dead: return self.handshakes > 1
        return self.burst and 1.5 <= time.monotonic()-self.started < 9.5

    def get_request(self):
        connection, address = self.socket.accept()
        with Handler.guard:
            self.handshakes += 1
            refuse = self.refusing()
            if refuse: self.refused.append(time.monotonic())
        if refuse:
            # A reset in the middle of the handshake: curl reports "SSL connect error", as on the console.
            connection.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
            connection.close()
            raise OSError("handshake refused")
        return self.context.wrap_socket(connection, server_side=True, do_handshake_on_connect=False), address

    def finish_request(self, request, address):
        try: request.do_handshake()
        except OSError: return
        super().finish_request(request, address)


def compile_client(directory):
    binary = directory / "client"
    subprocess.run(["clang++", "-std=c++20", "-O2", "-Wall", "-Wextra", "-Werror", "-pthread",
                    "-I", str(ROOT / "native/shared"), str(ROOT / "tools/tests/network_client.cpp"),
                    str(ROOT / "native/shared/network.cpp"), str(ROOT / "tools/tests/network_test_platform.cpp"),
                    "-lcurl", "-o", str(binary)], check=True)
    return binary


def main():
    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    origin = f"http://127.0.0.1:{server.server_port}"
    with tempfile.TemporaryDirectory(prefix="ps5-react-network-") as temporary:
        directory = Path(temporary)
        binary = compile_client(directory)

        def run(endpoint, path="-", connections=4, max_bytes=1024*1024, cancel=0, digest="", method="GET", file_limit=0, stop=0, pieces=False, hashes=None, recover=False, reject_html=False, no_redirect=False, mirrors=(), adaptive=False, range_probe=False, range_bytes=0):
            environment = dict(os.environ)
            environment.pop("NETWORK_TEST_RANGE_PROBE", None)
            environment.pop("NETWORK_TEST_RANGE_BYTES", None)
            if range_bytes: environment["NETWORK_TEST_RANGE_BYTES"] = str(range_bytes)
            if range_probe: environment["NETWORK_TEST_RANGE_PROBE"] = "1"
            for name in ("NETWORK_TEST_RECOVER", "NETWORK_TEST_REJECT_HTML", "NETWORK_TEST_NO_REDIRECT", "NETWORK_TEST_ADAPTIVE"):
                environment.pop(name, None)
            if adaptive: environment["NETWORK_TEST_ADAPTIVE"] = "1"
            if reject_html: environment["NETWORK_TEST_REJECT_HTML"] = "1"
            if no_redirect: environment["NETWORK_TEST_NO_REDIRECT"] = "1"
            if recover:
                environment["NETWORK_TEST_RECOVER"] = "1"
            environment.pop("NETWORK_TEST_MIRRORS", None)
            if mirrors:
                environment["NETWORK_TEST_MIRRORS"] = ",".join(origin+m for m in mirrors)
            result = subprocess.run([str(binary), (endpoint if endpoint.startswith("https:") else origin+endpoint), str(path), str(connections), str(max_bytes),
                                     str(cancel), digest, method, str(file_limit), str(stop),
                                     "pieces" if pieces else "single", *(hashes or [])], capture_output=True, text=True, timeout=20, env=environment)
            if not result.stdout.strip():
                raise AssertionError(result.stderr or f"client exited {result.returncode}")
            out = json.loads(result.stdout)
            assert out["peakBuffer"] <= 16*1024*1024, out
            assert out["peakConnections"] <= connections, out
            return out

        assert json.loads(run("/json")["body"]) == {"value": 42}
        assert run("/not-found")["status"] == 404
        assert run("/json", method="HEAD")["body"] == ""
        metadata = run("/redirect", method="HEAD")
        assert metadata["url"] == origin + "/file", metadata
        assert "content-length" in metadata["headers"]
        assert "hx-redirect" not in metadata["headers"] and "set-cookie" not in metadata["headers"]
        probe = run("/modified", max_bytes=8192, range_probe=True)
        assert probe["status"] == 206 and len(probe["body"]) == 1, probe
        assert probe["headers"]["content-range"] == f"bytes 0-0/{len(DATA)}", probe
        assert probe["headers"]["last-modified"] == Handler.modified, probe
        # Custom public Range headers still require the caller to inspect redirects explicitly.
        probe_redirect = run("/redirect", max_bytes=8192, range_probe=True)
        assert probe_redirect["status"] == 302 and probe_redirect["headers"]["location"] == "/file", probe_redirect
        redirect = run("/redirect", method="HEAD", no_redirect=True)
        assert redirect["status"] == 302 and redirect["headers"]["location"] == "/file", redirect
        for endpoint in ("/html", "/html-no-type", "/html-fragmented"):
            html_path = directory / endpoint.removeprefix("/")
            rejected = run(endpoint, html_path, reject_html=True)
            assert rejected["state"] == "failed" and "browser verification required" in rejected["error"], rejected
            assert not html_path.exists()
        missing = run("/not-found", directory / "missing", reject_html=True)
        assert missing["state"] == "failed" and "HTTP 404" in missing["error"], missing
        # A published file can be recovered after queue-state loss without contacting its provider.
        path = directory / "published-recovery"
        assert run("/file", path, recover=True)["state"] == "completed"
        before = len(Handler.requests)
        recovered = run("/file", path, recover=True)
        assert recovered["state"] == "completed" and recovered["written"] == len(DATA), recovered
        assert len(Handler.requests) == before
        # A re-resolved link names the same file; another expected hash is another request.
        assert run("/no-range", path, recover=True)["state"] == "completed"
        assert run("/file", path, recover=True, digest="0"*64)["state"] == "failed"
        # Without an expected hash the receipt is trusted: no read-back of the whole file.
        path = directory / "hashed-recovery"
        digest = hashlib.sha256(DATA).hexdigest()
        assert run("/file", path, recover=True, digest=digest)["state"] == "completed"
        with path.open("r+b") as completed:
            completed.write(b"corrupt")  # An expected hash is checked again: same inode and size are not enough.
        assert "SHA-256 mismatch" in run("/file", path, recover=True, digest=digest)["error"]
        path = directory / "unowned-final"
        path.write_bytes(DATA)
        assert run("/file", path, recover=True)["state"] == "failed"
        assert path.read_bytes() == DATA
        path = directory / "sequential-recovery"
        assert run("/no-range", path, recover=True)["state"] == "completed"
        assert run("/no-range", path, recover=True)["state"] == "completed"
        for endpoint in ("/slow-no-range",):
            path = directory / endpoint.removeprefix("/")
            assert run(endpoint, path, cancel=180, recover=True)["state"] == "cancelled"
            assert run(endpoint, path, recover=True)["state"] == "failed"
            # Explicit app restart discards its owned partial, preserving final-file exclusivity.
            Path(str(path)+".part").unlink()
            assert run(endpoint, path, recover=True)["state"] == "completed"
        assert run("/large-json", max_bytes=256)["state"] == "failed"
        assembled = b"".join(PIECES)
        piece_hashes = [hashlib.sha1(piece).hexdigest() for piece in PIECES]
        for endpoint in ("/multipart", "/multipart-no-range", "/multipart-no-validator"):
            path = directory / endpoint.removeprefix("/")
            out = run(endpoint, path, pieces=True, hashes=piece_hashes, digest=hashlib.sha256(assembled).hexdigest(), recover=True)
            assert out["state"] == "completed", out
            assert path.read_bytes() == assembled
            assert out["written"] == len(assembled), out
            before = len(Handler.requests)
            assert run(endpoint, path, pieces=True, hashes=piece_hashes, digest=hashlib.sha256(assembled).hexdigest(), recover=True)["state"] == "completed"
            assert len(Handler.requests) == before
        path = directory / "multipart-sha1-error"
        out = run("/multipart", path, pieces=True, hashes=["0"*40, *piece_hashes[1:]])
        assert out["state"] == "failed" and "SHA-1 mismatch" in out["error"], out
        assert not path.exists()
        path = directory / "multipart-resume"
        assert run("/multipart-slow", path, pieces=True, hashes=piece_hashes, cancel=180)["state"] == "cancelled"
        before = len(Handler.requests)
        out = run("/multipart-slow", path, pieces=True, hashes=piece_hashes)
        assert out["state"] == "completed", out
        assert path.read_bytes() == assembled
        assert len([r for r in Handler.requests[before:] if r[1] != "bytes=0-0"]) < 18
        # 64 connections run on four lanes; every range lands once, retried ones included.
        for endpoint in ("/slow", "/flaky"):
            path = directory / f"lanes{endpoint.replace('/', '-')}"
            out = run(endpoint, path, connections=64, digest=hashlib.sha256(DATA).hexdigest())
            assert out["state"] == "completed", out
            assert path.read_bytes() == DATA
            path.unlink()
        # With more connections than ranges, idle ones split the ranges still running; every byte lands
        # once, and a split part that fails retries alone.
        for endpoint in ("/slow", "/flaky"):
            before = len(Handler.requests)
            path = directory / f"split{endpoint.replace('/', '-')}"
            out = run(endpoint, path, connections=32, digest=hashlib.sha256(DATA).hexdigest(), range_bytes=4*1024*1024)
            assert out["state"] == "completed", out
            assert path.read_bytes() == DATA
            starts = [int(r[1][6:].split("-")[0]) for r in Handler.requests[before:] if r[1] and r[1] != "bytes=0-0"]
            assert any(start % (4*1024*1024) for start in starts), starts
            path.unlink()
        # Each part of a split range has its own attempts: a range cut into many parts that each fail
        # twice still completes.
        path = directory / "split-twice"
        out = run("/slow-twice", path, connections=32, digest=hashlib.sha256(DATA).hexdigest(), range_bytes=len(DATA))
        assert out["state"] == "completed", out
        assert path.read_bytes() == DATA
        path.unlink()
        # Ranges spread over a mirror with the primary's size and ETag; one with another ETag gets none.
        before = len(Handler.requests)
        path = directory / "mirrored"
        out = run("/file", path, connections=8, mirrors=("/mirror", "/changed"), digest=hashlib.sha256(DATA).hexdigest())
        assert out["state"] == "completed", out
        assert path.read_bytes() == DATA
        ranges = [r for r in Handler.requests[before:] if r[1] and r[1] != "bytes=0-0"]
        assert any(r[0] == "/mirror" for r in ranges) and any(r[0] == "/file" for r in ranges), ranges
        assert not any(r[0] == "/changed" for r in ranges), ranges
        path.unlink()
        # A mirror that passes the probe but fails its ranges is dropped; they go back to the primary.
        out = run("/file", path, connections=8, mirrors=("/truncated",), digest=hashlib.sha256(DATA).hexdigest())
        assert out["state"] == "completed", out
        assert path.read_bytes() == DATA
        path.unlink()
        # /flaky fails each range once per server run; later checks count those retries afresh.
        Handler.retried.clear()
        for endpoint in ("/file", "/redirect", "/no-range", "/no-validator"):
            path = directory / endpoint.removeprefix("/")
            before = len(Handler.requests)
            out = run(endpoint, path)
            if endpoint == "/redirect":
                assert [r[0] for r in Handler.requests[before:]].count("/redirect") == 1, Handler.requests[before:]
            assert out["state"] == "completed", out
            assert hashlib.sha256(path.read_bytes()).digest() == hashlib.sha256(DATA).digest()
            assert out["written"] == len(DATA), out
            assert not Path(str(path)+".part").exists()
            path.unlink()
        # A range answered with the wrong span, or a 200 that ignores Range, is retried a few times
        # and then fails without blaming the file; validators that differ on the probed URL fail at once.
        for endpoint, error in (("/bad-range", "unexpected range response"), ("/changed", "ignored the byte range"),
                                ("/changed-etag", "changed the resource")):
            path = directory / endpoint.removeprefix("/")
            out = run(endpoint, path)
            assert out["state"] == "failed" and error in out["error"], out
            assert endpoint != "/changed-etag" or out["retries"] == 0, out
            assert not path.exists()
        # Refused and reset ranges are a busy server, not a damaged file: the origin's window shrinks
        # and every range lands once, byte-identical, at 64 connections.
        Handler.refused.clear()
        path = directory / "throttled"
        out = run("/throttled", path, connections=64, adaptive=True, reject_html=True, digest=hashlib.sha256(DATA).hexdigest())
        assert out["state"] == "completed", out
        assert path.read_bytes() == DATA
        assert {500, 503, 429} <= set(Handler.refused) and out["retries"] >= 3, (out, Handler.refused)
        path.unlink()
        # A pinned redirect target that keeps failing is resolved again, and the new target is pinned.
        path = directory / "hop"
        out = run("/hop", path, digest=hashlib.sha256(DATA).hexdigest())
        assert out["state"] == "completed", out
        assert path.read_bytes() == DATA
        assert Handler.hops <= 8, Handler.hops
        path.unlink()
        path = directory / "large-offset"
        size = 4*1024**3 + 1024**2
        # Seed completed sparse ranges to exercise the actual downloader beyond 4 GiB
        # without transferring or allocating gigabytes on the development machine.
        with Path(str(path)+".part").open("wb") as partial:
            partial.truncate(4*1024**3)
        identity = hashlib.sha256(f'etag:"fixture-v1"\0\0{size}\0'.encode()).hexdigest().encode()+b"\0"
        meta = struct.pack("<8sQQI65s3x", b"P5RDL001", size, 1024**2, 4097, identity)
        Path(str(path)+".resume").write_bytes(meta + b"\1"*4096 + b"\0")
        assert run("/large-offset", path)["state"] == "completed"
        assert path.stat().st_size == size
        with path.open("rb") as file:
            file.seek(4*1024**3)
            assert file.read() == DATA[:1024**2]
        path.unlink()
        path = directory / "flaky"
        before = len(Handler.requests)
        flaky = run("/flaky", path)
        assert flaky["state"] == "completed", flaky
        assert flaky["retries"] == 16, flaky
        assert path.read_bytes() == DATA
        assert len(Handler.requests[before:]) == 33  # Probe plus two attempts per range.
        path.unlink()
        path = directory / "flaky-probe"
        probe_retry = run("/flaky-probe", path)
        assert probe_retry["state"] == "completed" and probe_retry["retries"] == 17, probe_retry
        path.unlink()
        path = directory / "disk-error"
        failed = run("/file", path, file_limit=512*1024)
        assert failed["state"] == "failed" and "write partial file" in failed["error"], failed
        assert not path.exists()
        # Once space is freed, the partial resumes: ranges completed before the disk filled are kept.
        failed = run("/file", path, file_limit=4*1024*1024)
        assert failed["state"] == "failed" and "write partial file" in failed["error"], failed
        before = len(Handler.requests)
        assert run("/file", path)["state"] == "completed"
        assert path.read_bytes() == DATA
        assert len([r for r in Handler.requests[before:] if r[1] != "bytes=0-0"]) < 16
        path.unlink()
        # A killed process leaves a partial longer than its last checkpoint; unfinished ranges download again.
        path = directory / "killed"
        client = subprocess.Popen([str(binary), origin+"/slow", str(path), "4", str(1024*1024), "0", "", "GET", "0", "0", "single"],
                                  stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        deadline = time.monotonic()+5
        while not Path(str(path)+".resume").exists() and time.monotonic() < deadline:
            time.sleep(0.01)
        time.sleep(0.1)
        client.kill(); client.wait()
        assert Path(str(path)+".part").exists() and Path(str(path)+".resume").exists()
        assert run("/slow", path)["state"] == "completed"
        assert path.read_bytes() == DATA
        path.unlink()
        # Ranges whose connections died while the process was suspended (rest mode) are retried without
        # counting as a busy server's refusals; a server's HTTP error after it still counts.
        def suspended(refusals):
            path = directory / "suspended"
            Handler.suspend_refusals = 0
            client = subprocess.Popen([str(binary), origin+"/suspend", str(path), "4", str(1024*1024), "0", hashlib.sha256(DATA).hexdigest(),
                                       "GET", "0", "0", "single"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True,
                                      env={**os.environ, "NETWORK_TEST_ADAPTIVE": "1"})
            try:
                deadline = time.monotonic()+5
                while len(Handler.streams) < 4 and time.monotonic() < deadline:
                    time.sleep(0.01)
                client.send_signal(signal.SIGSTOP)
                with Handler.guard:
                    for stream in Handler.streams:
                        try: stream.shutdown(socket.SHUT_RDWR)
                        except OSError: pass
                    Handler.streams.clear()
                    Handler.suspend_refusals = refusals
                time.sleep(10.5)
                client.send_signal(signal.SIGCONT)
                out = json.loads(client.communicate(timeout=20)[0])
            finally:
                client.kill()
            assert out["state"] == "completed" and out["retries"] == refusals, out
            assert path.read_bytes() == DATA
            path.unlink()
        suspended(0)
        suspended(1)
        path = directory / "shutdown"
        assert run("/slow", path, stop=180)["state"] == "stopped"
        assert run("/slow", path)["state"] == "completed"
        assert path.read_bytes() == DATA
        path.unlink()
        path = directory / "resume"
        cancelled = run("/slow", path, cancel=180)
        assert cancelled["state"] == "cancelled", cancelled
        assert Path(str(path)+".resume").exists()
        before = len(Handler.requests)
        resumed = run("/slow", path)
        assert resumed["state"] == "completed", resumed
        assert path.read_bytes() == DATA
        requests = Handler.requests[before:]
        assert len([item for item in requests if item[1] != "bytes=0-0"]) < 16, requests
        assert all(item[2] == '"fixture-v1"' for item in requests if item[1] != "bytes=0-0")
        path.unlink()
        # A signed link that expires mid-download resumes from a freshly resolved one: the checkpoint
        # names the bytes (validator, size), not the URL.
        path = directory / "signed"
        assert run("/slow-signed?sig=old", path, cancel=180)["state"] == "cancelled"
        Handler.expired.add("sig=old")
        expired = run("/slow-signed?sig=old", path)
        assert expired["state"] == "failed" and "HTTP 403" in expired["error"], expired
        before = len(Handler.requests)
        resumed = run("/slow-signed?sig=new", path)
        assert resumed["state"] == "completed", resumed
        assert path.read_bytes() == DATA
        assert len([r for r in Handler.requests[before:] if r[1] != "bytes=0-0"]) < 16
        path.unlink()
        # Without a strong ETag, Last-Modified guards ranges and resumes; the kept tail is re-read first.
        path = directory / "slow-modified"
        assert run("/slow-modified", path, cancel=180)["state"] == "cancelled"
        before = len(Handler.requests)
        resumed = run("/slow-modified", path)
        assert resumed["state"] == "completed", resumed
        assert path.read_bytes() == DATA
        requests = Handler.requests[before:]
        def span(request):
            first, last = map(int, request[1][6:].split("-"))
            return last-first+1
        ranged = [r for r in requests if r[1] and r[1] != "bytes=0-0"]
        assert any(span(r) == 65536 and r[2] is None for r in ranged), requests  # The tail check.
        assert all(r[2] == Handler.modified for r in ranged if span(r) != 65536), requests
        assert len([r for r in requests if r[1] != "bytes=0-0"]) < 17, requests
        path.unlink()
        assert run("/slow-modified", path, cancel=180)["state"] == "cancelled"
        Handler.modified = "Thu, 02 Jan 2025 00:00:00 GMT"
        changed = run("/slow-modified", path)
        assert changed["state"] == "failed" and "checkpoint" in changed["error"], changed
        # No validator at all: ranges still resume once the server's tail matches the partial's.
        for endpoint in ("/slow-no-validator", "/slow-no-validator-mutable"):
            path = directory / endpoint.removeprefix("/")
            Handler.mutated = False
            assert run(endpoint, path, cancel=180)["state"] == "cancelled"
            Handler.mutated = endpoint.endswith("mutable")
            out = run(endpoint, path)
            if Handler.mutated:
                assert out["state"] == "failed" and "changed since this partial" in out["error"], out
                assert not path.exists()
            else:
                assert out["state"] == "completed", out
                assert path.read_bytes() == DATA
        Handler.mutated = False
        path = directory / "hash"
        assert run("/file", path, digest="0"*64)["state"] == "failed"
        assert not path.exists()
        # A complete but hash-mismatched partial can be retried with the same expected identity;
        # it must never be silently promoted to a destination file.
        assert run("/file", path, digest="0"*64)["state"] == "failed"
        path = directory / "correct-hash"
        assert run("/file", path, digest=hashlib.sha256(DATA).hexdigest())["state"] == "completed"
        path.unlink()
        certificate, key = directory / "certificate.pem", directory / "key.pem"
        subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
                        "-keyout", str(key), "-out", str(certificate), "-days", "1",
                        "-subj", "/CN=localhost", "-addext", "subjectAltName=IP:127.0.0.1"], check=True, capture_output=True)
        tls = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(certificate, key)
        tls.socket = context.wrap_socket(tls.socket, server_side=True)
        threading.Thread(target=tls.serve_forever, daemon=True).start()
        assert run(f"https://127.0.0.1:{tls.server_port}/json")["state"] == "failed"
        tls.shutdown()
        # An origin that refuses new TLS handshakes for eight seconds while idle connections split the tails of
        # its 32 MiB ranges: the download lands byte-identical without a handshake storm, new connections
        # one at a time. One that refuses for good fails cleanly in bounded time.
        refusing = Refusing(context)
        threading.Thread(target=refusing.serve_forever, daemon=True).start()

        def refused_run(path, digest, deadline):
            result = subprocess.run([str(binary), f"https://127.0.0.1:{refusing.server_port}/paced", str(path), "32", str(1024*1024),
                                     "0", digest, "GET", "0", "0", "single"], capture_output=True, text=True, timeout=deadline+10,
                                    env={**os.environ, "NETWORK_TEST_ADAPTIVE": "1", "NETWORK_TEST_CA": str(certificate),
                                         "NETWORK_TEST_RANGE_BYTES": str(32*1024*1024), "NETWORK_TEST_DEADLINE_MS": str(deadline*1000)})
            return json.loads(result.stdout), result.stderr

        path = directory / "refused-burst"
        refusing.started, refusing.burst = time.monotonic(), True
        out, log = refused_run(path, hashlib.sha256(LARGE).hexdigest(), 60)
        refusing.burst = False
        print(f"Refusal burst: {refusing.handshakes} handshakes, {len(refusing.refused)} refused, {out['retries']} retries, "
              f"{out['milliseconds']} ms")
        assert out["state"] == "completed", (out, log[-2000:])
        assert path.read_bytes() == LARGE
        path.unlink()
        # The handshakes in flight when the burst began fail together; after them, one at a time.
        later = [at for at in refusing.refused if at-refusing.refused[0] > 1]
        assert all(b-a > 0.1 for a, b in zip(later, later[1:])), (refusing.refused, log[-2000:])
        assert len(refusing.refused) <= 20 and refusing.handshakes <= 40, (refusing.handshakes, refusing.refused)

        path = directory / "refused-forever"
        refusing.handshakes, refusing.refused, refusing.dead = 0, [], True
        started = time.monotonic()
        out, log = refused_run(path, "", 150)
        elapsed = time.monotonic()-started
        print(f"Refused forever: failed after {elapsed:.1f} s, {refusing.handshakes} handshakes, {out['retries']} retries")
        assert out["state"] == "failed" and "SSL connect error" in out["error"], out
        assert elapsed < 120 and refusing.handshakes <= 200, (elapsed, refusing.handshakes, out)
        assert not path.exists()
        refusing.shutdown()
        existing = directory / "existing"
        existing.write_bytes(b"preserve")
        assert run("/file", existing)["state"] == "failed"
        assert existing.read_bytes() == b"preserve"
        print("Native networking: HTTP methods/status/limits, exact ranges, bounded buffering, redirects, "
              "sequential fallbacks, truncation, 64-bit offsets, retries, cancellation/resume, re-resolved and expired "
              "links, Last-Modified and validator-less resume, full-disk and killed-process resume, hashes, "
              "TLS verification and destination preservation passed.")
    server.shutdown()


if __name__ == "__main__":
    main()