File size: 4,866 Bytes
0b9787b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """API contract tests — run against the real precomputed artifacts.
Covers every route: response shape, parameter validation, 404s, cross-endpoint
consistency (the recommendation endpoint must agree with the raw forecast it is
derived from), and action-list invariants.
"""
import sys
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.main import app
KNOWN = ("CA_1", "FOODS_3_090")
@pytest.fixture(scope="module")
def client():
# context manager triggers the startup event that loads the parquet artifacts
with TestClient(app) as c:
yield c
class TestHealth:
def test_health(self, client):
r = client.get("/health")
assert r.status_code == 200
assert r.json() == {"status": "ok"}
class TestSkus:
def test_shape(self, client):
r = client.get("/api/skus").json()
assert len(r["stores"]) == 10
assert len(r["items"]) == 3049
assert KNOWN[0] in r["stores"] and KNOWN[1] in r["items"]
class TestForecast:
def test_arrays_aligned(self, client):
r = client.get(f"/api/forecast/{KNOWN[0]}/{KNOWN[1]}")
assert r.status_code == 200
b = r.json()
n = len(b["dates"])
assert n == 28
assert len(b["p10"]) == len(b["p50"]) == len(b["p90"]) == len(b["actual"]) == n
assert len(b["history_dates"]) == len(b["history_actual"]) == 56
def test_quantiles_ordered(self, client):
b = client.get(f"/api/forecast/{KNOWN[0]}/{KNOWN[1]}").json()
for lo, mid, hi in zip(b["p10"], b["p50"], b["p90"]):
assert lo <= mid <= hi
def test_unknown_series_404(self, client):
assert client.get("/api/forecast/XX_9/NOPE").status_code == 404
class TestRecommendation:
def test_defaults(self, client):
b = client.get(f"/api/recommendation/{KNOWN[0]}/{KNOWN[1]}").json()
assert b["service_level"] == 0.95
assert b["lead_time_days"] == 7
assert b["suggested_order_qty"] >= 0
def test_inventory_reduces_order(self, client):
base = f"/api/recommendation/{KNOWN[0]}/{KNOWN[1]}"
q0 = client.get(base, params={"current_inventory": 0}).json()
q100 = client.get(base, params={"current_inventory": 100}).json()
assert q100["suggested_order_qty"] <= q0["suggested_order_qty"]
# reorder point is independent of what's on the shelf
assert q100["reorder_point"] == q0["reorder_point"]
def test_higher_service_level_bigger_cushion(self, client):
base = f"/api/recommendation/{KNOWN[0]}/{KNOWN[1]}"
lo = client.get(base, params={"service_level": 0.90}).json()
hi = client.get(base, params={"service_level": 0.99}).json()
assert hi["safety_stock"] > lo["safety_stock"]
def test_agrees_with_forecast(self, client):
"""Recommendation must be derivable from the served forecast (no drift
between endpoints)."""
fc = client.get(f"/api/forecast/{KNOWN[0]}/{KNOWN[1]}").json()
rec = client.get(f"/api/recommendation/{KNOWN[0]}/{KNOWN[1]}").json()
expected_avg = sum(fc["p50"][:7]) / 7
assert rec["avg_daily_demand"] == pytest.approx(expected_avg, rel=1e-3)
@pytest.mark.parametrize("bad", [{"service_level": 0}, {"service_level": 1.5},
{"lead_time_days": 0}, {"current_inventory": -5}])
def test_validation_422(self, client, bad):
r = client.get(f"/api/recommendation/{KNOWN[0]}/{KNOWN[1]}", params=bad)
assert r.status_code == 422
class TestActionList:
def test_counts_sum_to_total(self, client):
b = client.get(f"/api/action_list/{KNOWN[0]}").json()
assert b["total_items"] == 3049
assert sum(b["counts"].values()) == b["total_items"]
def test_sorted_by_order_size(self, client):
items = client.get(f"/api/action_list/{KNOWN[0]}?limit=50").json()["items"]
qtys = [i["suggested_order_qty"] for i in items]
assert qtys == sorted(qtys, reverse=True)
def test_zero_inventory_means_everything_ordered(self, client):
b = client.get(f"/api/action_list/{KNOWN[0]}?current_inventory=0").json()
assert b["counts"].get("order_now", 0) + b["counts"].get("order_soon", 0) >= 3000
def test_cached_response_identical(self, client):
url = f"/api/action_list/{KNOWN[0]}?current_inventory=42"
assert client.get(url).json() == client.get(url).json()
class TestBacktest:
def test_summary_shape(self, client):
b = client.get("/api/backtest_summary").json()
for policy in ("naive", "recommended"):
p = b["policies"][policy]
assert 0 <= p["stockout_rate_pct"] <= 100
assert p["avg_holding_units"] >= 0
assert len(b["top_skus"]) == 20
|