Hands-on lab — Day 126: A Reproducible Cleaning Pipeline

"""Shared fixtures. pytest finds this file by itself -- nothing imports it.

`raw_orders` returns a FRESH copy of the raw data on every test, so one
test's mutation can never leak into the next. `config` returns a fresh
COPY of the dict too, for the same reason -- a test that mutates its own
config must never affect another test's.
"""

import copy

import pytest

from data import CONFIG, build_raw_orders


@pytest.fixture
def raw_orders():
    return build_raw_orders()


@pytest.fixture
def config():
    return copy.deepcopy(CONFIG)

"""Raw messy orders data and pipeline configuration for Day 126's lab.

`build_raw_orders()` returns a small, deliberately messy DataFrame built to
exercise every property this lab tests:

- `amount` arrives as currency-formatted strings ("$120.50"), including one
  genuinely missing value -- the input a real intake system would hand you.
- `region` carries whitespace and inconsistent casing (" north", "South").
- Two rows (order_id 1 and 3) are the SAME real-world order resubmitted
  under a new order_id, differing only in region's whitespace/casing --
  they become visible as duplicates only after normalisation. This is
  exercise 6's order-dependence case, built into the data on purpose.
- `priority` is a nullable pandas Int64 column with one missing value
  (order_id 4), carried through the whole pipeline untouched, so exercise 8
  can prove a Parquet checkpoint preserves it exactly (Day 121's result,
  used in anger).

CONFIG carries every threshold, mapping and column list the pipeline reads,
so re-running with different parameters never means editing steps.py or
pipeline.py -- only this dict.
"""

from __future__ import annotations

import pandas as pd

CONFIG: dict = {
    # Fixed ceiling for outlier clipping. NEVER recomputed from the data at
    # run time -- that is exactly the mistake exercise 1's broken step
    # demonstrates. Chosen so two rows collide exactly at the ceiling,
    # which is what makes exercise 2's explicit tie-break necessary.
    "amount_clip_max": 900.0,
    # Columns compared when deciding whether two rows describe the same
    # real-world order. Deliberately excludes order_id, because a
    # resubmitted order is assigned a NEW order_id by the intake system.
    "dedupe_subset": ["region", "amount", "priority"],
    # Fixed reference statistics for the z-score step, computed once from
    # this lab's known-good raw data and frozen here -- never recomputed
    # from whatever frame happens to be passing through the pipeline.
    "amount_reference_mean": 300.0,
    "amount_reference_std": 150.0,
}


def build_raw_orders() -> pd.DataFrame:
    """Seven rows, one duplicate pair, one missing amount, one missing
    priority. Every value below is a literal -- nothing here is randomised,
    so every hash and every assertion in this lab is exactly reproducible.
    """
    return pd.DataFrame(
        {
            "order_id": pd.array([1, 2, 3, 4, 5, 6, 7], dtype="int64"),
            "region": [
                " north",
                "South",
                "north",
                "EAST ",
                "South",
                "west",
                "East",
            ],
            "amount": [
                "$120.50",
                "$980.00",
                "$120.50",
                "$75.00",
                None,
                "$60.00",
                "$1,250.00",
            ],
            "priority": pd.array([1, 2, 1, None, 3, 2, 1], dtype="Int64"),
        }
    )

"""Pipeline orchestration: contracts at both ends, a step log that
reconciles, two equivalent ways to compose the steps, content hashing,
a Parquet checkpoint, and a manifest tying an output back to the input
and configuration that produced it.

The declared, required order is:

    parse_currency_amount
    normalize_region_strings
    dedupe_orders
    impute_missing_amount
    clip_amount_to_fixed_ceiling
    add_amount_zscore
    sort_deterministic

`normalize_region_strings` before `dedupe_orders` is not arbitrary: this
lab's data contains a resubmitted order whose region string is only
recognisable as a duplicate of an earlier order once whitespace and casing
are normalised (see `data.py` and exercise 6). Swapping those two steps
changes which rows survive.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path

import pandas as pd

from steps import (
    add_amount_zscore,
    clip_amount_to_fixed_ceiling,
    dedupe_orders,
    impute_missing_amount,
    normalize_region_strings,
    parse_currency_amount,
    sort_deterministic,
)

# --------------------------------------------------------------------------
# Contracts at both ends. A step that cannot meet its contract fails loudly
# -- ContractError, naming the offending column -- rather than passing a
# subtly wrong frame further down the pipeline.
# --------------------------------------------------------------------------

REQUIRED_INPUT_COLUMNS: dict[str, str] = {
    "order_id": "int64",
    # pandas 3.0's default string-inference gives plain Python string
    # columns pandas' own dedicated "str" extension dtype, not "object" --
    # a real, version-specific fact confirmed in this lab's own run, and
    # recorded again in expected-output/FIELDS.md.
    "region": "str",
    "amount": "str",
    "priority": "Int64",
}


class ContractError(ValueError):
    """Raised when a frame fails an input or output contract."""


def check_input_contract(df: pd.DataFrame) -> None:
    for column, expected_dtype in REQUIRED_INPUT_COLUMNS.items():
        if column not in df.columns:
            raise ContractError(f"input contract violated: missing required column '{column}'")
        actual_dtype = str(df[column].dtype)
        if actual_dtype != expected_dtype:
            raise ContractError(
                f"input contract violated: column '{column}' has dtype "
                f"'{actual_dtype}', expected '{expected_dtype}'"
            )


def check_output_contract(df: pd.DataFrame, config: dict) -> None:
    if df["amount"].isna().any():
        raise ContractError("output contract violated: 'amount' still has missing values")
    if not pd.api.types.is_float_dtype(df["amount"]):
        raise ContractError(f"output contract violated: 'amount' has dtype '{df['amount'].dtype}', expected float")
    ceiling = config["amount_clip_max"]
    if df["amount"].max() > ceiling:
        raise ContractError(
            f"output contract violated: 'amount' has a value above the clip ceiling "
            f"{ceiling} (max seen: {df['amount'].max()})"
        )
    if "amount_zscore" not in df.columns:
        raise ContractError("output contract violated: missing required output column 'amount_zscore'")


# --------------------------------------------------------------------------
# The step log. Every step records rows in, rows out, and the resulting
# net change -- read after the fact, it is how you discover that a step is
# quietly discarding rows nobody meant to discard.
# --------------------------------------------------------------------------


def _run_logged(name: str, func, df: pd.DataFrame, log: list[dict]) -> pd.DataFrame:
    rows_in = len(df)
    result = func(df)
    rows_out = len(result)
    log.append({"step": name, "rows_in": rows_in, "rows_out": rows_out, "delta": rows_out - rows_in})
    return result


def apply_steps_logged(df: pd.DataFrame, config: dict) -> tuple[pd.DataFrame, list[dict]]:
    """The pipeline itself: the seven steps, in the declared order, with NO
    contract checks. This is deliberately the function idempotence is
    checked against (exercise 1), because the input contract below is a
    check on raw, freshly-ingested data specifically (`amount` still a
    string) -- feeding a pipeline's OWN output back into that same contract
    would fail for a reason that has nothing to do with idempotence
    (`amount` is now float64, exactly as the pipeline is supposed to leave
    it). Idempotence is a property of the TRANSFORMATION, checked by
    applying it twice in a row; the input contract is a property of
    external data arriving from outside the pipeline for the first time.
    Both matter; they are not the same check.
    """
    log: list[dict] = []
    out = df
    out = _run_logged("parse_currency_amount", parse_currency_amount, out, log)
    out = _run_logged("normalize_region_strings", normalize_region_strings, out, log)
    out = _run_logged("dedupe_orders", lambda d: dedupe_orders(d, config), out, log)
    out = _run_logged("impute_missing_amount", lambda d: impute_missing_amount(d, config), out, log)
    out = _run_logged("clip_amount_to_fixed_ceiling", lambda d: clip_amount_to_fixed_ceiling(d, config), out, log)
    out = _run_logged("add_amount_zscore", lambda d: add_amount_zscore(d, config), out, log)
    out = _run_logged("sort_deterministic", sort_deterministic, out, log)
    return out, log


def run_pipeline(raw_df: pd.DataFrame, config: dict) -> tuple[pd.DataFrame, list[dict]]:
    """The pipeline's real entry point for freshly-ingested data: checks
    the input contract, runs `apply_steps_logged`, checks the output
    contract, and returns the result and its step log.
    """
    check_input_contract(raw_df)
    df, log = apply_steps_logged(raw_df, config)
    check_output_contract(df, config)
    return df, log


def run_pipeline_swapped_order(raw_df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """The SAME seven steps, with `dedupe_orders` run BEFORE
    `normalize_region_strings` instead of after -- the declared order,
    reversed. Used only to demonstrate order-dependence (exercise 6); never
    called from `run_pipeline`.
    """
    check_input_contract(raw_df)
    df = parse_currency_amount(raw_df)
    df = dedupe_orders(df, config)  # swapped: dedupe before normalising
    df = normalize_region_strings(df)
    df = impute_missing_amount(df, config)
    df = clip_amount_to_fixed_ceiling(df, config)
    df = add_amount_zscore(df, config)
    df = sort_deterministic(df)
    return df


def run_pipeline_via_pipe(raw_df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """The same seven steps, in the same declared order, composed with
    `DataFrame.pipe` instead of sequential assignment. Must produce a
    frame identical to `run_pipeline`'s -- exercise 4 proves it.

    `.pipe()` chaining reads well top to bottom, but it comes at a real
    cost: there is nowhere to put a breakpoint or a `print(df.shape)`
    between two links in the chain without breaking the chain apart again,
    which `run_pipeline`'s sequential form gives you for free. That
    tradeoff -- readability against inspectability -- is real, not
    cosmetic, and this lab ran both forms to show it rather than asserting
    it.
    """
    check_input_contract(raw_df)
    df = (
        raw_df.pipe(parse_currency_amount)
        .pipe(normalize_region_strings)
        .pipe(dedupe_orders, config)
        .pipe(impute_missing_amount, config)
        .pipe(clip_amount_to_fixed_ceiling, config)
        .pipe(add_amount_zscore, config)
        .pipe(sort_deterministic)
    )
    check_output_contract(df, config)
    return df


# --------------------------------------------------------------------------
# Determinism: content hashing, a Parquet checkpoint, and the manifest that
# ties an output back to the input and configuration that produced it.
# --------------------------------------------------------------------------


def content_hash(df: pd.DataFrame) -> str:
    """A SHA-256 hex digest of the frame's exact CSV bytes.

    Deterministic for a given pandas/NumPy version and a given frame's
    values, column order and dtypes -- change any one value, one column's
    dtype, or the column or row order, and the digest changes. See this
    lab's `expected-output/FIELDS.md` for exactly what is and is not
    guaranteed to be identical on a different machine.
    """
    payload = df.to_csv(index=False).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def config_hash(config: dict) -> str:
    """A SHA-256 hex digest of the config, serialised with sorted keys so
    the digest does not depend on the dict's insertion order.
    """
    payload = json.dumps(config, sort_keys=True).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def checkpoint_to_parquet(df: pd.DataFrame, path: Path) -> None:
    """Write a checkpoint. Parquet, not CSV, because Parquet preserves
    dtypes exactly -- including a nullable Int64 column's missing values
    -- where CSV round-trips everything through text and loses them
    (Day 121).
    """
    df.to_parquet(path, index=False)


def load_checkpoint(path: Path) -> pd.DataFrame:
    return pd.read_parquet(path)


def build_manifest(raw_df: pd.DataFrame, config: dict, step_log: list[dict], output_df: pd.DataFrame) -> dict:
    """The provenance record: which input, which configuration, which
    steps ran and what each one did to the row count, and what came out --
    everything needed, months later, to answer "which data produced this
    number?" without guessing.
    """
    return {
        "input_hash": content_hash(raw_df),
        "config_hash": config_hash(config),
        "steps": step_log,
        "output_hash": content_hash(output_df),
    }

"""Individual pipeline steps.

Each step is a pure function from frame to frame, named for what it does,
and testable on its own against a small fixture -- the cure for the
notebook's defining hazard, out-of-order cell execution. No step mutates
its input in place; every step returns a new frame built with `.copy()`.

Two steps clip `amount` to an upper bound, and only one of them belongs in
the real pipeline:

- `clip_amount_to_fixed_ceiling` reads its threshold from `config` --
  computed once, outside the pipeline, and frozen. Idempotent: clipping
  already-clipped data to the same fixed number changes nothing.
- `clip_amount_to_recomputed_percentile` is kept here ONLY as the lesson's
  worked failure. It recomputes its threshold from whatever data happens to
  be passing through it, which means a second call sees already-clipped
  data and computes a NEW, lower threshold from it -- non-idempotent by
  construction. `pipeline.py` never calls it; `test_pipeline.py` calls it
  directly to prove the failure before the fix.
"""

from __future__ import annotations

import pandas as pd


def parse_currency_amount(df: pd.DataFrame) -> pd.DataFrame:
    """Strip '$' and ',' from `amount` and convert to float64.

    Idempotent by inspection, not just by luck: once `amount` is numeric,
    the "still text" guard below is False and the step is a no-op on a
    second call, rather than re-stripping characters that are no longer
    there. The guard checks for a non-numeric dtype rather than the
    specific `object` dtype, because pandas 3.0's default string inference
    gives a plain Python string column its own `str` extension dtype, not
    `object` -- a version-specific fact this lab's own run confirmed.
    """
    df = df.copy()
    if not pd.api.types.is_numeric_dtype(df["amount"]):
        cleaned = df["amount"].str.replace(r"[$,]", "", regex=True)
        df["amount"] = pd.to_numeric(cleaned, errors="coerce")
    return df


def normalize_region_strings(df: pd.DataFrame) -> pd.DataFrame:
    """Strip whitespace and title-case `region` (" north" -> "North").

    Idempotent: an already-normalised string title-cases to itself.
    """
    df = df.copy()
    df["region"] = df["region"].str.strip().str.title()
    return df


def dedupe_orders(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Drop rows that describe the same real-world order, keeping the
    first occurrence after an explicit, deterministic sort by `order_id`.

    The sort matters: `drop_duplicates(keep="first")` depends on row
    order, and pandas does not promise the input arrived in `order_id`
    order. Sorting first makes "first occurrence" mean the same thing on
    every run, not whatever order the source happened to hand rows over
    in.

    Order-dependent by design (exercise 6): comparing `region` before it
    has been normalised treats " north" and "north" as different values,
    so the resubmitted order in this lab's data is NOT recognised as a
    duplicate unless `normalize_region_strings` already ran.
    """
    df = df.sort_values("order_id", kind="stable").reset_index(drop=True)
    df = df.drop_duplicates(subset=config["dedupe_subset"], keep="first")
    return df.reset_index(drop=True)


def impute_missing_amount(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Fill a missing `amount` with the column's own mean.

    Idempotent in the sense this pipeline relies on: once every `amount`
    is filled, `fillna` on a column with no missing values is a no-op, so
    a second call changes nothing -- provided no later step reintroduces a
    missing value, which none of this pipeline's steps do.
    """
    df = df.copy()
    df["amount"] = df["amount"].fillna(df["amount"].mean())
    return df


def clip_amount_to_fixed_ceiling(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Clip `amount` to `config["amount_clip_max"]` -- a fixed number read
    from configuration, never recomputed from the data. This is the
    correct, idempotent version. Compare `clip_amount_to_recomputed_percentile`
    below, which is not.
    """
    df = df.copy()
    df["amount"] = df["amount"].clip(upper=config["amount_clip_max"])
    return df


def clip_amount_to_recomputed_percentile(df: pd.DataFrame) -> pd.DataFrame:
    """DELIBERATELY NON-IDEMPOTENT. Never called by `pipeline.run_pipeline`.

    Computes its own threshold -- the 99th percentile of `amount` -- from
    whatever frame is passed in, every time it runs. The first call clips
    the true outliers down to the raw data's 99th percentile. Because the
    ceiling is now lower than it was, the SECOND call sees a narrower
    column and computes a new, lower 99th percentile from it, clipping
    again. `pipeline(pipeline(df))` therefore does not equal `pipeline(df)`
    when this step is used -- kept here only so
    `test_pipeline.py::test_1` can run it, watch it fail, and then run the
    fixed version and watch it pass.
    """
    df = df.copy()
    ceiling = df["amount"].quantile(0.99)
    df["amount"] = df["amount"].clip(upper=ceiling)
    return df


def add_amount_zscore(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Attach a z-score computed against FIXED reference statistics in
    `config`, never against this frame's own (possibly already-clipped,
    already-deduplicated) mean and standard deviation. Recomputing the
    reference from the current frame would make this step non-idempotent
    for exactly the same reason the broken clip step is.
    """
    df = df.copy()
    mean = config["amount_reference_mean"]
    std = config["amount_reference_std"]
    df["amount_zscore"] = (df["amount"] - mean) / std
    return df


def sort_deterministic(df: pd.DataFrame) -> pd.DataFrame:
    """Sort by `amount` with `order_id` as an explicit tie-break.

    Two rows in this lab's data collide exactly at the clip ceiling
    (900.0), so sorting by `amount` alone leaves their relative order
    unspecified -- a "stable" sort only preserves whatever order the ROWS
    ARRIVED in, which is not the same thing as a deterministic order
    across two independently built frames. Naming `order_id` as the
    tie-break makes the final row order a fact about the values, not
    about arrival order.
    """
    df = df.copy()
    return df.sort_values(["amount", "order_id"], kind="stable").reset_index(drop=True)

"""The worked reference suite for Day 126 -- "A Pipeline You Can Re-run".

Nine exercises, each proving one real property of a reproducible pandas
3.0.5 cleaning pipeline by running code and reading real values -- never by
reading source. Run it:

    pytest examples

Every table and every configuration value these tests use comes from
`data.py`. `steps.py` holds the seven pure step functions plus the one
deliberately broken step. `pipeline.py` composes them, checks contracts,
logs steps, hashes content, checkpoints to Parquet, and builds the
manifest. Read `starter/00_brief.md` for the exercise-by-exercise
explanation; this file is the answer key.
"""

from __future__ import annotations

import copy
import json
import tempfile
from pathlib import Path

import pandas as pd
import pytest

import pipeline as P
import steps as S
from data import CONFIG, build_raw_orders

# --------------------------------------------------------------------------
# Exercise 1 -- idempotence. pipeline(pipeline(df)) must equal pipeline(df)
# exactly. First, the worked failure: a step that recomputes its threshold
# from whatever data is currently passing through it is NOT idempotent.
# --------------------------------------------------------------------------


def test_1_broken_clip_step_is_not_idempotent(raw_orders, config):
    # Get to the point in the pipeline right before clipping, using the
    # real, correct earlier steps.
    prepared = S.parse_currency_amount(raw_orders)
    prepared = S.normalize_region_strings(prepared)
    prepared = S.dedupe_orders(prepared, config)
    prepared = S.impute_missing_amount(prepared, config)

    once = S.clip_amount_to_recomputed_percentile(prepared)
    twice = S.clip_amount_to_recomputed_percentile(once)

    # The two calls do NOT agree: the second call sees already-clipped
    # data, computes a new (lower) 99th percentile from it, and clips
    # again. This is exactly the failure the lesson opens with.
    assert not once.equals(twice), "the recomputing-percentile step is expected to NOT be idempotent"

    # Order 7's amount was clipped twice, to two different values.
    order_7_once = once.loc[once["order_id"] == 7, "amount"].iloc[0]
    order_7_twice = twice.loc[twice["order_id"] == 7, "amount"].iloc[0]
    assert order_7_once == pytest.approx(1236.5)
    assert order_7_twice == pytest.approx(1223.675)
    assert order_7_once != order_7_twice


def test_1_real_pipeline_is_idempotent(raw_orders, config):
    once, _ = P.apply_steps_logged(raw_orders, config)
    twice, _ = P.apply_steps_logged(once, config)

    # Every step in the real pipeline reads its thresholds from `config`,
    # never from the frame passing through it -- applying the whole
    # pipeline to its own output changes nothing.
    assert once.equals(twice)


# --------------------------------------------------------------------------
# Exercise 2 -- determinism. Two independent runs on the same input must
# produce an identical content hash, including after an explicit
# tie-breaking sort.
# --------------------------------------------------------------------------


def test_2_two_independent_runs_produce_an_identical_hash(config):
    df_a, _ = P.run_pipeline(build_raw_orders(), config)
    df_b, _ = P.run_pipeline(build_raw_orders(), config)

    assert df_a.equals(df_b)
    assert P.content_hash(df_a) == P.content_hash(df_b)


def test_2_tie_break_makes_the_final_order_deterministic_regardless_of_arrival_order(raw_orders, config):
    prepared, _ = P.apply_steps_logged(raw_orders, config)
    # Two rows -- order_id 2 and order_id 7 -- collide exactly at the clip
    # ceiling (900.0), so their relative order after a sort by `amount`
    # alone is not determined by the VALUES.
    tied = prepared.loc[prepared["amount"] == 900.0, "order_id"].tolist()
    assert sorted(tied) == [2, 7]

    # A stable sort by amount ALONE merely preserves whatever order the
    # rows happened to arrive in -- shuffle the arrival order and the tie
    # is broken differently.
    forward = prepared.sort_values("amount", kind="stable").reset_index(drop=True)
    reversed_input = prepared.iloc[::-1].reset_index(drop=True)
    backward = reversed_input.sort_values("amount", kind="stable").reset_index(drop=True)
    forward_tied_order = forward.loc[forward["amount"] == 900.0, "order_id"].tolist()
    backward_tied_order = backward.loc[backward["amount"] == 900.0, "order_id"].tolist()
    assert forward_tied_order != backward_tied_order, (
        "a sort with no tie-break is expected to depend on arrival order among tied rows"
    )

    # sort_deterministic names order_id as an explicit tie-break, so the
    # SAME two arrival orders now agree with each other.
    forward_final = S.sort_deterministic(prepared)
    backward_final = S.sort_deterministic(reversed_input)
    assert forward_final.equals(backward_final)
    assert forward_final.loc[forward_final["amount"] == 900.0, "order_id"].tolist() == [2, 7]


# --------------------------------------------------------------------------
# Exercise 3 -- the step log reconciles: every step's rows-out equals the
# next step's rows-in, and the total change equals the sum of the
# per-step changes.
# --------------------------------------------------------------------------


def test_3_step_log_reconciles_between_consecutive_steps(raw_orders, config):
    _, log = P.run_pipeline(raw_orders, config)

    for earlier, later in zip(log, log[1:]):
        assert earlier["rows_out"] == later["rows_in"], (
            f"{earlier['step']}'s rows_out must equal {later['step']}'s rows_in"
        )

    total_change = log[-1]["rows_out"] - log[0]["rows_in"]
    sum_of_deltas = sum(step["delta"] for step in log)
    assert total_change == sum_of_deltas == -1  # exactly one row deduplicated away


def test_3_step_log_shows_exactly_where_the_row_count_changed(raw_orders, config):
    _, log = P.run_pipeline(raw_orders, config)
    by_name = {step["step"]: step for step in log}

    assert by_name["dedupe_orders"]["delta"] == -1
    for name, step in by_name.items():
        if name != "dedupe_orders":
            assert step["delta"] == 0, f"{name} was expected to change no rows, delta was {step['delta']}"


# --------------------------------------------------------------------------
# Exercise 4 -- the input contract raises on a frame with a missing column
# or a wrong dtype, naming the offending column.
# --------------------------------------------------------------------------


def test_4_input_contract_raises_on_a_missing_column(raw_orders, config):
    broken = raw_orders.drop(columns=["priority"])
    with pytest.raises(P.ContractError, match="priority"):
        P.run_pipeline(broken, config)


def test_4_input_contract_raises_on_a_wrong_dtype(raw_orders, config):
    broken = raw_orders.copy()
    broken["order_id"] = broken["order_id"].astype("float64")
    with pytest.raises(P.ContractError, match="order_id"):
        P.run_pipeline(broken, config)


# --------------------------------------------------------------------------
# Exercise 5 -- the output contract raises when a step is sabotaged so its
# post-condition fails -- proving the contract can genuinely fail, not
# just pass.
# --------------------------------------------------------------------------


def test_5_output_contract_raises_when_the_clip_step_is_sabotaged(raw_orders, config, monkeypatch):
    def no_op_clip(df, config):  # a "clip" step that clips nothing
        return df

    monkeypatch.setattr(P, "clip_amount_to_fixed_ceiling", no_op_clip)

    with pytest.raises(P.ContractError, match="clip ceiling"):
        P.run_pipeline(raw_orders, config)


def test_5_output_contract_passes_once_the_step_is_restored(raw_orders, config):
    # Sanity check that the sabotage above is what triggers the failure,
    # not something else -- the unmodified pipeline must still pass.
    df, _ = P.run_pipeline(raw_orders, config)
    P.check_output_contract(df, config)  # raises nothing


# --------------------------------------------------------------------------
# Exercise 6 -- a .pipe() chain gives exactly the same frame as sequential
# application.
# --------------------------------------------------------------------------


def test_6_pipe_chain_equals_sequential_application(raw_orders, config):
    sequential, _ = P.run_pipeline(raw_orders, config)
    via_pipe = P.run_pipeline_via_pipe(raw_orders, config)
    assert sequential.equals(via_pipe)


# --------------------------------------------------------------------------
# Exercise 7 -- order dependence. Swapping normalize_region_strings and
# dedupe_orders changes the result: the declared order (normalise, then
# dedupe) catches a resubmitted order that arrives with different region
# casing; the reversed order misses it.
# --------------------------------------------------------------------------


def test_7_declared_order_catches_the_resubmitted_order(raw_orders, config):
    df, _ = P.run_pipeline(raw_orders, config)
    assert len(df) == 6
    assert 3 not in df["order_id"].tolist()  # the resubmission, order_id 3, is gone
    assert 1 in df["order_id"].tolist()  # the original, order_id 1, survives


def test_7_reversed_order_misses_the_resubmitted_order(raw_orders, config):
    swapped = P.run_pipeline_swapped_order(raw_orders, config)
    assert len(swapped) == 7  # nothing was deduplicated
    assert {1, 3}.issubset(set(swapped["order_id"].tolist()))  # BOTH survive


# --------------------------------------------------------------------------
# Exercise 8 -- a Parquet checkpoint round-trip preserves every dtype
# exactly, including a nullable Int64 column with a missing value.
# --------------------------------------------------------------------------


def test_8_parquet_checkpoint_preserves_every_dtype_exactly(raw_orders, tmp_path):
    before = S.normalize_region_strings(S.parse_currency_amount(raw_orders))
    checkpoint_path = tmp_path / "checkpoint.parquet"

    P.checkpoint_to_parquet(before, checkpoint_path)
    after = P.load_checkpoint(checkpoint_path)

    assert list(before.dtypes.astype(str)) == list(after.dtypes.astype(str))
    assert before.equals(after)

    # The nullable Int64 column, missing value included, round-trips exactly.
    assert str(before["priority"].dtype) == "Int64"
    assert str(after["priority"].dtype) == "Int64"
    assert before["priority"].isna().tolist() == after["priority"].isna().tolist()
    assert after.loc[after["order_id"] == 4, "priority"].isna().iloc[0]


# --------------------------------------------------------------------------
# Exercise 9 -- the manifest's hashes are stable across runs, and changing
# one input byte changes the input hash AND the output hash.
# --------------------------------------------------------------------------


def test_9_manifest_hashes_are_stable_across_independent_runs(config):
    df_a, log_a = P.run_pipeline(build_raw_orders(), config)
    manifest_a = P.build_manifest(build_raw_orders(), config, log_a, df_a)

    df_b, log_b = P.run_pipeline(build_raw_orders(), config)
    manifest_b = P.build_manifest(build_raw_orders(), config, log_b, df_b)

    assert manifest_a["input_hash"] == manifest_b["input_hash"]
    assert manifest_a["config_hash"] == manifest_b["config_hash"]
    assert manifest_a["output_hash"] == manifest_b["output_hash"]
    assert manifest_a["steps"] == manifest_b["steps"]


def test_9_changing_one_input_byte_changes_both_input_and_output_hash(raw_orders, config):
    original_df, original_log = P.run_pipeline(raw_orders, config)
    original_manifest = P.build_manifest(raw_orders, config, original_log, original_df)

    changed = raw_orders.copy()
    changed.loc[5, "amount"] = "$60.01"  # was "$60.00" -- one character different
    changed_df, changed_log = P.run_pipeline(changed, config)
    changed_manifest = P.build_manifest(changed, config, changed_log, changed_df)

    assert original_manifest["input_hash"] != changed_manifest["input_hash"]
    assert original_manifest["output_hash"] != changed_manifest["output_hash"]
    # The config did not change, so its hash must not change either.
    assert original_manifest["config_hash"] == changed_manifest["config_hash"]
    # Row count is unaffected -- only the one value changed.
    assert len(original_df) == len(changed_df) == 6


def test_9_manifest_is_json_serialisable(raw_orders, config):
    df, log = P.run_pipeline(raw_orders, config)
    manifest = P.build_manifest(raw_orders, config, log, df)
    # A manifest that cannot round-trip through JSON is not much of a
    # provenance record -- prove it actually can.
    serialised = json.dumps(manifest, sort_keys=True)
    reloaded = json.loads(serialised)
    assert reloaded["input_hash"] == manifest["input_hash"]
    assert reloaded["steps"] == manifest["steps"]

lesson_id: D126
day: 126
kind: guided-build
languages: [python, bash]
setup_commands:
  - cd labs/sections/math-statistics-and-data/day-126-a-reproducible-cleaning-pipeline
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/python3 -c "import pandas; print(pandas.__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-19'
executed_on: 'macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, pandas 3.0.5, pyarrow 25.0.1, numpy 2.5.2, pytest 9.1.1, bash 3.2.57 -- bash tests/run_tests.sh -> 16 checks, 0 failure(s), exit 0. pytest examples -> 17 passed. pytest starter -> 17 skipped (untouched checkout). pytest examples starter (one invocation) -> collection aborts with "import file mismatch", exit non-zero, confirmed directly rather than assumed, because every one of this lab''s five modules (data, steps, pipeline, conftest, test_pipeline) is defined identically in both directories. Section 5 of the harness solves every exercise in a scratch copy (17 passed), deliberately breaks exercise 1''s idempotence assertion (once.equals(twice) -> not once.equals(twice)), confirms the run exits non-zero with a printed FAIL, restores the file, and confirms 17 passed again. Separately, the coordinator broke the identical line directly in examples/test_pipeline.py (not a scratch copy) and re-ran the full harness: 1 failed, 16 passed from pytest, and 6 of the harness''s own 16 checks failed with overall exit 1; restoring the line returned the harness to 16 checks, 0 failure(s), exit 0. Everything was run through a real lab-local .venv created by the documented setup commands. Two honesty notes from this run. FIRST: pandas 3.0.5 infers the plain Python string columns `region` and `amount` as its own dedicated `str` extension dtype by default, not the historical `object` -- the same correction Day 124''s lab made for the identical reason -- and pipeline.py''s input contract and steps.py''s idempotence guard on parse_currency_amount are written against that fact rather than against `object`, confirmed by running the contract check and reading the raised error before writing the fix. SECOND: idempotence in this lab is checked against pipeline.apply_steps_logged (the seven steps with no contract checks), never against pipeline.run_pipeline directly on a pipeline''s own output -- feeding a pipeline''s OUTPUT (amount already float64) back into the INPUT contract (which requires amount to still be the str dtype of freshly-ingested data) raises ContractError for a reason that has nothing to do with idempotence, and conflating the two was an early design mistake caught by actually running the code, not assumed correct from the design. matplotlib, scipy, scikit-learn, polars, pandera and Great Expectations are not installed in this environment; pandera, Great Expectations and a workflow runner such as Prefect or Dagster are described in the lesson''s Tools section from public documentation only, and no output attributed to any of them is reproduced anywhere in this lab or its lesson.'

# What is installed, why, and what it costs

Four packages, all free and open source, installed into a lab-local
virtual environment that `rm -rf .venv` completely undoes.

| Package | Version pinned | Licence | What this lab uses it for |
| --- | --- | --- | --- |
| `pandas` | 3.0.5 | BSD 3-Clause | Every step function, `.pipe()` chain, `to_csv` (content hashing), `to_parquet`/`read_parquet` (checkpointing). |
| `pyarrow` | 25.0.1 | Apache 2.0 | pandas 3.0's Parquet engine — exercise 8's checkpoint round-trip goes through it. |
| `numpy` | 2.5.2 | BSD 3-Clause | Pulled in as a pandas dependency; nothing in this lab calls NumPy directly. |
| `pytest` | 9.1.1 | MIT | The test harness every exercise is written against, plus `monkeypatch` and `tmp_path` for exercises 5 and 8. |

`hashlib`, `json`, `logging` and `pathlib` — all standard library, no
install, no cost — do the rest: `hashlib` builds every hash in
`pipeline.py`, `json` serialises the manifest and the config for hashing,
and `pathlib.Path` is the type every checkpoint path uses.

There is no paid tier of anything in this lab, no account, no key and no
signup, personally or commercially.

## The one time the network is needed

```bash
.venv/bin/pip install -r requirements/requirements.txt
```

That is the only command in the lab that opens a connection. Every script
and test after that runs completely offline.

## What is deliberately *not* installed

**matplotlib**, **scipy**, **scikit-learn**, **polars**, **pandera** and
**Great Expectations** are not installed in this environment. The lesson's
Tools section describes `pandera`, Great Expectations, and a workflow
runner such as Prefect or Dagster from their public documentation as
design contrasts to this lab's hand-written contracts — no output from any
of them is reproduced anywhere in this lab or its lesson; every place they
are mentioned says so plainly.

## If you cannot install anything at all

pandas is not in the Python standard library, and there is no reduced path
through this lab without it. If pandas genuinely cannot be installed, read
the lesson's captured output and `expected-output/` directory instead;
every number there came from a real run and is not invented.

pandas==3.0.5
pyarrow==25.0.1
numpy==2.5.2
pytest==9.1.1

# Day 126 lab — the brief

Nine exercises, in order. Work top to bottom in `test_pipeline.py`. The
pipeline itself lives in three modules you read but do NOT edit:

- `data.py` — `build_raw_orders()`, the seven-row messy input, and
  `CONFIG`, every threshold and mapping the pipeline reads.
- `steps.py` — seven pure step functions, each frame-to-frame, plus one
  deliberately broken step (`clip_amount_to_recomputed_percentile`) kept
  only so exercise 1 can run it and watch it fail.
- `pipeline.py` — orchestration: `check_input_contract`,
  `check_output_contract`, `apply_steps_logged`, `run_pipeline`,
  `run_pipeline_swapped_order`, `run_pipeline_via_pipe`, `content_hash`,
  `config_hash`, `checkpoint_to_parquet`, `load_checkpoint`,
  `build_manifest`.

Read all three once before you start. Every fixture you need
(`raw_orders`, `config`) comes from `conftest.py`.

Check yourself at any point:

```bash
.venv/bin/pytest starter -v
```

On an untouched checkout that prints `17 skipped`. A **skip** means "not
attempted". Replace a `pytest.skip(...)` line with real assertions and
delete it — when every skip is gone and the suite is green, you are
finished:

```bash
.venv/bin/pytest starter -q
echo $?
```

Assert exact values everywhere. Nothing in this lab depends on timing.

---

## Exercise 1 — idempotence

`pipeline(pipeline(df))` must equal `pipeline(df)` exactly. First reproduce
the failure that makes this worth checking: `steps.clip_amount_to_recomputed_percentile`
computes its clip threshold — the 99th percentile — from whatever data is
CURRENTLY passing through it. Run the real early steps by hand (parse,
normalise, dedupe, impute), then call the broken clip step once and again
on its own output. Assert the two results are NOT equal, and that order 7's
amount differs between the two calls (`approx(1236.5)`, then
`approx(1223.675)`).

Then prove the real pipeline does not share this flaw:
`pipeline.apply_steps_logged` reads every threshold from `config`, never
from the frame passing through it. Call it once, then again on its own
output, and assert the two frames are `.equals()` — exactly, no `approx`.

## Exercise 2 — determinism

Run the real pipeline twice, each time on a **fresh** `build_raw_orders()`
call. Assert the two output frames are `.equals()` and that
`pipeline.content_hash` agrees on both.

Then the tie-break. Orders 2 and 7 both land on `amount == 900.0` after
clipping. Show that sorting by `amount` alone gives a DIFFERENT tie order
depending on whether the rows arrived forward or reversed
(`prepared.iloc[::-1]`), then show that `steps.sort_deterministic` gives
the SAME order either way, because it names `order_id` as an explicit
tie-break.

## Exercise 3 — the step log reconciles

Run the pipeline and read its step log. For every consecutive pair of
steps, assert the earlier one's `rows_out` equals the later one's
`rows_in`. Assert the total change (`last rows_out - first rows_in`)
equals the sum of every step's `delta`, and that this equals `-1` — one row
deduplicated away. Then assert `dedupe_orders` is the ONLY step with a
non-zero delta.

## Exercise 4 — the input contract

Drop the `priority` column from `raw_orders` and assert
`pipeline.run_pipeline` raises `pipeline.ContractError` naming `priority`
in its message. Then cast `order_id` to `float64` and assert it raises
naming `order_id`.

## Exercise 5 — the output contract

Use `monkeypatch.setattr(pipeline, "clip_amount_to_fixed_ceiling", ...)` to
replace the clip step with a function that returns its input unchanged —
sabotage the pipeline's own promise that `amount` never exceeds the
configured ceiling. Assert `run_pipeline` now raises `ContractError`
mentioning "clip ceiling". Then, as a sanity check, run the UNMODIFIED
pipeline and call `pipeline.check_output_contract` on its result directly —
it must raise nothing, confirming the sabotage above, not something else,
was what triggered the failure.

## Exercise 6 — `.pipe()` equivalence

Assert `pipeline.run_pipeline(raw_orders, config)[0]` and
`pipeline.run_pipeline_via_pipe(raw_orders, config)` are `.equals()`.

## Exercise 7 — order dependence

`normalize_region_strings` before `dedupe_orders` is the pipeline's
declared order, and this lab's data makes it matter: order_id 3 is a
resubmission of order_id 1, differing only in region's whitespace and
casing (`" north"` vs `"north"`). Run the real pipeline and assert the
result has 6 rows, order_id 3 is gone, and order_id 1 survives. Then run
`pipeline.run_pipeline_swapped_order` (the same steps, dedupe run first)
and assert it has 7 rows — nothing deduplicated — with BOTH order_id 1 and
order_id 3 present.

## Exercise 8 — the Parquet checkpoint

Parse and normalise `raw_orders` (stop there — do not dedupe yet, so
order_id 4's missing `priority` is still present). Checkpoint it to
`tmp_path / "checkpoint.parquet"` with `pipeline.checkpoint_to_parquet`,
reload it with `pipeline.load_checkpoint`, and assert every dtype matches
exactly, the two frames are `.equals()`, and the reloaded `priority` column
is still `Int64` with order_id 4's value still missing.

## Exercise 9 — the manifest

Build a manifest from two independent `build_raw_orders()` runs through the
pipeline (`pipeline.build_manifest`). Assert `input_hash`, `config_hash`,
`output_hash` and `steps` all agree between the two. Then build a manifest
for `raw_orders`, change order_id 6's amount from `"$60.00"` to `"$60.01"`
(one character), and build a second manifest — assert `input_hash` and
`output_hash` both differ, `config_hash` stays the same (the config did not
change), and the row count is unaffected. Finally, round-trip a manifest
through `json.dumps`/`json.loads` and assert the reloaded `input_hash` and
`steps` match the original.

---

Prove your suite is not vacuous once you are green: re-break one assertion
on purpose (flip a comparison, change an expected number), confirm the run
exits non-zero with a printed `FAIL`, then restore it and confirm green
again.

"""Shared fixtures. pytest finds this file by itself -- nothing imports it.

`raw_orders` returns a FRESH copy of the raw data on every test, so one
test's mutation can never leak into the next. `config` returns a fresh
COPY of the dict too, for the same reason -- a test that mutates its own
config must never affect another test's.
"""

import copy

import pytest

from data import CONFIG, build_raw_orders


@pytest.fixture
def raw_orders():
    return build_raw_orders()


@pytest.fixture
def config():
    return copy.deepcopy(CONFIG)

"""Raw messy orders data and pipeline configuration for Day 126's lab.

`build_raw_orders()` returns a small, deliberately messy DataFrame built to
exercise every property this lab tests:

- `amount` arrives as currency-formatted strings ("$120.50"), including one
  genuinely missing value -- the input a real intake system would hand you.
- `region` carries whitespace and inconsistent casing (" north", "South").
- Two rows (order_id 1 and 3) are the SAME real-world order resubmitted
  under a new order_id, differing only in region's whitespace/casing --
  they become visible as duplicates only after normalisation. This is
  exercise 6's order-dependence case, built into the data on purpose.
- `priority` is a nullable pandas Int64 column with one missing value
  (order_id 4), carried through the whole pipeline untouched, so exercise 8
  can prove a Parquet checkpoint preserves it exactly (Day 121's result,
  used in anger).

CONFIG carries every threshold, mapping and column list the pipeline reads,
so re-running with different parameters never means editing steps.py or
pipeline.py -- only this dict.
"""

from __future__ import annotations

import pandas as pd

CONFIG: dict = {
    # Fixed ceiling for outlier clipping. NEVER recomputed from the data at
    # run time -- that is exactly the mistake exercise 1's broken step
    # demonstrates. Chosen so two rows collide exactly at the ceiling,
    # which is what makes exercise 2's explicit tie-break necessary.
    "amount_clip_max": 900.0,
    # Columns compared when deciding whether two rows describe the same
    # real-world order. Deliberately excludes order_id, because a
    # resubmitted order is assigned a NEW order_id by the intake system.
    "dedupe_subset": ["region", "amount", "priority"],
    # Fixed reference statistics for the z-score step, computed once from
    # this lab's known-good raw data and frozen here -- never recomputed
    # from whatever frame happens to be passing through the pipeline.
    "amount_reference_mean": 300.0,
    "amount_reference_std": 150.0,
}


def build_raw_orders() -> pd.DataFrame:
    """Seven rows, one duplicate pair, one missing amount, one missing
    priority. Every value below is a literal -- nothing here is randomised,
    so every hash and every assertion in this lab is exactly reproducible.
    """
    return pd.DataFrame(
        {
            "order_id": pd.array([1, 2, 3, 4, 5, 6, 7], dtype="int64"),
            "region": [
                " north",
                "South",
                "north",
                "EAST ",
                "South",
                "west",
                "East",
            ],
            "amount": [
                "$120.50",
                "$980.00",
                "$120.50",
                "$75.00",
                None,
                "$60.00",
                "$1,250.00",
            ],
            "priority": pd.array([1, 2, 1, None, 3, 2, 1], dtype="Int64"),
        }
    )

"""Pipeline orchestration: contracts at both ends, a step log that
reconciles, two equivalent ways to compose the steps, content hashing,
a Parquet checkpoint, and a manifest tying an output back to the input
and configuration that produced it.

The declared, required order is:

    parse_currency_amount
    normalize_region_strings
    dedupe_orders
    impute_missing_amount
    clip_amount_to_fixed_ceiling
    add_amount_zscore
    sort_deterministic

`normalize_region_strings` before `dedupe_orders` is not arbitrary: this
lab's data contains a resubmitted order whose region string is only
recognisable as a duplicate of an earlier order once whitespace and casing
are normalised (see `data.py` and exercise 6). Swapping those two steps
changes which rows survive.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path

import pandas as pd

from steps import (
    add_amount_zscore,
    clip_amount_to_fixed_ceiling,
    dedupe_orders,
    impute_missing_amount,
    normalize_region_strings,
    parse_currency_amount,
    sort_deterministic,
)

# --------------------------------------------------------------------------
# Contracts at both ends. A step that cannot meet its contract fails loudly
# -- ContractError, naming the offending column -- rather than passing a
# subtly wrong frame further down the pipeline.
# --------------------------------------------------------------------------

REQUIRED_INPUT_COLUMNS: dict[str, str] = {
    "order_id": "int64",
    # pandas 3.0's default string-inference gives plain Python string
    # columns pandas' own dedicated "str" extension dtype, not "object" --
    # a real, version-specific fact confirmed in this lab's own run, and
    # recorded again in expected-output/FIELDS.md.
    "region": "str",
    "amount": "str",
    "priority": "Int64",
}


class ContractError(ValueError):
    """Raised when a frame fails an input or output contract."""


def check_input_contract(df: pd.DataFrame) -> None:
    for column, expected_dtype in REQUIRED_INPUT_COLUMNS.items():
        if column not in df.columns:
            raise ContractError(f"input contract violated: missing required column '{column}'")
        actual_dtype = str(df[column].dtype)
        if actual_dtype != expected_dtype:
            raise ContractError(
                f"input contract violated: column '{column}' has dtype "
                f"'{actual_dtype}', expected '{expected_dtype}'"
            )


def check_output_contract(df: pd.DataFrame, config: dict) -> None:
    if df["amount"].isna().any():
        raise ContractError("output contract violated: 'amount' still has missing values")
    if not pd.api.types.is_float_dtype(df["amount"]):
        raise ContractError(f"output contract violated: 'amount' has dtype '{df['amount'].dtype}', expected float")
    ceiling = config["amount_clip_max"]
    if df["amount"].max() > ceiling:
        raise ContractError(
            f"output contract violated: 'amount' has a value above the clip ceiling "
            f"{ceiling} (max seen: {df['amount'].max()})"
        )
    if "amount_zscore" not in df.columns:
        raise ContractError("output contract violated: missing required output column 'amount_zscore'")


# --------------------------------------------------------------------------
# The step log. Every step records rows in, rows out, and the resulting
# net change -- read after the fact, it is how you discover that a step is
# quietly discarding rows nobody meant to discard.
# --------------------------------------------------------------------------


def _run_logged(name: str, func, df: pd.DataFrame, log: list[dict]) -> pd.DataFrame:
    rows_in = len(df)
    result = func(df)
    rows_out = len(result)
    log.append({"step": name, "rows_in": rows_in, "rows_out": rows_out, "delta": rows_out - rows_in})
    return result


def apply_steps_logged(df: pd.DataFrame, config: dict) -> tuple[pd.DataFrame, list[dict]]:
    """The pipeline itself: the seven steps, in the declared order, with NO
    contract checks. This is deliberately the function idempotence is
    checked against (exercise 1), because the input contract below is a
    check on raw, freshly-ingested data specifically (`amount` still a
    string) -- feeding a pipeline's OWN output back into that same contract
    would fail for a reason that has nothing to do with idempotence
    (`amount` is now float64, exactly as the pipeline is supposed to leave
    it). Idempotence is a property of the TRANSFORMATION, checked by
    applying it twice in a row; the input contract is a property of
    external data arriving from outside the pipeline for the first time.
    Both matter; they are not the same check.
    """
    log: list[dict] = []
    out = df
    out = _run_logged("parse_currency_amount", parse_currency_amount, out, log)
    out = _run_logged("normalize_region_strings", normalize_region_strings, out, log)
    out = _run_logged("dedupe_orders", lambda d: dedupe_orders(d, config), out, log)
    out = _run_logged("impute_missing_amount", lambda d: impute_missing_amount(d, config), out, log)
    out = _run_logged("clip_amount_to_fixed_ceiling", lambda d: clip_amount_to_fixed_ceiling(d, config), out, log)
    out = _run_logged("add_amount_zscore", lambda d: add_amount_zscore(d, config), out, log)
    out = _run_logged("sort_deterministic", sort_deterministic, out, log)
    return out, log


def run_pipeline(raw_df: pd.DataFrame, config: dict) -> tuple[pd.DataFrame, list[dict]]:
    """The pipeline's real entry point for freshly-ingested data: checks
    the input contract, runs `apply_steps_logged`, checks the output
    contract, and returns the result and its step log.
    """
    check_input_contract(raw_df)
    df, log = apply_steps_logged(raw_df, config)
    check_output_contract(df, config)
    return df, log


def run_pipeline_swapped_order(raw_df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """The SAME seven steps, with `dedupe_orders` run BEFORE
    `normalize_region_strings` instead of after -- the declared order,
    reversed. Used only to demonstrate order-dependence (exercise 6); never
    called from `run_pipeline`.
    """
    check_input_contract(raw_df)
    df = parse_currency_amount(raw_df)
    df = dedupe_orders(df, config)  # swapped: dedupe before normalising
    df = normalize_region_strings(df)
    df = impute_missing_amount(df, config)
    df = clip_amount_to_fixed_ceiling(df, config)
    df = add_amount_zscore(df, config)
    df = sort_deterministic(df)
    return df


def run_pipeline_via_pipe(raw_df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """The same seven steps, in the same declared order, composed with
    `DataFrame.pipe` instead of sequential assignment. Must produce a
    frame identical to `run_pipeline`'s -- exercise 4 proves it.

    `.pipe()` chaining reads well top to bottom, but it comes at a real
    cost: there is nowhere to put a breakpoint or a `print(df.shape)`
    between two links in the chain without breaking the chain apart again,
    which `run_pipeline`'s sequential form gives you for free. That
    tradeoff -- readability against inspectability -- is real, not
    cosmetic, and this lab ran both forms to show it rather than asserting
    it.
    """
    check_input_contract(raw_df)
    df = (
        raw_df.pipe(parse_currency_amount)
        .pipe(normalize_region_strings)
        .pipe(dedupe_orders, config)
        .pipe(impute_missing_amount, config)
        .pipe(clip_amount_to_fixed_ceiling, config)
        .pipe(add_amount_zscore, config)
        .pipe(sort_deterministic)
    )
    check_output_contract(df, config)
    return df


# --------------------------------------------------------------------------
# Determinism: content hashing, a Parquet checkpoint, and the manifest that
# ties an output back to the input and configuration that produced it.
# --------------------------------------------------------------------------


def content_hash(df: pd.DataFrame) -> str:
    """A SHA-256 hex digest of the frame's exact CSV bytes.

    Deterministic for a given pandas/NumPy version and a given frame's
    values, column order and dtypes -- change any one value, one column's
    dtype, or the column or row order, and the digest changes. See this
    lab's `expected-output/FIELDS.md` for exactly what is and is not
    guaranteed to be identical on a different machine.
    """
    payload = df.to_csv(index=False).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def config_hash(config: dict) -> str:
    """A SHA-256 hex digest of the config, serialised with sorted keys so
    the digest does not depend on the dict's insertion order.
    """
    payload = json.dumps(config, sort_keys=True).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def checkpoint_to_parquet(df: pd.DataFrame, path: Path) -> None:
    """Write a checkpoint. Parquet, not CSV, because Parquet preserves
    dtypes exactly -- including a nullable Int64 column's missing values
    -- where CSV round-trips everything through text and loses them
    (Day 121).
    """
    df.to_parquet(path, index=False)


def load_checkpoint(path: Path) -> pd.DataFrame:
    return pd.read_parquet(path)


def build_manifest(raw_df: pd.DataFrame, config: dict, step_log: list[dict], output_df: pd.DataFrame) -> dict:
    """The provenance record: which input, which configuration, which
    steps ran and what each one did to the row count, and what came out --
    everything needed, months later, to answer "which data produced this
    number?" without guessing.
    """
    return {
        "input_hash": content_hash(raw_df),
        "config_hash": config_hash(config),
        "steps": step_log,
        "output_hash": content_hash(output_df),
    }

"""Individual pipeline steps.

Each step is a pure function from frame to frame, named for what it does,
and testable on its own against a small fixture -- the cure for the
notebook's defining hazard, out-of-order cell execution. No step mutates
its input in place; every step returns a new frame built with `.copy()`.

Two steps clip `amount` to an upper bound, and only one of them belongs in
the real pipeline:

- `clip_amount_to_fixed_ceiling` reads its threshold from `config` --
  computed once, outside the pipeline, and frozen. Idempotent: clipping
  already-clipped data to the same fixed number changes nothing.
- `clip_amount_to_recomputed_percentile` is kept here ONLY as the lesson's
  worked failure. It recomputes its threshold from whatever data happens to
  be passing through it, which means a second call sees already-clipped
  data and computes a NEW, lower threshold from it -- non-idempotent by
  construction. `pipeline.py` never calls it; `test_pipeline.py` calls it
  directly to prove the failure before the fix.
"""

from __future__ import annotations

import pandas as pd


def parse_currency_amount(df: pd.DataFrame) -> pd.DataFrame:
    """Strip '$' and ',' from `amount` and convert to float64.

    Idempotent by inspection, not just by luck: once `amount` is numeric,
    the "still text" guard below is False and the step is a no-op on a
    second call, rather than re-stripping characters that are no longer
    there. The guard checks for a non-numeric dtype rather than the
    specific `object` dtype, because pandas 3.0's default string inference
    gives a plain Python string column its own `str` extension dtype, not
    `object` -- a version-specific fact this lab's own run confirmed.
    """
    df = df.copy()
    if not pd.api.types.is_numeric_dtype(df["amount"]):
        cleaned = df["amount"].str.replace(r"[$,]", "", regex=True)
        df["amount"] = pd.to_numeric(cleaned, errors="coerce")
    return df


def normalize_region_strings(df: pd.DataFrame) -> pd.DataFrame:
    """Strip whitespace and title-case `region` (" north" -> "North").

    Idempotent: an already-normalised string title-cases to itself.
    """
    df = df.copy()
    df["region"] = df["region"].str.strip().str.title()
    return df


def dedupe_orders(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Drop rows that describe the same real-world order, keeping the
    first occurrence after an explicit, deterministic sort by `order_id`.

    The sort matters: `drop_duplicates(keep="first")` depends on row
    order, and pandas does not promise the input arrived in `order_id`
    order. Sorting first makes "first occurrence" mean the same thing on
    every run, not whatever order the source happened to hand rows over
    in.

    Order-dependent by design (exercise 6): comparing `region` before it
    has been normalised treats " north" and "north" as different values,
    so the resubmitted order in this lab's data is NOT recognised as a
    duplicate unless `normalize_region_strings` already ran.
    """
    df = df.sort_values("order_id", kind="stable").reset_index(drop=True)
    df = df.drop_duplicates(subset=config["dedupe_subset"], keep="first")
    return df.reset_index(drop=True)


def impute_missing_amount(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Fill a missing `amount` with the column's own mean.

    Idempotent in the sense this pipeline relies on: once every `amount`
    is filled, `fillna` on a column with no missing values is a no-op, so
    a second call changes nothing -- provided no later step reintroduces a
    missing value, which none of this pipeline's steps do.
    """
    df = df.copy()
    df["amount"] = df["amount"].fillna(df["amount"].mean())
    return df


def clip_amount_to_fixed_ceiling(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Clip `amount` to `config["amount_clip_max"]` -- a fixed number read
    from configuration, never recomputed from the data. This is the
    correct, idempotent version. Compare `clip_amount_to_recomputed_percentile`
    below, which is not.
    """
    df = df.copy()
    df["amount"] = df["amount"].clip(upper=config["amount_clip_max"])
    return df


def clip_amount_to_recomputed_percentile(df: pd.DataFrame) -> pd.DataFrame:
    """DELIBERATELY NON-IDEMPOTENT. Never called by `pipeline.run_pipeline`.

    Computes its own threshold -- the 99th percentile of `amount` -- from
    whatever frame is passed in, every time it runs. The first call clips
    the true outliers down to the raw data's 99th percentile. Because the
    ceiling is now lower than it was, the SECOND call sees a narrower
    column and computes a new, lower 99th percentile from it, clipping
    again. `pipeline(pipeline(df))` therefore does not equal `pipeline(df)`
    when this step is used -- kept here only so
    `test_pipeline.py::test_1` can run it, watch it fail, and then run the
    fixed version and watch it pass.
    """
    df = df.copy()
    ceiling = df["amount"].quantile(0.99)
    df["amount"] = df["amount"].clip(upper=ceiling)
    return df


def add_amount_zscore(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Attach a z-score computed against FIXED reference statistics in
    `config`, never against this frame's own (possibly already-clipped,
    already-deduplicated) mean and standard deviation. Recomputing the
    reference from the current frame would make this step non-idempotent
    for exactly the same reason the broken clip step is.
    """
    df = df.copy()
    mean = config["amount_reference_mean"]
    std = config["amount_reference_std"]
    df["amount_zscore"] = (df["amount"] - mean) / std
    return df


def sort_deterministic(df: pd.DataFrame) -> pd.DataFrame:
    """Sort by `amount` with `order_id` as an explicit tie-break.

    Two rows in this lab's data collide exactly at the clip ceiling
    (900.0), so sorting by `amount` alone leaves their relative order
    unspecified -- a "stable" sort only preserves whatever order the ROWS
    ARRIVED in, which is not the same thing as a deterministic order
    across two independently built frames. Naming `order_id` as the
    tie-break makes the final row order a fact about the values, not
    about arrival order.
    """
    df = df.copy()
    return df.sort_values(["amount", "order_id"], kind="stable").reset_index(drop=True)

"""YOUR test suite for Day 126 -- "A Pipeline You Can Re-run". Nine exercises.

Run it from the lab directory, not from here:

    pytest starter -v

Every exercise below ends in a `pytest.skip(...)` line. pytest reports a
skip as `s` in the dot line and moves on, so an unfinished suite still
exits 0. Replace each skip with real assertions -- deleting the skip line
is part of the exercise. `starter/00_brief.md` explains each exercise in
full; `data.py`, `steps.py` and `pipeline.py` are the pipeline itself, not
exercises -- read them before you start.

Assert exact values everywhere; nothing in this lab depends on timing.
"""

from __future__ import annotations

import json

import pandas as pd
import pytest

import pipeline as P
import steps as S
from data import CONFIG, build_raw_orders

# --------------------------------------------------------------------------
# EXERCISE 1 -- idempotence. pipeline(pipeline(df)) must equal pipeline(df)
# exactly. First reproduce the worked failure, then prove the real
# pipeline does not share it. See starter/00_brief.md exercise 1.
#
# Check with:   pytest starter -v -k test_1
# --------------------------------------------------------------------------


def test_1_broken_clip_step_is_not_idempotent(raw_orders, config):
    pytest.skip(
        "exercise 1a: run parse_currency_amount, normalize_region_strings, dedupe_orders "
        "and impute_missing_amount, then call clip_amount_to_recomputed_percentile once and "
        "again on its own output. Assert the two results are NOT equal, and that order_id 7's "
        "amount differs between the two calls (approx 1236.5, then approx 1223.675)."
    )


def test_1_real_pipeline_is_idempotent(raw_orders, config):
    pytest.skip(
        "exercise 1b: call pipeline.apply_steps_logged on raw_orders, then call it again on "
        "that result. Assert the two frames are equal exactly (.equals(), no approx)."
    )


# --------------------------------------------------------------------------
# EXERCISE 2 -- determinism. Two independent runs on the same input must
# hash identically, and an explicit tie-break must make the final row
# order independent of arrival order.
#
# Check with:   pytest starter -v -k test_2
# --------------------------------------------------------------------------


def test_2_two_independent_runs_produce_an_identical_hash(config):
    pytest.skip(
        "exercise 2a: run the pipeline twice, each time on a FRESH build_raw_orders() call. "
        "Assert the two output frames are .equals() and that pipeline.content_hash agrees on both."
    )


def test_2_tie_break_makes_the_final_order_deterministic_regardless_of_arrival_order(raw_orders, config):
    pytest.skip(
        "exercise 2b: order_id 2 and order_id 7 both land on amount 900.0 after clipping. "
        "Show that sorting by 'amount' ALONE gives a different tie order depending on whether "
        "the rows arrived forward or reversed (prepared.iloc[::-1]), then show that "
        "steps.sort_deterministic gives the SAME order either way, because it names order_id "
        "as an explicit tie-break."
    )


# --------------------------------------------------------------------------
# EXERCISE 3 -- the step log reconciles: every step's rows-out equals the
# next step's rows-in, and the total change equals the sum of the
# per-step changes.
#
# Check with:   pytest starter -v -k test_3
# --------------------------------------------------------------------------


def test_3_step_log_reconciles_between_consecutive_steps(raw_orders, config):
    pytest.skip(
        "exercise 3a: run the pipeline and get its step log. For every consecutive pair of "
        "steps, assert the earlier one's rows_out equals the later one's rows_in. Assert the "
        "total change (last rows_out minus first rows_in) equals the sum of every step's delta, "
        "and that this equals -1."
    )


def test_3_step_log_shows_exactly_where_the_row_count_changed(raw_orders, config):
    pytest.skip(
        "exercise 3b: from the same step log, assert 'dedupe_orders' has delta -1 and every "
        "other step has delta 0."
    )


# --------------------------------------------------------------------------
# EXERCISE 4 -- the input contract raises on a frame with a missing column
# or a wrong dtype, naming the offending column.
#
# Check with:   pytest starter -v -k test_4
# --------------------------------------------------------------------------


def test_4_input_contract_raises_on_a_missing_column(raw_orders, config):
    pytest.skip(
        "exercise 4a: drop the 'priority' column from raw_orders and assert pipeline.run_pipeline "
        "raises pipeline.ContractError with 'priority' in the message (pytest.raises(..., match=...))."
    )


def test_4_input_contract_raises_on_a_wrong_dtype(raw_orders, config):
    pytest.skip(
        "exercise 4b: cast 'order_id' to float64 and assert run_pipeline raises ContractError "
        "with 'order_id' in the message."
    )


# --------------------------------------------------------------------------
# EXERCISE 5 -- the output contract raises when a step is sabotaged so its
# post-condition fails -- proving the contract can genuinely fail.
#
# Check with:   pytest starter -v -k test_5
# --------------------------------------------------------------------------


def test_5_output_contract_raises_when_the_clip_step_is_sabotaged(raw_orders, config, monkeypatch):
    pytest.skip(
        "exercise 5a: use monkeypatch.setattr(pipeline, 'clip_amount_to_fixed_ceiling', a function "
        "that returns its input unchanged), then assert run_pipeline raises ContractError "
        "mentioning 'clip ceiling'."
    )


def test_5_output_contract_passes_once_the_step_is_restored(raw_orders, config):
    pytest.skip(
        "exercise 5b: run the UNMODIFIED pipeline and call pipeline.check_output_contract on "
        "its result directly -- it must raise nothing, confirming the sabotage above, not "
        "something else, was what triggered exercise 5a's failure."
    )


# --------------------------------------------------------------------------
# EXERCISE 6 -- a .pipe() chain gives exactly the same frame as sequential
# application.
#
# Check with:   pytest starter -v -k test_6
# --------------------------------------------------------------------------


def test_6_pipe_chain_equals_sequential_application(raw_orders, config):
    pytest.skip(
        "exercise 6: assert pipeline.run_pipeline(raw_orders, config)[0] and "
        "pipeline.run_pipeline_via_pipe(raw_orders, config) are .equals()."
    )


# --------------------------------------------------------------------------
# EXERCISE 7 -- order dependence. normalize_region_strings before
# dedupe_orders is the declared order, and it is not arbitrary.
#
# Check with:   pytest starter -v -k test_7
# --------------------------------------------------------------------------


def test_7_declared_order_catches_the_resubmitted_order(raw_orders, config):
    pytest.skip(
        "exercise 7a: run the real pipeline. Assert the result has 6 rows, order_id 3 (the "
        "resubmission) is gone, and order_id 1 (the original) survives."
    )


def test_7_reversed_order_misses_the_resubmitted_order(raw_orders, config):
    pytest.skip(
        "exercise 7b: run pipeline.run_pipeline_swapped_order. Assert it has 7 rows -- nothing "
        "was deduplicated -- and that BOTH order_id 1 and order_id 3 survive."
    )


# --------------------------------------------------------------------------
# EXERCISE 8 -- a Parquet checkpoint round-trip preserves every dtype
# exactly, including a nullable Int64 column with a missing value.
#
# Check with:   pytest starter -v -k test_8
# --------------------------------------------------------------------------


def test_8_parquet_checkpoint_preserves_every_dtype_exactly(raw_orders, tmp_path):
    pytest.skip(
        "exercise 8: parse and normalise raw_orders (do not dedupe yet, so order_id 4's missing "
        "priority is still present), checkpoint it to tmp_path / 'checkpoint.parquet' with "
        "pipeline.checkpoint_to_parquet, reload it with pipeline.load_checkpoint, and assert "
        "every dtype matches exactly, the frames are .equals(), and the reloaded 'priority' "
        "column is still Int64 with order_id 4's value still missing."
    )


# --------------------------------------------------------------------------
# EXERCISE 9 -- the manifest's hashes are stable across runs, and changing
# one input byte changes the input hash AND the output hash.
#
# Check with:   pytest starter -v -k test_9
# --------------------------------------------------------------------------


def test_9_manifest_hashes_are_stable_across_independent_runs(config):
    pytest.skip(
        "exercise 9a: build a manifest from two independent build_raw_orders() runs through the "
        "pipeline. Assert input_hash, config_hash, output_hash and steps all agree between the two."
    )


def test_9_changing_one_input_byte_changes_both_input_and_output_hash(raw_orders, config):
    pytest.skip(
        "exercise 9b: build a manifest for raw_orders, then change order_id 6's amount from "
        "'$60.00' to '$60.01' (one character) and build a second manifest. Assert input_hash and "
        "output_hash both differ, config_hash stays the same, and the row count is unaffected."
    )


def test_9_manifest_is_json_serialisable(raw_orders, config):
    pytest.skip(
        "exercise 9c: build a manifest, round-trip it through json.dumps/json.loads, and assert "
        "the reloaded input_hash and steps match the original manifest's."
    )

#!/usr/bin/env bash
# Tests for the Day 126 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# The harness proves the lesson's claims by running code and reading real
# values, never by reading source:
#
#   * a step that recomputes its clip threshold from whatever data is
#     currently passing through it is NOT idempotent -- pipeline(pipeline(df))
#     changes order 7's amount from 1236.5 to 1223.675 -- while the real
#     pipeline, which reads its threshold from config, is idempotent exactly;
#   * two independent runs on the same input hash identically, and an
#     explicit order_id tie-break makes the final row order the same
#     regardless of which order two tied rows arrived in;
#   * the step log reconciles: every step's rows-out equals the next step's
#     rows-in, and the total change equals the sum of the per-step deltas;
#   * the input contract raises, naming the column, on a missing column or a
#     wrong dtype;
#   * the output contract raises, naming the violated condition, when the
#     clip step is sabotaged into a no-op;
#   * a .pipe() chain produces a frame identical to sequential application;
#   * normalising region strings before deduplicating catches a resubmitted
#     order that reversing the order misses entirely;
#   * a Parquet checkpoint round-trip preserves every dtype exactly,
#     including a nullable Int64 column's missing value;
#   * a manifest's input, config and output hashes are stable across
#     independent runs, and changing one input byte changes both the input
#     hash and the output hash;
#   * 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 FAIL, then restoring it;
#   * nothing -- no .parquet, .json or .csv file, no __pycache__ -- is left
#     behind by this run.
#
# Everything after the one-time install runs offline. Nothing binds a port,
# nothing writes outside the lab, nothing needs a key. Deterministic,
# non-interactive, exits 0 only if every check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

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

# Bytecode left by an EARLIER command is not this run's litter. The README
# documents `pytest starter -q` and `pytest examples -q` separately, and
# running either writes .pyc files that would then fail the cleanliness
# check at the end of this script -- failing the reader for following the
# instructions. Clearing them here makes that final check measure what it
# claims to: what THIS run left behind. `.venv` is untouched, because the
# packages' own bytecode is theirs, not ours.
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
}

check_eq() {
  # check_eq <label> <expected> <actual>
  if [ "$2" = "$3" ]; then
    check "$1" "yes"
  else
    check "$1 (expected [$2], got [$3])" "no"
  fi
}

# Resolve pytest: an explicit override, then this lab's .venv, then PATH.
# Fails loudly with instructions rather than silently skipping checks.
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 126 — A Pipeline You Can Re-run"
echo

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

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

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

pandas_version="$("${python_bin}" -c "import pandas; print(pandas.__version__)" 2>/dev/null || echo "")"
pinned_pandas="$(grep -m1 '^pandas==' "${lab_dir}/requirements/requirements.txt" | cut -d= -f3)"
check_eq "installed pandas matches requirements.txt exactly" "${pinned_pandas}" "${pandas_version}"
echo

# --------------------------------------------------------------------------
echo "2. 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 17 passed, 0 failed" "$( echo "${examples_passed_line}" | grep -qE '^17 passed' && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "3. 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 17 skipped, 0 failed" "$( echo "${starter_output}" | grep -qE '^17 skipped' && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "4. Never run 'pytest examples starter' in one invocation -- every"
echo "   module name (data, steps, pipeline, conftest, test_pipeline) is"
echo "   shared between both directories, so the second collected can"
echo "   shadow, or outright collide with, the first. Checked below."
# --------------------------------------------------------------------------

both_output="$(cd "${lab_dir}" && "${pytest_bin}" examples starter -q 2>&1)"
both_status=$?
check "pytest examples starter (one invocation) does NOT exit 0" "$( [ ${both_status} -ne 0 ] && echo yes || echo no )"
check "pytest examples starter reports an import file mismatch, not a quiet partial run" "$( echo "${both_output}" | grep -qi 'import file mismatch' && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "5. 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 FAIL, then restore."
# --------------------------------------------------------------------------

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

cp "${lab_dir}/examples/test_pipeline.py" "${scratch_dir}/test_pipeline.py"
cp "${lab_dir}/examples/data.py" "${scratch_dir}/data.py"
cp "${lab_dir}/examples/steps.py" "${scratch_dir}/steps.py"
cp "${lab_dir}/examples/pipeline.py" "${scratch_dir}/pipeline.py"
cp "${lab_dir}/examples/conftest.py" "${scratch_dir}/conftest.py"

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 17 passed" "$( echo "${solved_output}" | grep -qE '^17 passed' && echo yes || echo no )"

# Break test_1's exact idempotence assertion on purpose.
sed -i.bak 's/assert once\.equals(twice)/assert not once.equals(twice)/' "${scratch_dir}/test_pipeline.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 FAIL/failed line" "$( echo "${broken_output}" | grep -qiE 'failed|assert' && echo yes || echo no )"

mv "${scratch_dir}/test_pipeline.py.bak" "${scratch_dir}/test_pipeline.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 17 passed again" "$( echo "${restored_output}" | grep -qE '^17 passed' && echo yes || echo no )"

cleanup_scratch
trap - EXIT
echo

# --------------------------------------------------------------------------
echo "6. Nothing in examples/ or starter/ opens a network connection"
# --------------------------------------------------------------------------

url_hits="$(grep -rEl 'https?://|ftp://' "${lab_dir}/examples" "${lab_dir}/starter" 2>/dev/null || true)"
check "no URLs inside examples/ or starter/" "$( [ -z "${url_hits}" ] && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "7. A pipeline day that litters would be embarrassing -- confirm no"
echo "   .parquet, .json or .csv artifact is left behind anywhere in the lab"
# --------------------------------------------------------------------------

artifact_hits="$(find "${lab_dir}" -name '.venv' -prune -o -name 'expected-output' -prune -o \
  \( -type f \( -name '*.parquet' -o -name '*.json' -o -name '*.csv' \) -print \) 2>/dev/null || true)"
check "no stray .parquet/.json/.csv files under the lab (outside expected-output/)" "$( [ -z "${artifact_hits}" ] && echo yes || echo no )"
echo

# --------------------------------------------------------------------------
echo "8. Cleanliness -- nothing left behind by THIS run"
# --------------------------------------------------------------------------

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 )"
echo

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