File size: 8,472 Bytes
d2d1903 | 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 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 | """
Tests for Stage 67 — CSV variant of backfill.
GET /tenants/{tid}/backfill.csv?entity_type&from&until&step_days
Wide table, sorted as_of_day ASC so a chart in Excel reads
left-to-right. Each row carries per-cutoff counts + the single
top issue flattened to three columns (top_entity_id, top_score,
top_title) — Excel doesn't love nested arrays.
"""
import csv
import io
import pytest
pytest.importorskip("fastapi")
pytest.importorskip("httpx")
pytest.importorskip("yaml")
from fastapi.testclient import TestClient
from infra import OrgStateService
from infra.api import create_app, handlers
from orgstate_client import Client
def _seed_calibrated(dbfile, tenant_id="acme"):
from verticals import get_vertical_config, get_vertical_observation_loader
svc = OrgStateService(dbfile)
try:
svc.register_tenant(tenant_id, tenant_id,
vertical="logistics")
cfg = get_vertical_config("logistics").entity_type("warehouse")
obs = get_vertical_observation_loader(
"logistics")()["warehouse"]
svc.ingest_observations(tenant_id, "warehouse", [
{"entity_id": o.entity_id, "day": o.day, "values": o.values}
for o in obs
])
svc.calibrate_and_store(tenant_id, obs, cfg,
vertical="logistics")
finally:
svc.close()
def _bootstrap(tmp_path):
dbfile = str(tmp_path / "bf_csv.sqlite3")
_seed_calibrated(dbfile, "acme")
_seed_calibrated(dbfile, "globex")
svc = OrgStateService(dbfile)
try:
keys = {
"ro": svc.create_api_key("acme", role="readonly").raw,
"admin": svc.create_admin_key().raw,
}
finally:
svc.close()
return dbfile, keys
def _client(dbfile):
return TestClient(create_app(dbfile))
def _auth(k):
return {"Authorization": f"Bearer {k}"}
# --- pure handler ----------------------------------------------------
@pytest.fixture
def seeded_dbfile(tmp_path):
dbfile = str(tmp_path / "h.sqlite3")
_seed_calibrated(dbfile, "acme")
return dbfile
def test_handler_header_row(seeded_dbfile):
svc = OrgStateService(seeded_dbfile)
try:
text = handlers.backfill_csv(svc, "acme", "warehouse")
finally:
svc.close()
rows = list(csv.reader(io.StringIO(text)))
assert rows[0] == [
"as_of_day", "n_states", "n_issues", "n_decisions",
"top_severity",
"top_entity_id", "top_score", "top_title",
]
def test_handler_body_sorted_by_date_ascending(seeded_dbfile):
"""Chart-friendly order — Excel x-axis left-to-right is past-to-
present."""
svc = OrgStateService(seeded_dbfile)
try:
text = handlers.backfill_csv(svc, "acme", "warehouse")
finally:
svc.close()
rows = list(csv.reader(io.StringIO(text)))
body = rows[1:]
days = [r[0] for r in body]
assert days == sorted(days), (
f"days must be sorted ascending; got {days}"
)
def test_handler_top_issue_flattens_to_three_columns(seeded_dbfile):
"""When a cutoff has issues, the top one gets flattened into
top_entity_id / top_score / top_title columns."""
svc = OrgStateService(seeded_dbfile)
try:
text = handlers.backfill_csv(svc, "acme", "warehouse")
finally:
svc.close()
rows = list(csv.reader(io.StringIO(text)))
# find a row that has issues
with_issue = [r for r in rows[1:] if int(r[2]) > 0]
assert with_issue, "expected at least one cutoff with issues"
r = with_issue[0]
assert r[5] # top_entity_id non-empty
assert r[6] # top_score non-empty
assert r[7] # top_title non-empty
def test_handler_empty_top_renders_as_blank_cells(seeded_dbfile):
"""When a cutoff has zero issues, the top_* cells are blank
(NOT 'None' literal). Excel cells stay empty visually."""
svc = OrgStateService(seeded_dbfile)
try:
text = handlers.backfill_csv(svc, "acme", "warehouse")
finally:
svc.close()
rows = list(csv.reader(io.StringIO(text)))
zero_issue = [r for r in rows[1:] if int(r[2]) == 0]
if not zero_issue:
pytest.skip("no zero-issue cutoffs in this seed")
r = zero_issue[0]
assert r[4] == "" # top_severity blank
assert r[5] == "" # top_entity_id blank
assert r[6] == "" # top_score blank
assert r[7] == "" # top_title blank
def test_handler_unknown_tenant_404(tmp_path):
svc = OrgStateService(str(tmp_path / "empty.sqlite3"))
try:
from infra.api.errors import ApiError
with pytest.raises(ApiError):
handlers.backfill_csv(svc, "ghost", "warehouse")
finally:
svc.close()
# --- HTTP route ------------------------------------------------------
def test_route_returns_text_csv(tmp_path):
dbfile, keys = _bootstrap(tmp_path)
client = _client(dbfile)
r = client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse",
headers=_auth(keys["ro"]),
)
assert r.status_code == 200
assert r.headers["content-type"].startswith("text/csv")
assert "utf-8" in r.headers["content-type"]
def test_route_attachment_filename_carries_tenant_and_entity_type(tmp_path):
dbfile, keys = _bootstrap(tmp_path)
client = _client(dbfile)
r = client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse",
headers=_auth(keys["ro"]),
)
cd = r.headers["content-disposition"]
assert "attachment" in cd
assert "acme" in cd
assert "warehouse" in cd
def test_route_body_parses_as_csv(tmp_path):
dbfile, keys = _bootstrap(tmp_path)
client = _client(dbfile)
text = client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse",
headers=_auth(keys["ro"]),
).text
rows = list(csv.reader(io.StringIO(text)))
assert rows[0][0] == "as_of_day"
assert len(rows) > 1
def test_route_filters_narrow_body(tmp_path):
"""from/until/step_days actually limit the rows that come back."""
dbfile, keys = _bootstrap(tmp_path)
client = _client(dbfile)
weekly = client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse&step_days=7",
headers=_auth(keys["ro"]),
).text
daily = client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse",
headers=_auth(keys["ro"]),
).text
assert len(weekly.splitlines()) < len(daily.splitlines())
def test_route_no_DB_writes(tmp_path):
"""Stage 65/66 contract — CSV variant honors it too."""
dbfile, keys = _bootstrap(tmp_path)
svc = OrgStateService(dbfile)
try:
before = svc.db.query_one(
"SELECT COUNT(*) AS n FROM runs"
)["n"]
finally:
svc.close()
client = _client(dbfile)
client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse",
headers=_auth(keys["ro"]),
)
svc = OrgStateService(dbfile)
try:
after = svc.db.query_one(
"SELECT COUNT(*) AS n FROM runs"
)["n"]
finally:
svc.close()
assert before == after
def test_route_no_key_401(tmp_path):
dbfile, _ = _bootstrap(tmp_path)
client = _client(dbfile)
r = client.get(
"/tenants/acme/backfill.csv?entity_type=warehouse",
)
assert r.status_code == 401
def test_route_unknown_entity_type_404(tmp_path):
dbfile, keys = _bootstrap(tmp_path)
client = _client(dbfile)
r = client.get(
"/tenants/acme/backfill.csv?entity_type=ghost",
headers=_auth(keys["ro"]),
)
assert r.status_code == 404
# --- SDK -----------------------------------------------------------
def test_sdk_download_backfill_csv(tmp_path):
dbfile, keys = _bootstrap(tmp_path)
c = Client(base_url="http://test", tenant_id="acme",
api_key=keys["ro"],
transport=_client(dbfile))
text = c.download_backfill_csv("warehouse")
assert text.startswith("as_of_day,n_states,n_issues,")
def test_sdk_download_backfill_csv_passes_filters(tmp_path):
dbfile, keys = _bootstrap(tmp_path)
c = Client(base_url="http://test", tenant_id="acme",
api_key=keys["ro"],
transport=_client(dbfile))
text = c.download_backfill_csv(
"warehouse",
from_day="2026-02-01", until_day="2026-02-15",
)
rows = list(csv.reader(io.StringIO(text)))
for r in rows[1:]:
assert "2026-02-01" <= r[0] <= "2026-02-15"
|