Testing and Deploying the App

76.1 What can break in a causal tool (40 min)

Ordinary software breaks loudly: it crashes. A causal tool usually breaks quietly: it returns a plausible wrong number. The tests worth writing are about properties the economics guarantees:

Tests use a tiny parameters fixture with two cells, not the real file: they must pass whatever tomorrow's estimates are.

76.2 Testing the engine with pytest (50 min)

Create pytest.ini in the repository root, then tests/conftest.py (the fixture) and tests/test_engine.py:

[pytest]
testpaths = tests
filterwarnings = ignore::DeprecationWarning

"""Shared test fixtures: a tiny, hand-checkable parameter set, so tests don't depend on estimation."""
import pytest


@pytest.fixture
def tiny_params():
    return {
        "version": "test", "created": "2026-01-01", "sources": {},
        "pricing": [
            {"segment": "a", "category": "x", "elasticity": -2.0, "se": 0.0, "margin": 0.3, "price": 10.0,
             "weekly_units": 100.0, "valid_change": 0.1},
            {"segment": "b", "category": "x", "elasticity": -1.0, "se": 0.5, "margin": 0.3, "price": 10.0,
             "weekly_units": 100.0, "valid_change": 0.1},
        ],
        "coupons": {"customers": 1000, "campaigns_per_year": 12, "margin": 0.3, "shares": [0.0, 0.5, 1.0],
                    "uplift": [0.0, 10.0, 15.0], "uplift_se": [0.0, 1.0, 1.0]},
        "assumptions": {"pull_forward": 0.0},
    }

"""Unit tests for the scenario engine (Day 76). Run: pytest"""
import pytest
from pydantic import ValidationError

from engine import Scenario, simulate
from engine.parameters import validate


def test_no_change_means_no_change(tiny_params):
    r = simulate(tiny_params, Scenario())
    assert r["change"]["gross_profit"]["median"] == 0
    assert r["change"]["revenue"]["p95"] == 0
    assert r["change"]["prob_profit_up"] is None


def test_pricing_matches_the_formula(tiny_params):
    # cell a has se = 0, so it's deterministic: profit = (m + d) * p * units * 52 * (1 + d) ** e
    r = simulate(tiny_params, Scenario(price_change={"a": 0.05}))
    cell = next(c for c in r["cells"] if c["segment"] == "a")
    expected = (0.3 + 0.05) * 10 * 100 * 52 * 1.05 ** -2.0 - 0.3 * 10 * 100 * 52
    assert cell["profit_change_median"] == pytest.approx(expected)


def test_most_specific_key_wins():
    sc = Scenario(price_change={"all": 0.05, "a": 0.0, "b/x": 0.02})
    assert sc.change_for("a", "x") == 0.0
    assert sc.change_for("b", "x") == 0.02
    assert sc.change_for("c", "y") == 0.05


def test_coupon_cost_lowers_profit(tiny_params):
    cheap = simulate(tiny_params, Scenario(coupon_share=0.5, coupon_cost=2, seed=1))
    dear = simulate(tiny_params, Scenario(coupon_share=0.5, coupon_cost=8, seed=1))
    assert dear["change"]["gross_profit"]["median"] < cheap["change"]["gross_profit"]["median"]


def test_same_seed_same_answer(tiny_params):
    sc = Scenario(price_change={"all": 0.05}, coupon_share=0.3, seed=7)
    assert simulate(tiny_params, sc) == simulate(tiny_params, sc)


def test_extrapolation_is_flagged(tiny_params):
    assert simulate(tiny_params, Scenario(price_change={"all": 0.05}))["warnings"] == []
    assert len(simulate(tiny_params, Scenario(price_change={"all": 0.15}))["warnings"]) == 2


def test_correlated_errors_widen_the_range(tiny_params):
    tiny_params["pricing"][0]["se"] = 0.5
    width = {}
    for rho in [0.0, 1.0]:
        g = simulate(tiny_params, Scenario(price_change={"all": 0.05}, elasticity_correlation=rho))["change"]["gross_profit"]
        width[rho] = g["p95"] - g["p5"]
    assert width[1.0] > width[0.0]


def test_bad_scenarios_are_rejected():
    with pytest.raises(ValidationError):
        Scenario(price_change={"all": 0.8})
    with pytest.raises(ValidationError):
        Scenario(coupon_share=1.5)


def test_bad_parameters_are_rejected(tiny_params):
    tiny_params["pricing"][0]["elasticity"] = 0.5
    with pytest.raises(ValueError, match="non-negative elasticity"):
        validate(tiny_params)

python -m pytest -q

The engine tests run in under a second. When one fails, the name says which promise was broken.

Try it. Add tests/test_more_properties.py with two more promises: a 5% rise in a cell with elasticity −2 lowers revenue but raises profit; and more pull-forward always means less coupon profit.

# tests/test_more_properties.py   (python -m pytest -q)
from engine import Scenario, simulate


def test_price_rise_lowers_revenue_when_demand_is_elastic(tiny_params):
    r = simulate(tiny_params, Scenario(price_change={"a": 0.05}))       # cell a: elasticity -2, no noise
    assert r["change"]["revenue"]["median"] < 0
    assert r["change"]["gross_profit"]["median"] > 0                   # ... while profit rises


def test_pull_forward_lowers_coupon_profit(tiny_params):
    # spend that is only pulled forward isn't extra: the more of it, the less coupons earn (same draws, seed 0)
    profits = [simulate(tiny_params, Scenario(coupon_share=0.5, pull_forward=p))["change"]["gross_profit"]["median"]
               for p in [0.0, 0.3, 0.6, 0.9]]
    assert profits == sorted(profits, reverse=True)

python -m pytest -q

16 passed

76.3 Testing the API and the app (45 min)

The API tests point the service at the tiny fixture by setting PARAMS_PATH with pytest's monkeypatch and clearing the cached parameters. The app test is Day 74's AppTest, with assertions. Create tests/test_api.py and tests/test_app.py:

"""API tests (Day 76): the contract other teams rely on."""
import json

import pytest
from fastapi.testclient import TestClient


@pytest.fixture
def client(tiny_params, tmp_path, monkeypatch):
    path = tmp_path / "params.json"
    path.write_text(json.dumps(tiny_params))
    monkeypatch.setenv("PARAMS_PATH", str(path))
    from service import app as app_module
    app_module.params.cache_clear()
    return TestClient(app_module.app)


def test_health_reports_the_parameters_version(client):
    assert client.get("/health").json() == {"status": "ok", "parameters_version": "test"}


def test_scenario_returns_projection_and_counterfactual(client):
    body = client.post("/scenario", json={"price_change": {"all": 0.05}}).json()
    assert {"counterfactual", "projected", "change", "cells", "warnings"} <= body.keys()
    assert "draws" not in body


def test_unknown_segment_is_a_422(client):
    r = client.post("/scenario", json={"price_change": {"zzz": 0.05}})
    assert r.status_code == 422
    assert "zzz" in r.json()["detail"]


def test_compare_keeps_order(client):
    rows = client.post("/compare", json=[{"name": "one", "scenario": {}},
                                         {"name": "two", "scenario": {"coupon_share": 0.5}}]).json()["rows"]
    assert [r["name"] for r in rows] == ["one", "two"]

"""Streamlit app tests (Day 76), with Streamlit's own AppTest: runs the script without a browser."""
from streamlit.testing.v1 import AppTest


def test_app_runs_and_reacts_to_a_slider():
    at = AppTest.from_file("../app/streamlit_app.py", default_timeout=60).run()   # relative to this file
    assert not at.exception
    assert "Move a slider" in at.info[0].value
    at.slider[0].set_value(5).run()                      # price change everywhere: +5%
    assert not at.exception
    assert at.metric[0].label == "Gross profit, change per year"
    assert at.metric[2].value == "100%"

python -m pytest -q

16 passed, 1 warning in 8.41s

(The warning is a deprecation notice from the test client's HTTP library, not from our code.)

76.4 Continuous integration with GitHub Actions (40 min)

Tests only protect you if they run. GitHub Actions runs them on every push and pull request, on a clean machine, with the pinned requirements. Create .github/workflows/ci.yml:

# Runs the tests on every push and pull request (Day 76).
name: tests
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.13"
          cache: pip
      - run: pip install -r requirements.txt
      - run: python -m pytest -q

Commit and push, then open the repository's Actions tab: a green tick means a fresh machine could install everything and every promise held. Break one on purpose (make simulate return revenue for profit), push, watch it go red, and revert. From now on, a red cross on a pull request blocks the merge.

76.5 Day 76 mini-project: a container and a smoke test (75 min)

Deployment means running the API somewhere other teams can reach. Whatever the platform (Cloud Run, Azure Container Apps, Fly.io, an internal Kubernetes cluster), the unit you hand over is a container image. The API needs only FastAPI, numpy and pydantic at run time, not DoubleML or LightGBM, so it gets its own small requirements file. Create requirements-serve.txt and Dockerfile:

# Only what the API needs at run time (Day 76): the estimation stack stays out of the container.
fastapi==0.150.0
uvicorn==0.54.0
numpy==2.4.6
pydantic==2.14.2

# The scenario API as a container (Day 76).
#   docker build -t northmart-scenarios .
#   docker run -p 8001:8001 northmart-scenarios
FROM python:3.13-slim
WORKDIR /srv
COPY requirements-serve.txt .
RUN pip install --no-cache-dir -r requirements-serve.txt
COPY engine/ engine/
COPY service/ service/
COPY models/causal_parameters.json models/causal_parameters.json
EXPOSE 8001
CMD ["uvicorn", "service.app:app", "--host", "0.0.0.0", "--port", "8001"]

Codespaces has Docker:

docker build -t northmart-scenarios .
docker run -d -p 8001:8001 --name scenarios northmart-scenarios

Note what goes into the image: the engine, the service and one parameters file. The image is the model version. Rebuilding the parameters means building and deploying a new image, which is exactly the audit trail you want.

After every deployment, run a smoke test against the live URL: is it up, is it serving the parameters version we expect, and does a known scenario give the known answer? Create day76/smoke_test.py:

# day76/smoke_test.py   (python -m day76.smoke_test https://your-api.example.com)
"""
A smoke test for a DEPLOYED API: is it up, is it serving the parameters version we expect, and does a
known scenario give the known answer? Run it after every deployment (and from the scheduler, daily).
Exits with an error code if anything is wrong, so CI or a scheduler can alert on it.
"""
import sys

import requests

from engine import Scenario, load_parameters, simulate

url = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8001"
expected = load_parameters()
problems = []
health = requests.get(f"{url}/health", timeout=10).json()
if health.get("parameters_version") != expected["version"]:
    problems.append(f"serving parameters {health.get('parameters_version')}, expected {expected['version']}")
known = Scenario(price_change={"all": 0.05}, seed=0)
local = simulate(expected, known)["change"]["gross_profit"]["median"]
remote = requests.post(f"{url}/scenario", json=known.model_dump(), timeout=30).json()["change"]["gross_profit"]["median"]
if abs(local - remote) > 1:
    problems.append(f"known scenario gives {remote:,.0f}, expected {local:,.0f}")
print(f"{url}: " + ("OK" if not problems else "PROBLEMS: " + "; ".join(problems)))
sys.exit(1 if problems else 0)

python -m day76.smoke_test            # defaults to http://127.0.0.1:8001

http://127.0.0.1:8001: OK

(If nothing is listening on the port you'll get a connection error: start the container, or uvicorn service.app:app --port 8001 &.) Now rebuild the parameters (python -m engine.build) on different data and rerun the smoke test without rebuilding the image: it reports that the server is serving an older version than the repository expects. That mismatch is the most common production bug in model serving, and now it's caught in seconds. Stop the container with docker rm -f scenarios.

76.6 Save your work with Git (20 min)

git add tests pytest.ini .github Dockerfile requirements-serve.txt day76
git commit -m "Day 76: engine, API and app tests; CI; container and smoke test"
git push

Day 76 checklist

• My tests check properties the economics guarantees, on a tiny fixture
• The API and the app are tested without starting a server or a browser
• Tests run automatically on every push
• The API runs in a container with only its run-time dependencies
• A smoke test confirms the deployed version and a known answer