Hands-on lab — Day 134: Finding Data: Open Datasets and APIs

"""Shared fixtures: two mock API instances with different rate-limit patience.

``mock_api`` relents after 2 requests -- the "source that eventually lets
you through" case. ``stubborn_mock_api`` relents after 10, which is more
than any test's attempt budget -- the "source that never relents within a
sane budget" case. Both are started and stopped inside the fixture, so a
test that never touches them never binds a port, and one that does leaves
nothing running when it finishes.
"""

import pytest

from mock_server import serve_mock_api


@pytest.fixture
def mock_api():
    with serve_mock_api(rate_limit_trigger_count=2) as api:
        yield api


@pytest.fixture
def stubborn_mock_api():
    with serve_mock_api(rate_limit_trigger_count=10) as api:
        yield api

"""The client and judgement functions Day 134's exercises test.

Everything here runs on the standard library's ``urllib.request`` -- no
third-party HTTP client is a dependency of this lab. Day 135 owns turning
the JSON these functions return into a tidy DataFrame; this module stops at
"assembled rows" and "a verdict about whether the source deserves your
time in the first place".
"""

from __future__ import annotations

import hashlib
import json
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


class RateLimitExceeded(RuntimeError):
    """Raised when a source keeps answering 429 past the retry budget.

    A client that retries forever is not polite, it is a denial-of-service
    tool pointed at someone else's server. Giving up loudly, with a count
    of how many times it tried, is the correct behaviour.
    """


def fetch_raw(url: str, headers: dict[str, str] | None = None) -> tuple[int, dict[str, str], bytes]:
    """One GET via ``urllib.request``. Returns ``(status, headers, body)``.

    ``urllib.request`` raises ``HTTPError`` for any status >= 400, including
    304 in some circumstances and always 429 -- this unwraps that so callers
    can inspect the status code like any other response instead of writing
    a try/except around every call site.
    """
    request = urllib.request.Request(url, headers=headers or {})
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            return response.status, dict(response.headers), response.read()
    except urllib.error.HTTPError as exc:
        return exc.code, dict(exc.headers or {}), exc.read()


def fetch_all_pages(base_url: str, path: str = "/dataset") -> list[dict]:
    """Follow ``has_more`` until the source itself says stop.

    Never loops a fixed number of times and never trusts a total handed to
    it in advance -- the stopping condition is the server's own word for
    "nothing left", which is the only thing that can't drift out of sync
    with what actually got paginated.
    """
    rows: list[dict] = []
    page = 1
    while True:
        status, _, body = fetch_raw(f"{base_url}{path}?page={page}")
        if status != 200:
            raise RuntimeError(f"unexpected status {status} fetching page {page}")
        payload = json.loads(body)
        rows.extend(payload["items"])
        if not payload.get("has_more"):
            break
        page += 1
    return rows


def fetch_with_backoff(
    base_url: str,
    path: str,
    max_attempts: int = 5,
    base_delay: float = 0.01,
) -> tuple[bytes, int]:
    """GET with bounded exponential backoff on 429. Returns ``(body, attempts)``.

    Attempts are capped at ``max_attempts``; a source that never relents
    raises ``RateLimitExceeded`` rather than looping forever.
    """
    attempts = 0
    while attempts < max_attempts:
        attempts += 1
        status, headers, body = fetch_raw(f"{base_url}{path}")
        if status == 200:
            return body, attempts
        if status == 429:
            server_hint = headers.get("Retry-After")
            delay = float(server_hint) if server_hint not in (None, "0") else base_delay * (2 ** (attempts - 1))
            time.sleep(delay)
            continue
        raise RuntimeError(f"unexpected status {status}")
    raise RateLimitExceeded(f"gave up after {attempts} attempts, source still rate-limiting")


@dataclass
class CacheEntry:
    """One cached response, keyed by the ETag the source sent with it."""

    etag: str
    body: bytes


def fetch_with_etag(
    base_url: str, path: str, cache: dict[str, CacheEntry]
) -> tuple[bytes, bool, int]:
    """Conditional GET. Returns ``(body, served_from_cache, bytes_over_wire)``.

    The second call for the same path sends ``If-None-Match`` with the
    stored ETag. A ``304`` means the cached copy is still current: the
    caller gets it back with zero bytes counted against the wire, which is
    the entire point of a conditional request -- a re-run costs nothing.
    """
    headers = {}
    cached = cache.get(path)
    if cached is not None:
        headers["If-None-Match"] = cached.etag

    status, response_headers, body = fetch_raw(f"{base_url}{path}", headers=headers)

    if status == 304:
        if cached is None:
            raise RuntimeError("received 304 with nothing cached to serve")
        return cached.body, True, len(body)

    if status == 200:
        etag = response_headers.get("ETag", "")
        cache[path] = CacheEntry(etag=etag, body=body)
        return body, False, len(body)

    raise RuntimeError(f"unexpected status {status}")


def sha256_of(path: Path) -> str:
    """The SHA-256 hex digest of a file's bytes, for pinning a download."""
    return hashlib.sha256(path.read_bytes()).hexdigest()


@dataclass
class SourceVerdict:
    """The result of the five-minute assessment: what's here, what's missing."""

    granularity: str | None
    coverage: str | None
    licence: str | None
    dictionary_present: bool
    problems: list[str] = field(default_factory=list)

    @property
    def ready(self) -> bool:
        return not self.problems


def assess_source(metadata: dict[str, Any]) -> SourceVerdict:
    """Run the five-minute checklist against a source's stated metadata.

    Checks presence, not truth -- it cannot tell you the stated granularity
    is accurate, only that the source bothered to state one. That is still
    the majority of what separates a source worth trusting from one that
    is not, because most low-quality sources fail at "bothered to state it".
    """
    problems: list[str] = []
    granularity = metadata.get("granularity")
    coverage = metadata.get("coverage")
    licence = metadata.get("licence")
    dictionary_present = bool(metadata.get("dictionary"))

    if not granularity:
        problems.append("no stated granularity")
    if not coverage:
        problems.append("no stated coverage")
    if not licence:
        problems.append("no stated licence")
    if not dictionary_present:
        problems.append("no data dictionary")
    if not metadata.get("update_cadence"):
        problems.append("no update cadence")
    if "known_issues" not in metadata:
        problems.append("known issues undocumented")

    return SourceVerdict(
        granularity=granularity,
        coverage=coverage,
        licence=licence,
        dictionary_present=dictionary_present,
        problems=problems,
    )


REDISTRIBUTABLE_LICENCES = {"CC0", "CC-BY", "CC-BY-4.0", "ODbL"}
NON_REDISTRIBUTABLE_LICENCES = {"All rights reserved", "Proprietary"}


def check_licence(licence: str, purpose: str = "redistribution") -> dict[str, Any]:
    """Whether ``licence`` permits ``purpose``, with the reason spelled out.

    Returns a reason rather than a bare boolean on purpose: "allowed" and
    "allowed, with attribution required" are both True and both very
    different obligations for whoever ships the result.
    """
    if licence in REDISTRIBUTABLE_LICENCES:
        reason = f"{licence} permits {purpose}"
        if licence.startswith("CC-BY"):
            reason += ", with attribution to the source"
        if licence == "ODbL":
            reason += ", with share-alike terms for the derived database"
        return {"allowed": True, "reason": reason}

    if licence in NON_REDISTRIBUTABLE_LICENCES:
        return {
            "allowed": False,
            "reason": f"{licence} forbids {purpose}; analysis of the data is not the same permission as republishing it",
        }

    return {
        "allowed": False,
        "reason": f"unrecognised licence '{licence}' -- treat as not redistributable until confirmed",
    }


def check_coverage(dictionary: dict[str, Any], data_keys: set[str]) -> dict[str, Any]:
    """Compare a dataset's actual keys against the dictionary's expected list.

    Detects a gap by comparing sets, not by eyeballing a chart -- a region
    with zero rows looks identical to a region that was never collected
    unless something checks for its *absence* from the key list.
    """
    expected = set(dictionary.get("expected_regions", []))
    missing = sorted(expected - data_keys)
    return {"complete": not missing, "missing": missing, "expected": sorted(expected)}


def definitions_match(dictionary_a: dict[str, Any], dictionary_b: dict[str, Any], column: str) -> bool:
    """Whether two dictionaries define ``column`` the same way."""
    definition_a = dictionary_a["fields"][column]["definition"]
    definition_b = dictionary_b["fields"][column]["definition"]
    return definition_a == definition_b


def naive_join_check(series_a, series_b) -> dict[str, Any]:
    """What most people check before joining two columns: dtype and range.

    Deliberately weak. This is the check that passes on the two
    ``unemployment_rate`` columns the day's opening story describes, which
    is exactly why it is not sufficient on its own.
    """
    dtype_match = series_a.dtype == series_b.dtype
    range_a = (series_a.min(), series_a.max())
    range_b = (series_b.min(), series_b.max())
    ranges_overlap = not (range_a[1] < range_b[0] or range_b[1] < range_a[0])
    return {
        "dtype_match": dtype_match,
        "ranges_overlap": ranges_overlap,
        "would_pass_naive_check": dtype_match and ranges_overlap,
    }


def dictionary_aware_join_check(
    dictionary_a: dict[str, Any], dictionary_b: dict[str, Any], column: str
) -> dict[str, Any]:
    """The check the naive one is missing: do the two sources mean the same thing?"""
    same_definition = definitions_match(dictionary_a, dictionary_b, column)
    if same_definition:
        reason = "definitions match"
    else:
        definition_a = dictionary_a["fields"][column]["definition"]
        definition_b = dictionary_b["fields"][column]["definition"]
        reason = f"definitions differ: {definition_a!r} vs {definition_b!r}"
    return {"same_definition": same_definition, "safe_to_join": same_definition, "reason": reason}


def record_provenance(url: str, checksum: str, retrieved_at: datetime | None = None) -> dict[str, str]:
    """Build the record that makes a download reproducible: url, when, checksum.

    ``retrieved_at`` defaults to now, which is exactly the part that would
    make two calls compare unequal in a test -- callers that need a stable
    comparison (this lab's tests included) pass a fixed timestamp in rather
    than letting the clock make the suite flaky.
    """
    stamp = retrieved_at or datetime.now(timezone.utc)
    return {
        "url": url,
        "retrieved_at": stamp.replace(microsecond=0).isoformat(),
        "sha256": checksum,
    }

"""Fixture data for Day 134's exercises. Not itself an exercise -- read
alongside ``00_brief.md`` when the tests reference these names.

``DICTIONARY_A`` and ``DICTIONARY_B`` are two data dictionaries (codebooks)
for two fictional sources that both publish a column called
``unemployment_rate``. The columns below are built so that a naive check
-- same dtype, overlapping range -- passes on both, and only the
dictionaries' prose reveals that they are not the same measurement. That
gap is the lesson's opening failure story, made into fixtures.
"""

from __future__ import annotations

import pandas as pd

DICTIONARY_A: dict = {
    "name": "national-labour-force-survey",
    "fields": {
        "unemployment_rate": {
            "definition": (
                "share of the labour force not employed, actively seeking "
                "work in the last 4 weeks, and available to start within 2 weeks"
            ),
            "unit": "percent",
        }
    },
    "expected_regions": ["north", "south", "east", "west"],
}

DICTIONARY_B: dict = {
    "name": "administrative-benefit-claims-index",
    "fields": {
        "unemployment_rate": {
            "definition": (
                "share of the working-age population not currently employed, "
                "counted regardless of whether they are searching for work"
            ),
            "unit": "percent",
        }
    },
    "expected_regions": ["north", "south", "east", "west"],
}


def unemployment_series_a() -> "pd.Series":
    """From the labour-force-survey definition: job seekers only."""
    return pd.Series([4.1, 4.3, 4.0, 4.4, 5.2, 4.8], name="unemployment_rate")


def unemployment_series_b() -> "pd.Series":
    """From the administrative-claims definition: everyone not employed.

    Same dtype as series A, and its range (4.9-5.6) overlaps series A's
    range (4.0-5.2) -- the naive dtype-and-range check has nothing to flag.
    """
    return pd.Series([5.0, 5.4, 4.9, 5.6, 5.1, 5.3], name="unemployment_rate")


GOOD_SOURCE_METADATA: dict = {
    "granularity": "monthly, per region",
    "coverage": "national, 4 regions",
    "licence": "CC-BY-4.0",
    "dictionary": DICTIONARY_A,
    "update_cadence": "monthly, published on the 15th",
    "known_issues": ["one region was backfilled quarterly before 2020"],
}

DEFICIENT_SOURCE_METADATA: dict = {
    "granularity": "monthly",
    # coverage, licence, dictionary, update_cadence and known_issues all absent
}

NATIONAL_DATASET_KEYS: set = {"north", "south", "east"}
"""What a dataset that claims national coverage actually delivers here --
missing 'west', the gap Exercise 8 detects."""

"""A local mock API for Day 134 -- "Judge the Source Before the Data".

Nothing in this lab talks to the internet. Instead it serves a paginated
dataset, a rate-limited endpoint and an ETag-aware resource over real HTTP
from 127.0.0.1 on an ephemeral port, so the client code you exercise is a
real HTTP client -- it just cannot reach anyone else's machine.

Three endpoints:

* ``GET /dataset?page=N`` -- a paginated JSON collection of ``TOTAL_ROWS``
  rows, ``PAGE_SIZE`` at a time, each page carrying ``has_more`` so a client
  can tell when it has everything without knowing the total in advance.
* ``GET /dataset.csv`` -- the same rows as one CSV document, for the
  ``pandas.read_csv(url)`` demonstration.
* ``GET /ratelimited`` -- returns ``429`` with a ``Retry-After`` header for
  the first ``rate_limit_trigger_count`` requests this server instance has
  seen, then ``200``. Every server carries its own counter, so tests that
  want a source which never relents just ask for a trigger count higher
  than the client's attempt budget.
* ``GET /etag-resource`` -- returns ``200`` with an ``ETag`` header on the
  first request, and ``304`` with an empty body when the caller sends a
  matching ``If-None-Match``.

Standard library only: ``http.server``, ``threading``, ``json``.
"""

from __future__ import annotations

import json
import threading
from contextlib import contextmanager
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Iterator
from urllib.parse import parse_qs, urlparse

PAGE_SIZE = 10
TOTAL_ROWS = 25

ETAG_VALUE = '"d134-etag-v1"'
_ETAG_PAYLOAD = {
    "resource": "codebook",
    "version": 1,
    "fields": {"unemployment_rate": {"unit": "percent"}},
}
ETAG_BODY = json.dumps(_ETAG_PAYLOAD).encode("utf-8")


def _rows(start: int, stop: int) -> list[dict]:
    return [{"id": i, "value": i * 2} for i in range(start, stop)]


@dataclass
class MockAPI:
    """A running mock server plus the log of everything it was asked for."""

    host: str
    port: int
    rate_limit_hits: list[int] = field(default_factory=list)
    request_log: list[str] = field(default_factory=list)

    @property
    def base_url(self) -> str:
        return f"http://{self.host}:{self.port}"


def _make_handler(state: dict):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, fmt, *args):  # noqa: A003 - silence stderr access log
            pass

        def do_GET(self) -> None:  # noqa: N802 - name fixed by http.server
            parsed = urlparse(self.path)
            state["log"].append(parsed.path)
            if parsed.path == "/dataset":
                self._dataset(parse_qs(parsed.query))
            elif parsed.path == "/dataset.csv":
                self._dataset_csv()
            elif parsed.path == "/ratelimited":
                self._ratelimited()
            elif parsed.path == "/etag-resource":
                self._etag_resource()
            else:
                self._write(404, b"not found")

        def _write(self, status: int, body: bytes, headers: dict[str, str] | None = None) -> None:
            self.send_response(status)
            for key, value in (headers or {}).items():
                self.send_header(key, value)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            if body:
                self.wfile.write(body)

        def _dataset(self, query: dict[str, list[str]]) -> None:
            page = int(query.get("page", ["1"])[0])
            start = (page - 1) * PAGE_SIZE
            items = _rows(start, min(start + PAGE_SIZE, TOTAL_ROWS))
            has_more = start + PAGE_SIZE < TOTAL_ROWS
            body = json.dumps(
                {"items": items, "page": page, "has_more": has_more, "total": TOTAL_ROWS}
            ).encode("utf-8")
            self._write(200, body, {"Content-Type": "application/json"})

        def _dataset_csv(self) -> None:
            lines = ["id,value"] + [f"{r['id']},{r['value']}" for r in _rows(0, TOTAL_ROWS)]
            body = ("\n".join(lines) + "\n").encode("utf-8")
            self._write(200, body, {"Content-Type": "text/csv"})

        def _ratelimited(self) -> None:
            state["attempts"] += 1
            attempt = state["attempts"]
            if attempt <= state["trigger_count"]:
                state["rate_limit_hits"].append(attempt)
                self._write(429, b"", {"Retry-After": "0"})
                return
            body = json.dumps({"ok": True, "attempt": attempt}).encode("utf-8")
            self._write(200, body, {"Content-Type": "application/json"})

        def _etag_resource(self) -> None:
            if self.headers.get("If-None-Match") == ETAG_VALUE:
                self._write(304, b"", {"ETag": ETAG_VALUE})
                return
            self._write(200, ETAG_BODY, {"Content-Type": "application/json", "ETag": ETAG_VALUE})

    return Handler


@contextmanager
def serve_mock_api(rate_limit_trigger_count: int = 2) -> Iterator[MockAPI]:
    """Serve the mock API on 127.0.0.1 and an ephemeral port for the block.

    ``rate_limit_trigger_count`` is how many requests to ``/ratelimited``
    this instance answers with 429 before it starts answering 200 -- a
    higher count than a test's attempt budget models a source that never
    relents.
    """
    state = {
        "attempts": 0,
        "trigger_count": rate_limit_trigger_count,
        "rate_limit_hits": [],
        "log": [],
    }
    handler = _make_handler(state)
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    host, port = httpd.server_address[0], httpd.server_address[1]

    thread = threading.Thread(
        target=httpd.serve_forever, kwargs={"poll_interval": 0.02}, name="d134-mock-api"
    )
    thread.daemon = True
    thread.start()
    try:
        yield MockAPI(
            host=host,
            port=port,
            rate_limit_hits=state["rate_limit_hits"],
            request_log=state["log"],
        )
    finally:
        httpd.shutdown()
        httpd.server_close()
        thread.join(timeout=5)


if __name__ == "__main__":  # A quick manual check of the server itself.
    import urllib.request

    with serve_mock_api() as api:
        with urllib.request.urlopen(f"{api.base_url}/dataset?page=1") as response:
            print("page 1:", response.read().decode()[:80])
        print("requests so far:", api.request_log)

"""Reference solutions -- Day 134, "Judge the Source Before the Data".

Nine exercises. Each asserts on real behaviour: a real mock HTTP server
answering real requests, or a real checksum of real bytes -- never on
source code, and never on a value nobody computed.

Run with: pytest examples -q
"""

from __future__ import annotations

import json
from datetime import datetime, timezone

import pytest

import datasource as ds
import fixtures as fx
from mock_server import PAGE_SIZE, TOTAL_ROWS


def test_01_the_definition_trap():
    """Two columns named unemployment_rate, defined differently.

    The naive dtype-and-range check passes on both -- nothing mechanical
    would flag the join. Only the dictionary-aware check, which reads the
    prose definition rather than the numbers, refuses it.
    """
    series_a = fx.unemployment_series_a()
    series_b = fx.unemployment_series_b()

    naive = ds.naive_join_check(series_a, series_b)
    assert naive["dtype_match"] is True
    assert naive["ranges_overlap"] is True
    assert naive["would_pass_naive_check"] is True

    aware = ds.dictionary_aware_join_check(fx.DICTIONARY_A, fx.DICTIONARY_B, "unemployment_rate")
    assert aware["same_definition"] is False
    assert aware["safe_to_join"] is False
    assert "differ" in aware["reason"]
    assert "actively seeking" in aware["reason"]
    assert "regardless of whether" in aware["reason"]


def test_02_pagination_to_exhaustion(mock_api):
    """The client follows pages until the source says stop, not a fixed count."""
    rows = ds.fetch_all_pages(mock_api.base_url, "/dataset")

    assert len(rows) == TOTAL_ROWS
    assert [row["id"] for row in rows] == list(range(TOTAL_ROWS))

    expected_pages = -(-TOTAL_ROWS // PAGE_SIZE)  # ceiling division
    assert mock_api.request_log.count("/dataset") == expected_pages


def test_03_rate_limiting(mock_api, stubborn_mock_api):
    """A 429 triggers bounded backoff and eventual success; the client gives up."""
    body, attempts = ds.fetch_with_backoff(
        mock_api.base_url, "/ratelimited", max_attempts=5, base_delay=0.01
    )
    payload = json.loads(body)

    assert payload["ok"] is True
    assert attempts == 3  # 2 rejections (mock_api's trigger count) then success
    assert mock_api.rate_limit_hits == [1, 2]

    with pytest.raises(ds.RateLimitExceeded):
        ds.fetch_with_backoff(
            stubborn_mock_api.base_url, "/ratelimited", max_attempts=3, base_delay=0.01
        )
    # It tried exactly 3 times (the budget) and no more, against a source
    # that would have kept saying 429 forever.
    assert stubborn_mock_api.rate_limit_hits == [1, 2, 3]


def test_04_conditional_request(mock_api):
    """A second fetch with the stored ETag returns 304 and costs zero bytes."""
    cache: dict[str, ds.CacheEntry] = {}

    first_body, first_from_cache, first_bytes = ds.fetch_with_etag(
        mock_api.base_url, "/etag-resource", cache
    )
    assert first_from_cache is False
    assert first_bytes > 0  # the real payload went over the wire

    second_body, second_from_cache, second_bytes = ds.fetch_with_etag(
        mock_api.base_url, "/etag-resource", cache
    )
    assert second_from_cache is True
    assert second_body == first_body  # the cached copy is served, unchanged
    assert second_bytes == 0  # a 304 carries no body -- the re-run cost nothing

    assert mock_api.request_log.count("/etag-resource") == 2


def test_05_checksum_pinning(tmp_path):
    """The SHA-256 of a fixture matches a recorded value; one byte breaks it."""
    original = tmp_path / "dataset.csv"
    original.write_text("id,value\n1,10\n2,20\n3,30\n")

    digest = ds.sha256_of(original)
    assert digest == "4c0610aa92b75ca794ceec30068934fc6bc3d2fbff87969a15977f8fcf96f13f"

    altered = tmp_path / "altered.csv"
    altered.write_text("id,value\n1,10\n2,20\n3,31\n")  # one digit changed
    altered_digest = ds.sha256_of(altered)

    assert altered_digest == "9352ed755477b7af1eefd6e473c3880dd49e0a5d368846f51f8d96519d2bcf50"
    assert altered_digest != digest


def test_06_five_minute_source_assessment():
    """The structured verdict distinguishes a documented source from an undocumented one."""
    good_verdict = ds.assess_source(fx.GOOD_SOURCE_METADATA)
    assert good_verdict.ready is True
    assert good_verdict.problems == []
    assert good_verdict.granularity == "monthly, per region"
    assert good_verdict.dictionary_present is True

    deficient_verdict = ds.assess_source(fx.DEFICIENT_SOURCE_METADATA)
    assert deficient_verdict.ready is False
    assert "no stated coverage" in deficient_verdict.problems
    assert "no stated licence" in deficient_verdict.problems
    assert "no data dictionary" in deficient_verdict.problems
    assert "no update cadence" in deficient_verdict.problems
    assert "known issues undocumented" in deficient_verdict.problems


def test_07_licence_gate():
    """Redistribution passes for CC0, fails for 'all rights reserved' -- with a reason."""
    cc0 = ds.check_licence("CC0", purpose="redistribution")
    assert cc0["allowed"] is True
    assert "CC0" in cc0["reason"]

    ccby = ds.check_licence("CC-BY-4.0", purpose="redistribution")
    assert ccby["allowed"] is True
    assert "attribution" in ccby["reason"]

    odbl = ds.check_licence("ODbL", purpose="redistribution")
    assert odbl["allowed"] is True
    assert "share-alike" in odbl["reason"]

    all_rights = ds.check_licence("All rights reserved", purpose="redistribution")
    assert all_rights["allowed"] is False
    assert "forbids" in all_rights["reason"]
    assert all_rights["reason"] != ""  # a reason, never a bare boolean


def test_08_coverage_check():
    """A dataset claiming national coverage is missing a region -- caught by key comparison."""
    result = ds.check_coverage(fx.DICTIONARY_A, fx.NATIONAL_DATASET_KEYS)

    assert result["complete"] is False
    assert result["missing"] == ["west"]
    assert result["expected"] == ["east", "north", "south", "west"]

    complete_result = ds.check_coverage(fx.DICTIONARY_A, {"north", "south", "east", "west"})
    assert complete_result["complete"] is True
    assert complete_result["missing"] == []


def test_09_provenance_record(tmp_path):
    """The record carries url, retrieval timestamp and checksum, and is stable
    once the timestamp is held fixed rather than left to the clock."""
    payload = tmp_path / "dataset.csv"
    payload.write_text("id,value\n1,10\n2,20\n3,30\n")
    checksum = ds.sha256_of(payload)
    fixed_moment = datetime(2026, 8, 20, 12, 0, 0, tzinfo=timezone.utc)

    record_one = ds.record_provenance(
        "http://example.test/dataset.csv", checksum, retrieved_at=fixed_moment
    )
    record_two = ds.record_provenance(
        "http://example.test/dataset.csv", checksum, retrieved_at=fixed_moment
    )

    assert set(record_one) == {"url", "retrieved_at", "sha256"}
    assert record_one["url"] == "http://example.test/dataset.csv"
    assert record_one["sha256"] == checksum
    # Regenerating from the same fixture with the same injected timestamp
    # is byte-identical -- the flaky part (the real clock) is handled by
    # letting the caller pin it explicitly rather than asserting on `now()`.
    assert record_one == record_two

    # Without a pinned timestamp the function still returns a well-formed
    # ISO-8601 string -- it does not crash, it just won't compare equal
    # to a call made a second later, which is expected and not tested here.
    natural_record = ds.record_provenance("http://example.test/dataset.csv", checksum)
    assert natural_record["url"] == record_one["url"]
    assert natural_record["sha256"] == record_one["sha256"]
    datetime.fromisoformat(natural_record["retrieved_at"])  # parses without error

lesson_id: D134
day: 134
kind: guided-build
languages:
  - python
  - bash
setup_commands:
  - cd labs/sections/math-statistics-and-data/day-134-finding-data-open-datasets-and-apis
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - '.venv/bin/python3 -c "import pandas, pytest; print(pandas.__version__, pytest.__version__)"'
run_commands:
  - .venv/bin/pytest examples
  - .venv/bin/pytest starter
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - >-
    find . -path ./.venv -prune -o -type d -name '__pycache__' -print -exec rm -rf -- {} +
  - rm -rf .pytest_cache
  - 'rm -rf .venv  # optional: removes the lab virtual environment'
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-20'
executed_on: >-
  macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, pandas 3.0.5, pytest 9.1.1, bash
  3.2.57 -- bash tests/run_tests.sh -> 38 checks, 0 failure(s), exit=0. pytest examples
  -> 9 passed in 0.40s. pytest starter -> 9 skipped in 0.27s (untouched checkout).
  Everything was run through a real lab-local .venv created by the documented setup
  commands, against a real mock HTTP server (http.server.ThreadingHTTPServer bound to
  127.0.0.1 on an ephemeral port) implementing pagination, a 429 rate limit and an ETag
  -- never against the internet. Section 6 of the harness solves every exercise in a
  scratch copy (9 passed), deliberately breaks exercise 8's exact missing-region
  assertion (result["missing"] == ["west"] -> == ["nowhere"]), confirms a non-zero exit
  and a printed failure, restores the file, and confirms 9 passed again, so the suite is
  demonstrated to be capable of failing rather than merely claimed to be. Separately, a
  real assertion inside examples/datasource.py (check_coverage's missing computation)
  was broken directly and the whole harness was re-run: it reported 38 checks, 7
  failure(s) and exited 1 across the behaviour script, examples/, and both halves of the
  fail-then-restore proof; the file was restored and the harness returned to 38 checks,
  0 failure(s), exit=0. Section 5 confirms directly that `pytest examples starter` in
  one invocation aborts collection with `import file mismatch` (both directories define
  a module named test_datasource.py) rather than silently letting one shadow the other.
  MEASURED RESULTS from the mock server: pagination assembled 25 rows across 3 requests
  (TOTAL_ROWS=25, PAGE_SIZE=10); rate limiting against a relenting source succeeded on
  the 3rd attempt with rejections logged at attempts 1 and 2, and against a stubborn
  source (relents after 10) gave up after exactly 3 attempts (its own max_attempts
  budget) rather than retrying forever; the conditional ETag request downloaded 92 bytes
  on the first call and 0 bytes on the second, served from cache; pandas.read_csv(url)
  against the mock's /dataset.csv endpoint returned 25 rows. TWO HONESTY CALLS. FIRST:
  requests is not a dependency of this lab and was not installed in this lab's own
  .venv; the client code runs entirely on urllib.request from the standard library, and
  the lesson describes requests from its own documentation without reproducing any
  output from it here (it IS installed and pinned at 2.34.2 in the separate authoring
  tools venv used elsewhere in this repository, which is a different thing from this
  lab's own dependencies). SECOND: Hugging Face's `datasets` library is described from
  its published documentation only; it is not installed anywhere in this repository's
  authoring environment and no output from it is reproduced.

# Requirements

`requirements.txt` pins the exact versions this lab was written and run
against on 2026-08-20. Everything else it uses — `http.server`,
`urllib.request`, `hashlib`, `json`, `threading`, `dataclasses`, `datetime`
— is in the Python standard library. That is deliberate: the client code
this lab tests needs no third-party HTTP library at all.

Install into a lab-local virtual environment so the pins cannot collide
with anything else on your machine:

```bash
cd labs/sections/math-statistics-and-data/day-134-finding-data-open-datasets-and-apis
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
```

Only that install step needs the network. Everything after it runs
offline: the mock API in `mock_server.py` binds `127.0.0.1` on an
ephemeral port, so the tests exercise real HTTP without reaching anyone
else's machine.

## Why pandas is here at all

`fixtures.py` builds the two `unemployment_rate` series as `pandas.Series`
so Exercise 1's dtype-and-range check is the same kind of check you would
run on a real DataFrame column, not a hand-rolled substitute. `requests`
is not a dependency of this lab; the lesson describes it from its own
documentation and says plainly that it was not installed for this lab's
runs.

## If a pin will not install

Any recent pandas 2.2+ and pytest 7+ will almost certainly run this lab
unchanged — nothing here depends on a pandas 3.0-specific behaviour.

pandas==3.0.5
pytest==9.1.1

# Judge the Source Before the Data — the nine exercises

Today's lab is not about the shape of a dataset. It is about deciding
**whether a source deserves the hour you are about to spend on it**, and
about the handful of client behaviours — pagination, backoff, caching,
checksums — that make working with a live source safe and reproducible
rather than fragile and rude.

Five files sit in this directory:

| File | What it is |
| --- | --- |
| `mock_server.py` | A mock API on `127.0.0.1` and an ephemeral port: a paginated dataset, a rate-limited endpoint, and an ETag-aware resource. Infrastructure, not an exercise |
| `datasource.py` | The client and judgement functions your exercises test. Read this first |
| `fixtures.py` | The two `unemployment_rate` dictionaries and series, and the good/deficient source metadata used below |
| `conftest.py` | Two server fixtures: `mock_api` (relents after 2 requests) and `stubborn_mock_api` (relents after 10 — more than any test's budget) |
| `test_datasource.py` | Your nine exercises. Each currently calls `pytest.skip` |

Replace each `pytest.skip(...)` with real assertions, and delete the skip
line. Run `pytest starter -v` as often as you like. Never run
`pytest examples starter` in one command — both directories hold a module
named `test_datasource.py`, and pytest aborts collection on the clash. Run
the two directories as two separate commands.

The mock server is started and stopped inside each fixture. When the
suite finishes, no port is bound and no process from this lab is left
running.

---

## Exercise 1 — the definition trap

`fixtures.unemployment_series_a()` and `unemployment_series_b()` are two
columns both called `unemployment_rate`, both `float64`, with overlapping
ranges. `datasource.naive_join_check` — the dtype-and-range check most
people would actually run — passes on both. `datasource.dictionary_aware_join_check`
reads the two dictionaries' *prose definitions* instead, and refuses the
join. This is the day's centrepiece: prove the naive check passes and the
dictionary-aware check does not, on the same two columns.

## Exercise 2 — pagination to exhaustion

`datasource.fetch_all_pages` follows the mock API's `has_more` flag rather
than a page count you supply. Prove the assembled row count equals
`mock_server.TOTAL_ROWS`, that the row ids arrive in order, and that the
number of requests the server logged for `/dataset` matches how many
pages it actually took to cover `TOTAL_ROWS` at `PAGE_SIZE` per page.

## Exercise 3 — rate limiting

`datasource.fetch_with_backoff` retries a `429` with a bounded, growing
delay. Against `mock_api` (relents after 2 rejections), prove it succeeds
on the third attempt and that the server's own log of rejected attempts
is `[1, 2]`. Against `stubborn_mock_api` (relents after 10, more than any
sane budget), prove that calling it with `max_attempts=3` raises
`datasource.RateLimitExceeded` rather than retrying forever, and that it
made exactly 3 attempts — no more.

## Exercise 4 — conditional request

`datasource.fetch_with_etag` sends `If-None-Match` on the second call for
a path it has already cached. Prove the first call is not served from
cache and downloads real bytes; prove the second call **is** served from
cache, returns the same body, and — because a `304` carries no body — costs
zero bytes over the wire. Report the byte counts in your assertions, not
just booleans.

## Exercise 5 — checksum pinning

Write a small CSV to `tmp_path`, compute its SHA-256 with
`datasource.sha256_of`, and assert it equals a value you compute once
(with `hashlib.sha256(...).hexdigest()`, not by guessing) and record here.
Then change a single byte in a copy of the file and assert the checksum
changes. "Downloaded from X" is not reproducible. "Downloaded from X,
checksum `4c0610aa...`" is.

## Exercise 6 — the five-minute source assessment

`datasource.assess_source` takes a metadata dict and returns a
`SourceVerdict`: what granularity, coverage and licence were stated,
whether a dictionary exists, and a list of everything missing. Run it on
`fixtures.GOOD_SOURCE_METADATA` and assert `.ready` is `True` with no
problems. Run it on `fixtures.DEFICIENT_SOURCE_METADATA` and assert
`.ready` is `False`, with the specific missing items named in `.problems`.

## Exercise 7 — the licence gate

`datasource.check_licence` returns `{"allowed": bool, "reason": str}` —
never a bare boolean. Assert `CC0` is allowed for redistribution, and that
`"All rights reserved"` is refused with a reason that names why. A licence
permitting *analysis* is not the same permission as one permitting
*republishing the data* — the reason string is where that distinction
lives.

## Exercise 8 — coverage check

`datasource.check_coverage` compares a dictionary's `expected_regions`
against the keys a dataset actually delivers. Run it against
`fixtures.NATIONAL_DATASET_KEYS`, which is missing one of the four
expected regions, and assert the gap is detected by name — `"west"` — not
merely flagged as "incomplete".

## Exercise 9 — provenance record

`datasource.record_provenance(url, checksum, retrieved_at=...)` returns a
dict with exactly `url`, `retrieved_at` and `sha256`. Call it twice with
the **same** injected timestamp and assert the two records are equal —
regenerating from the same fixture is stable once the clock is pinned.
That is the honest way to test something that includes "now" without the
test becoming flaky: hold the clock still rather than asserting on a
moving target.

"""Shared fixtures: two mock API instances with different rate-limit patience.

``mock_api`` relents after 2 requests -- the "source that eventually lets
you through" case. ``stubborn_mock_api`` relents after 10, which is more
than any test's attempt budget -- the "source that never relents within a
sane budget" case. Both are started and stopped inside the fixture, so a
test that never touches them never binds a port, and one that does leaves
nothing running when it finishes.
"""

import pytest

from mock_server import serve_mock_api


@pytest.fixture
def mock_api():
    with serve_mock_api(rate_limit_trigger_count=2) as api:
        yield api


@pytest.fixture
def stubborn_mock_api():
    with serve_mock_api(rate_limit_trigger_count=10) as api:
        yield api

"""The client and judgement functions Day 134's exercises test.

Everything here runs on the standard library's ``urllib.request`` -- no
third-party HTTP client is a dependency of this lab. Day 135 owns turning
the JSON these functions return into a tidy DataFrame; this module stops at
"assembled rows" and "a verdict about whether the source deserves your
time in the first place".
"""

from __future__ import annotations

import hashlib
import json
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


class RateLimitExceeded(RuntimeError):
    """Raised when a source keeps answering 429 past the retry budget.

    A client that retries forever is not polite, it is a denial-of-service
    tool pointed at someone else's server. Giving up loudly, with a count
    of how many times it tried, is the correct behaviour.
    """


def fetch_raw(url: str, headers: dict[str, str] | None = None) -> tuple[int, dict[str, str], bytes]:
    """One GET via ``urllib.request``. Returns ``(status, headers, body)``.

    ``urllib.request`` raises ``HTTPError`` for any status >= 400, including
    304 in some circumstances and always 429 -- this unwraps that so callers
    can inspect the status code like any other response instead of writing
    a try/except around every call site.
    """
    request = urllib.request.Request(url, headers=headers or {})
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            return response.status, dict(response.headers), response.read()
    except urllib.error.HTTPError as exc:
        return exc.code, dict(exc.headers or {}), exc.read()


def fetch_all_pages(base_url: str, path: str = "/dataset") -> list[dict]:
    """Follow ``has_more`` until the source itself says stop.

    Never loops a fixed number of times and never trusts a total handed to
    it in advance -- the stopping condition is the server's own word for
    "nothing left", which is the only thing that can't drift out of sync
    with what actually got paginated.
    """
    rows: list[dict] = []
    page = 1
    while True:
        status, _, body = fetch_raw(f"{base_url}{path}?page={page}")
        if status != 200:
            raise RuntimeError(f"unexpected status {status} fetching page {page}")
        payload = json.loads(body)
        rows.extend(payload["items"])
        if not payload.get("has_more"):
            break
        page += 1
    return rows


def fetch_with_backoff(
    base_url: str,
    path: str,
    max_attempts: int = 5,
    base_delay: float = 0.01,
) -> tuple[bytes, int]:
    """GET with bounded exponential backoff on 429. Returns ``(body, attempts)``.

    Attempts are capped at ``max_attempts``; a source that never relents
    raises ``RateLimitExceeded`` rather than looping forever.
    """
    attempts = 0
    while attempts < max_attempts:
        attempts += 1
        status, headers, body = fetch_raw(f"{base_url}{path}")
        if status == 200:
            return body, attempts
        if status == 429:
            server_hint = headers.get("Retry-After")
            delay = float(server_hint) if server_hint not in (None, "0") else base_delay * (2 ** (attempts - 1))
            time.sleep(delay)
            continue
        raise RuntimeError(f"unexpected status {status}")
    raise RateLimitExceeded(f"gave up after {attempts} attempts, source still rate-limiting")


@dataclass
class CacheEntry:
    """One cached response, keyed by the ETag the source sent with it."""

    etag: str
    body: bytes


def fetch_with_etag(
    base_url: str, path: str, cache: dict[str, CacheEntry]
) -> tuple[bytes, bool, int]:
    """Conditional GET. Returns ``(body, served_from_cache, bytes_over_wire)``.

    The second call for the same path sends ``If-None-Match`` with the
    stored ETag. A ``304`` means the cached copy is still current: the
    caller gets it back with zero bytes counted against the wire, which is
    the entire point of a conditional request -- a re-run costs nothing.
    """
    headers = {}
    cached = cache.get(path)
    if cached is not None:
        headers["If-None-Match"] = cached.etag

    status, response_headers, body = fetch_raw(f"{base_url}{path}", headers=headers)

    if status == 304:
        if cached is None:
            raise RuntimeError("received 304 with nothing cached to serve")
        return cached.body, True, len(body)

    if status == 200:
        etag = response_headers.get("ETag", "")
        cache[path] = CacheEntry(etag=etag, body=body)
        return body, False, len(body)

    raise RuntimeError(f"unexpected status {status}")


def sha256_of(path: Path) -> str:
    """The SHA-256 hex digest of a file's bytes, for pinning a download."""
    return hashlib.sha256(path.read_bytes()).hexdigest()


@dataclass
class SourceVerdict:
    """The result of the five-minute assessment: what's here, what's missing."""

    granularity: str | None
    coverage: str | None
    licence: str | None
    dictionary_present: bool
    problems: list[str] = field(default_factory=list)

    @property
    def ready(self) -> bool:
        return not self.problems


def assess_source(metadata: dict[str, Any]) -> SourceVerdict:
    """Run the five-minute checklist against a source's stated metadata.

    Checks presence, not truth -- it cannot tell you the stated granularity
    is accurate, only that the source bothered to state one. That is still
    the majority of what separates a source worth trusting from one that
    is not, because most low-quality sources fail at "bothered to state it".
    """
    problems: list[str] = []
    granularity = metadata.get("granularity")
    coverage = metadata.get("coverage")
    licence = metadata.get("licence")
    dictionary_present = bool(metadata.get("dictionary"))

    if not granularity:
        problems.append("no stated granularity")
    if not coverage:
        problems.append("no stated coverage")
    if not licence:
        problems.append("no stated licence")
    if not dictionary_present:
        problems.append("no data dictionary")
    if not metadata.get("update_cadence"):
        problems.append("no update cadence")
    if "known_issues" not in metadata:
        problems.append("known issues undocumented")

    return SourceVerdict(
        granularity=granularity,
        coverage=coverage,
        licence=licence,
        dictionary_present=dictionary_present,
        problems=problems,
    )


REDISTRIBUTABLE_LICENCES = {"CC0", "CC-BY", "CC-BY-4.0", "ODbL"}
NON_REDISTRIBUTABLE_LICENCES = {"All rights reserved", "Proprietary"}


def check_licence(licence: str, purpose: str = "redistribution") -> dict[str, Any]:
    """Whether ``licence`` permits ``purpose``, with the reason spelled out.

    Returns a reason rather than a bare boolean on purpose: "allowed" and
    "allowed, with attribution required" are both True and both very
    different obligations for whoever ships the result.
    """
    if licence in REDISTRIBUTABLE_LICENCES:
        reason = f"{licence} permits {purpose}"
        if licence.startswith("CC-BY"):
            reason += ", with attribution to the source"
        if licence == "ODbL":
            reason += ", with share-alike terms for the derived database"
        return {"allowed": True, "reason": reason}

    if licence in NON_REDISTRIBUTABLE_LICENCES:
        return {
            "allowed": False,
            "reason": f"{licence} forbids {purpose}; analysis of the data is not the same permission as republishing it",
        }

    return {
        "allowed": False,
        "reason": f"unrecognised licence '{licence}' -- treat as not redistributable until confirmed",
    }


def check_coverage(dictionary: dict[str, Any], data_keys: set[str]) -> dict[str, Any]:
    """Compare a dataset's actual keys against the dictionary's expected list.

    Detects a gap by comparing sets, not by eyeballing a chart -- a region
    with zero rows looks identical to a region that was never collected
    unless something checks for its *absence* from the key list.
    """
    expected = set(dictionary.get("expected_regions", []))
    missing = sorted(expected - data_keys)
    return {"complete": not missing, "missing": missing, "expected": sorted(expected)}


def definitions_match(dictionary_a: dict[str, Any], dictionary_b: dict[str, Any], column: str) -> bool:
    """Whether two dictionaries define ``column`` the same way."""
    definition_a = dictionary_a["fields"][column]["definition"]
    definition_b = dictionary_b["fields"][column]["definition"]
    return definition_a == definition_b


def naive_join_check(series_a, series_b) -> dict[str, Any]:
    """What most people check before joining two columns: dtype and range.

    Deliberately weak. This is the check that passes on the two
    ``unemployment_rate`` columns the day's opening story describes, which
    is exactly why it is not sufficient on its own.
    """
    dtype_match = series_a.dtype == series_b.dtype
    range_a = (series_a.min(), series_a.max())
    range_b = (series_b.min(), series_b.max())
    ranges_overlap = not (range_a[1] < range_b[0] or range_b[1] < range_a[0])
    return {
        "dtype_match": dtype_match,
        "ranges_overlap": ranges_overlap,
        "would_pass_naive_check": dtype_match and ranges_overlap,
    }


def dictionary_aware_join_check(
    dictionary_a: dict[str, Any], dictionary_b: dict[str, Any], column: str
) -> dict[str, Any]:
    """The check the naive one is missing: do the two sources mean the same thing?"""
    same_definition = definitions_match(dictionary_a, dictionary_b, column)
    if same_definition:
        reason = "definitions match"
    else:
        definition_a = dictionary_a["fields"][column]["definition"]
        definition_b = dictionary_b["fields"][column]["definition"]
        reason = f"definitions differ: {definition_a!r} vs {definition_b!r}"
    return {"same_definition": same_definition, "safe_to_join": same_definition, "reason": reason}


def record_provenance(url: str, checksum: str, retrieved_at: datetime | None = None) -> dict[str, str]:
    """Build the record that makes a download reproducible: url, when, checksum.

    ``retrieved_at`` defaults to now, which is exactly the part that would
    make two calls compare unequal in a test -- callers that need a stable
    comparison (this lab's tests included) pass a fixed timestamp in rather
    than letting the clock make the suite flaky.
    """
    stamp = retrieved_at or datetime.now(timezone.utc)
    return {
        "url": url,
        "retrieved_at": stamp.replace(microsecond=0).isoformat(),
        "sha256": checksum,
    }

"""Fixture data for Day 134's exercises. Not itself an exercise -- read
alongside ``00_brief.md`` when the tests reference these names.

``DICTIONARY_A`` and ``DICTIONARY_B`` are two data dictionaries (codebooks)
for two fictional sources that both publish a column called
``unemployment_rate``. The columns below are built so that a naive check
-- same dtype, overlapping range -- passes on both, and only the
dictionaries' prose reveals that they are not the same measurement. That
gap is the lesson's opening failure story, made into fixtures.
"""

from __future__ import annotations

import pandas as pd

DICTIONARY_A: dict = {
    "name": "national-labour-force-survey",
    "fields": {
        "unemployment_rate": {
            "definition": (
                "share of the labour force not employed, actively seeking "
                "work in the last 4 weeks, and available to start within 2 weeks"
            ),
            "unit": "percent",
        }
    },
    "expected_regions": ["north", "south", "east", "west"],
}

DICTIONARY_B: dict = {
    "name": "administrative-benefit-claims-index",
    "fields": {
        "unemployment_rate": {
            "definition": (
                "share of the working-age population not currently employed, "
                "counted regardless of whether they are searching for work"
            ),
            "unit": "percent",
        }
    },
    "expected_regions": ["north", "south", "east", "west"],
}


def unemployment_series_a() -> "pd.Series":
    """From the labour-force-survey definition: job seekers only."""
    return pd.Series([4.1, 4.3, 4.0, 4.4, 5.2, 4.8], name="unemployment_rate")


def unemployment_series_b() -> "pd.Series":
    """From the administrative-claims definition: everyone not employed.

    Same dtype as series A, and its range (4.9-5.6) overlaps series A's
    range (4.0-5.2) -- the naive dtype-and-range check has nothing to flag.
    """
    return pd.Series([5.0, 5.4, 4.9, 5.6, 5.1, 5.3], name="unemployment_rate")


GOOD_SOURCE_METADATA: dict = {
    "granularity": "monthly, per region",
    "coverage": "national, 4 regions",
    "licence": "CC-BY-4.0",
    "dictionary": DICTIONARY_A,
    "update_cadence": "monthly, published on the 15th",
    "known_issues": ["one region was backfilled quarterly before 2020"],
}

DEFICIENT_SOURCE_METADATA: dict = {
    "granularity": "monthly",
    # coverage, licence, dictionary, update_cadence and known_issues all absent
}

NATIONAL_DATASET_KEYS: set = {"north", "south", "east"}
"""What a dataset that claims national coverage actually delivers here --
missing 'west', the gap Exercise 8 detects."""

"""A local mock API for Day 134 -- "Judge the Source Before the Data".

Nothing in this lab talks to the internet. Instead it serves a paginated
dataset, a rate-limited endpoint and an ETag-aware resource over real HTTP
from 127.0.0.1 on an ephemeral port, so the client code you exercise is a
real HTTP client -- it just cannot reach anyone else's machine.

Three endpoints:

* ``GET /dataset?page=N`` -- a paginated JSON collection of ``TOTAL_ROWS``
  rows, ``PAGE_SIZE`` at a time, each page carrying ``has_more`` so a client
  can tell when it has everything without knowing the total in advance.
* ``GET /dataset.csv`` -- the same rows as one CSV document, for the
  ``pandas.read_csv(url)`` demonstration.
* ``GET /ratelimited`` -- returns ``429`` with a ``Retry-After`` header for
  the first ``rate_limit_trigger_count`` requests this server instance has
  seen, then ``200``. Every server carries its own counter, so tests that
  want a source which never relents just ask for a trigger count higher
  than the client's attempt budget.
* ``GET /etag-resource`` -- returns ``200`` with an ``ETag`` header on the
  first request, and ``304`` with an empty body when the caller sends a
  matching ``If-None-Match``.

Standard library only: ``http.server``, ``threading``, ``json``.
"""

from __future__ import annotations

import json
import threading
from contextlib import contextmanager
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Iterator
from urllib.parse import parse_qs, urlparse

PAGE_SIZE = 10
TOTAL_ROWS = 25

ETAG_VALUE = '"d134-etag-v1"'
_ETAG_PAYLOAD = {
    "resource": "codebook",
    "version": 1,
    "fields": {"unemployment_rate": {"unit": "percent"}},
}
ETAG_BODY = json.dumps(_ETAG_PAYLOAD).encode("utf-8")


def _rows(start: int, stop: int) -> list[dict]:
    return [{"id": i, "value": i * 2} for i in range(start, stop)]


@dataclass
class MockAPI:
    """A running mock server plus the log of everything it was asked for."""

    host: str
    port: int
    rate_limit_hits: list[int] = field(default_factory=list)
    request_log: list[str] = field(default_factory=list)

    @property
    def base_url(self) -> str:
        return f"http://{self.host}:{self.port}"


def _make_handler(state: dict):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, fmt, *args):  # noqa: A003 - silence stderr access log
            pass

        def do_GET(self) -> None:  # noqa: N802 - name fixed by http.server
            parsed = urlparse(self.path)
            state["log"].append(parsed.path)
            if parsed.path == "/dataset":
                self._dataset(parse_qs(parsed.query))
            elif parsed.path == "/dataset.csv":
                self._dataset_csv()
            elif parsed.path == "/ratelimited":
                self._ratelimited()
            elif parsed.path == "/etag-resource":
                self._etag_resource()
            else:
                self._write(404, b"not found")

        def _write(self, status: int, body: bytes, headers: dict[str, str] | None = None) -> None:
            self.send_response(status)
            for key, value in (headers or {}).items():
                self.send_header(key, value)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            if body:
                self.wfile.write(body)

        def _dataset(self, query: dict[str, list[str]]) -> None:
            page = int(query.get("page", ["1"])[0])
            start = (page - 1) * PAGE_SIZE
            items = _rows(start, min(start + PAGE_SIZE, TOTAL_ROWS))
            has_more = start + PAGE_SIZE < TOTAL_ROWS
            body = json.dumps(
                {"items": items, "page": page, "has_more": has_more, "total": TOTAL_ROWS}
            ).encode("utf-8")
            self._write(200, body, {"Content-Type": "application/json"})

        def _dataset_csv(self) -> None:
            lines = ["id,value"] + [f"{r['id']},{r['value']}" for r in _rows(0, TOTAL_ROWS)]
            body = ("\n".join(lines) + "\n").encode("utf-8")
            self._write(200, body, {"Content-Type": "text/csv"})

        def _ratelimited(self) -> None:
            state["attempts"] += 1
            attempt = state["attempts"]
            if attempt <= state["trigger_count"]:
                state["rate_limit_hits"].append(attempt)
                self._write(429, b"", {"Retry-After": "0"})
                return
            body = json.dumps({"ok": True, "attempt": attempt}).encode("utf-8")
            self._write(200, body, {"Content-Type": "application/json"})

        def _etag_resource(self) -> None:
            if self.headers.get("If-None-Match") == ETAG_VALUE:
                self._write(304, b"", {"ETag": ETAG_VALUE})
                return
            self._write(200, ETAG_BODY, {"Content-Type": "application/json", "ETag": ETAG_VALUE})

    return Handler


@contextmanager
def serve_mock_api(rate_limit_trigger_count: int = 2) -> Iterator[MockAPI]:
    """Serve the mock API on 127.0.0.1 and an ephemeral port for the block.

    ``rate_limit_trigger_count`` is how many requests to ``/ratelimited``
    this instance answers with 429 before it starts answering 200 -- a
    higher count than a test's attempt budget models a source that never
    relents.
    """
    state = {
        "attempts": 0,
        "trigger_count": rate_limit_trigger_count,
        "rate_limit_hits": [],
        "log": [],
    }
    handler = _make_handler(state)
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    host, port = httpd.server_address[0], httpd.server_address[1]

    thread = threading.Thread(
        target=httpd.serve_forever, kwargs={"poll_interval": 0.02}, name="d134-mock-api"
    )
    thread.daemon = True
    thread.start()
    try:
        yield MockAPI(
            host=host,
            port=port,
            rate_limit_hits=state["rate_limit_hits"],
            request_log=state["log"],
        )
    finally:
        httpd.shutdown()
        httpd.server_close()
        thread.join(timeout=5)


if __name__ == "__main__":  # A quick manual check of the server itself.
    import urllib.request

    with serve_mock_api() as api:
        with urllib.request.urlopen(f"{api.base_url}/dataset?page=1") as response:
            print("page 1:", response.read().decode()[:80])
        print("requests so far:", api.request_log)

"""Your exercises for Day 134 -- "Judge the Source Before the Data".

Nine exercises. Every test below currently calls `pytest.skip(...)` --
replace the skip with real assertions and delete the skip line. Read
`00_brief.md` for the exercise-by-exercise explanation, `datasource.py`
for the client and judgement functions you are testing, `mock_server.py`
for the mock API they talk to, and `fixtures.py` for the fixture data.

Check yourself at any point:

    pytest starter -v

The reference answer key lives in `examples/test_datasource.py` -- read it
AFTER you have tried, never before.
"""

from __future__ import annotations

import json
from datetime import datetime, timezone

import pytest

import datasource as ds
import fixtures as fx
from mock_server import PAGE_SIZE, TOTAL_ROWS


def test_01_the_definition_trap():
    pytest.skip(
        "Take fx.unemployment_series_a() and fx.unemployment_series_b(). Run "
        "ds.naive_join_check(series_a, series_b) and assert dtype_match, "
        "ranges_overlap and would_pass_naive_check are all True -- nothing "
        "mechanical flags the join. Then run "
        "ds.dictionary_aware_join_check(fx.DICTIONARY_A, fx.DICTIONARY_B, "
        "'unemployment_rate') and assert same_definition and safe_to_join are "
        "both False, and that the reason string contains 'differ'"
    )


def test_02_pagination_to_exhaustion(mock_api):
    pytest.skip(
        "Call ds.fetch_all_pages(mock_api.base_url, '/dataset') and assert the "
        "number of rows returned equals TOTAL_ROWS, and that the ids form "
        "range(TOTAL_ROWS) in order. Then assert "
        "mock_api.request_log.count('/dataset') equals the number of pages "
        "needed to cover TOTAL_ROWS at PAGE_SIZE per page (ceiling division)"
    )


def test_03_rate_limiting(mock_api, stubborn_mock_api):
    pytest.skip(
        "Call ds.fetch_with_backoff(mock_api.base_url, '/ratelimited', "
        "max_attempts=5, base_delay=0.01) and assert the decoded JSON body has "
        "ok=True and that attempts equals 3 (2 rejections then success). Assert "
        "mock_api.rate_limit_hits == [1, 2]. Then assert that calling "
        "ds.fetch_with_backoff(stubborn_mock_api.base_url, '/ratelimited', "
        "max_attempts=3, base_delay=0.01) raises ds.RateLimitExceeded, and that "
        "stubborn_mock_api.rate_limit_hits == [1, 2, 3] -- it tried exactly the "
        "budget and no more"
    )


def test_04_conditional_request(mock_api):
    pytest.skip(
        "Create an empty cache dict. Call ds.fetch_with_etag(mock_api.base_url, "
        "'/etag-resource', cache) once and assert served_from_cache is False and "
        "bytes_over_wire is greater than 0. Call it again with the same cache "
        "and assert served_from_cache is True, the returned body is unchanged, "
        "and bytes_over_wire equals 0 -- a 304 carries no body, so the re-run "
        "cost nothing"
    )


def test_05_checksum_pinning(tmp_path):
    pytest.skip(
        "Write 'id,value\\n1,10\\n2,20\\n3,30\\n' to a file in tmp_path, compute "
        "ds.sha256_of(that_file), and assert it equals the recorded 64-character "
        "hex digest (compute it once with hashlib.sha256(...).hexdigest() and "
        "pin the value here -- do not guess it). Then write the same content "
        "with one digit changed to a second file, compute its digest, and "
        "assert it differs from the first"
    )


def test_06_five_minute_source_assessment():
    pytest.skip(
        "Call ds.assess_source(fx.GOOD_SOURCE_METADATA) and assert .ready is "
        "True and .problems is empty. Call ds.assess_source("
        "fx.DEFICIENT_SOURCE_METADATA) and assert .ready is False and that "
        "'no stated coverage', 'no stated licence', 'no data dictionary' and "
        "'no update cadence' are all in .problems"
    )


def test_07_licence_gate():
    pytest.skip(
        "Call ds.check_licence('CC0', purpose='redistribution') and assert "
        "allowed is True. Call ds.check_licence('All rights reserved', "
        "purpose='redistribution') and assert allowed is False and that the "
        "reason string is non-empty and contains 'forbids' -- the function "
        "must return a reason, never a bare boolean"
    )


def test_08_coverage_check():
    pytest.skip(
        "Call ds.check_coverage(fx.DICTIONARY_A, fx.NATIONAL_DATASET_KEYS) and "
        "assert complete is False and missing equals ['west'] -- detected by "
        "comparing the dictionary's expected_regions against the actual keys, "
        "not by looking at a chart. Then call it again with all four regions "
        "present and assert complete is True"
    )


def test_09_provenance_record(tmp_path):
    pytest.skip(
        "Write a small CSV to tmp_path and compute its checksum. Call "
        "ds.record_provenance(url, checksum, retrieved_at=some_fixed_datetime) "
        "twice with the SAME fixed timestamp and assert the two records are "
        "equal -- regenerating from the same fixture is stable once the clock "
        "is pinned. Assert the record has exactly the keys url, retrieved_at "
        "and sha256"
    )

#!/usr/bin/env bash
# Tests for the Day 134 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# The harness proves the day's claims by driving the real client functions
# against a real mock HTTP server on 127.0.0.1 and an ephemeral port --
# never by reading source and never by asserting on a timing:
#
#   * a naive dtype-and-range check passes on two unemployment_rate
#     columns with different definitions, and a dictionary-aware check
#     refuses the join between the same two columns;
#   * pagination follows the source's own has_more flag until exhaustion,
#     and the assembled row count equals the advertised total;
#   * a 429 triggers bounded backoff and eventual success against a source
#     that relents, with a named attempt count; a source that never
#     relents is refused after a bounded number of attempts, not retried
#     forever;
#   * a second request carrying the stored ETag returns 304 and is served
#     from cache at zero bytes over the wire, reported as byte counts;
#   * the SHA-256 of a fixture matches a recorded value, and a single
#     altered byte changes it;
#   * the five-minute source assessment returns a structured verdict for
#     a well-documented source and a deficient one;
#   * a licence gate passes CC0 for redistribution and refuses "all rights
#     reserved", with a reason rather than a bare boolean;
#   * a dataset claiming national coverage is missing a documented region,
#     detected by comparing key sets against the dictionary;
#   * a provenance record carries url, retrieval timestamp and checksum,
#     and regenerating it from the same fixture with the same pinned
#     timestamp is stable;
#   * the reference suite (examples/) passes in full;
#   * the exercise suite (starter/) is all-skip on an untouched checkout,
#     and the harness proves it can genuinely FAIL by solving every
#     exercise in a scratch copy, breaking one assertion on purpose,
#     confirming a non-zero exit and a printed failure, then restoring it;
#   * nothing is left listening, and no __pycache__ or .pytest_cache
#     survives the run.
#
# Everything after the one-time install runs offline. The mock server
# binds 127.0.0.1 on an ephemeral port and is shut down inside every
# fixture, including on failure. Deterministic, non-interactive, exits 0
# only if every check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

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

find "${lab_dir}" -name '.venv' -prune -o -type d -name '__pycache__' -exec rm -rf {} + 2>/dev/null || true
find "${lab_dir}" -name '.venv' -prune -o -type d -name '.pytest_cache' -exec rm -rf {} + 2>/dev/null || true

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_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 the 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:" >&2
  echo "    PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
  exit 1
}

python_bin="$(dirname "${pytest_bin}")/python3"
if [ ! -x "${python_bin}" ]; then
  python_bin="$(command -v python3 || true)"
fi
if [ -z "${python_bin}" ]; then
  echo "FAIL: python3 not found on PATH." >&2
  exit 1
fi

if ! "${python_bin}" -c "import pandas" >/dev/null 2>&1; then
  echo "FAIL: pandas is not importable from ${python_bin}." >&2
  echo "  Install the lab's dependencies with:" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
  exit 1
fi

echo "Day 134 — Judge the Source Before the Data"
echo

# --------------------------------------------------------------------------
echo "1. The tools and the versions this lab was written against"
# --------------------------------------------------------------------------

versions="$("${python_bin}" - <<'PY'
import platform
from importlib.metadata import version

print(f"python     {platform.python_version()}")
for name in ("pandas", "pytest"):
    try:
        print(f"{name:<10} {version(name)}")
    except Exception as exc:  # pragma: no cover
        print(f"{name:<10} NOT INSTALLED ({exc})")
PY
)"
echo "${versions}"
echo

mismatch=0
while IFS= read -r line; do
  [ -z "${line}" ] && continue
  pkg="${line%%==*}"
  pinned="${line#*==}"
  installed="$("${python_bin}" -c "from importlib.metadata import version; print(version('${pkg}'))" 2>/dev/null || echo "MISSING")"
  if [ "${installed}" != "${pinned}" ]; then
    mismatch=1
    echo "  version mismatch: ${pkg} pinned ${pinned}, installed ${installed}"
  fi
done < "${lab_dir}/requirements/requirements.txt"
check "installed packages match requirements.txt exactly" "$( [ ${mismatch} -eq 0 ] && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "2. The client and judgement functions, exercised directly against a"
echo "   real mock HTTP server on 127.0.0.1 and an ephemeral port"
# --------------------------------------------------------------------------

behaviour="$(cd "${lab_dir}/examples" && "${python_bin}" - <<'PY'
"""Drive datasource.py against mock_server.py and print one machine-readable
line per claim, including the attempt and byte counts the day brief asks for."""
import hashlib
from datetime import datetime, timezone

import datasource as ds
import fixtures as fx
import mock_server as ms

results = {}


def record(key, value):
    results[key] = value


# -- exercise 1: the definition trap ----------------------------------------
series_a = fx.unemployment_series_a()
series_b = fx.unemployment_series_b()
naive = ds.naive_join_check(series_a, series_b)
record("naive_dtype_match", "yes" if naive["dtype_match"] else "no")
record("naive_ranges_overlap", "yes" if naive["ranges_overlap"] else "no")
record("naive_would_pass", "yes" if naive["would_pass_naive_check"] else "no")
aware = ds.dictionary_aware_join_check(fx.DICTIONARY_A, fx.DICTIONARY_B, "unemployment_rate")
record("dictionary_aware_safe_to_join", "yes" if aware["safe_to_join"] else "no")
record("dictionary_aware_reason_has_differ", "yes" if "differ" in aware["reason"] else "no")

with ms.serve_mock_api(rate_limit_trigger_count=2) as api:
    # -- exercise 2: pagination to exhaustion -------------------------------
    rows = ds.fetch_all_pages(api.base_url, "/dataset")
    record("rows_fetched", str(len(rows)))
    record("rows_are_advertised_total", "yes" if len(rows) == ms.TOTAL_ROWS else "no")
    record("row_ids_in_order", "yes" if [r["id"] for r in rows] == list(range(ms.TOTAL_ROWS)) else "no")
    expected_pages = -(-ms.TOTAL_ROWS // ms.PAGE_SIZE)
    record("dataset_requests_made", str(api.request_log.count("/dataset")))
    record("dataset_requests_matches_pages_needed", "yes" if api.request_log.count("/dataset") == expected_pages else "no")

    # -- exercise 3: rate limiting (relenting source) -----------------------
    body, attempts = ds.fetch_with_backoff(api.base_url, "/ratelimited", max_attempts=5, base_delay=0.01)
    record("relenting_attempts", str(attempts))
    record("relenting_rejections_logged", ",".join(str(x) for x in api.rate_limit_hits))

with ms.serve_mock_api(rate_limit_trigger_count=10) as stubborn:
    # -- exercise 3: rate limiting (a source that never relents in budget) -
    gave_up = False
    attempts_before_giving_up = None
    try:
        ds.fetch_with_backoff(stubborn.base_url, "/ratelimited", max_attempts=3, base_delay=0.01)
    except ds.RateLimitExceeded:
        gave_up = True
        attempts_before_giving_up = len(stubborn.rate_limit_hits)
    record("gave_up_rather_than_retry_forever", "yes" if gave_up else "no")
    record("stubborn_attempts_made", str(attempts_before_giving_up))

with ms.serve_mock_api() as api:
    # -- exercise 4: conditional request -------------------------------------
    cache = {}
    first_body, first_cached, first_bytes = ds.fetch_with_etag(api.base_url, "/etag-resource", cache)
    second_body, second_cached, second_bytes = ds.fetch_with_etag(api.base_url, "/etag-resource", cache)
    record("first_fetch_from_cache", "yes" if first_cached else "no")
    record("first_fetch_bytes", str(first_bytes))
    record("second_fetch_from_cache", "yes" if second_cached else "no")
    record("second_fetch_bytes", str(second_bytes))
    record("cached_body_matches_original", "yes" if second_body == first_body else "no")

    # -- pandas.read_csv reading a URL, against the local mock ---------------
    import pandas as pd
    frame = pd.read_csv(f"{api.base_url}/dataset.csv")
    record("read_csv_row_count", str(len(frame)))
    record("read_csv_matches_total_rows", "yes" if len(frame) == ms.TOTAL_ROWS else "no")

# -- exercise 5: checksum pinning -------------------------------------------
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory(prefix="d134-checksum-") as tmp:
    original = Path(tmp) / "dataset.csv"
    original.write_text("id,value\n1,10\n2,20\n3,30\n")
    digest = ds.sha256_of(original)
    record("checksum_matches_recorded", "yes" if digest == "4c0610aa92b75ca794ceec30068934fc6bc3d2fbff87969a15977f8fcf96f13f" else "no")
    altered = Path(tmp) / "altered.csv"
    altered.write_text("id,value\n1,10\n2,20\n3,31\n")
    altered_digest = ds.sha256_of(altered)
    record("altered_checksum_differs", "yes" if altered_digest != digest else "no")

# -- exercise 6: five-minute source assessment -------------------------------
good = ds.assess_source(fx.GOOD_SOURCE_METADATA)
deficient = ds.assess_source(fx.DEFICIENT_SOURCE_METADATA)
record("good_source_ready", "yes" if good.ready else "no")
record("deficient_source_ready", "yes" if deficient.ready else "no")
record("deficient_problem_count", str(len(deficient.problems)))

# -- exercise 7: licence gate -------------------------------------------------
cc0 = ds.check_licence("CC0")
arr = ds.check_licence("All rights reserved")
record("cc0_allowed", "yes" if cc0["allowed"] else "no")
record("all_rights_reserved_allowed", "yes" if arr["allowed"] else "no")
record("all_rights_reserved_has_reason", "yes" if arr["reason"] else "no")

# -- exercise 8: coverage check -----------------------------------------------
coverage = ds.check_coverage(fx.DICTIONARY_A, fx.NATIONAL_DATASET_KEYS)
record("coverage_complete", "yes" if coverage["complete"] else "no")
record("coverage_missing", ",".join(coverage["missing"]))

# -- exercise 9: provenance record --------------------------------------------
fixed = datetime(2026, 8, 20, 12, 0, 0, tzinfo=timezone.utc)
p1 = ds.record_provenance("http://example.test/dataset.csv", digest, retrieved_at=fixed)
p2 = ds.record_provenance("http://example.test/dataset.csv", digest, retrieved_at=fixed)
record("provenance_stable_with_pinned_clock", "yes" if p1 == p2 else "no")
record("provenance_has_required_keys", "yes" if set(p1) == {"url", "retrieved_at", "sha256"} else "no")

for key, value in results.items():
    print(f"{key}={value}")
PY
)"
behaviour_status=$?
echo "${behaviour}"
echo

value_of() { echo "${behaviour}" | grep "^$1=" | cut -d= -f2-; }

check "the behaviour script ran without error" "$( [ ${behaviour_status} -eq 0 ] && echo yes || echo no )"
check "naive check: dtype matches and ranges overlap on the two unemployment_rate columns" "$( [ "$(value_of naive_dtype_match)" = yes ] && [ "$(value_of naive_ranges_overlap)" = yes ] && echo yes || echo no )"
check "naive check would pass the join -- nothing mechanical flags it" "$( [ "$(value_of naive_would_pass)" = yes ] && echo yes || echo no )"
check "dictionary-aware check refuses the same join" "$( [ "$(value_of dictionary_aware_safe_to_join)" = no ] && [ "$(value_of dictionary_aware_reason_has_differ)" = yes ] && echo yes || echo no )"
check "pagination assembled $(value_of rows_fetched) rows, matching the advertised total" "$( [ "$(value_of rows_are_advertised_total)" = yes ] && echo yes || echo no )"
check "row ids arrived in order" "$( [ "$(value_of row_ids_in_order)" = yes ] && echo yes || echo no )"
check "dataset requests made ($(value_of dataset_requests_made)) match pages actually needed" "$( [ "$(value_of dataset_requests_matches_pages_needed)" = yes ] && echo yes || echo no )"
check "rate limiting: relenting source succeeded after $(value_of relenting_attempts) attempts (rejections logged: $(value_of relenting_rejections_logged))" "$( [ "$(value_of relenting_attempts)" = 3 ] && [ "$(value_of relenting_rejections_logged)" = "1,2" ] && echo yes || echo no )"
check "rate limiting: client gave up against a source that never relents, after $(value_of stubborn_attempts_made) attempts" "$( [ "$(value_of gave_up_rather_than_retry_forever)" = yes ] && [ "$(value_of stubborn_attempts_made)" = 3 ] && echo yes || echo no )"
check "conditional request: first fetch was not from cache, $(value_of first_fetch_bytes) bytes over the wire" "$( [ "$(value_of first_fetch_from_cache)" = no ] && [ "$(value_of first_fetch_bytes)" -gt 0 ] && echo yes || echo no )"
check "conditional request: second fetch was served from cache at $(value_of second_fetch_bytes) bytes over the wire" "$( [ "$(value_of second_fetch_from_cache)" = yes ] && [ "$(value_of second_fetch_bytes)" = 0 ] && echo yes || echo no )"
check "the cached body matches the original" "$( [ "$(value_of cached_body_matches_original)" = yes ] && echo yes || echo no )"
check "pandas.read_csv against the local mock returned $(value_of read_csv_row_count) rows" "$( [ "$(value_of read_csv_matches_total_rows)" = yes ] && echo yes || echo no )"
check "checksum of the fixture matches the recorded SHA-256" "$( [ "$(value_of checksum_matches_recorded)" = yes ] && echo yes || echo no )"
check "a single altered byte changes the checksum" "$( [ "$(value_of altered_checksum_differs)" = yes ] && echo yes || echo no )"
check "the well-documented source assesses as ready" "$( [ "$(value_of good_source_ready)" = yes ] && echo yes || echo no )"
check "the deficient source assesses as not ready, with $(value_of deficient_problem_count) named problems" "$( [ "$(value_of deficient_source_ready)" = no ] && [ "$(value_of deficient_problem_count)" -ge 5 ] && echo yes || echo no )"
check "CC0 is allowed for redistribution" "$( [ "$(value_of cc0_allowed)" = yes ] && echo yes || echo no )"
check "'all rights reserved' is refused, with a reason rather than a bare boolean" "$( [ "$(value_of all_rights_reserved_allowed)" = no ] && [ "$(value_of all_rights_reserved_has_reason)" = yes ] && echo yes || echo no )"
check "coverage check finds the national dataset incomplete, missing: $(value_of coverage_missing)" "$( [ "$(value_of coverage_complete)" = no ] && [ "$(value_of coverage_missing)" = "west" ] && echo yes || echo no )"
check "the provenance record is stable once the clock is pinned" "$( [ "$(value_of provenance_stable_with_pinned_clock)" = yes ] && echo yes || echo no )"
check "the provenance record has exactly url, retrieved_at and sha256" "$( [ "$(value_of provenance_has_required_keys)" = yes ] && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "3. Reference suite -- examples/ must pass in full"
# --------------------------------------------------------------------------

examples_output="$(cd "${lab_dir}" && "${pytest_bin}" examples -q 2>&1)"
examples_status=$?
echo "${examples_output}" | tail -5
check "examples/ exits 0" "$( [ ${examples_status} -eq 0 ] && echo yes || echo no )"
examples_passed_line="$(echo "${examples_output}" | grep -E '^[0-9]+ passed' || true)"
check "examples/ reports 9 passed, 0 failed" "$( echo "${examples_passed_line}" | grep -qE '^9 passed' && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "4. Exercise suite -- starter/ is all-skip on an untouched checkout"
# --------------------------------------------------------------------------

starter_output="$(cd "${lab_dir}" && "${pytest_bin}" starter -q 2>&1)"
starter_status=$?
echo "${starter_output}" | tail -5
check "starter/ (untouched) exits 0" "$( [ ${starter_status} -eq 0 ] && echo yes || echo no )"
check "starter/ (untouched) reports 9 skipped, 0 failed" "$( echo "${starter_output}" | grep -qE '^9 skipped' && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "5. Never run 'pytest examples starter' in one invocation -- both"
echo "   directories define a module named test_datasource.py, and pytest"
echo "   collects by dotted module name. Documented, and run as two commands."
# --------------------------------------------------------------------------

combined_output="$(cd "${lab_dir}" && "${pytest_bin}" examples starter -q 2>&1)"
combined_status=$?
check "'pytest examples starter' aborts rather than silently passing" "$( [ ${combined_status} -ne 0 ] && echo yes || echo no )"
check "the collision is reported as an import file mismatch" "$( echo "${combined_output}" | grep -qi 'import file mismatch' && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "6. Prove the suite can genuinely FAIL: solve every exercise in a"
echo "   scratch copy, confirm green, break one assertion on purpose,"
echo "   confirm a non-zero exit and a printed failure, then restore."
# --------------------------------------------------------------------------

scratch_dir="$(mktemp -d "${TMPDIR:-/tmp}/d134-scratch.XXXXXX")"
cleanup_scratch() { rm -rf "${scratch_dir}"; }
trap cleanup_scratch EXIT

for module in test_datasource.py datasource.py mock_server.py fixtures.py conftest.py; do
  cp "${lab_dir}/examples/${module}" "${scratch_dir}/${module}"
done

solved_output="$("${pytest_bin}" "${scratch_dir}" -q 2>&1)"
solved_status=$?
check "scratch copy of the solved suite exits 0" "$( [ ${solved_status} -eq 0 ] && echo yes || echo no )"
check "scratch copy reports 9 passed" "$( echo "${solved_output}" | grep -qE '^9 passed' && echo yes || echo no )"

# Break exercise 8's exact missing-region assertion on purpose.
sed -i.bak "s/result\[\"missing\"\] == \[\"west\"\]/result[\"missing\"] == [\"nowhere\"]/" "${scratch_dir}/test_datasource.py"

broken_output="$("${pytest_bin}" "${scratch_dir}" -q 2>&1)"
broken_status=$?
check "broken scratch copy exits non-zero" "$( [ ${broken_status} -ne 0 ] && echo yes || echo no )"
check "broken scratch copy prints a failure" "$( echo "${broken_output}" | grep -qiE 'failed|assert' && echo yes || echo no )"

mv "${scratch_dir}/test_datasource.py.bak" "${scratch_dir}/test_datasource.py"
restored_output="$("${pytest_bin}" "${scratch_dir}" -q 2>&1)"
restored_status=$?
check "restored scratch copy exits 0 again" "$( [ ${restored_status} -eq 0 ] && echo yes || echo no )"
check "restored scratch copy reports 9 passed again" "$( echo "${restored_output}" | grep -qE '^9 passed' && echo yes || echo no )"

cleanup_scratch
trap - EXIT
echo

# --------------------------------------------------------------------------
echo "7. Offline, and nothing left behind"
# --------------------------------------------------------------------------

localhost_hits="$(grep -rl 'localhost:' "${lab_dir}/examples" "${lab_dir}/starter" 2>/dev/null || true)"
check "no literal 'localhost:port' string anywhere -- 127.0.0.1 only" "$( [ -z "${localhost_hits}" ] && echo yes || echo no )"

find "${lab_dir}" -name '.venv' -prune -o -type d -name '__pycache__' -exec rm -rf {} + 2>/dev/null || true
find "${lab_dir}" -name '.venv' -prune -o -type d -name '.pytest_cache' -exec rm -rf {} + 2>/dev/null || true

stray="$(find "${lab_dir}" -name '.venv' -prune -o \( -type d -name '__pycache__' -print -o -type d -name '.pytest_cache' -print \) 2>/dev/null || true)"
check "no __pycache__ or .pytest_cache left behind" "$( [ -z "${stray}" ] && echo yes || echo no )"

leftover_tmp="$(find "${TMPDIR:-/tmp}" -maxdepth 1 -name 'd134-*' -print 2>/dev/null || true)"
check "no d134 temporary directory left in the system temp directory" "$( [ -z "${leftover_tmp}" ] && echo yes || echo no )"
echo

echo "-------------------------------------------------------------"
echo "${checks} checks, ${failures} failure(s)"
if [ "${failures}" -gt 0 ]; then
  exit 1
fi
exit 0