Hands-on lab — Day 78: HTTP in Python with requests

"""The reference client — the shape every HTTP call in this course will take.

Three rules are baked into every function here, and they are the three that
separate a script from something you can leave running:

  1. Every request has a TIMEOUT. `requests` has no default one. A call
     without a timeout can wait forever on a socket that will never answer,
     and "forever" is not an exaggeration.
  2. Every function takes its `session` as a PARAMETER. That is Day 74's
     boundary argument applied to the network: a function that calls
     `requests.get` directly can only be tested with a real server or a
     patch, while a function that takes a session can be tested with a
     fifteen-line fake and no server at all.
  3. Retries are only for the failures worth retrying — 429 and 5xx. A 400,
     a 401 or a 404 will give the same answer however many times you ask.

Nothing here knows anything about a particular host. `base_url` is passed
in, which is why the same client works against the local test server and
would work against a real API.
"""

from __future__ import annotations

import hashlib
import os
import random
import time
from dataclasses import dataclass
from typing import Any, Callable, Iterable, Protocol

import requests

# 10 seconds to finish reading the body, 3.05 to get the connection open.
# The odd 3.05 is the documented habit: connect timeouts slightly larger
# than a multiple of 3 line up with the TCP retransmission window.
DEFAULT_TIMEOUT: tuple[float, float] = (3.05, 10.0)

# The families worth trying again. Everything else is a permanent answer.
RETRY_STATUSES = frozenset({429, 500, 502, 503, 504})


class ReadingsError(Exception):
    """Base class for everything this client raises on purpose."""


class ReadingsUnavailable(ReadingsError):
    """The server could not be reached, or kept failing."""


class StationNotFound(ReadingsError):
    """The server answered clearly: there is no such station."""


class HttpSession(Protocol):
    """The slice of `requests.Session` this client actually uses.

    Writing the boundary down as a Protocol costs three lines and buys two
    things: mypy checks that a fake really implements it, and a reader can
    see exactly how much of `requests` the code depends on.
    """

    def get(self, url: str, **kwargs: Any) -> Any: ...

    def post(self, url: str, **kwargs: Any) -> Any: ...


@dataclass(frozen=True)
class Reading:
    station: str
    hour: int
    celsius: float


def make_session(user_agent: str = "day078-lab/1.0 (course exercise)") -> requests.Session:
    """A Session with the headers every request from this client should carry.

    A Session is two things at once: a place to put shared configuration,
    and a pool of open TCP connections. The second is the one people forget,
    and it is worth roughly the whole cost of the handshake per request.
    """
    session = requests.Session()
    session.headers.update(
        {
            "User-Agent": user_agent,
            "Accept": "application/json",
        }
    )
    token = os.environ.get("READINGS_TOKEN")
    if token:
        # Read from the environment, never written in the file. A token in
        # source control is a token you have to rotate.
        session.headers["Authorization"] = f"Bearer {token}"
    return session


def fetch_readings(
    base_url: str,
    station: str,
    *,
    session: HttpSession,
    timeout: tuple[float, float] | float = DEFAULT_TIMEOUT,
) -> list[Reading]:
    """Fetch one station's readings. The whole point is the `session=` parameter.

    Note `params=` rather than string concatenation. `requests` percent-encodes
    the values for you, so a station called `ALPHA ONE&BRAVO` produces a legal
    URL instead of a second query parameter you did not mean to send.
    """
    response = session.get(
        f"{base_url}/api/readings",
        params={"station": station},
        timeout=timeout,
    )
    if response.status_code == 404:
        raise StationNotFound(f"no station named {station!r}")
    response.raise_for_status()
    payload = response.json()
    return [
        Reading(station=row["station"], hour=int(row["hour"]), celsius=float(row["celsius"]))
        for row in payload["readings"]
    ]


def describe_failure(response: Any) -> str:
    """Turn a failed response into one sentence a human can act on.

    A traceback is for the programmer. This is for the person running the
    program, and it is the difference between "it crashed" and "the server
    said the station does not exist".
    """
    families = {
        3: "redirection",
        4: "your request was rejected",
        5: "the server failed",
    }
    family = families.get(response.status_code // 100, "unexpected status")
    detail = ""
    content_type = response.headers.get("Content-Type", "")
    if content_type.startswith("application/json"):
        try:
            body = response.json()
        except ValueError:
            body = None
        if isinstance(body, dict) and "detail" in body:
            detail = f" — {body['detail']}"
        elif isinstance(body, dict) and "error" in body:
            detail = f" — {body['error']}"
    return f"HTTP {response.status_code} ({family}){detail}"


def backoff_delays(
    attempts: int,
    *,
    base: float = 0.5,
    cap: float = 8.0,
    jitter: Callable[[], float] = random.random,
) -> list[float]:
    """The waits between `attempts` tries: exponential, capped, with jitter.

    The jitter is not decoration. Without it, a hundred clients that all saw
    the same outage retry at the same instant, and the server that was
    recovering is knocked over again by a synchronised wave. `jitter` is a
    parameter so a test can pass `lambda: 0.0` and assert exact numbers.
    """
    if attempts < 1:
        raise ValueError("attempts must be at least 1")
    delays = []
    for i in range(attempts - 1):
        raw = min(cap, base * (2**i))
        delays.append(round(raw * (0.5 + 0.5 * jitter()), 4))
    return delays


def get_with_retry(
    url: str,
    *,
    session: HttpSession,
    attempts: int = 4,
    timeout: tuple[float, float] | float = DEFAULT_TIMEOUT,
    sleep: Callable[[float], None] = time.sleep,
    jitter: Callable[[], float] = random.random,
    params: dict[str, Any] | None = None,
) -> Any:
    """GET with retries on 429 and 5xx — and on nothing else.

    `sleep` and `jitter` arrive as parameters for exactly the reason Day 74
    gave: a test can pass a recording sleep and prove the schedule in
    microseconds, without anything ever waiting.
    """
    delays = backoff_delays(attempts, jitter=jitter)
    last_error: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            response = session.get(url, params=params, timeout=timeout)
        except requests.exceptions.RequestException as exc:
            # A transport-level failure: DNS, connection refused, timeout.
            # There is no status code here because there is no response.
            last_error = exc
        else:
            if response.status_code not in RETRY_STATUSES:
                return response
            last_error = ReadingsUnavailable(describe_failure(response))
            # A server that says how long to wait knows better than we do.
            retry_after = response.headers.get("Retry-After")
            if retry_after is not None and attempt <= len(delays):
                try:
                    delays[attempt - 1] = min(float(retry_after), 8.0)
                except ValueError:
                    pass
        if attempt <= len(delays):
            sleep(delays[attempt - 1])
    raise ReadingsUnavailable(f"gave up after {attempts} attempts: {last_error}")


def stream_to_file(
    url: str,
    destination: str,
    *,
    session: HttpSession,
    chunk_size: int = 8192,
    timeout: tuple[float, float] | float = DEFAULT_TIMEOUT,
) -> tuple[int, int, str]:
    """Download a body without ever holding all of it in memory.

    Returns (bytes written, chunks read, sha256 hex digest). `stream=True`
    means the response headers have arrived and the body has not; the body
    is pulled a chunk at a time as you iterate.
    """
    digest = hashlib.sha256()
    total = 0
    chunks = 0
    with session.get(url, stream=True, timeout=timeout) as response:
        response.raise_for_status()
        with open(destination, "wb") as handle:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if not chunk:
                    continue
                handle.write(chunk)
                digest.update(chunk)
                total += len(chunk)
                chunks += 1
    return total, chunks, digest.hexdigest()


def summarise(readings: Iterable[Reading]) -> dict[str, float]:
    """Pure function, no network. The part worth testing without a server."""
    values = [r.celsius for r in readings]
    if not values:
        raise ReadingsError("cannot summarise an empty set of readings")
    return {
        "count": float(len(values)),
        "min": min(values),
        "max": max(values),
        "mean": round(sum(values) / len(values), 4),
    }

"""Fixtures for the example suite: one local server for the whole session.

The server is started once, on an ephemeral port, and shut down when the
last test finishes. Starting it per test would be correct but wasteful; the
tests that need a clean counter call `/control/reset` themselves.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Iterator

import pytest

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

from demo_server import CountingServer, base_url, running_server  # noqa: E402


@pytest.fixture(scope="session")
def server() -> Iterator[CountingServer]:
    with running_server() as srv:
        yield srv


@pytest.fixture(scope="session")
def base(server: CountingServer) -> str:
    """The address the local test server actually bound to, this run."""
    return base_url(server)

"""A small HTTP server you control, built only from the standard library.

Everything in this lab talks to THIS server, on the loopback address
127.0.0.1, on a port the operating system picks at run time. Nothing here
opens a connection to the internet, and nothing here needs one.

Why a local server instead of a real public API:

  * it is fast — no DNS lookup, no round trip across the world;
  * it is deterministic — the same bytes every run, so a test can assert;
  * it is honest — you can ask it for a 500, a 429 or a two-second delay,
    and no real service has to be harmed to produce them;
  * it works on a plane, and it will still work in five years.

The endpoints exist to produce the interesting cases:

  GET  /api/readings          200 with a JSON body
  GET  /api/readings?station= 200, filtered; also echoes the parsed query
  GET  /api/missing           404 with a JSON error body
  GET  /api/broken            500 with a JSON error body
  GET  /old/readings          301 permanent redirect to /api/readings
  GET  /api/flaky             429 with Retry-After for the first N calls,
                              then 200 (call /control/reset to arm it)
  GET  /api/slow?seconds=2    sleeps, then 200 — lets a timeout really fire
  GET  /api/large?kb=512      a large body, for streaming
  POST /api/echo              echoes method, headers and body back as JSON
  GET  /control/reset?fail=2  arms the flaky endpoint, resets counters
  GET  /control/stats         how many TCP connections and requests so far

Run it on its own if you want to poke at it by hand:

    python3 examples/demo_server.py

It prints the address it bound to and serves until you press Ctrl-C.
"""

from __future__ import annotations

import contextlib
import json
import socket
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Iterator
from urllib.parse import parse_qs, urlparse

READINGS: list[dict[str, object]] = [
    {"station": "ALPHA", "hour": 0, "celsius": 12.0},
    {"station": "ALPHA", "hour": 6, "celsius": 14.0},
    {"station": "ALPHA", "hour": 12, "celsius": 20.0},
    {"station": "ALPHA", "hour": 18, "celsius": 22.0},
    {"station": "BRAVO", "hour": 0, "celsius": 3.0},
    {"station": "BRAVO", "hour": 12, "celsius": 9.0},
]


class CountingServer(ThreadingHTTPServer):
    """A threading server that counts the TCP connections it accepts.

    The connection count is what makes "a Session reuses one connection"
    something a test can PROVE rather than something a lesson asserts.
    `get_request` is called once per accepted connection, not once per
    request, so the two numbers diverge exactly when keep-alive works.
    """

    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.lock = threading.Lock()
        self.connections = 0
        self.requests = 0
        self.flaky_remaining = 0
        self.flaky_calls = 0

    def get_request(self):  # noqa: D102 - inherited contract
        conn, addr = super().get_request()
        with self.lock:
            self.connections += 1
        return conn, addr


class DemoHandler(BaseHTTPRequestHandler):
    # HTTP/1.1 is what makes keep-alive possible. With HTTP/1.0 the server
    # closes after every response and connection reuse cannot be shown.
    protocol_version = "HTTP/1.1"
    server_version = "DayLab/1.0"
    sys_version = ""

    # ---- plumbing ---------------------------------------------------------

    def log_message(self, fmt: str, *args) -> None:
        """Silence the default stderr access log; tests want clean output."""

    def version_string(self) -> str:
        """The Server header. Kept short and stable so captures are diffable."""
        return self.server_version

    def _send(
        self,
        status: int,
        body: bytes,
        content_type: str = "application/json",
        extra_headers: dict[str, str] | None = None,
    ) -> None:
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        for name, value in (extra_headers or {}).items():
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(body)

    def _send_json(
        self,
        status: int,
        payload: object,
        extra_headers: dict[str, str] | None = None,
    ) -> None:
        body = json.dumps(payload, indent=None, separators=(",", ":")).encode("utf-8")
        self._send(status, body, "application/json; charset=utf-8", extra_headers)

    # ---- routing ----------------------------------------------------------

    def do_GET(self) -> None:  # noqa: N802 - name fixed by BaseHTTPRequestHandler
        parsed = urlparse(self.path)
        query = parse_qs(parsed.query)
        with self.server.lock:
            self.server.requests += 1

        if parsed.path == "/api/readings":
            self._readings(query)
        elif parsed.path == "/api/search":
            # Echoes the query string exactly as the server parsed it. This
            # is how you see what your client really sent, which is the
            # whole argument for `params=` over string concatenation.
            self._send_json(200, {"raw_query": parsed.query, "parsed": dict(query)})
        elif parsed.path == "/api/missing":
            self._send_json(
                404,
                {"error": "not_found", "detail": "no such station", "path": parsed.path},
            )
        elif parsed.path == "/api/broken":
            self._send_json(500, {"error": "internal", "detail": "the server fell over"})
        elif parsed.path == "/old/readings":
            self._send(
                301,
                b"",
                "text/plain; charset=utf-8",
                {"Location": "/api/readings"},
            )
        elif parsed.path == "/api/flaky":
            self._flaky()
        elif parsed.path == "/api/slow":
            time.sleep(float(query.get("seconds", ["2"])[0]))
            self._send_json(200, {"slept": float(query.get("seconds", ["2"])[0])})
        elif parsed.path == "/api/large":
            self._large(query)
        elif parsed.path == "/control/reset":
            with self.server.lock:
                self.server.flaky_remaining = int(query.get("fail", ["2"])[0])
                self.server.flaky_calls = 0
            self._send_json(200, {"flaky_remaining": self.server.flaky_remaining})
        elif parsed.path == "/control/stats":
            with self.server.lock:
                stats = {
                    "connections": self.server.connections,
                    "requests": self.server.requests,
                    "flaky_calls": self.server.flaky_calls,
                }
            self._send_json(200, stats)
        else:
            self._send_json(404, {"error": "not_found", "path": parsed.path})

    def do_POST(self) -> None:  # noqa: N802 - name fixed by the base class
        parsed = urlparse(self.path)
        with self.server.lock:
            self.server.requests += 1
        length = int(self.headers.get("Content-Length", "0"))
        raw = self.rfile.read(length) if length else b""

        if parsed.path != "/api/echo":
            self._send_json(404, {"error": "not_found", "path": parsed.path})
            return

        try:
            parsed_body: object = json.loads(raw.decode("utf-8")) if raw else None
        except json.JSONDecodeError:
            parsed_body = None

        self._send_json(
            201,
            {
                "method": "POST",
                "path": parsed.path,
                "content_type": self.headers.get("Content-Type"),
                "user_agent": self.headers.get("User-Agent"),
                "authorization_seen": self.headers.get("Authorization") is not None,
                "body_bytes": length,
                "json": parsed_body,
            },
        )

    # ---- individual endpoints --------------------------------------------

    def _readings(self, query: dict[str, list[str]]) -> None:
        station = query.get("station", [None])[0]
        known = {str(r["station"]) for r in READINGS}
        if station is not None and station not in known:
            self._send_json(
                404,
                {
                    "error": "not_found",
                    "detail": f"no station named {station}",
                    "known": sorted(known),
                },
            )
            return
        rows = [r for r in READINGS if station is None or r["station"] == station]
        self._send_json(
            200,
            {
                "station": station,
                "count": len(rows),
                "query_seen": {k: v for k, v in query.items()},
                "readings": rows,
            },
        )

    def _flaky(self) -> None:
        with self.server.lock:
            self.server.flaky_calls += 1
            attempt = self.server.flaky_calls
            if self.server.flaky_remaining > 0:
                self.server.flaky_remaining -= 1
                remaining = self.server.flaky_remaining
                rate_limited = True
            else:
                remaining = 0
                rate_limited = False
        if rate_limited:
            self._send_json(
                429,
                {"error": "rate_limited", "attempt": attempt, "still_failing": remaining},
                {"Retry-After": "1"},
            )
        else:
            self._send_json(200, {"ok": True, "attempt": attempt})

    def _large(self, query: dict[str, list[str]]) -> None:
        kilobytes = int(query.get("kb", ["512"])[0])
        # A repeating 1 KiB line, so the body is large but perfectly
        # predictable: every byte is derivable from `kb`.
        line = (b"x" * 1023) + b"\n"
        body = line * kilobytes
        self._send(200, body, "text/plain; charset=utf-8")


def wait_until_accepting(host: str, port: int, timeout: float = 5.0) -> None:
    """Poll the port until a connection succeeds, or give up loudly.

    This is the readiness loop that replaces `time.sleep(1)`. A fixed sleep
    is both too slow (usually) and too short (sometimes), which is the
    recipe for a test that fails once a fortnight on a loaded machine.
    """
    deadline = time.monotonic() + timeout
    last_error: OSError | None = None
    while time.monotonic() < deadline:
        try:
            with socket.create_connection((host, port), timeout=0.25):
                return
        except OSError as exc:  # not up yet
            last_error = exc
            time.sleep(0.01)
    raise RuntimeError(f"the local test server never became ready on port {port}: {last_error}")


@contextlib.contextmanager
def running_server() -> Iterator[CountingServer]:
    """Start the server on an ephemeral port and shut it down afterwards.

    Binding port 0 asks the operating system for any free port, which is the
    only way to avoid colliding with whatever the learner already has
    running. The real port is read back from `server_address` afterwards.
    """
    server = CountingServer(("127.0.0.1", 0), DemoHandler)
    thread = threading.Thread(target=server.serve_forever, name="demo-server", daemon=True)
    thread.start()
    try:
        wait_until_accepting(*server.server_address[:2])
        yield server
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def base_url(server: CountingServer) -> str:
    host, port = server.server_address[:2]
    return f"http://{host}:{port}"


if __name__ == "__main__":
    with running_server() as srv:
        print(f"serving on {base_url(srv)} — press Ctrl-C to stop")
        try:
            while True:
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("\nstopped.")

"""The whole lab in one run: eight sections, one local server, no internet.

    python3 examples/demo.py

Each section is one of the operational points from the lesson, demonstrated
rather than asserted. Read the timings — they are where the argument lives.
"""

from __future__ import annotations

import sys
import tempfile
import time
from pathlib import Path

import requests

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

from client import (  # noqa: E402
    DEFAULT_TIMEOUT,
    StationNotFound,
    backoff_delays,
    describe_failure,
    fetch_readings,
    get_with_retry,
    make_session,
    stream_to_file,
    summarise,
)
from demo_server import base_url, running_server  # noqa: E402


class RecordingSleep:
    """A spy, straight out of Day 74. It records waits and never waits."""

    def __init__(self) -> None:
        self.waits: list[float] = []

    def __call__(self, seconds: float) -> None:
        self.waits.append(round(seconds, 4))


def section(title: str) -> None:
    print()
    print(title)
    print("=" * len(title))


def main() -> int:
    with running_server() as server:
        root = base_url(server)
        session = make_session()

        section("1. A request and a response, in the pieces that matter")
        response = session.get(
            f"{root}/api/readings", params={"station": "ALPHA"}, timeout=DEFAULT_TIMEOUT
        )
        print(f"  request method   : {response.request.method}")
        print(f"  request path     : {response.request.path_url}")
        print(f"  request headers  : {len(response.request.headers)} sent")
        print(f"  status code      : {response.status_code} {response.reason}")
        print(f"  content type     : {response.headers['Content-Type']}")
        print(f"  .content is      : {type(response.content).__name__}, {len(response.content)} bytes")
        print(f"  .text is         : {type(response.text).__name__}, {len(response.text)} characters")
        print(f"  .json() is       : {type(response.json()).__name__} with keys {sorted(response.json())}")
        print(f"  elapsed          : {response.elapsed.total_seconds():.4f}s")

        section("2. params= versus gluing strings together")
        awkward = "ALPHA ONE&station=BRAVO"
        good = session.get(f"{root}/api/search", params={"station": awkward}, timeout=DEFAULT_TIMEOUT)
        bad = session.get(f"{root}/api/search?station={awkward}", timeout=DEFAULT_TIMEOUT)
        print(f"  station value    : {awkward!r}")
        print(f"  params=          : {good.request.path_url}")
        print(f"    server parsed  : {good.json()['parsed']}")
        print(f"  f-string         : {bad.request.path_url}")
        print(f"    server parsed  : {bad.json()['parsed']}")
        print("  the f-string smuggled a second parameter in. params= encoded it.")

        section("3. Status codes: a 404 is a successful response")
        missing = session.get(f"{root}/api/missing", timeout=DEFAULT_TIMEOUT)
        print(f"  the call itself  : returned normally, no exception")
        print(f"  status_code      : {missing.status_code}")
        print(f"  bool(response)   : {bool(missing)}   <- False for 4xx and 5xx")
        print(f"  described        : {describe_failure(missing)}")
        try:
            missing.raise_for_status()
        except requests.exceptions.HTTPError as exc:
            print(f"  raise_for_status : {type(exc).__name__}: {str(exc).split(' for url')[0]}")
        try:
            fetch_readings(root, "NOWHERE", session=session)
        except StationNotFound as exc:
            print(f"  the client raises: StationNotFound: {exc}")

        section("4. A redirect, followed and unfollowed")
        followed = session.get(f"{root}/old/readings", timeout=DEFAULT_TIMEOUT)
        print(f"  final status     : {followed.status_code}")
        print(f"  final url path   : {followed.url.rsplit('/', 1)[-1]}")
        print(f"  history          : {[r.status_code for r in followed.history]}")
        raw = session.get(f"{root}/old/readings", timeout=DEFAULT_TIMEOUT, allow_redirects=False)
        print(f"  unfollowed       : {raw.status_code}, Location: {raw.headers['Location']}")
        print("  301 is permanent — a client may cache it. 302 is temporary.")

        section("5. The timeout that is not there by default")
        started = time.monotonic()
        try:
            session.get(f"{root}/api/slow", params={"seconds": 3}, timeout=(3.05, 0.5))
        except requests.exceptions.Timeout as exc:
            waited = time.monotonic() - started
            print(f"  asked for        : 3 seconds of server work")
            print(f"  read timeout     : 0.5s")
            print(f"  raised after     : {waited:.2f}s — {type(exc).__name__}")
        print("  with no timeout= at all, that call waits for the full 3s, and")
        print("  against a server that never answers it waits forever.")

        section("6. Retry with backoff — and what must never be retried")
        session.get(f"{root}/control/reset", params={"fail": 2}, timeout=DEFAULT_TIMEOUT)
        sleeper = RecordingSleep()
        started = time.monotonic()
        result = get_with_retry(
            f"{root}/api/flaky",
            session=session,
            attempts=4,
            sleep=sleeper,
            jitter=lambda: 1.0,
        )
        print(f"  server sent      : 429, 429, then 200")
        print(f"  final status     : {result.status_code} on attempt {result.json()['attempt']}")
        print(f"  waits requested  : {sleeper.waits}  (Retry-After: 1 overrode the schedule)")
        print(f"  real time taken  : {time.monotonic() - started:.3f}s — the sleep was injected")
        print(f"  schedule alone   : {backoff_delays(5, jitter=lambda: 1.0)}")
        print(f"  with half jitter : {backoff_delays(5, jitter=lambda: 0.0)}")
        not_retried = session.get(f"{root}/api/missing", timeout=DEFAULT_TIMEOUT)
        print(f"  404 retryable?   : {not_retried.status_code in {429, 500, 502, 503, 504}}")
        print("  a 404 will be a 404 on the tenth try. Retrying it is a bug.")

        section("7. Session and connection reuse")
        before = server.connections
        with requests.Session() as pooled:
            for _ in range(5):
                pooled.get(f"{root}/api/readings", timeout=DEFAULT_TIMEOUT).close()
        with_session = server.connections - before
        before = server.connections
        for _ in range(5):
            requests.get(f"{root}/api/readings", timeout=DEFAULT_TIMEOUT).close()
        without_session = server.connections - before
        print(f"  5 calls, one Session      : {with_session} TCP connection(s)")
        print(f"  5 calls, requests.get()   : {without_session} TCP connection(s)")
        print("  each extra connection is a handshake — and, over TLS, several.")

        section("8. Streaming a large body instead of loading it")
        with tempfile.TemporaryDirectory() as tmp:
            destination = str(Path(tmp) / "large.txt")
            total, chunks, digest = stream_to_file(
                f"{root}/api/large?kb=512", destination, session=session, chunk_size=8192
            )
            print(f"  bytes written    : {total}")
            print(f"  chunks read      : {chunks} of at most 8192 bytes")
            print(f"  peak held        : one chunk, not {total} bytes")
            print(f"  sha256           : {digest[:32]}...")
        readings = fetch_readings(root, "ALPHA", session=session)
        print(f"  and the summary  : {summarise(readings)}")
        session.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""A fake session — the Day 74 payoff, in about forty lines.

`fetch_readings` and `get_with_retry` both take a `session` parameter. That
one design decision means the network can be replaced by this file, and the
tests that use it need no server, no socket, no port, and no waiting.

Compare the two shapes:

    def fetch(station):                      # untestable without a network
        return requests.get(URL, params=...) # or without patching

    def fetch(station, *, session):          # testable with FakeSession
        return session.get(URL, params=...)

FakeResponse implements only the slice of `requests.Response` the client
actually touches: `.status_code`, `.headers`, `.json()`, `.text`, and
`raise_for_status()`. That slice being small is itself information — it
tells you how little of `requests` your code depends on.
"""

from __future__ import annotations

import json as _json
from typing import Any

import requests


class FakeResponse:
    """The part of a response this client uses, and nothing else."""

    def __init__(
        self,
        status_code: int,
        payload: Any = None,
        headers: dict[str, str] | None = None,
        text: str | None = None,
    ) -> None:
        self.status_code = status_code
        self._payload = payload
        self.headers = headers or {"Content-Type": "application/json; charset=utf-8"}
        self.text = text if text is not None else _json.dumps(payload)

    def json(self) -> Any:
        if self._payload is None:
            raise ValueError("no JSON body")
        return self._payload

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            raise requests.exceptions.HTTPError(f"{self.status_code} error")

    def __enter__(self) -> "FakeResponse":
        return self

    def __exit__(self, *exc_info: object) -> None:
        return None


class FakeSession:
    """A scripted stand-in for `requests.Session`.

    It is a fake, a spy and a stub at once, exactly as Day 74 described:
    it answers from a script, it records every call, and a scripted item
    that happens to be an exception instance is raised instead of returned —
    which is how a test produces a ConnectionError or a Timeout on demand.
    """

    def __init__(self, script: list[Any] | None = None) -> None:
        self._script = list(script or [])
        self.calls: list[dict[str, Any]] = []

    def _next(self, method: str, url: str, kwargs: dict[str, Any]) -> Any:
        self.calls.append({"method": method, "url": url, **kwargs})
        if not self._script:
            raise AssertionError(
                f"the fake session ran out of scripted responses at call {len(self.calls)}"
            )
        item = self._script.pop(0)
        if isinstance(item, Exception):
            raise item
        return item

    def get(self, url: str, **kwargs: Any) -> Any:
        return self._next("GET", url, kwargs)

    def post(self, url: str, **kwargs: Any) -> Any:
        return self._next("POST", url, kwargs)

    @property
    def timeouts(self) -> list[Any]:
        """Every timeout value passed in — so a test can prove one was set."""
        return [call.get("timeout") for call in self.calls]

"""The same calls in httpx — the modern alternative, if you have it.

httpx is NOT in this lab's `requirements/requirements.txt`, deliberately:
the lab's argument works with `requests` alone, and a lab should not make
you install a package to make a point about a package. If httpx happens to
be installed, this file runs and shows the differences; if it is not, it
says so and exits 0.

    python3 examples/httpx_demo.py

Two differences are worth watching for:

  * `httpx.Client` has a DEFAULT timeout of five seconds. `requests` has
    none. That single design decision is httpx's strongest argument.
  * the same code shape works with `httpx.AsyncClient` and `await`, which
    is what you want the day you need fifty model calls in flight at once.
"""

from __future__ import annotations

import sys
from pathlib import Path

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

from demo_server import base_url, running_server  # noqa: E402

try:
    import httpx
except ImportError:
    print("httpx is not installed in this environment, so this demo has nothing")
    print("to run. That is expected: httpx is not one of this lab's dependencies.")
    print("Install it with `.venv/bin/pip install httpx` if you want to compare.")
    raise SystemExit(0)


def section(title: str) -> None:
    print()
    print(title)
    print("=" * len(title))


def main() -> int:
    print(f"httpx {httpx.__version__}")
    with running_server() as server:
        root = base_url(server)

        section("1. The requests-shaped API, unchanged")
        with httpx.Client(headers={"User-Agent": "day078-httpx/1.0"}, timeout=10.0) as client:
            response = client.get(f"{root}/api/readings", params={"station": "ALPHA"})
            print(f"  status        : {response.status_code}")
            print(f"  path sent     : {response.request.url.raw_path.decode()}")
            print(f"  count         : {response.json()['count']}")
            print(f"  .text / .content / .json() all exist, same as requests")

            section("2. The default timeout — the difference that matters")
            default = httpx.Client()
            print(f"  httpx.Client() default timeout : {default.timeout}")
            default.close()
            print("  requests has no default timeout at all. A missing timeout=")
            print("  in requests hangs; in httpx it gives up after 5 seconds.")

            section("3. raise_for_status, and a 404")
            missing = client.get(f"{root}/api/missing")
            print(f"  status        : {missing.status_code}")
            print(f"  is_success    : {missing.is_success}")
            try:
                missing.raise_for_status()
            except httpx.HTTPStatusError as exc:
                print(f"  raised        : {type(exc).__name__}")

            section("4. Streaming")
            total = 0
            chunks = 0
            with client.stream("GET", f"{root}/api/large?kb=256") as stream:
                for chunk in stream.iter_bytes(8192):
                    total += len(chunk)
                    chunks += 1
            print(f"  bytes         : {total} in {chunks} chunk(s)")

            section("5. What httpx adds")
            print("  * a default timeout;")
            print("  * the same API in sync and async (httpx.AsyncClient);")
            print("  * HTTP/2 support, if you install the extra: pip install 'httpx[http2]'")
            print("    and pass http2=True. Without that extra it speaks HTTP/1.1,")
            print("    exactly like requests.")
            print("  * a strict URL and header model, which catches some mistakes")
            print("    requests would let through.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""HTTP with no HTTP library at all — just a socket and some text.

This is the demystifying one. `requests` is a convenience; underneath it,
an HTTP request is a few lines of ASCII sent down a TCP connection, and an
HTTP response is a few lines of ASCII sent back. You can type it by hand,
and here we do.

Run it:

    python3 examples/raw_socket_demo.py

Everything it prints was really sent and really received over a loopback
connection to the local test server started by this script.
"""

from __future__ import annotations

import socket
import sys
from pathlib import Path

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

from demo_server import running_server  # noqa: E402


def show(label: str, raw: bytes) -> None:
    print(f"  {label} ({len(raw)} bytes)")
    print("  " + "-" * (len(label) + 12))
    text = raw.decode("utf-8", errors="replace")
    lines = text.split("\r\n")
    while lines and lines[-1] == "":
        lines.pop()
    for line in lines:
        print(f"    {line}\\r\\n")
    print("    \\r\\n          <- the blank line: headers finished")
    print()


def main() -> int:
    with running_server() as server:
        host, port = server.server_address[:2]

        print("1. The request, typed out by hand")
        print("=================================")
        print("  Four parts: a request line, some headers, a blank line, and")
        print("  (for a GET) no body. Every line ends with carriage return +")
        print("  line feed, and the blank line is what says 'headers finished'.")
        print()

        request = (
            "GET /api/readings?station=ALPHA HTTP/1.1\r\n"
            f"Host: {host}:{port}\r\n"
            "User-Agent: day078-raw-socket/1.0\r\n"
            "Accept: application/json\r\n"
            "Connection: close\r\n"
            "\r\n"
        ).encode("ascii")
        show("bytes sent", request)

        with socket.create_connection((host, port), timeout=5.0) as sock:
            sock.sendall(request)
            received = b""
            while True:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                received += chunk

        print("2. The response, exactly as it arrived")
        print("======================================")
        print("  Three parts: a status line, some headers, a blank line, then")
        print("  the body. The body here is JSON, but HTTP neither knows nor")
        print("  cares — Content-Type is what says so.")
        print()

        head, _, body = received.partition(b"\r\n\r\n")
        show("status line and headers", head + b"\r\n")
        print(f"  body ({len(body)} bytes)")
        print("  " + "-" * 18)
        print(f"    {body.decode('utf-8')}")
        print()

        print("3. Reading the pieces back")
        print("==========================")
        lines = head.decode("ascii").split("\r\n")
        version, status, reason = lines[0].split(" ", 2)
        print(f"    HTTP version : {version}")
        print(f"    status code  : {status}")
        print(f"    reason phrase: {reason}")
        for line in lines[1:]:
            name, _, value = line.partition(": ")
            print(f"    header       : {name} = {value}")
        print()
        print("  That is the entire protocol. Everything `requests` adds is")
        print("  convenience on top of these bytes: building the request line,")
        print("  encoding the query string, pooling the connection, decoding")
        print("  the body, and turning a status code into an exception.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""The same four calls, without installing anything.

`requests` is not on the machine of every person who will run your script.
`urllib.request` and `http.client` are, because they ship with Python. This
file makes the three-way comparison concrete by doing the same work with
each of them against the same local test server.

Run it:

    python3 examples/stdlib_demo.py

Nothing here imports `requests`, and nothing here leaves 127.0.0.1.
"""

from __future__ import annotations

import http.client
import json
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

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

from demo_server import base_url, running_server  # noqa: E402


def section(title: str) -> None:
    print()
    print(title)
    print("=" * len(title))


def main() -> int:
    with running_server() as server:
        root = base_url(server)
        host, port = server.server_address[:2]

        section("1. urllib.request — a GET with a query string")
        query = urllib.parse.urlencode({"station": "ALPHA"})
        url = f"{root}/api/readings?{query}"
        req = urllib.request.Request(
            url,
            headers={"User-Agent": "day078-urllib/1.0", "Accept": "application/json"},
        )
        # A timeout is a parameter here too, and it is just as optional and
        # just as necessary. Note what you have to do by hand: build the
        # query string, set the headers, decode the bytes, parse the JSON.
        with urllib.request.urlopen(req, timeout=10.0) as response:
            payload = json.loads(response.read().decode("utf-8"))
        print(f"  status      : {response.status} {response.reason}")
        print(f"  content type: {response.headers['Content-Type']}")
        print(f"  count       : {payload['count']}")
        print(f"  first row   : {payload['readings'][0]}")
        print("  note        : you encoded the query, decoded the bytes, and")
        print("                parsed the JSON yourself. requests does all three.")

        section("2. urllib.request — a 404 is an EXCEPTION, not a status")
        try:
            with urllib.request.urlopen(f"{root}/api/missing", timeout=10.0):
                print("  unreachable")
        except urllib.error.HTTPError as exc:
            body = json.loads(exc.read().decode("utf-8"))
            print(f"  raised      : {type(exc).__name__}")
            print(f"  status      : {exc.code}")
            print(f"  detail      : {body['detail']}")
            print("  note        : this is the big behavioural difference. urllib")
            print("                raises on 4xx and 5xx; requests returns a")
            print("                response and lets you decide.")

        section("3. urllib.request — a POST with a JSON body")
        data = json.dumps({"station": "ALPHA", "note": "hand rolled"}).encode("utf-8")
        req = urllib.request.Request(
            f"{root}/api/echo",
            data=data,
            method="POST",
            headers={
                "Content-Type": "application/json",
                "User-Agent": "day078-urllib/1.0",
            },
        )
        with urllib.request.urlopen(req, timeout=10.0) as response:
            echoed = json.loads(response.read().decode("utf-8"))
        print(f"  status      : {response.status}")
        print(f"  server saw  : method={echoed['method']} bytes={echoed['body_bytes']}")
        print(f"  server saw  : content_type={echoed['content_type']}")
        print(f"  echoed json : {echoed['json']}")

        section("4. http.client — the layer underneath both")
        before = server.connections
        conn = http.client.HTTPConnection(host, port, timeout=10.0)
        try:
            conn.request(
                "GET",
                "/api/readings?station=BRAVO",
                headers={"Accept": "application/json", "User-Agent": "day078-httpclient/1.0"},
            )
            response = conn.getresponse()
            raw = response.read()
            print(f"  status      : {response.status} {response.reason}")
            print(f"  headers     : {len(response.getheaders())} of them")
            print(f"  body bytes  : {len(raw)}")
            print(f"  count       : {json.loads(raw.decode('utf-8'))['count']}")
            # The same connection, used twice. This is what a Session does
            # for you automatically; here it is manual.
            conn.request("GET", "/control/stats", headers={"Accept": "application/json"})
            conn.getresponse().read()
            print(f"  connections : {server.connections - before} opened for those 2 requests")
            print("  note        : http.client speaks the protocol and nothing")
            print("                more — no redirects, no pooling, no decoding.")
        finally:
            conn.close()

        section("5. What each layer costs you")
        print("  http.client     : the protocol, exactly. You manage everything.")
        print("  urllib.request  : redirects and a bit of convenience; verbose,")
        print("                    and it raises on 4xx/5xx.")
        print("  requests        : params=, .json(), Session pooling, retries via")
        print("                    an adapter, streaming. One pip install away.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""The reference suite, run against the local test server on 127.0.0.1.

Every test here opens a real socket — to a server this process started, on
a port the operating system chose, serving fixture data. None of them
touches the internet, and none of them is slow: the whole file finishes in
well under a second because the only "slow" endpoint is used precisely to
prove that a timeout fires.
"""

from __future__ import annotations

import time

import pytest
import requests

from client import (
    DEFAULT_TIMEOUT,
    RETRY_STATUSES,
    Reading,
    ReadingsUnavailable,
    StationNotFound,
    backoff_delays,
    describe_failure,
    fetch_readings,
    get_with_retry,
    make_session,
    stream_to_file,
    summarise,
)


class RecordingSleep:
    def __init__(self) -> None:
        self.waits: list[float] = []

    def __call__(self, seconds: float) -> None:
        self.waits.append(round(seconds, 4))


@pytest.fixture
def session():
    s = make_session()
    yield s
    s.close()


# --- 1. fetch and parse JSON ------------------------------------------------


def test_fetch_readings_returns_parsed_objects(base, session):
    readings = fetch_readings(base, "ALPHA", session=session)
    assert len(readings) == 4
    assert all(isinstance(r, Reading) for r in readings)
    assert readings[0] == Reading(station="ALPHA", hour=0, celsius=12.0)


def test_summary_of_the_fetched_readings(base, session):
    assert summarise(fetch_readings(base, "ALPHA", session=session)) == {
        "count": 4.0,
        "min": 12.0,
        "max": 22.0,
        "mean": 17.0,
    }


def test_params_are_encoded_not_concatenated(base, session):
    awkward = "ALPHA ONE&station=BRAVO"
    good = session.get(f"{base}/api/search", params={"station": awkward}, timeout=DEFAULT_TIMEOUT)
    assert good.json()["parsed"]["station"] == [awkward]
    assert good.json()["raw_query"] == "station=ALPHA+ONE%26station%3DBRAVO"
    bad = session.get(f"{base}/api/search?station={awkward}", timeout=DEFAULT_TIMEOUT)
    # The unencoded ampersand became a second parameter the caller never meant.
    assert bad.json()["parsed"]["station"] == ["ALPHA ONE", "BRAVO"]


def test_text_content_and_json_are_three_different_things(base, session):
    response = session.get(f"{base}/api/readings", timeout=DEFAULT_TIMEOUT)
    assert isinstance(response.content, bytes)
    assert isinstance(response.text, str)
    assert isinstance(response.json(), dict)
    assert response.content.decode("utf-8") == response.text


# --- 2. a 404 handled without a traceback -----------------------------------


def test_a_404_is_a_successful_response_not_an_exception(base, session):
    response = session.get(f"{base}/api/missing", timeout=DEFAULT_TIMEOUT)
    assert response.status_code == 404
    assert bool(response) is False


def test_a_missing_station_raises_a_domain_error_with_a_clean_message(base, session):
    with pytest.raises(StationNotFound) as caught:
        fetch_readings(base, "NOWHERE", session=session)
    assert "NOWHERE" in str(caught.value)
    assert "Traceback" not in str(caught.value)


def test_describe_failure_says_something_a_human_can_act_on(base, session):
    missing = session.get(f"{base}/api/missing", timeout=DEFAULT_TIMEOUT)
    assert describe_failure(missing) == "HTTP 404 (your request was rejected) — no such station"
    broken = session.get(f"{base}/api/broken", timeout=DEFAULT_TIMEOUT)
    assert describe_failure(broken) == "HTTP 500 (the server failed) — the server fell over"


def test_raise_for_status_raises_on_500_and_is_silent_on_200(base, session):
    with pytest.raises(requests.exceptions.HTTPError):
        session.get(f"{base}/api/broken", timeout=DEFAULT_TIMEOUT).raise_for_status()
    assert session.get(f"{base}/api/readings", timeout=DEFAULT_TIMEOUT).raise_for_status() is None


# --- 3. the timeout really fires --------------------------------------------


def test_a_read_timeout_fires_against_the_slow_endpoint(base, session):
    started = time.monotonic()
    with pytest.raises(requests.exceptions.Timeout):
        session.get(f"{base}/api/slow", params={"seconds": 3}, timeout=(3.05, 0.4))
    elapsed = time.monotonic() - started
    # It gave up on its own schedule, not the server's: well under 3 seconds.
    assert elapsed < 2.0


def test_a_timeout_is_a_requestexception_so_one_except_catches_the_family(base, session):
    with pytest.raises(requests.exceptions.RequestException):
        session.get(f"{base}/api/slow", params={"seconds": 3}, timeout=(3.05, 0.4))


# --- 4. retry with backoff --------------------------------------------------


def test_retry_succeeds_after_exactly_the_expected_number_of_attempts(base, session):
    session.get(f"{base}/control/reset", params={"fail": 2}, timeout=DEFAULT_TIMEOUT)
    sleeper = RecordingSleep()
    response = get_with_retry(
        f"{base}/api/flaky", session=session, attempts=4, sleep=sleeper, jitter=lambda: 1.0
    )
    assert response.status_code == 200
    assert response.json()["attempt"] == 3
    assert len(sleeper.waits) == 2


def test_the_retry_honours_retry_after_and_never_actually_sleeps(base, session):
    session.get(f"{base}/control/reset", params={"fail": 1}, timeout=DEFAULT_TIMEOUT)
    sleeper = RecordingSleep()
    started = time.monotonic()
    get_with_retry(f"{base}/api/flaky", session=session, attempts=3, sleep=sleeper, jitter=lambda: 1.0)
    assert sleeper.waits == [1.0]  # the server's Retry-After: 1
    assert time.monotonic() - started < 0.5


def test_retry_gives_up_and_says_how_many_attempts_it_made(base, session):
    session.get(f"{base}/control/reset", params={"fail": 99}, timeout=DEFAULT_TIMEOUT)
    sleeper = RecordingSleep()
    with pytest.raises(ReadingsUnavailable) as caught:
        get_with_retry(
            f"{base}/api/flaky", session=session, attempts=3, sleep=sleeper, jitter=lambda: 1.0
        )
    assert "after 3 attempts" in str(caught.value)
    assert len(sleeper.waits) == 2
    session.get(f"{base}/control/reset", params={"fail": 0}, timeout=DEFAULT_TIMEOUT)


def test_a_500_is_retried_and_a_404_is_not(base, session):
    assert 500 in RETRY_STATUSES and 429 in RETRY_STATUSES
    assert 404 not in RETRY_STATUSES and 400 not in RETRY_STATUSES and 401 not in RETRY_STATUSES
    sleeper = RecordingSleep()
    response = get_with_retry(
        f"{base}/api/missing", session=session, attempts=3, sleep=sleeper, jitter=lambda: 1.0
    )
    assert response.status_code == 404
    assert sleeper.waits == []  # returned on the first attempt, no waiting


@pytest.mark.parametrize(
    "attempts,expected",
    [(1, []), (2, [0.5]), (4, [0.5, 1.0, 2.0]), (6, [0.5, 1.0, 2.0, 4.0, 8.0])],
)
def test_the_backoff_schedule_doubles_and_caps(attempts, expected):
    assert backoff_delays(attempts, jitter=lambda: 1.0) == expected


def test_jitter_spreads_each_wait_over_half_its_slot():
    assert backoff_delays(4, jitter=lambda: 0.0) == [0.25, 0.5, 1.0]
    assert backoff_delays(4, jitter=lambda: 1.0) == [0.5, 1.0, 2.0]


# --- 5. Session and connection reuse ----------------------------------------


def test_one_session_reuses_one_connection_for_many_requests(base, server):
    before = server.connections
    with requests.Session() as pooled:
        for _ in range(5):
            pooled.get(f"{base}/api/readings", timeout=DEFAULT_TIMEOUT).close()
    assert server.connections - before == 1


def test_requests_get_without_a_session_opens_a_connection_every_time(base, server):
    before = server.connections
    for _ in range(5):
        requests.get(f"{base}/api/readings", timeout=DEFAULT_TIMEOUT).close()
    assert server.connections - before == 5


def test_the_session_carries_shared_headers_to_every_request(base, session):
    echoed = session.post(f"{base}/api/echo", json={"hello": "world"}, timeout=DEFAULT_TIMEOUT)
    assert echoed.status_code == 201
    assert echoed.json()["user_agent"].startswith("day078-lab/1.0")
    assert echoed.json()["content_type"] == "application/json"
    assert echoed.json()["json"] == {"hello": "world"}


def test_no_authorization_header_is_sent_when_the_environment_has_no_token(
    base, session, monkeypatch
):
    monkeypatch.delenv("READINGS_TOKEN", raising=False)
    echoed = session.post(f"{base}/api/echo", json={}, timeout=DEFAULT_TIMEOUT)
    assert echoed.json()["authorization_seen"] is False


def test_a_token_in_the_environment_becomes_an_authorization_header(base, monkeypatch):
    monkeypatch.setenv("READINGS_TOKEN", "not-a-real-secret")
    with make_session() as tokened:
        echoed = tokened.post(f"{base}/api/echo", json={}, timeout=DEFAULT_TIMEOUT)
    assert echoed.json()["authorization_seen"] is True


# --- 6. redirects -----------------------------------------------------------


def test_a_301_is_followed_by_default_and_recorded_in_history(base, session):
    response = session.get(f"{base}/old/readings", timeout=DEFAULT_TIMEOUT)
    assert response.status_code == 200
    assert [r.status_code for r in response.history] == [301]


def test_allow_redirects_false_shows_the_301_and_its_location(base, session):
    response = session.get(f"{base}/old/readings", timeout=DEFAULT_TIMEOUT, allow_redirects=False)
    assert response.status_code == 301
    assert response.headers["Location"] == "/api/readings"
    assert response.history == []


# --- 7. streaming -----------------------------------------------------------


def test_streaming_writes_the_whole_body_in_many_small_chunks(base, session, tmp_path):
    destination = tmp_path / "large.txt"
    total, chunks, digest = stream_to_file(
        f"{base}/api/large?kb=512", str(destination), session=session, chunk_size=8192
    )
    assert total == 512 * 1024
    assert chunks == 64
    assert destination.stat().st_size == total
    assert len(digest) == 64


def test_the_chunk_size_decides_how_much_is_held_at_once(base, session, tmp_path):
    total, chunks, _ = stream_to_file(
        f"{base}/api/large?kb=64", str(tmp_path / "s.txt"), session=session, chunk_size=1024
    )
    assert (total, chunks) == (64 * 1024, 64)

"""The Day 74 payoff: the same client, tested with NO server at all.

Not a local server. Not a mock of `requests`. Not a patch. A forty-line
fake object passed in through the `session=` parameter that was put there
for exactly this purpose.

Every test in this file runs in microseconds, would run identically on a
machine with no network stack, and can produce a ConnectionError, a 429
storm or a malformed body on demand — states that are awkward to arrange
even against a server you control.

The one thing these tests CANNOT prove is that your understanding of the
real server is correct. That is what `test_client.py` is for. Keep both.
"""

from __future__ import annotations

import pytest
import requests

from client import (
    Reading,
    ReadingsUnavailable,
    StationNotFound,
    fetch_readings,
    get_with_retry,
    summarise,
)
from fake_session import FakeResponse, FakeSession

PAYLOAD = {
    "station": "ALPHA",
    "count": 2,
    "readings": [
        {"station": "ALPHA", "hour": 0, "celsius": 12.0},
        {"station": "ALPHA", "hour": 12, "celsius": 22.0},
    ],
}


class RecordingSleep:
    def __init__(self) -> None:
        self.waits: list[float] = []

    def __call__(self, seconds: float) -> None:
        self.waits.append(round(seconds, 4))


def test_fetch_readings_works_against_a_fake_session():
    session = FakeSession([FakeResponse(200, PAYLOAD)])
    assert fetch_readings("http://example.invalid", "ALPHA", session=session) == [
        Reading("ALPHA", 0, 12.0),
        Reading("ALPHA", 12, 22.0),
    ]


def test_the_url_and_params_the_client_would_have_sent():
    session = FakeSession([FakeResponse(200, PAYLOAD)])
    fetch_readings("http://example.invalid", "ALPHA", session=session)
    call = session.calls[0]
    assert call["url"] == "http://example.invalid/api/readings"
    assert call["params"] == {"station": "ALPHA"}


def test_every_call_carries_a_timeout():
    """The check worth having in a real codebase. A missing timeout hangs."""
    session = FakeSession([FakeResponse(200, PAYLOAD)])
    fetch_readings("http://example.invalid", "ALPHA", session=session)
    assert session.timeouts == [(3.05, 10.0)]
    assert all(t is not None for t in session.timeouts)


def test_a_404_becomes_a_domain_exception_with_no_server_involved():
    session = FakeSession([FakeResponse(404, {"error": "not_found"})])
    with pytest.raises(StationNotFound):
        fetch_readings("http://example.invalid", "GHOST", session=session)


def test_a_connection_error_can_be_produced_on_demand():
    session = FakeSession([requests.exceptions.ConnectionError("name resolution failed")])
    with pytest.raises(requests.exceptions.ConnectionError):
        fetch_readings("http://example.invalid", "ALPHA", session=session)


def test_retry_recovers_from_two_429s_and_never_waits():
    session = FakeSession(
        [
            FakeResponse(429, {"error": "rate_limited"}, {"Retry-After": "2"}),
            FakeResponse(429, {"error": "rate_limited"}, {"Retry-After": "2"}),
            FakeResponse(200, PAYLOAD),
        ]
    )
    sleeper = RecordingSleep()
    response = get_with_retry(
        "http://example.invalid/api/flaky",
        session=session,
        attempts=4,
        sleep=sleeper,
        jitter=lambda: 1.0,
    )
    assert response.status_code == 200
    assert len(session.calls) == 3
    assert sleeper.waits == [2.0, 2.0]


def test_retry_survives_a_transport_failure_then_succeeds():
    session = FakeSession(
        [requests.exceptions.ConnectTimeout("no route"), FakeResponse(200, PAYLOAD)]
    )
    sleeper = RecordingSleep()
    response = get_with_retry(
        "http://example.invalid/api/flaky",
        session=session,
        attempts=3,
        sleep=sleeper,
        jitter=lambda: 1.0,
    )
    assert response.status_code == 200
    assert sleeper.waits == [0.5]


def test_retry_gives_up_after_five_503s_and_names_the_count():
    session = FakeSession([FakeResponse(503, {"error": "unavailable"}) for _ in range(5)])
    sleeper = RecordingSleep()
    with pytest.raises(ReadingsUnavailable) as caught:
        get_with_retry(
            "http://example.invalid/api/flaky",
            session=session,
            attempts=5,
            sleep=sleeper,
            jitter=lambda: 1.0,
        )
    assert "after 5 attempts" in str(caught.value)
    assert sleeper.waits == [0.5, 1.0, 2.0, 4.0]


@pytest.mark.parametrize("status", [400, 401, 403, 404, 409, 422])
def test_a_client_error_is_returned_immediately_and_never_retried(status):
    session = FakeSession([FakeResponse(status, {"error": "no"})])
    sleeper = RecordingSleep()
    response = get_with_retry(
        "http://example.invalid/api/x",
        session=session,
        attempts=4,
        sleep=sleeper,
        jitter=lambda: 1.0,
    )
    assert response.status_code == status
    assert len(session.calls) == 1
    assert sleeper.waits == []


@pytest.mark.parametrize("status", [429, 500, 502, 503, 504])
def test_a_retryable_status_is_tried_again(status):
    session = FakeSession([FakeResponse(status, {"error": "later"}), FakeResponse(200, PAYLOAD)])
    get_with_retry(
        "http://example.invalid/api/x",
        session=session,
        attempts=3,
        sleep=RecordingSleep(),
        jitter=lambda: 1.0,
    )
    assert len(session.calls) == 2


def test_summarise_is_pure_and_needs_no_session_at_all():
    assert summarise([Reading("A", 0, 10.0), Reading("A", 1, 20.0)])["mean"] == 15.0

lesson_id: D078
day: 78
kind: python-program
languages: [python, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-078-http-in-python-with-requests
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/pytest --version
  - '.venv/bin/python3 -c "import requests; print(requests.__version__)"'
run_commands:
  - .venv/bin/python3 examples/raw_socket_demo.py
  - .venv/bin/python3 examples/demo.py
  - .venv/bin/python3 examples/stdlib_demo.py
  - '.venv/bin/python3 examples/httpx_demo.py   # exits 0 whether or not httpx is installed'
  - .venv/bin/pytest examples -q
  - '.venv/bin/pytest examples/test_without_a_server.py -q   # no server started at all'
  - 'PYTHONPATH=tests .venv/bin/pytest examples -q   # every non-loopback socket blocked'
  - .venv/bin/pytest starter -q
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -rf .venv
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-07-19'
executed_on: 'macOS 26.5.1 (Apple Silicon), Python 3.14.0, requests 2.34.2, pytest 9.1.1, httpx 0.28.1, bash 3.2.57 — bash tests/run_tests.sh -> 58 checks, 0 failure(s), exit 0 (53 checks, 0 failure(s) with the starter exercises completed). requires_network is true for the one-time dependency install ONLY; the tests themselves open no socket to anything but 127.0.0.1, and the suite proves it by re-running every example test with all non-loopback sockets blocked.'

# Dependencies — Day 078 lab

**Two third-party packages. The server you talk to is not one of them — it is
built entirely from the standard library.**

## The pinned list

`requirements.txt` contains exactly two lines:

```
requests==2.34.2
pytest==9.1.1
```

| Dependency | Version | Licence | Why this lab needs it |
| --- | --- | --- | --- |
| requests | 2.34.2 | Apache License 2.0 (declared in the project's own metadata) | The subject of the day. The lab uses `Session`, `params=`, `timeout=`, `.json()`, `raise_for_status()`, `stream=True` with `iter_content`, `allow_redirects=False`, and the `requests.exceptions` hierarchy. |
| pytest | 9.1.1 | MIT (the `License-Expression` field of the installed distribution's own metadata) | The test runner from Day 71. This lab needs `pytest.raises`, `@pytest.mark.parametrize`, `tmp_path`, `monkeypatch`, session-scoped fixtures, and its exit code. |

Both are free and open source. Neither has a paid tier, an account, or
telemetry. Both versions were installed and verified on the authoring machine
on 2026-07-19.

## What is NOT in the list, and why that matters

**The server.** `examples/demo_server.py` is a subclass of
`http.server.BaseHTTPRequestHandler` served by `ThreadingHTTPServer`, with
`socket`, `threading`, `json` and `urllib.parse` doing the rest. All standard
library. You already have it. That is not a shortcut — it is the point:
Python ships with a working HTTP server, which is why "start a server you
control and test against that" costs you nothing.

**The alternatives the lesson compares.** `urllib.request` and `http.client`
are standard library too, which is why `examples/stdlib_demo.py` runs on a
machine with nothing installed at all. Run it and see:

```bash
python3 examples/stdlib_demo.py
```

**httpx.** `examples/httpx_demo.py` is the fourth alternative, and httpx is
deliberately *not* pinned here. If it happens to be installed, the demo runs
and prints the comparison; if it is not, the demo says so and exits 0. It ran
on the authoring machine against httpx 0.28.1. Install it yourself if you want
to compare:

```bash
.venv/bin/pip install httpx
```

## Install once

From this lab's directory:

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
.venv/bin/pytest --version
.venv/bin/python3 -c "import requests; print(requests.__version__)"
```

Those last two commands should print `pytest 9.1.1` and `2.34.2`. You created
your first virtual environment on Day 43; this is the same procedure.
`.venv/` is ignored by version control — never commit it.

## The network, precisely

The install needs the network **once**, to download two packages from the
Python Package Index. That is the only moment any part of this lab reaches
beyond your own machine.

After that, **nothing in this lab touches the internet** — not the demos, not
the example suites, not `tests/run_tests.sh`. Every socket opened goes to
`127.0.0.1`, the loopback interface, on a port the operating system assigned
at run time. This is not a promise; the test suite proves it. Section 5 of
`tests/run_tests.sh` runs the entire example suite with `tests/sitecustomize.py`
loaded, which replaces `socket.connect` and `socket.getaddrinfo` so that any
attempt to resolve a hostname or reach a non-loopback address raises
immediately — and then proves the guard is not vacuous by confirming that a
request to a public site under the same guard is refused.

If you already have both packages somewhere else, skip the virtual
environment and point the suite at that pytest:

```bash
PYTEST=/path/to/pytest bash tests/run_tests.sh
```

The runner resolves `python3` from the same directory as the pytest it found,
so the demos import the same `requests` the tests do.

## Check your Python

```bash
python3 --version
```

Verified on 3.14.0. Python 3.10 or newer is required, because the code uses
the `X | None` annotation syntax throughout.

requests==2.34.2
pytest==9.1.1

"""YOUR FILE — exercises 1 to 6.

Six functions. Each one has a docstring saying exactly what it must do, a
signature that is already right, and a `raise NotImplementedError` you
delete. `examples/client.py` contains a complete reference implementation:
use it when you are stuck, but write yours first — reading a solution feels
like learning and is not.

Run your work at any time:

    .venv/bin/pytest starter -q

Unfinished exercises are skipped, so the suite exits 0 from the first
minute. Check everything at the end with:

    bash tests/run_tests.sh

Two rules that apply to every function below, and to every HTTP call you
ever write after today:

  * pass `timeout=` to every request. `requests` has no default;
  * take `session` as a parameter, never reach for `requests.get` inside
    the function. That parameter is what makes exercise 7 possible.
"""

from __future__ import annotations

import hashlib
import random
import time
from dataclasses import dataclass
from typing import Any, Callable

import requests

# (connect timeout, read timeout). Use this as your default.
DEFAULT_TIMEOUT: tuple[float, float] = (3.05, 10.0)

# The only statuses worth trying again. Everything else is a final answer.
RETRY_STATUSES = frozenset({429, 500, 502, 503, 504})


class ReadingsError(Exception):
    """Base class for this client's own exceptions."""


class ReadingsUnavailable(ReadingsError):
    """Could not be reached, or kept failing."""


class StationNotFound(ReadingsError):
    """The server said clearly: there is no such station."""


@dataclass(frozen=True)
class Reading:
    station: str
    hour: int
    celsius: float


# ---------------------------------------------------------------------------
# Exercise 1 — fetch and parse JSON
# ---------------------------------------------------------------------------
def fetch_readings(
    base_url: str,
    station: str,
    *,
    session: Any,
    timeout: tuple[float, float] | float = DEFAULT_TIMEOUT,
) -> list[Reading]:
    """GET {base_url}/api/readings with the station as a QUERY PARAMETER.

    Steps:
      1. `response = session.get(f"{base_url}/api/readings",
                                 params={"station": station}, timeout=timeout)`
         — use `params=`, not an f-string. Exercise 1b below shows why.
      2. If `response.status_code == 404`, raise `StationNotFound` with a
         message naming the station. This is exercise 2's requirement, and
         it belongs here.
      3. Call `response.raise_for_status()` for everything else that failed.
      4. `payload = response.json()`, then build one `Reading` per item in
         `payload["readings"]`. Each item has keys `station`, `hour`,
         `celsius`.

    Verify by hand once you have it:
        .venv/bin/pytest starter -q -k fetch_readings
    """
    raise NotImplementedError("exercise 1: fetch and parse the JSON body")


# ---------------------------------------------------------------------------
# Exercise 2 — a failure a human can act on, instead of a traceback
# ---------------------------------------------------------------------------
def describe_failure(response: Any) -> str:
    """Turn a failed response into ONE sentence, with no traceback in it.

    Required format, exactly:
        "HTTP 404 (your request was rejected) — no such station"
        "HTTP 500 (the server failed) — the server fell over"

    Steps:
      1. `response.status_code // 100` gives the family: 3, 4 or 5. Map
         3 -> "redirection", 4 -> "your request was rejected",
         5 -> "the server failed", anything else -> "unexpected status".
      2. If `response.headers.get("Content-Type", "")` starts with
         "application/json", try `response.json()` and append
         f" — {body['detail']}" if the body has a "detail" key, or
         f" — {body['error']}" if it only has "error". Wrap the `.json()`
         call in try/except ValueError: a Content-Type header is a claim,
         not a guarantee.
      3. Return the assembled string. Note the em dash: "—", not "-".
    """
    raise NotImplementedError("exercise 2: describe the failure in one sentence")


# ---------------------------------------------------------------------------
# Exercise 3 — the backoff schedule (used by exercise 4)
# ---------------------------------------------------------------------------
def backoff_delays(
    attempts: int,
    *,
    base: float = 0.5,
    cap: float = 8.0,
    jitter: Callable[[], float] = random.random,
) -> list[float]:
    """Return the waits BETWEEN `attempts` tries — so `attempts - 1` of them.

    Requirements:
      * raise `ValueError` if `attempts < 1`;
      * the raw delay for wait number i (counting from 0) is
        `min(cap, base * 2 ** i)` — 0.5, 1.0, 2.0, 4.0, 8.0, 8.0, ...;
      * multiply each raw delay by `(0.5 + 0.5 * jitter())`, so a wait lands
        somewhere in the top half of its slot, and round to 4 places;
      * `jitter` is a PARAMETER so a test can pass `lambda: 1.0` and get the
        exact schedule, or `lambda: 0.0` and get half of it.

    With `jitter=lambda: 1.0`:
        backoff_delays(1) == []
        backoff_delays(4) == [0.5, 1.0, 2.0]
        backoff_delays(6) == [0.5, 1.0, 2.0, 4.0, 8.0]
    """
    raise NotImplementedError("exercise 3: build the exponential backoff schedule")


# ---------------------------------------------------------------------------
# Exercise 4 — retry on 429 and 5xx, and on nothing else
# ---------------------------------------------------------------------------
def get_with_retry(
    url: str,
    *,
    session: Any,
    attempts: int = 4,
    timeout: tuple[float, float] | float = DEFAULT_TIMEOUT,
    sleep: Callable[[float], None] = time.sleep,
    jitter: Callable[[], float] = random.random,
    params: dict[str, Any] | None = None,
) -> Any:
    """GET `url`, retrying only what deserves it. Return the final response.

    Steps:
      1. `delays = backoff_delays(attempts, jitter=jitter)`.
      2. Loop `for attempt in range(1, attempts + 1)`:
         a. call `session.get(url, params=params, timeout=timeout)` inside
            `try: ... except requests.exceptions.RequestException as exc:`.
            A transport failure has NO status code, because there is no
            response — record it and fall through to the sleep;
         b. if the status is NOT in `RETRY_STATUSES`, RETURN the response
            immediately. A 404 is a final answer, and retrying it is a bug;
         c. otherwise remember the failure. If the response carries a
            `Retry-After` header, use `min(float(header), 8.0)` as this
            wait instead of your computed one — the server knows better
            than you do. Guard the float() with try/except ValueError,
            because Retry-After may legally be a date instead of seconds;
         d. if `attempt <= len(delays)`, call `sleep(delays[attempt - 1])`.
      3. After the loop, raise `ReadingsUnavailable` with a message
         containing the exact text f"after {attempts} attempts".

    `sleep` is a parameter for the Day 74 reason: a test passes a recorder
    and proves the schedule in microseconds without anything waiting.
    """
    raise NotImplementedError("exercise 4: retry 429 and 5xx with backoff")


# ---------------------------------------------------------------------------
# Exercise 5 — a Session with shared headers and a token from the environment
# ---------------------------------------------------------------------------
def make_session(user_agent: str = "day078-yours/1.0 (course exercise)") -> requests.Session:
    """Build a `requests.Session` carrying the headers every call should send.

    Steps:
      1. `session = requests.Session()`.
      2. `session.headers.update({...})` with "User-Agent": user_agent and
         "Accept": "application/json".
      3. Read `os.environ.get("READINGS_TOKEN")`. If it is set, add
         `session.headers["Authorization"] = f"Bearer {token}"`. If it is
         not set, add NO Authorization header at all.
      4. Return the session.

    Never write a token into this file. A secret in source control is a
    secret you have to rotate, and you will not enjoy the afternoon.
    """
    raise NotImplementedError("exercise 5: build the session")


# ---------------------------------------------------------------------------
# Exercise 6 — stream a large body instead of loading it
# ---------------------------------------------------------------------------
def stream_to_file(
    url: str,
    destination: str,
    *,
    session: Any,
    chunk_size: int = 8192,
    timeout: tuple[float, float] | float = DEFAULT_TIMEOUT,
) -> tuple[int, int, str]:
    """Download `url` to `destination` without holding the whole body at once.

    Return (bytes written, number of chunks read, sha256 hex digest).

    Steps:
      1. `digest = hashlib.sha256()`, and counters at zero.
      2. `with session.get(url, stream=True, timeout=timeout) as response:`
         — `stream=True` means the headers have arrived and the body has
         not. Call `response.raise_for_status()` inside the block.
      3. `with open(destination, "wb") as handle:` then
         `for chunk in response.iter_content(chunk_size=chunk_size):`
         — skip a falsy chunk, write it, feed it to the digest, and count
         both bytes and chunks.
      4. Return the three values.

    The point: at no moment does more than one chunk exist in memory. Try
    this against a four-gigabyte file without `stream=True` and your
    process will be killed by the operating system.
    """
    raise NotImplementedError("exercise 6: stream the body a chunk at a time")


# ---------------------------------------------------------------------------
# Provided, complete — the pure part, which needs no network at all.
# ---------------------------------------------------------------------------
def summarise(readings: list[Reading]) -> dict[str, float]:
    values = [r.celsius for r in readings]
    if not values:
        raise ReadingsError("cannot summarise an empty set of readings")
    return {
        "count": float(len(values)),
        "min": min(values),
        "max": max(values),
        "mean": round(sum(values) / len(values), 4),
    }


def _digest_unused() -> str:  # pragma: no cover - keeps hashlib imported for you
    return hashlib.sha256(b"").hexdigest()

"""Provided for you, complete. Do not edit — this is the harness, not the work.

It puts the lab's `examples/` directory on the import path (so you can import
`demo_server` and `fake_session`), and it starts ONE local HTTP server for the
whole test session on an ephemeral port on 127.0.0.1.

Read `examples/demo_server.py` once before you start. It is the thing you are
writing a client against, it is about two hundred lines of standard library,
and knowing what the server does makes every exercise below easier.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Iterator

import pytest

LAB_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(LAB_DIR / "examples"))
sys.path.insert(0, str(Path(__file__).resolve().parent))

from demo_server import CountingServer, base_url, running_server  # noqa: E402


@pytest.fixture(scope="session")
def server() -> Iterator[CountingServer]:
    """The local test server. Started once, shut down when the run ends."""
    with running_server() as srv:
        yield srv


@pytest.fixture(scope="session")
def base(server: CountingServer) -> str:
    """The address it actually bound to this run, e.g. http://127.0.0.1:51234"""
    return base_url(server)

# Your notes — exercise 8

Fill this in after the code works. Sentences, not single words: the writing
is the exercise. Numbers come from your own runs, not from the lesson.

## 1. The status codes you met

Run each of these against the local test server and record what came back.
`pt` is your pytest; use `python3 -c` or the demo, whichever you prefer.

| Endpoint | Status | Retryable? | What your client did |
| --- | --- | --- | --- |
| `/api/readings?station=ALPHA` |  |  |  |
| `/api/readings?station=NOWHERE` |  |  |  |
| `/api/broken` |  |  |  |
| `/old/readings` |  |  |  |
| `/api/flaky` (first call after reset) |  |  |  |

## 2. The timeout

Run the slow endpoint three ways and record the wall-clock time each took:

| Call | Seconds elapsed | What happened |
| --- | --- | --- |
| `timeout=(3.05, 0.4)` against `/api/slow?seconds=3` |  |  |
| `timeout=(3.05, 10.0)` against `/api/slow?seconds=3` |  |  |
| no `timeout=` at all against `/api/slow?seconds=3` |  |  |

Now answer in two or three sentences: what would the third row have done if
the server had accepted the connection and then never sent a single byte?

## 3. Connection reuse

Fill in from your own run of the exercise-5 test:

- five requests through one `Session`: ______ TCP connection(s)
- five calls to `requests.get`: ______ TCP connection(s)

In one sentence: why does this matter more over HTTPS than over plain HTTP?

## 4. Retry

- Your schedule with `jitter=lambda: 1.0` and `attempts=6`: ______
- The same with `jitter=lambda: 0.0`: ______
- Wall-clock time of `test_retry_recovers_from_two_429s_on_the_third_attempt`:
  ______ seconds.

In two or three sentences: why is the jitter there? Describe what happens to
a server that has just come back up when a thousand clients all retry on an
identical, un-jittered schedule.

Then: name one status code you were tempted to retry and should not, and say
what retrying it would actually accomplish.

## 5. Streaming

- Bytes written by exercise 6: ______
- Chunks read: ______
- The most memory your process held at any one moment, approximately: ______

In one sentence: what would have happened if the body had been 4 GB and you
had called `response.content`?

## 6. The boundary — the point of the day

Count these yourself:

| | `test_client.py` (against the server) | your exercise-7 tests (fake session) |
| --- | --- | --- |
| tests |  |  |
| wall-clock seconds |  |  |
| sockets opened |  |  |
| failures they can produce on demand |  |  |

Now the three questions worth writing down properly:

1. What can the server-backed tests prove that the fake-backed ones cannot?
2. What can the fake-backed tests prove that the server-backed ones cannot,
   or can only prove awkwardly?
3. Rewrite `fetch_readings` in your head so it calls `requests.get` directly
   instead of taking `session`. Exactly which of your tests would still be
   possible, and what would you have had to do instead? Answer in sentences,
   and name the Day 74 idea this is an instance of.

## 7. One sentence for a code review

Write the single sentence you would leave on a pull request that adds a call
to a third-party API with no `timeout=` argument.

"""YOUR FILE — exercise 7, plus the checks that grade exercises 1 to 6.

Run it at any time:

    .venv/bin/pytest starter -q

Every test for an unfinished exercise is SKIPPED, so this file exits 0 from
the first minute and turns green one exercise at a time. The skip is
decided by reading your `client.py`: as soon as a function no longer says
`raise NotImplementedError`, its tests start running.

The last section is exercise 7, and it is the point of the whole lab.
"""

from __future__ import annotations

import inspect
import time

import pytest
import requests

import client
from fake_session import FakeResponse, FakeSession


def unfinished(fn) -> bool:
    """True while `fn` still contains its `raise NotImplementedError` line."""
    try:
        return "raise NotImplementedError" in inspect.getsource(fn)
    except OSError:  # pragma: no cover - source always available here
        return False


def needs(fn):
    return pytest.mark.skipif(unfinished(fn), reason=f"{fn.__name__} is not written yet")


class RecordingSleep:
    """A spy from Day 74: it records what it was asked to wait, and waits 0."""

    def __init__(self) -> None:
        self.waits: list[float] = []

    def __call__(self, seconds: float) -> None:
        self.waits.append(round(seconds, 4))


@pytest.fixture
def session():
    """A plain Session while exercise 5 is unfinished; yours once it is."""
    s = requests.Session() if unfinished(client.make_session) else client.make_session()
    yield s
    s.close()


# --- provided, and already passing: proof the server is up ------------------


def test_the_local_test_server_answers(base):
    """This one is written for you. If it fails, nothing else will work."""
    response = requests.get(f"{base}/api/readings", timeout=(3.05, 10.0))
    assert response.status_code == 200
    assert response.json()["count"] == 6


# --- exercise 1 -------------------------------------------------------------


@needs(client.fetch_readings)
def test_fetch_readings_parses_four_rows(base, session):
    readings = client.fetch_readings(base, "ALPHA", session=session)
    assert len(readings) == 4
    assert readings[0] == client.Reading("ALPHA", 0, 12.0)
    assert client.summarise(readings)["mean"] == 17.0


@needs(client.fetch_readings)
def test_fetch_readings_uses_params_so_the_value_is_encoded(base, session):
    awkward = "ALPHA ONE&station=BRAVO"
    echoed = session.get(f"{base}/api/search", params={"station": awkward}, timeout=(3.05, 10.0))
    assert echoed.json()["parsed"]["station"] == [awkward], (
        "if this fails you concatenated the query string instead of using params="
    )


# --- exercise 2 -------------------------------------------------------------


@needs(client.fetch_readings)
def test_an_unknown_station_raises_your_exception_not_a_traceback(base, session):
    with pytest.raises(client.StationNotFound) as caught:
        client.fetch_readings(base, "NOWHERE", session=session)
    assert "NOWHERE" in str(caught.value)


@needs(client.describe_failure)
def test_describe_failure_formats_a_404_and_a_500(base, session):
    missing = session.get(f"{base}/api/missing", timeout=(3.05, 10.0))
    assert client.describe_failure(missing) == (
        "HTTP 404 (your request was rejected) — no such station"
    )
    broken = session.get(f"{base}/api/broken", timeout=(3.05, 10.0))
    assert client.describe_failure(broken) == "HTTP 500 (the server failed) — the server fell over"


# --- exercise 3 -------------------------------------------------------------


@needs(client.backoff_delays)
def test_the_schedule_doubles_caps_and_jitters():
    assert client.backoff_delays(1, jitter=lambda: 1.0) == []
    assert client.backoff_delays(4, jitter=lambda: 1.0) == [0.5, 1.0, 2.0]
    assert client.backoff_delays(6, jitter=lambda: 1.0) == [0.5, 1.0, 2.0, 4.0, 8.0]
    assert client.backoff_delays(4, jitter=lambda: 0.0) == [0.25, 0.5, 1.0]
    with pytest.raises(ValueError):
        client.backoff_delays(0)


# --- exercise 4 -------------------------------------------------------------


@needs(client.get_with_retry)
def test_retry_recovers_from_two_429s_on_the_third_attempt(base, session):
    session.get(f"{base}/control/reset", params={"fail": 2}, timeout=(3.05, 10.0))
    sleeper = RecordingSleep()
    started = time.monotonic()
    response = client.get_with_retry(
        f"{base}/api/flaky", session=session, attempts=4, sleep=sleeper, jitter=lambda: 1.0
    )
    assert response.status_code == 200
    assert response.json()["attempt"] == 3
    assert len(sleeper.waits) == 2
    assert time.monotonic() - started < 0.5, "nothing should really have slept"


@needs(client.get_with_retry)
def test_retry_gives_up_and_names_the_attempt_count(base, session):
    session.get(f"{base}/control/reset", params={"fail": 99}, timeout=(3.05, 10.0))
    with pytest.raises(client.ReadingsUnavailable) as caught:
        client.get_with_retry(
            f"{base}/api/flaky", session=session, attempts=3, sleep=RecordingSleep(),
            jitter=lambda: 1.0,
        )
    assert "after 3 attempts" in str(caught.value)
    session.get(f"{base}/control/reset", params={"fail": 0}, timeout=(3.05, 10.0))


@needs(client.get_with_retry)
def test_a_404_is_returned_at_once_and_never_retried(base, session):
    sleeper = RecordingSleep()
    response = client.get_with_retry(
        f"{base}/api/missing", session=session, attempts=4, sleep=sleeper, jitter=lambda: 1.0
    )
    assert response.status_code == 404
    assert sleeper.waits == []


# --- exercise 5 -------------------------------------------------------------


@needs(client.make_session)
def test_the_session_sends_your_user_agent_and_no_token_by_default(base, monkeypatch):
    monkeypatch.delenv("READINGS_TOKEN", raising=False)
    with client.make_session() as s:
        echoed = s.post(f"{base}/api/echo", json={}, timeout=(3.05, 10.0))
    assert echoed.json()["user_agent"].startswith("day078-")
    assert echoed.json()["authorization_seen"] is False


@needs(client.make_session)
def test_a_token_in_the_environment_becomes_an_authorization_header(base, monkeypatch):
    monkeypatch.setenv("READINGS_TOKEN", "not-a-real-secret")
    with client.make_session() as s:
        echoed = s.post(f"{base}/api/echo", json={}, timeout=(3.05, 10.0))
    assert echoed.json()["authorization_seen"] is True


@needs(client.make_session)
def test_one_session_opens_one_connection_for_five_requests(base, server):
    before = server.connections
    with client.make_session() as s:
        for _ in range(5):
            s.get(f"{base}/api/readings", timeout=(3.05, 10.0)).close()
    assert server.connections - before == 1
    before = server.connections
    for _ in range(5):
        requests.get(f"{base}/api/readings", timeout=(3.05, 10.0)).close()
    assert server.connections - before == 5


# --- exercise 6 -------------------------------------------------------------


@needs(client.stream_to_file)
def test_streaming_writes_512_kib_in_64_chunks(base, session, tmp_path):
    destination = tmp_path / "large.txt"
    total, chunks, digest = client.stream_to_file(
        f"{base}/api/large?kb=512", str(destination), session=session, chunk_size=8192
    )
    assert total == 512 * 1024
    assert chunks == 64
    assert destination.stat().st_size == total
    assert len(digest) == 64


# --- the timeout, which needs no exercise: it is one keyword ----------------


def test_a_timeout_really_fires_against_the_slow_endpoint(base, session):
    """Provided and already passing. Read it — this is the habit of the day."""
    started = time.monotonic()
    with pytest.raises(requests.exceptions.Timeout):
        session.get(f"{base}/api/slow", params={"seconds": 3}, timeout=(3.05, 0.4))
    assert time.monotonic() - started < 2.0


# --- exercise 7 — the payoff: tests with NO server at all -------------------
#
# `fake_session.FakeSession` is provided in examples/. Because every function
# in client.py takes `session` as a parameter, these tests need no server, no
# socket and no port. Write at least three more of your own below:
#
#   * one that proves a ConnectionError propagates (script the exception);
#   * one that proves a 503 is retried and a 403 is not;
#   * one that proves EVERY call your client makes carries a timeout.
#
# The last one is the check worth stealing for your real projects.


@needs(client.fetch_readings)
def test_fetch_readings_works_with_no_server_in_sight():
    """Provided, as the model for the ones you write."""
    payload = {"readings": [{"station": "ALPHA", "hour": 0, "celsius": 12.0}]}
    fake = FakeSession([FakeResponse(200, payload)])
    assert client.fetch_readings("http://example.invalid", "ALPHA", session=fake) == [
        client.Reading("ALPHA", 0, 12.0)
    ]
    assert fake.calls[0]["params"] == {"station": "ALPHA"}
    assert fake.timeouts == [(3.05, 10.0)], "every call must carry a timeout"

#!/usr/bin/env bash
# Tests for the Day 078 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# The suite asserts real behaviour, not the presence of files. The five checks
# worth reading before the rest:
#
#   * "the whole example suite passes with the internet blocked" runs every
#     example test under tests/sitecustomize.py, which replaces
#     socket.connect and socket.getaddrinfo so that any attempt to resolve a
#     name or reach a non-loopback address raises. That turns "this lab is
#     offline" from a promise into a fact;
#   * "the fake-session suite passes with no server module present at all"
#     copies three files to a temporary directory — client.py,
#     fake_session.py and test_without_a_server.py — leaving demo_server.py
#     and conftest.py behind, and runs pytest there. It passes because every
#     client function takes `session` as a parameter. That is Day 74's
#     argument, proved rather than asserted;
#   * "a read timeout really fires" asserts that a request against a
#     three-second endpoint gives up in well under two seconds;
#   * "retry succeeds on exactly the third attempt" pins the count, not just
#     the outcome;
#   * "one Session opens one connection where five bare calls open five"
#     reads the server's own accept counter.
#
# No network, non-interactive, deterministic. Exits 0 only if every check
# passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
failures=0
checks=0

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

# Resolve a tool: an explicit override, then this lab's .venv, then PATH.
# Fails loudly with install instructions rather than skipping silently.
resolve_tool() {
  local tool="$1" override="$2"
  if [ -n "${override}" ] && [ -x "${override}" ]; then echo "${override}"; return 0; fi
  if [ -x "${lab_dir}/.venv/bin/${tool}" ]; then echo "${lab_dir}/.venv/bin/${tool}"; return 0; fi
  if command -v "${tool}" >/dev/null 2>&1; then command -v "${tool}"; return 0; fi
  return 1
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  echo "  Install this lab's dependencies with:" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
  echo "  Or point this suite at an existing pytest: PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
  exit 1
}

# python3 must be the SAME environment pytest lives in, or `import requests`
# will fail in the demo scripts. Prefer the venv interpreter beside pytest.
python_bin="$(dirname "${pytest_bin}")/python3"
[ -x "${python_bin}" ] || python_bin="$(command -v python3 || true)"
if [ -z "${python_bin}" ]; then
  echo "FAIL: python3 not found on PATH." >&2
  exit 1
fi

if ! "${python_bin}" -c "import requests" >/dev/null 2>&1; then
  echo "FAIL: the 'requests' package is not importable by ${python_bin}." >&2
  echo "  Install it with: .venv/bin/pip install -r requirements/requirements.txt" >&2
  exit 1
fi

echo "Day 078 — Talk to a Server You Control"
echo

# --------------------------------------------------------------------------
echo "1. The tools"
# --------------------------------------------------------------------------

version_line="$("${pytest_bin}" --version 2>&1 | head -1)"
case "${version_line}" in
  pytest*) check "pytest --version reports a pytest ( ${version_line} )" "yes" ;;
  *) check "pytest --version reports a pytest ( ${version_line} )" "no" ;;
esac

requests_version="$("${python_bin}" -c "import requests; print(requests.__version__)" 2>&1)"
case "${requests_version}" in
  2.*) check "requests is importable ( ${requests_version} )" "yes" ;;
  *) check "requests is importable ( ${requests_version} )" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "2. The local test server behaves as the lab claims"
# --------------------------------------------------------------------------

server_probe="$(cd "${lab_dir}/examples" && "${python_bin}" - <<'PY' 2>&1
import json
import requests
from demo_server import base_url, running_server

with running_server() as srv:
    root = base_url(srv)
    host, port = srv.server_address[:2]
    s = requests.Session()
    out = {
        "host": host,
        "ephemeral": port != 0 and port != 8000,
        "readings": s.get(f"{root}/api/readings", timeout=5).status_code,
        "missing": s.get(f"{root}/api/missing", timeout=5).status_code,
        "broken": s.get(f"{root}/api/broken", timeout=5).status_code,
        "redirect": s.get(f"{root}/old/readings", timeout=5, allow_redirects=False).status_code,
        "echo": s.post(f"{root}/api/echo", json={"a": 1}, timeout=5).status_code,
    }
    s.get(f"{root}/control/reset", params={"fail": 1}, timeout=5)
    limited = s.get(f"{root}/api/flaky", timeout=5)
    out["flaky"] = limited.status_code
    out["retry_after"] = limited.headers.get("Retry-After")
    out["large"] = len(s.get(f"{root}/api/large?kb=8", timeout=5).content)
    s.close()
print(json.dumps(out))
PY
)"

expect_field() {
  local field="$1" want="$2" label="$3"
  case "${server_probe}" in
    *"\"${field}\": ${want}"*|*"\"${field}\": \"${want}\""*) check "${label}" "yes" ;;
    *) check "${label} (probe said: ${server_probe})" "no" ;;
  esac
}

expect_field host "127.0.0.1" "the server binds 127.0.0.1 and nothing else"
expect_field ephemeral "true" "the port is ephemeral, not a hard-coded 8000"
expect_field readings "200" "/api/readings answers 200"
expect_field missing "404" "/api/missing answers 404"
expect_field broken "500" "/api/broken answers 500"
expect_field redirect "301" "/old/readings answers 301"
expect_field echo "201" "POST /api/echo answers 201"
expect_field flaky "429" "/api/flaky answers 429 once armed"
expect_field retry_after "1" "the 429 carries a Retry-After header"
expect_field large "8192" "/api/large?kb=8 returns exactly 8192 bytes"

# --------------------------------------------------------------------------
echo
echo "3. The demonstrations run"
# --------------------------------------------------------------------------

raw_out="$(cd "${lab_dir}" && "${python_bin}" examples/raw_socket_demo.py 2>&1)"
raw_exit=$?
if [ "${raw_exit}" -eq 0 ]; then
  check "examples/raw_socket_demo.py exits 0" "yes"
else
  check "examples/raw_socket_demo.py exits 0 (got ${raw_exit})" "no"
fi
case "${raw_out}" in
  *"GET /api/readings?station=ALPHA HTTP/1.1"*)
    check "the hand-typed request line really is HTTP text" "yes" ;;
  *) check "the hand-typed request line really is HTTP text" "no" ;;
esac
case "${raw_out}" in
  *"HTTP/1.1 200 OK"*) check "the raw response begins with a status line" "yes" ;;
  *) check "the raw response begins with a status line" "no" ;;
esac

stdlib_out="$(cd "${lab_dir}" && "${python_bin}" examples/stdlib_demo.py 2>&1)"
stdlib_exit=$?
if [ "${stdlib_exit}" -eq 0 ]; then
  check "examples/stdlib_demo.py exits 0 — the standard library really can do this" "yes"
else
  check "examples/stdlib_demo.py exits 0 (got ${stdlib_exit})" "no"
fi
case "${stdlib_out}" in
  *"raised      : HTTPError"*)
    check "urllib.request raises on a 404 where requests returns a response" "yes" ;;
  *) check "urllib.request raises on a 404 where requests returns a response" "no" ;;
esac
case "${stdlib_out}" in
  *"connections : 1 opened for those 2 requests"*)
    check "http.client reuses one connection for two requests" "yes" ;;
  *) check "http.client reuses one connection for two requests" "no" ;;
esac

demo_out="$(cd "${lab_dir}" && "${python_bin}" examples/demo.py 2>&1)"
demo_exit=$?
if [ "${demo_exit}" -eq 0 ]; then
  check "examples/demo.py exits 0" "yes"
else
  check "examples/demo.py exits 0 (got ${demo_exit})" "no"
  printf '%s\n' "${demo_out}" | tail -20
fi
case "${demo_out}" in
  *"5 calls, one Session      : 1 TCP connection(s)"*)
    check "demo.py shows one Session using one connection for five calls" "yes" ;;
  *) check "demo.py shows one Session using one connection for five calls" "no" ;;
esac
case "${demo_out}" in
  *"5 calls, requests.get()   : 5 TCP connection(s)"*)
    check "demo.py shows five bare calls opening five connections" "yes" ;;
  *) check "demo.py shows five bare calls opening five connections" "no" ;;
esac
case "${demo_out}" in
  *"ReadTimeout"*) check "demo.py's timeout section really raises ReadTimeout" "yes" ;;
  *) check "demo.py's timeout section really raises ReadTimeout" "no" ;;
esac
case "${demo_out}" in
  *"station=ALPHA+ONE%26station%3DBRAVO"*)
    check "params= percent-encodes a value containing a space and an ampersand" "yes" ;;
  *) check "params= percent-encodes a value containing a space and an ampersand" "no" ;;
esac
case "${demo_out}" in
  *"chunks read      : 64 of at most 8192 bytes"*)
    check "streaming reads 512 KiB as 64 chunks, not one body" "yes" ;;
  *) check "streaming reads 512 KiB as 64 chunks, not one body" "no" ;;
esac

# httpx is NOT a dependency of this lab. If it happens to be installed the
# comparison runs; if not, the demo says so and exits 0 either way.
httpx_out="$(cd "${lab_dir}" && "${python_bin}" examples/httpx_demo.py 2>&1)"
httpx_exit=$?
if [ "${httpx_exit}" -eq 0 ]; then
  check "examples/httpx_demo.py exits 0 whether or not httpx is installed" "yes"
else
  check "examples/httpx_demo.py exits 0 (got ${httpx_exit})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "4. The reference suite"
# --------------------------------------------------------------------------

examples_out="$(cd "${lab_dir}" && "${pytest_bin}" examples -q -p no:cacheprovider 2>&1)"
examples_exit=$?
if [ "${examples_exit}" -eq 0 ]; then
  check "pytest examples exits 0" "yes"
else
  check "pytest examples exits 0 (got ${examples_exit})" "no"
  printf '%s\n' "${examples_out}" | tail -25
fi
case "${examples_out}" in
  *"48 passed"*) check "pytest examples reports 48 passed" "yes" ;;
  *) check "pytest examples reports 48 passed" "no" ;;
esac

for selection in \
  "a_read_timeout_fires_against_the_slow_endpoint" \
  "retry_succeeds_after_exactly_the_expected_number_of_attempts" \
  "a_missing_station_raises_a_domain_error_with_a_clean_message" \
  "one_session_reuses_one_connection_for_many_requests" \
  "streaming_writes_the_whole_body_in_many_small_chunks"
do
  if (cd "${lab_dir}" && "${pytest_bin}" examples -q -p no:cacheprovider \
        -k "${selection}" >/dev/null 2>&1); then
    check "behaviour asserted: ${selection}" "yes"
  else
    check "behaviour asserted: ${selection}" "no"
  fi
done

# The timeout check is worth timing as well as running: it must give up on
# its own schedule (0.4 s) rather than waiting for the server's 3 s.
timeout_seconds="$(cd "${lab_dir}" && "${python_bin}" - <<'PY' 2>&1
import time
import requests
from pathlib import Path
import sys
sys.path.insert(0, str(Path("examples").resolve()))
from demo_server import base_url, running_server

with running_server() as srv:
    started = time.monotonic()
    try:
        requests.get(f"{base_url(srv)}/api/slow", params={"seconds": 3}, timeout=(3.05, 0.4))
        print("NO-TIMEOUT-RAISED")
    except requests.exceptions.Timeout:
        print(f"{time.monotonic() - started:.2f}")
PY
)"
awk_ok="$("${python_bin}" -c "
v='''${timeout_seconds}'''.strip()
try:
    print('yes' if 0.1 < float(v) < 2.0 else 'no')
except ValueError:
    print('no')
")"
check "a read timeout of 0.4s fires in ${timeout_seconds}s against a 3s endpoint" "${awk_ok}"

# --------------------------------------------------------------------------
echo
echo "5. Nothing here touches the internet — proved, not promised"
# --------------------------------------------------------------------------

guarded_out="$(cd "${lab_dir}" && PYTHONPATH="${lab_dir}/tests" \
  "${pytest_bin}" examples -q -p no:cacheprovider 2>&1)"
guarded_exit=$?
if [ "${guarded_exit}" -eq 0 ]; then
  check "the whole example suite passes with all non-loopback sockets blocked" "yes"
else
  check "the whole example suite passes with all non-loopback sockets blocked" "no"
  printf '%s\n' "${guarded_out}" | tail -25
fi

# And the guard is not vacuous: with it loaded, a real request must fail.
if (cd "${lab_dir}" && PYTHONPATH="${lab_dir}/tests" "${python_bin}" -c "
import requests
requests.get('https://example.com', timeout=2)
" >/dev/null 2>&1); then
  check "the offline guard is real (a request to a public site is blocked)" "no"
else
  check "the offline guard is real (a request to a public site is blocked)" "yes"
fi

# No lab file may name a real remote host. `example.invalid` is allowed: the
# .invalid top-level domain is reserved by RFC 2606 precisely so that it can
# never resolve, which is why the fake-session tests use it.
offenders="$(grep -rn 'http://\|https://' "${lab_dir}/examples" "${lab_dir}/starter" \
  --include='*.py' \
  | grep -v '127\.0\.0\.1' | grep -v 'example\.invalid' | grep -v '{base' | grep -v '{root' \
  | grep -v '{host}' || true)"
if [ -z "${offenders}" ]; then
  check "no example or starter file names a real remote host" "yes"
else
  check "no example or starter file names a real remote host" "no"
  printf '%s\n' "${offenders}"
fi

if grep -rn ':8000' "${lab_dir}/examples" "${lab_dir}/starter" >/dev/null 2>&1; then
  check "no file hard-codes port 8000 (the collision waiting to happen)" "no"
else
  check "no file hard-codes port 8000 (the collision waiting to happen)" "yes"
fi

# --------------------------------------------------------------------------
echo
echo "6. The Day 74 payoff: tests that need no server at all"
# --------------------------------------------------------------------------

# Copy three files — and deliberately NOT demo_server.py or conftest.py — to a
# temporary directory, and run the fake-session suite there. If it passes, the
# client genuinely has an injectable boundary.
solo="$(mktemp -d "${TMPDIR:-/tmp}/day078-solo.XXXXXX")"
cp "${lab_dir}/examples/client.py" "${lab_dir}/examples/fake_session.py" \
   "${lab_dir}/examples/test_without_a_server.py" "${solo}/"
solo_out="$(cd "${solo}" && "${pytest_bin}" . -q -p no:cacheprovider 2>&1)"
solo_exit=$?
if [ "${solo_exit}" -eq 0 ]; then
  check "the fake-session suite passes with no server module present at all" "yes"
else
  check "the fake-session suite passes with no server module present at all" "no"
  printf '%s\n' "${solo_out}" | tail -25
fi
case "${solo_out}" in
  *"20 passed"*) check "that suite is 20 real tests, not a placeholder" "yes" ;;
  *) check "that suite is 20 real tests, not a placeholder (got: $(printf '%s' "${solo_out}" | tail -1))" "no" ;;
esac
if [ -e "${solo}/demo_server.py" ] || [ -e "${solo}/conftest.py" ]; then
  check "the isolated directory really lacks the server" "no"
else
  check "the isolated directory really lacks the server" "yes"
fi
# And it must run under the offline guard too.
if (cd "${solo}" && PYTHONPATH="${lab_dir}/tests" "${pytest_bin}" . -q -p no:cacheprovider \
      >/dev/null 2>&1); then
  check "and it passes with every non-loopback socket blocked" "yes"
else
  check "and it passes with every non-loopback socket blocked" "no"
fi
rm -rf "${solo}"

# The signatures that make all of that possible.
signature_report="$("${python_bin}" - "${lab_dir}" <<'PY' 2>&1
import inspect
import sys
sys.path.insert(0, f"{sys.argv[1]}/examples")
import client

problems = []
for name in ("fetch_readings", "get_with_retry", "stream_to_file"):
    params = inspect.signature(getattr(client, name)).parameters
    if "session" not in params:
        problems.append(f"{name} has no session parameter")
    if params.get("session") and params["session"].kind is not inspect.Parameter.KEYWORD_ONLY:
        problems.append(f"{name}'s session is not keyword-only")
for name in ("get_with_retry", "backoff_delays"):
    params = inspect.signature(getattr(client, name)).parameters
    if "jitter" not in params:
        problems.append(f"{name} has no injectable jitter")
if "sleep" not in inspect.signature(client.get_with_retry).parameters:
    problems.append("get_with_retry has no injectable sleep")
print("OK" if not problems else "; ".join(problems))
PY
)"
if [ "${signature_report}" = "OK" ]; then
  check "every networked function takes session (and sleep, and jitter) as parameters" "yes"
else
  check "every networked function takes session as a parameter (${signature_report})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "7. Retry policy: the statuses, checked one at a time"
# --------------------------------------------------------------------------

policy="$("${python_bin}" - "${lab_dir}" <<'PY' 2>&1
import sys
sys.path.insert(0, f"{sys.argv[1]}/examples")
from client import RETRY_STATUSES

should = {429, 500, 502, 503, 504}
should_not = {200, 201, 204, 301, 304, 400, 401, 403, 404, 409, 422}
wrong = [s for s in should if s not in RETRY_STATUSES]
wrong += [s for s in should_not if s in RETRY_STATUSES]
print("OK" if not wrong else f"wrong: {sorted(wrong)}")
PY
)"
if [ "${policy}" = "OK" ]; then
  check "429 and 5xx are retryable; 4xx and every success code are not" "yes"
else
  check "429 and 5xx are retryable; 4xx and every success code are not (${policy})" "no"
fi

schedule="$("${python_bin}" - "${lab_dir}" <<'PY' 2>&1
import sys
sys.path.insert(0, f"{sys.argv[1]}/examples")
from client import backoff_delays

full = backoff_delays(6, jitter=lambda: 1.0)
half = backoff_delays(4, jitter=lambda: 0.0)
ok = full == [0.5, 1.0, 2.0, 4.0, 8.0] and half == [0.25, 0.5, 1.0]
try:
    backoff_delays(0)
    ok = False
except ValueError:
    pass
print("OK" if ok else f"full={full} half={half}")
PY
)"
if [ "${schedule}" = "OK" ]; then
  check "the backoff doubles, caps at 8s, jitters into the top half of each slot" "yes"
else
  check "the backoff schedule is wrong (${schedule})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "8. Your work in starter/"
# --------------------------------------------------------------------------

starter_out="$(cd "${lab_dir}" && "${pytest_bin}" starter -q -p no:cacheprovider 2>&1)"
starter_exit=$?
if [ "${starter_exit}" -eq 0 ]; then
  check "pytest starter exits 0" "yes"
else
  check "pytest starter exits 0 (got ${starter_exit})" "no"
  printf '%s\n' "${starter_out}" | tail -25
fi

if grep -q 'raise NotImplementedError' "${lab_dir}/starter/client.py"; then
  echo "  (exercises unfinished — structural checks only)"
  for fn in fetch_readings describe_failure backoff_delays get_with_retry \
            make_session stream_to_file
  do
    if grep -q "^def ${fn}(\|^def ${fn}$" "${lab_dir}/starter/client.py"; then
      check "starter/client.py defines ${fn} for you to fill in" "yes"
    else
      check "starter/client.py defines ${fn} for you to fill in" "no"
    fi
  done
  case "${starter_out}" in
    *skipped*) check "unfinished exercises are skipped, so the suite is green from minute one" "yes" ;;
    *) check "unfinished exercises are skipped, so the suite is green from minute one" "no" ;;
  esac
  if grep -q 'exercise 7' "${lab_dir}/starter/test_client.py"; then
    check "starter/test_client.py carries the exercise-7 fake-session section" "yes"
  else
    check "starter/test_client.py carries the exercise-7 fake-session section" "no"
  fi
else
  echo "  (exercises finished — behavioural checks)"
  case "${starter_out}" in
    *skipped*) check "no exercise is still skipped" "no" ;;
    *) check "no exercise is still skipped" "yes" ;;
  esac
  # Your client must pass the reference suite's own fake-session tests.
  yours="$(mktemp -d "${TMPDIR:-/tmp}/day078-yours.XXXXXX")"
  cp "${lab_dir}/starter/client.py" "${lab_dir}/examples/fake_session.py" \
     "${lab_dir}/examples/test_without_a_server.py" "${yours}/"
  if (cd "${yours}" && "${pytest_bin}" . -q -p no:cacheprovider >/dev/null 2>&1); then
    check "YOUR client passes the reference fake-session suite, with no server" "yes"
  else
    check "YOUR client passes the reference fake-session suite, with no server" "no"
    (cd "${yours}" && "${pytest_bin}" . -q -p no:cacheprovider 2>&1 | tail -20)
  fi
  rm -rf "${yours}"
  if grep -qE '^\s*-\s*five requests through one .Session.: _{4,}' "${lab_dir}/starter/NOTES.md"; then
    check "starter/NOTES.md is filled in rather than left with blanks" "no"
  else
    check "starter/NOTES.md is filled in rather than left with blanks" "yes"
  fi
fi

# --------------------------------------------------------------------------
echo
echo "9. The captured output matches what the code does now"
# --------------------------------------------------------------------------

for capture in sample-run.txt pytest-runs.txt test-run.txt FIELDS.md; do
  if [ -s "${lab_dir}/expected-output/${capture}" ]; then
    check "expected-output/${capture} exists and is not empty" "yes"
  else
    check "expected-output/${capture} exists and is not empty" "no"
  fi
done

if grep -q '48 passed' "${lab_dir}/expected-output/pytest-runs.txt" 2>/dev/null; then
  check "the captured pytest run agrees with today's count of 48" "yes"
else
  check "the captured pytest run agrees with today's count of 48" "no"
fi

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]

"""A guard that makes "this lab never touches the internet" a fact, not a claim.

Python imports `sitecustomize` automatically at interpreter start-up if it can
find it on the import path. `tests/run_tests.sh` puts this directory on
PYTHONPATH for one check, then runs the whole example suite under it. Any
attempt to resolve a hostname or connect to an address that is not the
loopback interface raises immediately and fails the run.

If you ever add an example that talks to a real site, this file will tell you
in the least ambiguous way available.
"""

from __future__ import annotations

import socket

LOOPBACK = {"127.0.0.1", "::1", "0.0.0.0", ""}

_real_connect = socket.socket.connect
_real_connect_ex = socket.socket.connect_ex
_real_getaddrinfo = socket.getaddrinfo


class NetworkBlocked(RuntimeError):
    """Raised instead of opening a connection to anything but the loopback."""


def _check(address: object) -> None:
    if isinstance(address, tuple) and address:
        host = address[0]
        if isinstance(host, str) and host not in LOOPBACK:
            raise NetworkBlocked(f"blocked a connection to {host!r} — this lab is offline")


def _connect(self, address):  # type: ignore[no-untyped-def]
    _check(address)
    return _real_connect(self, address)


def _connect_ex(self, address):  # type: ignore[no-untyped-def]
    _check(address)
    return _real_connect_ex(self, address)


def _getaddrinfo(host, *args, **kwargs):  # type: ignore[no-untyped-def]
    if isinstance(host, str) and host not in LOOPBACK:
        raise NetworkBlocked(f"blocked a name lookup for {host!r} — this lab is offline")
    return _real_getaddrinfo(host, *args, **kwargs)


socket.socket.connect = _connect  # type: ignore[method-assign]
socket.socket.connect_ex = _connect_ex  # type: ignore[method-assign]
socket.getaddrinfo = _getaddrinfo  # type: ignore[assignment]