"""End-to-end exercise of the API against a throwaway database.""" import os import tempfile TMP = tempfile.mkdtemp() os.environ["PARTS_DB"] = os.path.join(TMP, "test.db") os.environ["PARTS_PASSWORD"] = "hunter2" os.environ["PARTS_SECRET"] = "test-secret" os.environ["PARTS_AUTH"] = "on" from fastapi.testclient import TestClient # noqa: E402 from app.main import app # noqa: E402 failures = [] def check(label, condition, detail=""): print((" PASS " if condition else " FAIL ") + label + (f" [{detail}]" if detail and not condition else "")) if not condition: failures.append(label) with TestClient(app) as client: # --- session tokens survive every signature byte value --- import app.auth as _auth import base64 as _b64, hashlib as _hl, hmac as _hm def _issue_at(exp, epoch=0): payload = f"{exp}:{epoch}".encode() return _b64.urlsafe_b64encode( payload + _hm.new(_auth._secret(), payload, _hl.sha256).digest()).decode() _base = int(__import__("time").time()) + 90 * 86400 _bad = [e for e in range(_base, _base + 3000) if not _auth.token_valid(_issue_at(e))] check("no issued token fails its own validator", not _bad, f"{len(_bad)}/3000 failed") check("tampered token rejected", not _auth.token_valid(_issue_at(_base)[:-2] + "AA")) check("expired token rejected", not _auth.token_valid(_issue_at(int(__import__("time").time()) - 5))) check("garbage token rejected", not _auth.token_valid("not-a-token")) # --- auth gate --- check("unauthenticated list is 401", client.get("/api/parts").status_code == 401) check("wrong password rejected", client.post("/api/login", json={"password": "nope"}).status_code == 401) check("healthz open", client.get("/healthz").json() == {"ok": True}) r = client.post("/api/login", json={"password": "hunter2"}) check("login succeeds", r.status_code == 200, r.text) check("session cookie set", "parts_session" in client.cookies) check("me reports authenticated", client.get("/api/me").json()["authenticated"] is True) # --- seeded taxonomy --- cats = client.get("/api/categories").json()["items"] check("categories seeded", len(cats) > 30, str(len(cats))) filament = next(c for c in cats if c["name"] == "Filament") check("filament nests under 3D Printing", filament["path"] == "3D Printing / Filament", filament["path"]) check("filament default unit is grams", filament["unit"] == "g") check("filament has spec template", any(s["key"] == "Diameter" for s in filament["spec_template"])) resistors = next(c for c in cats if c["name"] == "Resistors") electronics = next(c for c in cats if c["name"] == "Electronics") locs = client.get("/api/locations").json()["items"] workshop = next(l for l in locs if l["name"] == "Workshop") r = client.post("/api/locations", json={"name": "Bin A3", "parent_id": workshop["id"]}) check("nested location created", r.status_code == 201, r.text) bin_a3 = r.json()["id"] paths = {l["id"]: l["path"] for l in client.get("/api/locations").json()["items"]} check("location path nests", paths[bin_a3] == "Workshop / Bin A3", paths[bin_a3]) check("duplicate location rejected", client.post("/api/locations", json={"name": "Bin A3", "parent_id": workshop["id"]}).status_code == 409) # --- create parts --- r = client.post("/api/parts", json={ "name": "10k resistor 0603", "category_id": resistors["id"], "location_id": bin_a3, "manufacturer": "Yageo", "mpn": "RC0603FR-0710KL", "quantity": 480, "unit": "pcs", "min_quantity": 50, "cost_each": 0.01, "specs": [{"key": "Resistance", "value": "10k"}, {"key": "Package", "value": "0603"}, {"key": "Tolerance", "value": "1%"}], "tags": ["smd", "passives"], }) check("part created", r.status_code == 201, r.text) resistor = r.json() check("specs round-trip", len(resistor["specs"]) == 3) check("tags round-trip", resistor["tags"] == ["passives", "smd"], str(resistor["tags"])) check("category path resolved", resistor["category_path"] == "Electronics / Resistors", str(resistor["category_path"])) check("not low stock at 480/50", resistor["low_stock"] is False) r = client.post("/api/parts", json={ "name": "PLA Black", "category_id": filament["id"], "location_id": workshop["id"], "manufacturer": "Prusament", "quantity": 640, "unit": "g", "min_quantity": 200, "specs": [{"key": "Material", "value": "PLA"}, {"key": "Colour", "value": "Black"}, {"key": "Diameter", "value": "1.75mm"}], "tags": ["3d-printing"], }) check("filament created with gram unit", r.status_code == 201 and r.json()["unit"] == "g", r.text) pla = r.json() r = client.post("/api/parts", json={ "name": "M3x8 socket cap screw", "location_id": bin_a3, "quantity": 12, "unit": "pcs", "min_quantity": 20, "specs": [{"key": "Thread", "value": "M3"}, {"key": "Length", "value": "8mm"}], }) screw = r.json() check("low stock flagged at 12/20", screw["low_stock"] is True) # --- search --- def ids(q, **kw): params = {"q": q, **kw} return [p["name"] for p in client.get("/api/parts", params=params).json()["items"]] check("search by name", "10k resistor 0603" in ids("10k")) check("search by mpn", "10k resistor 0603" in ids("RC0603")) check("search by manufacturer", "PLA Black" in ids("prusament")) check("search by spec value", "PLA Black" in ids("1.75mm")) check("search by tag", "10k resistor 0603" in ids("smd")) check("prefix search while typing", "PLA Black" in ids("prus")) check("search by location name", "10k resistor 0603" in ids("Bin A3")) check("nonsense search returns nothing", ids("zzzznope") == []) check("punctuation does not crash FTS", isinstance(ids('0.1uF "quoted" AND OR *'), list)) # --- filters --- parent_filtered = client.get("/api/parts", params={"category_id": electronics["id"]}).json() check("parent category catches children", parent_filtered["total"] == 1, str(parent_filtered["total"])) check("location filter includes descendants", client.get("/api/parts", params={"location_id": workshop["id"]}).json()["total"] == 3) check("low stock filter", [p["name"] for p in client.get("/api/parts", params={"low_stock": "true"}).json()["items"]] == ["M3x8 socket cap screw"]) check("tag filter", client.get("/api/parts", params={"tag": "passives"}).json()["total"] == 1) # --- rolled-up counts match what the filter returns --- cats_now = {c["name"]: c for c in client.get("/api/categories").json()["items"]} check("parent count rolls up children", cats_now["Electronics"]["part_count"] == 1, str(cats_now["Electronics"]["part_count"])) check("parent has no direct parts", cats_now["Electronics"]["direct_count"] == 0) check("leaf count is its own", cats_now["Resistors"]["part_count"] == 1) locs_now = {l["name"]: l for l in client.get("/api/locations").json()["items"]} check("location count rolls up", locs_now["Workshop"]["part_count"] == 3, str(locs_now["Workshop"]["part_count"])) # --- sorting & paging --- names = [p["name"] for p in client.get("/api/parts", params={"sort": "name"}).json()["items"]] check("sort by name", names == sorted(names, key=str.lower), str(names)) check("sort by quantity ascending", [p["quantity"] for p in client.get("/api/parts", params={"sort": "quantity"}).json()["items"]] == [12, 480, 640]) page = client.get("/api/parts", params={"sort": "name", "limit": 2, "offset": 0}).json() check("paging returns total plus page", page["total"] == 3 and len(page["items"]) == 2) # --- stock adjustments --- r = client.post(f"/api/parts/{pla['id']}/adjust", json={"delta": -140, "reason": "bracket print"}) check("adjust decrements", r.json()["quantity"] == 500, r.text) r = client.post(f"/api/parts/{pla['id']}/adjust", json={"delta": -400, "reason": "big print"}) check("adjust flags low stock", r.json()["low_stock"] is True and r.json()["quantity"] == 100) r = client.post(f"/api/parts/{screw['id']}/adjust", json={"delta": -999}) check("quantity floors at zero", r.json()["quantity"] == 0) history = client.get(f"/api/parts/{pla['id']}/history").json()["items"] check("history records initial stock plus both adjustments", len(history) == 3, str(len(history))) check("history is newest first", history[0]["reason"] == "big print") # --- updates --- r = client.patch(f"/api/parts/{resistor['id']}", json={"name": "10k resistor 0603 1%", "specs": [{"key": "Resistance", "value": "10 kilohm"}]}) check("patch renames", r.json()["name"] == "10k resistor 0603 1%") check("patch replaces specs", len(r.json()["specs"]) == 1) check("patch leaves quantity alone", r.json()["quantity"] == 480) check("reindex picks up new spec text", "10k resistor 0603 1%" in ids("kilohm")) check("old spec value no longer matches", ids("0603") != [] and "kilohm" not in str(ids("10k"))) r = client.patch(f"/api/parts/{resistor['id']}", json={"quantity": 5}) check("patch quantity alone works", r.json()["quantity"] == 5) # --- taxonomy edits keep parts --- r = client.post("/api/categories", json={"name": "Salvage", "unit": "pcs", "spec_template": [{"key": "Source", "value": ""}]}) check("custom category created", r.status_code == 201, r.text) salvage_id = r.json()["id"] check("custom template stored", any(c["id"] == salvage_id and c["spec_template"] == [{"key": "Source"}] for c in client.get("/api/categories").json()["items"])) client.delete(f"/api/categories/{salvage_id}") client.delete(f"/api/locations/{bin_a3}") survivor = client.get(f"/api/parts/{resistor['id']}").json() check("deleting a location keeps the part", survivor["location_id"] is None) check("part still searchable after location delete", "10k resistor 0603 1%" in ids("10k") or "10k resistor 0603 1%" in ids("resistor")) # --- deletion --- check("delete part", client.delete(f"/api/parts/{screw['id']}").status_code == 200) check("deleted part is gone", client.get(f"/api/parts/{screw['id']}").status_code == 404) check("deleted part leaves the index", "M3x8 socket cap screw" not in ids("M3x8")) check("missing part 404s", client.get("/api/parts/999999").status_code == 404) check("history of a missing part 404s", client.get("/api/parts/999999/history").status_code == 404) # httpx refuses to serialise inf/nan, so these go as raw bodies — which is # exactly how a real client would smuggle them in: Python's json.loads # accepts the bare `Infinity` and `NaN` tokens. JSONH = {"Content-Type": "application/json"} def raw_post(path, body): return client.post(path, content=body, headers=JSONH) def raw_patch(path, body): return client.patch(path, content=body, headers=JSONH) check("Infinity quantity rejected", raw_post("/api/parts", '{"name":"inf","quantity":Infinity}').status_code == 422) check("-Infinity quantity rejected", raw_post("/api/parts", '{"name":"ninf","quantity":-Infinity}').status_code == 422) check("NaN quantity rejected", raw_post("/api/parts", '{"name":"nan","quantity":NaN}').status_code == 422) check("Infinity cost rejected", raw_post("/api/parts", '{"name":"inf2","cost_each":Infinity}').status_code == 422) check("Infinity min_quantity rejected", raw_post("/api/parts", '{"name":"inf3","min_quantity":Infinity}').status_code == 422) _infp = client.post("/api/parts", json={"name": "inf target", "quantity": 5}).json()["id"] check("Infinity adjustment delta rejected", raw_post(f"/api/parts/{_infp}/adjust", '{"delta":Infinity}').status_code == 422) check("NaN adjustment delta rejected", raw_post(f"/api/parts/{_infp}/adjust", '{"delta":NaN}').status_code == 422) check("Infinity PATCH quantity rejected", raw_patch(f"/api/parts/{_infp}", '{"quantity":Infinity}').status_code == 422) check("the part is untouched after a rejected Infinity", client.get(f"/api/parts/{_infp}").json()["quantity"] == 5) check("adjusting a missing part 404s", client.post("/api/parts/999999/adjust", json={"delta": 1}).status_code == 404) # --- stats & validation --- # Derived from the listing rather than a magic number, so adding a fixture # part earlier in the file can't silently invalidate it. s = client.get("/api/stats").json() listed = client.get("/api/parts", params={"limit": 1000}).json() check("stats part count matches the listing", s["parts"] == listed["total"], str(s)) check("stats low-stock count matches the listing", s["low_stock"] == sum(1 for p in listed["items"] if p["low_stock"]), str(s)) check("stats value matches the listing", s["estimated_value"] == round(sum(p["quantity"] * (p["cost_each"] or 0) for p in listed["items"]), 2), str(s)) check("whitespace-only name rejected", client.post("/api/parts", json={"name": " "}).status_code == 422) check("negative quantity rejected", client.post("/api/parts", json={"name": "neg", "quantity": -5}).status_code == 422) check("negative min_quantity rejected", client.post("/api/parts", json={"name": "neg2", "min_quantity": -1}).status_code == 422) check("blank unit rejected", client.post("/api/parts", json={"name": "u", "unit": " "}).status_code == 422) check("unknown category on create 404s", client.post("/api/parts", json={"name": "ghost", "category_id": 999999}).status_code == 404) # --- regression: quantity changes are always logged --- r = client.post("/api/parts", json={"name": "log target", "quantity": 10}) logged = r.json()["id"] before = len(client.get(f"/api/parts/{logged}/history").json()["items"]) client.patch(f"/api/parts/{logged}", json={"quantity": 3, "reason": "stocktake"}) hist = client.get(f"/api/parts/{logged}/history").json()["items"] check("PATCH quantity writes history", len(hist) == before + 1, f"{before} -> {len(hist)}") check("PATCH history records the signed delta", hist[0]["delta"] == -7, str(hist[0]["delta"])) check("PATCH history keeps the reason", hist[0]["reason"] == "stocktake") n_before = len(client.get(f"/api/parts/{logged}/history").json()["items"]) client.patch(f"/api/parts/{logged}", json={"name": "log target renamed"}) check("PATCH without quantity writes no history", len(client.get(f"/api/parts/{logged}/history").json()["items"]) == n_before) client.patch(f"/api/parts/{logged}", json={"quantity": 3}) check("PATCH to the same quantity writes no history", len(client.get(f"/api/parts/{logged}/history").json()["items"]) == n_before) # --- regression: a floored adjustment logs what was applied, not requested --- r = client.post("/api/parts", json={"name": "floor target", "quantity": 7}) floored = r.json()["id"] client.post(f"/api/parts/{floored}/adjust", json={"delta": -999, "reason": "floor"}) fh = client.get(f"/api/parts/{floored}/history").json()["items"] check("floored adjustment logs the applied delta", fh[0]["delta"] == -7, str(fh[0]["delta"])) check("floored adjustment lands at zero", fh[0]["quantity_after"] == 0) check("stock log sums to current quantity", sum(h["delta"] for h in fh) == client.get(f"/api/parts/{floored}").json()["quantity"]) # --- regression: renaming taxonomy rebuilds the search index --- shelf = client.post("/api/locations", json={"name": "Old Shelf"}).json()["id"] client.post("/api/parts", json={"name": "shelf widget", "location_id": shelf, "quantity": 1}) check("part found by its location name", client.get("/api/parts", params={"q": "Old Shelf"}).json()["total"] == 1) client.patch(f"/api/locations/{shelf}", json={"name": "New Shelf"}) check("rename drops the stale location term", client.get("/api/parts", params={"q": "Old Shelf"}).json()["total"] == 0) check("rename indexes the new location term", client.get("/api/parts", params={"q": "New Shelf"}).json()["total"] == 1) # Nested: renaming a PARENT has to reindex everything beneath it, because # the full path is what gets indexed. outer = client.post("/api/locations", json={"name": "Old Room"}).json()["id"] inner = client.post("/api/locations", json={"name": "Inner Bin", "parent_id": outer}).json()["id"] client.post("/api/parts", json={"name": "nested widget", "location_id": inner, "quantity": 1}) check("part found by its parent location", client.get("/api/parts", params={"q": "Old Room"}).json()["total"] == 1) client.patch(f"/api/locations/{outer}", json={"name": "New Room"}) check("parent rename reindexes descendants' parts", client.get("/api/parts", params={"q": "Old Room"}).json()["total"] == 0 and client.get("/api/parts", params={"q": "New Room"}).json()["total"] == 1) client.delete(f"/api/locations/{outer}") check("deleting a location clears its term from the index", client.get("/api/parts", params={"q": "New Room"}).json()["total"] == 0) check("the part itself survives the delete", client.get("/api/parts", params={"q": "nested widget"}).json()["total"] == 1) # --- regression: PATCH null handling --- check("explicit null name is a 422, not a 500", client.patch(f"/api/parts/{logged}", json={"name": None}).status_code == 422) check("explicit null unit is a 422", client.patch(f"/api/parts/{logged}", json={"unit": None}).status_code == 422) check("explicit null category_id is allowed", client.patch(f"/api/parts/{logged}", json={"category_id": None}).status_code == 200) check("explicit null cost_each is allowed", client.patch(f"/api/parts/{logged}", json={"cost_each": None}).status_code == 200) # --- regression: taxonomy cycles and missing ids --- a = client.post("/api/categories", json={"name": "CycleA"}).json()["id"] b = client.post("/api/categories", json={"name": "CycleB", "parent_id": a}).json()["id"] c = client.post("/api/categories", json={"name": "CycleC", "parent_id": b}).json()["id"] check("direct self-parent rejected", client.patch(f"/api/categories/{a}", json={"name": "CycleA", "parent_id": a}).status_code == 400) check("indirect cycle rejected", client.patch(f"/api/categories/{a}", json={"name": "CycleA", "parent_id": b}).status_code == 400) check("deep indirect cycle rejected", client.patch(f"/api/categories/{a}", json={"name": "CycleA", "parent_id": c}).status_code == 400) check("legitimate re-parent still allowed", client.patch(f"/api/categories/{c}", json={"name": "CycleC", "parent_id": a}).status_code == 200) check("PATCH missing category 404s", client.patch("/api/categories/999999", json={"name": "ghost"}).status_code == 404) check("DELETE missing category 404s", client.delete("/api/categories/999999").status_code == 404) check("PATCH missing location 404s", client.patch("/api/locations/999999", json={"name": "ghost"}).status_code == 404) check("DELETE missing location 404s", client.delete("/api/locations/999999").status_code == 404) check("unknown parent 404s", client.post("/api/categories", json={"name": "orphan", "parent_id": 999999}).status_code == 404) check("blank category name rejected", client.post("/api/categories", json={"name": " "}).status_code == 422) dup = client.post("/api/categories", json={"name": "DupTarget"}).json()["id"] client.post("/api/categories", json={"name": "DupOther"}) check("PATCH into a duplicate name is a 409, not a 500", client.patch(f"/api/categories/{dup}", json={"name": "DupOther"}).status_code == 409) # --- security headers and asset caching --- r = client.get("/") check("nosniff header", r.headers.get("x-content-type-options") == "nosniff") check("frame-options header", r.headers.get("x-frame-options") == "DENY") check("referrer-policy header", r.headers.get("referrer-policy") == "no-referrer") check("content-security-policy header", "default-src 'self'" in r.headers.get("content-security-policy", "")) check("index is not cacheable", "no-store" in r.headers.get("cache-control", "")) check("api responses are not cacheable", "no-store" in client.get("/api/stats").headers.get("cache-control", "")) check("index references content-hashed assets", "/static/app.js?v=" in r.text and "/static/app.css?v=" in r.text) import re as _re digest = _re.search(r"/static/app\.js\?v=([0-9a-f]+)", r.text).group(1) check("versioned asset is immutably cacheable", "immutable" in client.get(f"/static/app.js?v={digest}").headers.get("cache-control", "")) check("unversioned asset is not cached", client.get("/static/app.js").headers.get("cache-control") == "no-cache") check("hsts only when the original request was https", "strict-transport-security" not in r.headers and "strict-transport-security" in client.get("/", headers={"X-Forwarded-Proto": "https"}).headers) # --- frontend & logout --- check("index served", client.get("/").status_code == 200) check("css served", client.get("/static/app.css").status_code == 200) check("missing static 404s", client.get("/static/nope.css").status_code == 404) check("unknown api path 404s json", client.get("/api/nope").status_code == 404) client.post("/api/logout") check("logout clears session", client.get("/api/parts").status_code == 401) # --- failed-login throttling (last: it trips global state deliberately) --- auth_mod = __import__("app.auth", fromlist=["auth"]) auth_mod.clear_failures() codes = [client.post("/api/login", json={"password": "wrong"}).status_code for _ in range(12)] check("repeated wrong passwords start returning 429", 429 in codes, str(codes)) check("throttle kicks in only after the configured budget", codes.index(429) == auth_mod._max_failures(), str(codes)) blocked = client.post("/api/login", json={"password": "hunter2"}) check("the correct password is refused while throttled", blocked.status_code == 429) check("429 tells the client when to retry", blocked.headers.get("retry-after", "").isdigit()) auth_mod.clear_failures() check("a successful login works again once the window clears", client.post("/api/login", json={"password": "hunter2"}).status_code == 200) check("a successful login resets the failure budget", auth_mod.login_retry_after() == 0) # --- password management through the API --- check("bootstrap password is flagged", client.get("/api/me").json()["using_bootstrap_password"] is True) check("min length is advertised", client.get("/api/me").json()["min_password_length"] == 8) check("wrong current password is refused", client.post("/api/password", json={"current_password": "nope", "new_password": "brand-new-secret"}).status_code == 403) # The change-password route verifies the same credential as the login form, # so it has to spend from the same budget instead of only topping it up. auth_mod.clear_failures() pw_codes = [client.post("/api/password", json={"current_password": "wrong", "new_password": "brand-new-secret"}).status_code for _ in range(auth_mod._max_failures() + 3)] check("guessing through change-password eventually 429s", 429 in pw_codes, str(pw_codes)) check("change-password throttles on the same budget as login", pw_codes.index(429) == auth_mod._max_failures(), str(pw_codes)) check("a throttled change-password blocks even the right password", client.post("/api/password", json={"current_password": "hunter2", "new_password": "brand-new-secret"}).status_code == 429) check("login is throttled too once the budget is spent", client.post("/api/login", json={"password": "hunter2"}).status_code == 429) auth_mod.clear_failures() check("clearing the budget restores change-password", client.post("/api/password", json={"current_password": "wrong", "new_password": "x"}).status_code in (403, 422)) auth_mod.clear_failures() check("short new password rejected", client.post("/api/password", json={"current_password": "hunter2", "new_password": "short"}).status_code == 422) check("reusing the current password rejected", client.post("/api/password", json={"current_password": "hunter2", "new_password": "hunter2"}).status_code == 422) check("still logged in after failed attempts", client.get("/api/parts").status_code == 200) # A second, independent session that should be cut off by the change. other = TestClient(app) other.post("/api/login", json={"password": "hunter2"}) check("second session works before the change", other.get("/api/parts").status_code == 200) r = client.post("/api/password", json={"current_password": "hunter2", "new_password": "correct-horse-battery"}) check("password change succeeds", r.status_code == 200, r.text) check("change reports other sessions signed out", r.json()["other_sessions_signed_out"] is True) check("the changing session stays signed in", client.get("/api/parts").status_code == 200) check("other sessions are signed out", other.get("/api/parts").status_code == 401) check("bootstrap flag clears", client.get("/api/me").json()["using_bootstrap_password"] is False) fresh = TestClient(app) check("the old password no longer works", fresh.post("/api/login", json={"password": "hunter2"}).status_code == 401) auth_mod.clear_failures() check("the bootstrap env password is ignored once one is set", fresh.post("/api/login", json={"password": os.environ["PARTS_PASSWORD"]}).status_code == 401) auth_mod.clear_failures() check("the new password works", fresh.post("/api/login", json={"password": "correct-horse-battery"}).status_code == 200) check("the new session can read data", fresh.get("/api/parts").status_code == 200) # --- sign out other devices, without signing out this one --- third = TestClient(app) third.post("/api/login", json={"password": "correct-horse-battery"}) check("third session works", third.get("/api/parts").status_code == 200) check("revoke succeeds", client.post("/api/sessions/revoke").status_code == 200) check("revoke keeps the calling session", client.get("/api/parts").status_code == 200) check("revoke cuts off the other session", third.get("/api/parts").status_code == 401) # --- epoch bumps must not lose increments --- from concurrent.futures import ThreadPoolExecutor as _Pool import app.db as _dbmod def _bump_once(_): with _dbmod.session() as _c: auth_mod.bump_epoch(_c) with _dbmod.session() as _c: _epoch_before = auth_mod.current_epoch(_c) with _Pool(max_workers=20) as _ex: list(_ex.map(_bump_once, range(20))) with _dbmod.session() as _c: _epoch_after = auth_mod.current_epoch(_c) check("20 concurrent epoch bumps advance it by exactly 20", _epoch_after - _epoch_before == 20, f"{_epoch_before} -> {_epoch_after}") # --- the hash itself --- import app.db as _db with _db.session() as _conn: stored = auth_mod.stored_hash(_conn) check("password is stored hashed, not in the clear", stored is not None and stored.startswith("scrypt$") and "correct-horse-battery" not in stored) check("the stored hash verifies", auth_mod.verify_hash("correct-horse-battery", stored)) check("the stored hash rejects a wrong password", not auth_mod.verify_hash("wrong", stored)) check("two hashes of the same password differ (salted)", auth_mod.hash_password("same") != auth_mod.hash_password("same")) # --- recovery CLI --- import app.admin as _admin check("admin set-password works", _admin.main(["set-password", "cli-set-password"]) == 0) cli = TestClient(app) check("CLI-set password logs in", cli.post("/api/login", json={"password": "cli-set-password"}).status_code == 200) auth_mod.clear_failures() check("CLI change signed out the API session", client.get("/api/parts").status_code == 401) check("admin rejects a short password", _admin.main(["set-password", "abc"]) == 1) check("admin show-status works", _admin.main(["show-status"]) == 0) check("admin rejects an unknown command", _admin.main(["nonsense"]) == 2) check("admin clear-password works", _admin.main(["clear-password"]) == 0) back = TestClient(app) check("clearing restores the env password", back.post("/api/login", json={"password": os.environ["PARTS_PASSWORD"]}).status_code == 200) check("bootstrap flag returns after clearing", back.get("/api/me").json()["using_bootstrap_password"] is True) print() if failures: print(f"{len(failures)} FAILED: " + "; ".join(failures)) raise SystemExit(1) print("all checks passed")