"""2026-07-28 — BE↔FE contract: an ``@idempotent`` endpoint must be CALLED with an ``Idempotency-Key``. ## The failure this exists to prevent PR3 put ``@idempotent`` on 14 endpoints whose replay would produce a second external side effect (a second cloud grant, a second Google group, a second role). The mobile write primitive mints a key for every write. Both halves looked correct — and six of the fourteen were **dead**, because the API client function had no way to accept the key: // web/src/api/cloudLifecycle.ts (before) async decommissionScope(id: string, reason: string) { const r = await apiClient.post(`/cmdb/scopes/${id}/decommission`, { reason }) // ^ no header } The hook minted a key, the call site dropped it, the request went out bare, and the decorator's very first branch is ``if key is None: return await func(...)``. The guard was a no-op on the *irreversible* operation the PR body specifically named — and nothing anywhere failed. This is the same shape as the mobile deep-link bug class: two halves that are each individually correct and are not wired to each other. The fix is the same too — pin the seam with a test, because no amount of care at review time catches a missing third argument. ## What is asserted For every ``@idempotent``-decorated route, find the frontend API-client function(s) that call its path. If one exists, its body **must** mention ``Idempotency-Key``. An endpoint with no frontend caller at all is fine (CLI/MCP-only surfaces exist) and is reported, not failed — the point is that a caller may not silently drop the key. The frontend is parsed with a brace-matched block scan rather than a real TS parser (no node dependency in the backend test env). That is adequate here: the API client layer is uniform (``async name(...) { ... }`` one level deep in an exported object) and the scan asserts it found a plausible number of functions, so a format drift shows up as a failure rather than as silent under-collection. """ from __future__ import annotations import ast import re from pathlib import Path import pytest _REPO = Path(__file__).resolve().parents[3] _SERVER_APP = _REPO / "server" / "app" _WEB_SRC = _REPO / "web" / "src" _WEB_API = _WEB_SRC / "api" # Paths whose client callers legitimately live outside `web/src/api` (or # nowhere). Keep this EMPTY unless there is a real reason — an entry here is # a hole in the guard, so each one carries its justification. _NO_CLIENT_EXPECTED: dict[str, str] = {} def _norm(path: str) -> str: """Normalise a route/template path for cross-language comparison. ``/api/v1/cmdb/scopes/{scope_id}/decommission`` (FastAPI) and ``/cmdb/scopes/${id}/decommission`` (axios, baseURL carries /api/v1) both become ``/cmdb/scopes/*/decommission``. """ path = re.sub(r"^/api/v1", "", path) path = re.sub(r"\$\{[^}]*\}", "*", path) # axios template hole path = re.sub(r"\{[^}]*\}", "*", path) # FastAPI path param return path.rstrip("/") or "/" # -------------------------------------------------------------------------- # Backend: collect @idempotent routes # -------------------------------------------------------------------------- def _router_prefixes(tree: ast.AST) -> dict[str, str]: """Map router variable name → its ``prefix=`` literal.""" out: dict[str, str] = {} for node in ast.walk(tree): if not isinstance(node, ast.Assign) or len(node.targets) != 1: continue target = node.targets[0] if not isinstance(target, ast.Name): continue call = node.value if not isinstance(call, ast.Call): continue fn = call.func name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", None) if name != "APIRouter": continue prefix = "" for kw in call.keywords: if kw.arg == "prefix" and isinstance(kw.value, ast.Constant): prefix = str(kw.value.value) out[target.id] = prefix return out def _collect_idempotent_routes() -> list[tuple[str, str, str]]: """→ [(normalised_path, http_method, scope)] for every @idempotent route.""" found: list[tuple[str, str, str]] = [] for py in _SERVER_APP.rglob("*.py"): text = py.read_text(encoding="utf-8") if "@idempotent" not in text: continue tree = ast.parse(text, filename=str(py)) prefixes = _router_prefixes(tree) for node in ast.walk(tree): if not isinstance(node, ast.AsyncFunctionDef | ast.FunctionDef): continue scope: str | None = None route: tuple[str, str] | None = None for dec in node.decorator_list: if not isinstance(dec, ast.Call): continue fn = dec.func # @idempotent("scope.name", ...) if isinstance(fn, ast.Name) and fn.id == "idempotent": if dec.args and isinstance(dec.args[0], ast.Constant): scope = str(dec.args[0].value) # @router.post("/path") if ( isinstance(fn, ast.Attribute) and isinstance(fn.value, ast.Name) and fn.attr in {"get", "post", "put", "patch", "delete"} and dec.args and isinstance(dec.args[0], ast.Constant) ): prefix = prefixes.get(fn.value.id, "") route = (prefix + str(dec.args[0].value), fn.attr.upper()) if scope and route: found.append((_norm(route[0]), route[1], scope)) return found # -------------------------------------------------------------------------- # Frontend: collect API-client functions and the paths they call # -------------------------------------------------------------------------- _FUNC_RE = re.compile(r"^\s{2}(?:async\s+)?([A-Za-z_$][\w$]*)\s*[(<]", re.M) # Captures the verb too: a GET on the same path as an idempotent POST is a # different operation and must not be reported as a key-dropping caller. _PATH_RE = re.compile( r"apiClient\.(\w+)\s*(?:<[^()]*?>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]" ) def _match(text: str, i: int, open_ch: str, close_ch: str) -> int: """Index of the bracket closing the one at ``i``, or -1.""" depth = 0 while i < len(text): if text[i] == open_ch: depth += 1 elif text[i] == close_ch: depth -= 1 if depth == 0: return i i += 1 return -1 def _body_span(text: str, name_end: int) -> tuple[int, int]: """Locate a method body given the index just past its name. Finding the body is fiddlier than "first ``{`` after the name" — that lands inside an object-typed parameter (``body: { note?: string }``) and then brace-matching closes on the parameter list, so the body is never scanned and the function looks like it calls nothing. Two hazards, in order: the parameter list, then a return type that itself contains braces (``): Promise<{ data: T }> {``). Skip the parens, then take the first ``{`` that appears at angle-bracket depth zero. """ paren = text.find("(", name_end - 1) if paren == -1: return -1, -1 close = _match(text, paren, "(", ")") if close == -1: return -1, -1 angle, i = 0, close + 1 while i < len(text): ch = text[i] if ch == "<": angle += 1 elif ch == ">": angle = max(0, angle - 1) elif ch == "{" and angle == 0: end = _match(text, i, "{", "}") return (i, end) if end != -1 else (-1, -1) elif ch == ";": # an interface/type member, not an implementation return -1, -1 i += 1 return -1, -1 def _raw_page_calls() -> list[tuple[str, str, str, bool]]: """Direct ``apiClient.('/path')`` calls made OUTSIDE ``src/api``. The api-client layer is not the only place that writes. Two admin consoles called the seat-revoke endpoint straight from the page and so were invisible to the function scan below — both without a key, on an endpoint whose whole point is that a replay must not dispatch a second vendor-side removal. "The wrapper CAN forward a key" is a much weaker property than "every caller does". → [(relpath, METHOD, normalised path, sends_key)], one per call site. """ out: list[tuple[str, str, str, bool]] = [] for ts in sorted(_WEB_SRC.rglob("*.ts*")): if _WEB_API in ts.parents or ts.name.endswith((".test.ts", ".test.tsx")): continue text = ts.read_text(encoding="utf-8") if "apiClient." not in text: continue for m in _PATH_RE.finditer(text): verb, path = m.group(1), m.group(2) if verb.lower() not in {"post", "put", "patch", "delete"}: continue # The header may be on the same call, a few lines below (the # config object is often multi-line), or the call may be inside # a helper that received the key. Look at a generous window and # let a false negative be impossible rather than a false # positive be annoying. window = text[m.start() : m.start() + 600] out.append(( ts.relative_to(_REPO).as_posix(), verb.upper(), _norm(path), "Idempotency-Key" in window, )) return out def _client_functions() -> list[tuple[str, str, set[tuple[str, str]], bool]]: """→ [(file, func_name, {(METHOD, normalised path)}, sends_key)].""" out: list[tuple[str, str, set[tuple[str, str]], bool]] = [] for ts in sorted(_WEB_API.glob("*.ts")): text = ts.read_text(encoding="utf-8") for m in _FUNC_RE.finditer(text): start, end = _body_span(text, m.end()) if start == -1: continue body = text[start : end + 1] calls = { (verb.upper(), _norm(path)) for verb, path in _PATH_RE.findall(body) } if calls: out.append( (ts.name, m.group(1), calls, "Idempotency-Key" in body) ) return out # -------------------------------------------------------------------------- def test_scanners_find_a_plausible_amount() -> None: """Guard the guard: silent under-collection would make this test vacuous.""" routes = _collect_idempotent_routes() assert len(routes) >= 14, f"only found {len(routes)} @idempotent routes" funcs = _client_functions() # 639 at the time of writing. The floor is deliberately close to it: the # first version of this scanner silently collected 187 (it brace-matched # an object-typed parameter instead of the body) and every assertion # below was vacuously green. assert len(funcs) >= 550, f"only parsed {len(funcs)} API client functions" # And it must actually see the header where we know one exists. assert any(sends for _f, _n, _c, sends in funcs) def test_every_client_caller_of_an_idempotent_route_sends_the_key() -> None: routes = _collect_idempotent_routes() funcs = _client_functions() offenders: list[str] = [] uncalled: list[str] = [] for path, method, scope in routes: callers = [f for f in funcs if (method, path) in f[2]] if not callers: if f"{method} {path}" not in _NO_CLIENT_EXPECTED: uncalled.append(f"{scope} → {method} {path}") continue for file, name, _paths, sends in callers: if not sends: offenders.append( f"{file}::{name}() calls {method} {path} " f"(@idempotent '{scope}') " f"but never sets an Idempotency-Key header" ) # Direct page-level writes, which the function scan cannot see. for rel, method, path, sends in _raw_page_calls(): if (path, method) in {(p, m) for p, m, _s in [(r[0], r[1], r[2]) for r in routes]} and not sends: offenders.append( f"{rel} calls {method} {path} directly via apiClient " f"without an Idempotency-Key (the endpoint is @idempotent)" ) assert not offenders, ( "API client functions drop the idempotency key, making the backend " "guard a no-op:\n " + "\n ".join(sorted(offenders)) ) # Not a failure — CLI/MCP-only endpoints are legitimate — but surfaced so # a genuinely unreachable endpoint does not hide here forever. if uncalled: print("\n[info] @idempotent routes with no web API client caller:") for u in sorted(uncalled): print(" -", u) @pytest.mark.parametrize( ("raw", "expected"), [ ("/api/v1/cmdb/scopes/{scope_id}/decommission", "/cmdb/scopes/*/decommission"), ("/cmdb/scopes/${id}/decommission", "/cmdb/scopes/*/decommission"), ("/email-groups/requests", "/email-groups/requests"), ("/users/${userId}/roles", "/users/*/roles"), ], ) def test_path_normalisation(raw: str, expected: str) -> None: """The two languages must land on the same string, or the whole test is vacuously green.""" assert _norm(raw) == expected