diff --git a/README.md b/README.md index 9929e43..7a150c1 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,8 @@ creation, issue/work item creation and comments, git worktrees (creation and merged-worktree/orphaned-test-DB cleanup), a commit-and-push helper with pre-flight checks, Azure log queries, and static checks (`bdt lint`) for postgres/psycopg SQL rules, pydantic-model placement, hand-wired HTTP access in -TypeScript, and baseline tooling. +TypeScript and baseline tooling, plus `bdt dead-code` for `.sql` files nothing loads and +backend routes nothing calls. `bdt pr *` and `bdt issue *` auto-detect whether the current repo's `origin` remote is Azure DevOps or GitHub and use `az`/`gh` accordingly. Consolidates near-duplicate scripts that used to be copy-pasted across @@ -516,8 +517,9 @@ bdt logs fetch --env prod --out logs/ --keep-archive ## `bdt lint` Static checks (implementing [bmsuisse/skills#52](https://github.com/bmsuisse/skills/issues/52)) -for the `postgres-best-practices` skill's SQL rules, pydantic-model placement, and -that the repo has its baseline tooling actually set up: +for the `postgres-best-practices` skill's SQL rules, pydantic-model placement, hand-wired HTTP in +TypeScript, and that the repo has its baseline tooling actually set up. Unused `.sql` files and +dead backend routes are a separate command, [`bdt dead-code`](#bdt-dead-code): ```bash bdt lint # scan the current directory, recursively @@ -606,6 +608,112 @@ already pulled in transitively via `pgdevkit[db]`, but declared explicitly as the `bmsdna-devtools[lint]` extra; a clear install hint is printed (not a raw `ImportError`) if it's ever missing. +## `bdt dead-code` + +Dead code a linter can't see, in one command and one config table, `[tool.bdt.dead_code]`: + +- **`sql`** -- `.sql` files that no Python code loads; +- **`routes`** -- backend (FastAPI) routes that neither non-generated frontend code nor a `url_for(...)` + call uses. + +Each check runs when it is configured (`sql_roots`, resp. `[[tool.bdt.dead_code.apps]]`); `--only sql` / +`--only routes` (repeatable) narrows a run, e.g. to keep the slow one -- `routes` imports the app, so run +`bdt` with the repo's own interpreter (`uv run bdt dead-code`) -- out of a prek hook. Exit code is 0 (clean), 1 +(findings) or 2 (setup problem: invalid config or pyproject.toml, nothing configured, the app doesn't import +within 5 minutes -- the error shows the import's output). + +### `sql` -- unreferenced `.sql` files + +A `.sql` file that no Python code loads is a query left behind after its caller was deleted or renamed. Name the +folders that hold *loadable* SQL (not schema/migration scripts, which are applied rather than loaded) and every +file under them that nothing references is reported (`sql-file-unreferenced`): + +```toml +[tool.bdt.dead_code] +sql_roots = ["backend/db/queries"] # repo-relative; a typo'd root is itself reported +sql_loader_functions = ["load_sql"] # default; add your own loader's name if it differs +sql_unreferenced_ignore = ["backend/db/queries/legacy/*.sql"] # globs for files reached some other way +# exclude_dirs = ["generated"] # extra directory names to skip, beyond .venv/node_modules/etc. +``` + +A file counts as referenced by a literal `load_sql("topic", "name")` call (positional or `topic=`/`name=`; +topic = the file's parent directory, name = its stem); by a `load_sql("topic", some_var)` call in a Python file that +also contains the stem as a string literal (so `name = "a" if x else "b"` and lookup tables work); +by a string literal that is a path whose trailing segments equal the file's repo-relative path +(`get_sql_with_prm_list("backend/api/sql/x.sql")`); or by its bare filename as a literal in a +Python file under the SQL folder's parent (`_SQL_DIR / "x.sql"`). It never executes code, so a +file reached through a fully computed path is a false positive -- list it in +`sql_unreferenced_ignore`. References are searched across the whole repo, since the caller can live anywhere. +A Python file that can't be parsed (including syntax newer than the interpreter running `bdt`) is itself +reported as `sql-check-python-unparseable`, because references in it are unknown. + +### `routes` -- backend routes nobody uses + +Backend (FastAPI) operations that **no non-generated frontend code calls and no backend code references by +name** -- dead routes that still have to be maintained, secured and tested. + +```toml +[[tool.bdt.dead_code.apps]] +name = "akeneo" # label used in the report +app = "main_app:app" # module:attribute, imported in a subprocess ... +app_dir = "akeneo_editor/backend" # ... with this directory (repo-relative) as cwd/import root +# openapi = "frontend/openapi.json" # alternative to `app`: a committed OpenAPI document +frontends = ["akeneo_editor/frontend/src"] # repo-relative dirs or globs, e.g. "mdmapp/app/react_apps/*/src" +exclude_prefixes = ["/external_api"] # routes meant for other callers (external API, webhooks) +exclude_tags = ["agent"] # e.g. LLM/MCP tools +exclude_paths = ["/auth/*", "GET /health"] # fnmatch globs on "/path" or "METHOD /path" +baseline = "dead-code-baseline.txt" # optional ratchet, see below +# env = { SOME_REQUIRED_SETTING = "x" } # extra environment for importing the app +# exclude_frontend_globs = ["src/legacy/*"] # repo-relative frontend files to ignore as callers +# url_for_functions = ["url_for", "url_path_for"] # default; functions whose first argument names a route +``` + +Pair each backend app with *its own* frontends (one `[[...apps]]` entry per backend) -- a route +called only by another backend's frontend is still unused here. + +The backend inventory is `app.openapi()` of the app **and of every mounted sub-app** (with the +mount prefix), so it needs no committed schema and can't go stale; routes with +`include_in_schema=False` are not considered. + +**Generated code never counts as a caller.** A generated client lists *every* route, so it is +skipped (`generated/`, `*.gen.*`, `*.generated.*`, `api-types*`, `openapi_schema*`, `*.d.ts`), as are +tests and e2e specs (`*.test.*`, `*.spec.*`, `__tests__/`, `tests/`, `e2e/`) -- for TypeScript, +JavaScript and Vue files alike. Comments are blanked before scanning. An operation is called when +non-generated code of one of the app's frontends has (strongest first): + +- **sdk** -- a reference to a hey-api SDK function (read from the `sdk.gen.ts` under that same frontend + directory, matched on method *and* url) or one of its react-query helpers (`fooOptions`, `fooMutation`, + `fooQueryKey`, ...). Each `frontends` directory is scoped to its own SDK, so two apps that both generate + `listItems` aren't confused; +- **fetch** -- an openapi-fetch call `.GET("/path"` naming this method and path (a `DELETE` of the + same path with no call of its own is still reported); +- **url** -- a string/template literal equal to the path template (any method), e.g. a hand-written + `fetch(`/api/x/${id}`)`, `"/api/x/" + id` or an `` -- nested templates like + `${qs ? `?a=${b}` : ""}` are handled; +- **url-sfx** -- a literal (two or more segments) that is a suffix of the route (a client with a base URL, + or a `${base}` prefix). + +**Routes the backend refers to by name are used, too.** Auth redirects and OAuth callbacks have no frontend caller +by nature -- the backend builds their URL: `request.url_for("auth_callback")`, `app.url_path_for("login")`, +`{{ url_for('login') }}` in a template. A string literal passed (first argument, or `name=`) to one of +`url_for_functions` in a non-test Python file or Jinja-style template (`.html`, `.jinja`, `.jinja2`, `.j2`) under the app's `app_dir` +marks the route of that name as used (`"mount:name"` for a mounted sub-app counts as `name`). The route name is recovered from +its `operationId`: FastAPI's default `name + path + method` form, a bare `name`, or `-` from a custom +`generate_unique_id`; an OpenAPI file without operation ids can't be matched this way. It is a text match, so a call +in a comment counts too. + +It is a heuristic: no type information, a URL literal matches every method of that path (and so does an +SPA ``), and URLs assembled from non-literal pieces (including a non-literal route +name in `url_for`) or handed to the client by the server are invisible -- exclude those routes explicitly. An app +that exposes no documented operations (e.g. one wrapped in middleware that hides `openapi()`) is a setup error, +not a clean pass. Run against our repos, the usual legitimate exclusions are health checks, the SPA +catch-all, service-to-service endpoints and an external API folder. + +**Adopting it without fixing everything first:** set `baseline`, run +`uv run bdt dead-code --update-baseline` once (it concerns the `routes` check only), and commit the file. From +then on only *new* unused routes fail -- and a baseline line whose route has since been deleted, excluded or +used is reported as `api-route-baseline-stale`, so the list can only shrink. + ## `bdt find-injection` Scans backend and frontend code for injection risks, and warns when a web project has no diff --git a/bmsdna/devtools/_api_dump.py b/bmsdna/devtools/_api_dump.py new file mode 100644 index 0000000..60d2394 --- /dev/null +++ b/bmsdna/devtools/_api_dump.py @@ -0,0 +1,58 @@ +"""Standalone (stdlib-only) helper of the `routes` check of `bdt dead-code`: list the documented HTTP operations of a +FastAPI-like app, recursing into mounted sub-apps. + +It is executed in the *target repo's* interpreter via `python -c module:attr out.json` +and must therefore never import `bmsdna.devtools`: the working directory is first on `sys.path` there, and a repo may +ship its own top-level `bmsdna` package (CCMT2 does) that would shadow ours. `api_usage` imports it normally for +`ops_from_doc`, so the OpenAPI-document parsing exists exactly once. +""" + +import importlib +import json +import sys + +HTTP_METHODS = ("get", "post", "put", "patch", "delete", "options", "head") + + +def ops_from_doc(doc, mount=""): + """Operations of one OpenAPI document as dicts, with `mount` prepended to every path.""" + ops = [] + for path, item in (doc.get("paths") or {}).items(): + for method, op in item.items(): + if method in HTTP_METHODS and isinstance(op, dict): + ops.append( + { + "method": method.upper(), + "path": mount + path, + "tags": list(op.get("tags") or []), + "mount": mount, + "operation_id": str(op.get("operationId") or ""), + } + ) + return ops + + +def collect(app, mount=""): + """Operations of `app` and, recursively, of every mounted sub-app. Uses only `app.openapi()` and the + `.path`/`.app` attributes of mount routes -- not FastAPI internals, which change between releases (FastAPI 0.141 wraps + included routers in lazy objects that `app.routes` no longer lists as plain routes).""" + openapi = getattr(app, "openapi", None) + if not callable(openapi): + return [] + ops = ops_from_doc(openapi(), mount) + for route in getattr(app, "routes", None) or []: + sub = getattr(route, "app", None) + if sub is not None and callable(getattr(sub, "openapi", None)) and isinstance(getattr(route, "path", None), str): + ops.extend(collect(sub, mount + route.path.rstrip("/"))) + return ops + + +def main(argv): + module_name, _, attr = argv[0].partition(":") + app = getattr(importlib.import_module(module_name), attr) + with open(argv[1], "w", encoding="utf-8") as out: + json.dump(collect(app), out) + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/bmsdna/devtools/api_usage.py b/bmsdna/devtools/api_usage.py new file mode 100644 index 0000000..aca2318 --- /dev/null +++ b/bmsdna/devtools/api_usage.py @@ -0,0 +1,501 @@ +"""The `routes` check of `bdt dead-code`: backend (FastAPI) operations that no non-generated frontend code calls +and no backend code references by name (`url_for`). + +A route nobody calls is dead code that still has to be maintained, secured and tested. The check +needs two inventories: + +- **the backend's operations** -- from the live app (`app.openapi()`, recursing into mounted + sub-apps, run in a subprocess so two backends with a same-named package can't collide) or from a + committed `openapi.json`; +- **the frontend's call sites** -- found in *non-generated* TypeScript/JavaScript/Vue sources only. + Generated code (`*.gen.*`, `*.generated.*`, `generated/`, `*.d.ts`, ...), tests and e2e specs are + never counted: a generated client lists *every* route, so counting it would make everything look called. + Comments are blanked before scanning. + +An operation counts as called when, in non-generated code of one of its app's frontends, there is +(strongest first): + +- **sdk** -- a reference to a hey-api SDK function (read from that frontend's own `sdk.gen.ts`, matched on + method *and* url) or one of its react-query helpers (`fooOptions`, `fooMutation`, `fooQueryKey`, ...); +- **fetch** -- an openapi-fetch style `.GET("/path"` literal naming this method and path; +- **url** -- a string/template literal equal to the path template (any method), e.g. a hand-written + `fetch(`/api/x/${id}`)`, `"/api/x/" + id` or an ``; +- **url-sfx** -- a literal that is a suffix of the route (client with a base URL the scan can't see). + +An operation also counts as used when backend code (Python or Jinja-style templates under the app's `app_dir`, tests +excluded) names its route -- `request.url_for("auth_callback")`, `app.url_path_for("login")`, `{{ url_for('login') }}`. +That is how auth redirects and OAuth callbacks are wired, and they have no frontend caller by nature. + +Deliberately heuristic (no type information; a URL literal matches every method of its path, and so does an SPA +``). Routes that are legitimately not called by the frontend -- external APIs, auth +redirects, health checks, service-to-service calls, LLM/MCP tools -- are excluded by prefix/tag/glob, or ratcheted +through a baseline file. +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import tempfile +from dataclasses import dataclass, field +from fnmatch import fnmatch +from pathlib import Path + +from . import _api_dump +from .lint import _DEFAULT_EXCLUDE_DIR_NAMES, _iter_files +from .lint_findings import Finding +from .lint_typescript import _mask, is_excluded_ts_file + +RULE = "api-route-uncalled" +RULE_STALE_BASELINE = "api-route-baseline-stale" + +SOURCE_SUFFIXES = (".ts", ".tsx", ".mts", ".js", ".jsx", ".vue") +_EXCLUDE_DIR_NAMES = _DEFAULT_EXCLUDE_DIR_NAMES | {".output", ".nuxt", ".next", ".turbo", "storybook-static"} +# `is_excluded_ts_file` only knows .ts/.tsx/.mts names; the same generated/test conventions apply to JS and Vue files. +_EXCLUDE_FILE_RE = re.compile(r"\.(?:gen|generated|test|spec)\.(?:[cm]?[jt]sx?|vue)$|\.d\.[cm]?ts$") +_EXCLUDE_FILE_PREFIXES = ("api-types", "openapi_schema") +_REACT_QUERY_SUFFIXES = ("Options", "Mutation", "InfiniteOptions", "QueryKey", "InfiniteQueryKey", "Query") +_NAME_SOURCE_SUFFIXES = (".py", ".html", ".htm", ".jinja", ".jinja2", ".j2") +_TEST_DIR_NAMES = frozenset({"tests", "test", "__tests__"}) +_TEST_FILE_RE = re.compile(r"^(?:test_.*|.*_test|conftest)\.py$") +DEFAULT_URL_FOR_FUNCTIONS = ("url_for", "url_path_for") +_APP_IMPORT_TIMEOUT_SECONDS = 300 +_MAX_TEMPLATE_CHARS = 4000 # a URL template longer than this is not a URL; also bounds the rescan of an unclosed backtick +_MAX_TEMPLATE_DEPTH = 20 + +# Bounded character classes keep these linear on garbage input (a lone `'${` must not scan to end of file). +_QUOTED_PATH_RE = re.compile(r"""(?P['"])(?P(?:\$\{[^}\n'"]*\})*/[^'"\n]*)(?P=q)""") +_TEMPLATE_START_RE = re.compile(r"`(?=(?:\$\{[^}`\n]*\})*/)") +_FETCH_METHOD_RE = re.compile(r"\.(GET|POST|PUT|PATCH|DELETE)\s*(?:<[^()]*?>)?\(\s*$") +_IDENT_RE = re.compile(r"[A-Za-z_$][\w$]*") +_SDK_FN_RE = re.compile(r"export const (\w+) = ") +_SDK_URL_RE = re.compile(r"""\.(get|post|put|patch|delete|head|options)\b[^;]{0,4000}?url:\s*["']([^"']+)["']""", re.S) + + +class ApiUsageError(Exception): + """A configuration or environment problem (not a lint finding): the app wouldn't import, a path is missing, ...""" + + +@dataclass(frozen=True, slots=True) +class Operation: + method: str # upper-case + path: str # full path including any mount prefix, FastAPI `{param}` style + tags: tuple[str, ...] = () + mount: str = "" # the sub-app mount prefix part of `path` ("" for the root app) + operation_id: str = "" # OpenAPI operationId; carries the route name FastAPI derives it from (see `is_named_by`) + + @property + def key(self) -> str: + return f"{self.method} {self.path}" + + @classmethod + def from_dict(cls, d: dict) -> Operation: + return cls(d["method"], d["path"], tuple(d["tags"]), d["mount"], d.get("operation_id", "")) + + +# --------------------------------------------------------------------------- backend inventory + + +def load_app_operations(app_spec: str, *, cwd: Path, env: dict[str, str] | None = None) -> list[Operation]: + """Import `module:attr` in a fresh interpreter (same venv as bdt) with `cwd` as the working/import + directory, and return its operations. The child runs `_api_dump`'s source via `-c` instead of importing + anything from this package, so a repo-local `bmsdna` package can't shadow `bmsdna.devtools`.""" + if ":" not in app_spec: + raise ApiUsageError(f"app '{app_spec}' must look like 'module.path:attribute'") + if not cwd.is_dir(): + raise ApiUsageError(f"app_dir '{cwd}' is not a directory") + with tempfile.TemporaryDirectory() as tmp: + out = Path(tmp) / "ops.json" + try: + proc = subprocess.run( + [sys.executable, "-c", Path(_api_dump.__file__).read_text(encoding="utf-8"), app_spec, str(out)], + cwd=cwd, + env={**os.environ, **(env or {})}, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + encoding="utf-8", + timeout=_APP_IMPORT_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as exc: + raise ApiUsageError(f"importing '{app_spec}' took longer than {_APP_IMPORT_TIMEOUT_SECONDS}s (cwd {cwd})") from exc + if proc.returncode != 0 or not out.is_file(): + tail = "\n".join((proc.stderr or proc.stdout).strip().splitlines()[-15:]) + raise ApiUsageError(f"could not load operations from '{app_spec}' (cwd {cwd}):\n{tail}") + return [Operation.from_dict(d) for d in json.loads(out.read_text(encoding="utf-8"))] + + +def load_openapi_file(path: Path) -> list[Operation]: + try: + doc = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise ApiUsageError(f"could not read OpenAPI file {path}: {exc}") from exc + if not isinstance(doc, dict): + raise ApiUsageError(f"OpenAPI file {path} is not a JSON object") + return [Operation.from_dict(d) for d in _api_dump.ops_from_doc(doc)] + + +# --------------------------------------------------------------------------- frontend scan + + +def normalize_path(text: str) -> str: + """Comparable form of a path template or URL literal: `{x}`/`${expr}` -> `{}`, query/fragment dropped, + a trailing placeholder glued to the path (`/x${qs}`, a query suffix) dropped, no trailing slash.""" + text = re.sub(r"\$\{[^}]*\}", "{}", text) + text = re.sub(r"\{[^}/]*\}", "{}", text) + text = text.split("?")[0].split("#")[0] + text = re.sub(r"(\{\})+", "{}", text) + text = re.sub(r"(?<=[^/{}])\{\}$", "", text) + return text.rstrip("/") or "/" + + +def _read_template(text: str, i: int, end: int, depth: int = 0) -> tuple[str, int]: + """`text[i]` is an opening backtick. Returns (template text with every `${expr}` replaced by `{}`, + index after the closing backtick), handling templates/strings/braces nested inside `${...}`. Stops at `end` + (and at a nesting depth that only hostile input reaches) instead of scanning on.""" + out: list[str] = [] + i += 1 + while i < end: + c = text[i] + if c == "\\": + i += 2 + elif c == "`": + return "".join(out), i + 1 + elif c == "$" and text.startswith("${", i): + level, i = 1, i + 2 + while i < end and level: + c = text[i] + if c == "`": + if depth >= _MAX_TEMPLATE_DEPTH: + return "".join(out), end + _, i = _read_template(text, i, end, depth + 1) + continue + if c in "'\"": + j = i + 1 + while j < end and text[j] != c and text[j] != "\n": + j += 2 if text[j] == "\\" else 1 + i = j + 1 + continue + level += (c == "{") - (c == "}") + i += 1 + out.append("{}") + else: + out.append(c) + i += 1 + return "".join(out), end + + +@dataclass(slots=True) +class FrontendUsage: + files: int = 0 + identifiers: set[str] = field(default_factory=set) + fetch_calls: set[tuple[str, str]] = field(default_factory=set) # (METHOD, normalized path) from `.GET("/x"` + literals: set[str] = field(default_factory=set) # normalized path-like string/template literals + suffix_literals: set[str] = field(default_factory=set) # literals usable for url-sfx matching (derived, see scan_frontend) + + +@dataclass(slots=True) +class Frontend: + """One frontend source directory: what its non-generated code references, and its own generated hey-api SDK + (function -> (METHOD, normalized url)). Kept per directory so two apps that both generate `listItems` can't be confused.""" + + usage: FrontendUsage + sdk: dict[str, tuple[str, str]] = field(default_factory=dict) + + +def is_generated_or_test_file(path: Path, *, repo_root: Path, exclude_globs: list[str]) -> bool: + return ( + bool(_EXCLUDE_FILE_RE.search(path.name)) + or path.name.startswith(_EXCLUDE_FILE_PREFIXES) + or is_excluded_ts_file(path, repo_root=repo_root, exclude_globs=exclude_globs) + ) + + +def iter_frontend_files(directory: Path, *, repo_root: Path, exclude_globs: list[str]) -> list[Path]: + files, _ = _iter_files([directory], _EXCLUDE_DIR_NAMES, SOURCE_SUFFIXES) + return [f for f in files if not is_generated_or_test_file(f, repo_root=repo_root, exclude_globs=exclude_globs)] + + +def scan_frontend(files: list[Path]) -> FrontendUsage: + usage = FrontendUsage() + for path in files: + text = _mask(path.read_text(encoding="utf-8", errors="ignore")).code # comments blanked, strings kept + usage.files += 1 + usage.identifiers.update(_IDENT_RE.findall(text)) + found: list[tuple[int, str]] = [(m.start(), m.group("body")) for m in _QUOTED_PATH_RE.finditer(text)] + for m in _TEMPLATE_START_RE.finditer(text): + found.append((m.start(), _read_template(text, m.start(), min(len(text), m.start() + _MAX_TEMPLATE_CHARS))[0])) + for pos, body in found: + normalized = normalize_path(body) + call = _FETCH_METHOD_RE.search(text[max(0, pos - 40) : pos]) + if call: # method-exact evidence only; it must not also count as a method-agnostic URL literal + usage.fetch_calls.add((call.group(1), normalized)) + continue + if normalized != "/": + usage.literals.add(normalized) + if body.endswith("/") and body != "/": # `"/api/items/" + id` + usage.literals.add(normalize_path(body + "{}")) + for literal in usage.literals: + bare = re.sub(r"^(\{\})+", "", literal) + if bare.count("/") >= 2 and "{}" not in bare.split("/")[1:2]: + usage.suffix_literals.add(bare) + return usage + + +def read_sdk_functions(directory: Path) -> dict[str, tuple[str, str]]: + """hey-api `sdk.gen.ts` files under `directory`: exported function name -> (METHOD, normalized url as generated).""" + functions: dict[str, tuple[str, str]] = {} + for dirpath, dirnames, filenames in os.walk(directory): + dirnames[:] = [d for d in dirnames if d not in _EXCLUDE_DIR_NAMES] + if "sdk.gen.ts" not in filenames: + continue + text = Path(dirpath, "sdk.gen.ts").read_text(encoding="utf-8", errors="ignore") + for chunk in re.split(r"(?=^export const \w+ = )", text, flags=re.M): + name, url = _SDK_FN_RE.match(chunk), _SDK_URL_RE.search(chunk) + if name and url: + functions[name.group(1)] = (url.group(1).upper(), normalize_path(url.group(2))) + return functions + + +def build_frontend(directory: Path, *, repo_root: Path, exclude_globs: list[str]) -> Frontend: + files = iter_frontend_files(directory, repo_root=repo_root, exclude_globs=exclude_globs) + return Frontend(scan_frontend(files), read_sdk_functions(directory)) + + +# --------------------------------------------------------------------------- backend references by route name + + +def read_referenced_route_names(directory: Path, functions: tuple[str, ...] = DEFAULT_URL_FOR_FUNCTIONS) -> set[str]: + """Route names passed as a string literal to one of `functions` (`request.url_for("x")`, `app.url_path_for(name="x")`, + `{{ url_for('x', id=1) }}`) in the non-test Python files and templates under `directory`. A mounted sub-app's + `"mount:x"` yields `x`. Regex based, so a call in a comment counts too (that only ever hides a dead route, never invents one).""" + pattern = re.compile(r"\b(?:" + "|".join(map(re.escape, functions)) + r")\(\s*(?:\w+\s*=\s*)?(['\"])(?P[^'\"\n]+)\1") + files, _ = _iter_files([directory], _EXCLUDE_DIR_NAMES | _TEST_DIR_NAMES, _NAME_SOURCE_SUFFIXES) + names: set[str] = set() + for path in files: + if _TEST_FILE_RE.match(path.name): + continue + text = path.read_text(encoding="utf-8", errors="ignore") + names.update(m.group("name").rsplit(":", 1)[-1] for m in pattern.finditer(text)) + return names + + +def is_named_by(op: Operation, names: set[str]) -> bool: + """Whether `op`'s route is one of `names`. The OpenAPI document has no route name, but the operationId is derived + from it: `{name}{path}_{method}` with non-word characters as `_` by default (rebuilt here from the mount-relative + path), the bare name or `-` with a custom `generate_unique_id`.""" + if not op.operation_id: + return False + path = op.path[len(op.mount) :] if op.mount and op.path.startswith(op.mount) else op.path + return any( + op.operation_id in (n, re.sub(r"\W", "_", n + path) + "_" + op.method.lower()) or op.operation_id.endswith("-" + n) + for n in names + ) + + +# --------------------------------------------------------------------------- matching + +_EVIDENCE_RANK = {"sdk": 0, "fetch": 1, "url": 2, "url-sfx": 3} + + +def _evidence_in(op: Operation, frontend: Frontend, candidates: set[str]) -> str | None: + usage, sdk = frontend.usage, frontend.sdk + for fn, (method, url) in sdk.items(): + if method != op.method or url not in candidates: + continue + if fn in usage.identifiers: + return "sdk" + # `fooOptions`/`fooMutation`/... -- unless that name is itself another SDK function (`getUser` vs `getUserQuery`) + if any(fn + s in usage.identifiers and fn + s not in sdk for s in _REACT_QUERY_SUFFIXES): + return "sdk" + if any((op.method, c) in usage.fetch_calls for c in candidates): + return "fetch" + if candidates & usage.literals: + return "url" + if any(c.endswith(bare) for bare in usage.suffix_literals for c in candidates): + return "url-sfx" + return None + + +def call_evidence(op: Operation, frontends: list[Frontend]) -> str | None: + """The strongest kind of evidence (`sdk`/`fetch`/`url`/`url-sfx`) that `op` is called from any of `frontends`, or None.""" + full = normalize_path(op.path) + relative = (full[len(op.mount) :] if op.mount and full.startswith(op.mount) else full) or "/" + candidates = {full, relative} + found = [e for fe in frontends if (e := _evidence_in(op, fe, candidates))] + return min(found, key=_EVIDENCE_RANK.__getitem__) if found else None + + +def is_excluded(op: Operation, *, prefixes: list[str], tags: list[str], path_globs: list[str]) -> bool: + return ( + any(op.path.startswith(p) for p in prefixes) + or any(t in tags for t in op.tags) + or any(fnmatch(op.path, g) or fnmatch(op.key, g) for g in path_globs) + ) + + +# --------------------------------------------------------------------------- configuration + + +@dataclass(frozen=True, slots=True) +class AppConfig: + name: str + frontends: list[str] + app: str | None = None + openapi: str | None = None + app_dir: str = "." + env: dict[str, str] = field(default_factory=dict) + exclude_prefixes: list[str] = field(default_factory=list) + exclude_tags: list[str] = field(default_factory=list) + exclude_paths: list[str] = field(default_factory=list) + exclude_frontend_globs: list[str] = field(default_factory=list) + url_for_functions: list[str] = field(default_factory=lambda: list(DEFAULT_URL_FOR_FUNCTIONS)) + baseline: str | None = None + + +def _str_list(raw: dict, key: str, app: str) -> list[str]: + value = raw.get(key, []) + if not isinstance(value, list) or not all(isinstance(v, str) for v in value): + raise ApiUsageError(f"dead_code app '{app}': `{key}` must be a list of strings, got {value!r}") + return value + + +def parse_config(table: dict) -> list[AppConfig]: + apps = table.get("apps") or [] + if not apps: + raise ApiUsageError("no [[tool.bdt.dead_code.apps]] configured in pyproject.toml") + configs: list[AppConfig] = [] + for i, raw in enumerate(apps): + name = str(raw.get("name") or raw.get("app") or raw.get("openapi") or f"app{i}") + if bool(raw.get("app")) == bool(raw.get("openapi")): + raise ApiUsageError(f"dead_code app '{name}': set exactly one of `app` (module:attr) and `openapi` (path to openapi.json)") + frontends = _str_list(raw, "frontends", name) + if not frontends: + raise ApiUsageError(f"dead_code app '{name}': `frontends` (list of frontend source dirs/globs) is required") + env = raw.get("env") or {} + if not isinstance(env, dict): + raise ApiUsageError(f"dead_code app '{name}': `env` must be a table") + configs.append( + AppConfig( + name=name, + frontends=frontends, + app=raw.get("app"), + openapi=raw.get("openapi"), + app_dir=str(raw.get("app_dir", ".")), + env={str(k): str(v) for k, v in env.items()}, + exclude_prefixes=_str_list(raw, "exclude_prefixes", name), + exclude_tags=_str_list(raw, "exclude_tags", name), + exclude_paths=_str_list(raw, "exclude_paths", name), + exclude_frontend_globs=_str_list(raw, "exclude_frontend_globs", name), + url_for_functions=_str_list(raw, "url_for_functions", name) or list(DEFAULT_URL_FOR_FUNCTIONS), + baseline=raw.get("baseline"), + ) + ) + return configs + + +def _resolve_frontend_dirs(repo_root: Path, patterns: list[str]) -> list[Path]: + dirs: list[Path] = [] + for pattern in patterns: + if not pattern or Path(pattern).is_absolute(): + raise ApiUsageError(f"frontends entry '{pattern}' must be a non-empty repo-relative path or glob") + matches = sorted(p for p in repo_root.glob(pattern) if p.is_dir()) + if not matches: + raise ApiUsageError(f"frontends entry '{pattern}' matches no directory under {repo_root}") + dirs.extend(matches) + return dirs + + +# --------------------------------------------------------------------------- baseline + + +def read_baseline(path: Path) -> set[str]: + if not path.is_file(): + return set() + return {line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip() and not line.lstrip().startswith("#")} + + +def write_baseline(path: Path, keys: list[str]) -> None: + header = "# Backend operations known to be unused (bdt dead-code). Remove a line once the route is deleted or called.\n" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(header + "".join(f"{k}\n" for k in sorted(keys)), encoding="utf-8") + + +# --------------------------------------------------------------------------- orchestration + + +def check_app(config: AppConfig, *, repo_root: Path, update_baseline: bool = False) -> list[Finding]: + if config.openapi: + operations = load_openapi_file(repo_root / config.openapi) + else: + assert config.app is not None + operations = load_app_operations(config.app, cwd=repo_root / config.app_dir, env=config.env) + if not operations: + raise ApiUsageError( + f"dead_code app '{config.name}': no operations found in {config.openapi or config.app} " + "(is it a FastAPI app with documented routes, not a wrapped/middleware ASGI app?)" + ) + + frontends = [ + build_frontend(d, repo_root=repo_root, exclude_globs=config.exclude_frontend_globs) + for d in _resolve_frontend_dirs(repo_root, config.frontends) + ] + + def excluded(op: Operation) -> bool: + return is_excluded(op, prefixes=config.exclude_prefixes, tags=config.exclude_tags, path_globs=config.exclude_paths) + + names = read_referenced_route_names(repo_root / config.app_dir, tuple(config.url_for_functions)) + by_key = {op.key: op for op in operations} + uncalled = [op for op in operations if not excluded(op) and call_evidence(op, frontends) is None and not is_named_by(op, names)] + where = Path(config.name) + + if not config.baseline: + return [_finding(where, op) for op in uncalled] + baseline_path = repo_root / config.baseline + if update_baseline: + write_baseline(baseline_path, [op.key for op in uncalled]) + return [] + baseline = read_baseline(baseline_path) + uncalled_keys = {op.key for op in uncalled} + findings = [_finding(where, op) for op in uncalled if op.key not in baseline] + for key in sorted(baseline - uncalled_keys): + op = by_key.get(key) + reason = "no longer a backend route" if op is None else "now excluded" if excluded(op) else "now used" + findings.append( + Finding( + where, + 0, + RULE_STALE_BASELINE, + f"'{key}' is listed in {config.baseline} but is {reason} -- remove it (or run `bdt dead-code --update-baseline`).", + ) + ) + return findings + + +def _finding(where: Path, op: Operation) -> Finding: + tags = f" [{', '.join(op.tags)}]" if op.tags else "" + return Finding( + where, + 0, + RULE, + f"{op.key}{tags} is never called from non-generated frontend code nor referenced by url_for -- delete it, or " + "exclude it (exclude_prefixes/exclude_tags/exclude_paths) if something other than the frontend calls it.", + ) + + +def run(table: dict, *, repo_root: Path, update_baseline: bool = False) -> list[Finding]: + configs = parse_config(table) + if update_baseline: # validate everything first so a late failure can't leave some baselines rewritten and others not + paths = [c.baseline for c in configs] + if not all(paths): + missing = ", ".join(c.name for c in configs if not c.baseline) + raise ApiUsageError(f"--update-baseline needs a `baseline` for every app; missing for: {missing}") + if len(set(paths)) != len(paths): + raise ApiUsageError("--update-baseline: two apps share the same `baseline` file, each would overwrite the other") + findings: list[Finding] = [] + for config in configs: + findings.extend(check_app(config, repo_root=repo_root, update_baseline=update_baseline)) + return findings diff --git a/bmsdna/devtools/cli.py b/bmsdna/devtools/cli.py index 1f0bacb..bea4da1 100644 --- a/bmsdna/devtools/cli.py +++ b/bmsdna/devtools/cli.py @@ -3,6 +3,7 @@ import json import subprocess import sys +import tomllib from collections.abc import Callable from datetime import datetime, timedelta, timezone from importlib.metadata import version as _pkg_version @@ -12,7 +13,7 @@ import typer from pgdevkit.testdb import constants as pgdevkit_constants -from . import ado_issue, app_service_logs, commit as commit_mod +from . import ado_issue, api_usage as api_usage_mod, app_service_logs, commit as commit_mod, dead_code as dead_code_mod from . import env_config from . import find_repo as find_repo_mod from . import issue_do as issue_do_mod @@ -22,8 +23,10 @@ from . import logs as logs_mod from . import pr_build, pr_issue_link, pr_labels, pull as pull_mod, worktree as worktree_mod from .ado_auth import auth_header +from .bdt_config import find_pyproject, load_bdt_table from .cli_tools import detect_agent_session, require_az, require_gh from .gitrepo import AdoRemote, GitHubRemote, UnknownRemoteError, current_branch, current_remote, head_commit_subject +from .lint_findings import render_findings # Non-ASCII output (checkmarks, en-dashes in ADO project names, etc.) needs a # UTF-8 stream — the default Windows console codepage isn't UTF-8, and would @@ -947,12 +950,52 @@ def lint( """Static checks (bmsuisse/skills#52): postgres/psycopg SQL rules on every `.execute()` call (must use load_sql()/a .sql file, a t-string, or psycopg.sql for anything beyond a trivial query; never an f-string/concatenation/`%`-formatting), pydantic-model placement under api/ - directories, and that the repo declares/configures ty, ruff, pytest and prek. + directories, hand-wired HTTP in TypeScript, and that the repo declares/configures ty, ruff, pytest and prek. + For unused .sql files and dead backend routes see `dead-code`. """ result = lint_mod.run(paths or [], skip_tooling_check=no_tooling_check) raise typer.Exit(lint_mod.print_report(result)) +@app.command("dead-code") +def dead_code( + only: list[str] = typer.Option( + None, + "--only", + help="Run just this check (repeatable): `sql` (unreferenced .sql files) or `routes` (backend routes nothing " + "calls; imports the app, so it is the slow one). Default: every check that is configured.", + ), + update_baseline: bool = typer.Option( + False, + "--update-baseline", + help="Routes only: rewrite each app's `baseline` file with the routes currently uncalled (and report nothing). " + "Use once to adopt the check, then only to remove lines.", + ), +) -> None: + """Dead code that is invisible to a linter: `.sql` files no Python code loads (`sql_roots`) and backend + (FastAPI) routes that neither non-generated frontend code nor a `url_for(...)` call references (the `apps` + array of tables). Both are configured under tool.bdt.dead_code in pyproject.toml; generated API-client code and + tests never count as callers. Exit 0 clean, 1 findings, 2 setup error. + """ + root = Path.cwd() + pyproject = find_pyproject(root) + repo_root = pyproject.parent if pyproject else root + try: + findings = dead_code_mod.run(load_bdt_table("dead_code", root), repo_root=repo_root, only=only or [], update_baseline=update_baseline) + except (api_usage_mod.ApiUsageError, tomllib.TOMLDecodeError) as exc: + typer.echo(f"bdt dead-code: {exc}", err=True) + raise typer.Exit(2) from exc + if update_baseline: + typer.echo("bdt dead-code: baseline(s) updated") + raise typer.Exit(0) + if not findings: + typer.echo("bdt dead-code: no dead code found") + raise typer.Exit(0) + typer.echo(render_findings(findings)) + typer.echo(f"\n{len(findings)} issue(s) found.") + raise typer.Exit(1) + + @app.command("find-injection") def find_injection_cmd( paths: list[str] = typer.Argument(None, help="Files and/or directories to scan (default: current directory, recursive)."), diff --git a/bmsdna/devtools/dead_code.py b/bmsdna/devtools/dead_code.py new file mode 100644 index 0000000..30596e6 --- /dev/null +++ b/bmsdna/devtools/dead_code.py @@ -0,0 +1,82 @@ +"""`bdt dead-code`: the dead-code checks a linter can't do, behind one command and one config table. + +- **sql** (`lint_sql_files`) -- `.sql` files under `sql_roots` that no Python code references; +- **routes** (`api_usage`) -- backend (FastAPI) routes that no non-generated frontend code calls and no + `url_for(...)` references. + +Configuration lives in `[tool.bdt.dead_code]` of the consuming repo's pyproject.toml. A check runs when it is +configured (`sql_roots`, resp. `[[tool.bdt.dead_code.apps]]`); `only` narrows a run to one of them, e.g. to +keep the slow one (the routes check imports the app) out of a prek hook. +""" + +from __future__ import annotations + +from pathlib import Path + +from . import api_usage +from .api_usage import ApiUsageError +from .lint import _DEFAULT_EXCLUDE_DIR_NAMES, _iter_files, _iter_python_files +from .lint_findings import Finding +from .lint_sql_files import DEFAULT_LOADER_FUNCTIONS, check_unreferenced_sql_files + +CHECKS = ("sql", "routes") + + +def _as_list(value: str | list[str] | None) -> list[str]: + """A TOML list of strings; a bare string is accepted as a one-item list rather than iterated character by character.""" + return [value] if isinstance(value, str) else [str(v) for v in value or []] + + +def check_sql_roots(config: dict, *, repo_root: Path) -> list[Finding]: + """`sql-file-unreferenced` over `sql_roots`. Looks at every Python file of the repo: the caller of a query + can live anywhere.""" + exclude_dir_names = _DEFAULT_EXCLUDE_DIR_NAMES | set(_as_list(config.get("exclude_dirs"))) + findings: list[Finding] = [] + roots: list[Path] = [] + for root in _as_list(config.get("sql_roots")): + candidate = repo_root / root + if candidate.is_dir() and candidate.resolve().is_relative_to(repo_root.resolve()): + roots.append(candidate) + else: + findings.append(Finding(candidate, 0, "lint-path-not-found", f"sql_roots entry '{root}' is not a directory inside {repo_root}.")) + sql_files, _ = _iter_files(roots, exclude_dir_names, (".sql",)) + python_files, _ = _iter_python_files([repo_root], exclude_dir_names) + findings.extend( + check_unreferenced_sql_files( + repo_root=repo_root, + sql_files=sql_files, + python_files=python_files, + loader_functions=_as_list(config.get("sql_loader_functions", list(DEFAULT_LOADER_FUNCTIONS))), + ignore_globs=_as_list(config.get("sql_unreferenced_ignore")), + ) + ) + return findings + + +def run(table: dict, *, repo_root: Path, only: list[str] | None = None, update_baseline: bool = False) -> list[Finding]: + """Every configured check (or just those named in `only`). `update_baseline` concerns the routes check alone and + skips the sql one.""" + unknown = [c for c in only or [] if c not in CHECKS] + if unknown: + raise ApiUsageError(f"unknown check {', '.join(unknown)} -- choose from: {', '.join(CHECKS)}") + if update_baseline and only and "routes" not in only: + raise ApiUsageError("--update-baseline only applies to the `routes` check") + wanted = set(only or CHECKS) + run_sql = "sql" in wanted and not update_baseline and bool(table.get("sql_roots")) + run_routes = "routes" in wanted and bool(table.get("apps")) + if not run_sql and not run_routes: + missing = " / ".join( + part + for part, wanted_here in ( + ("`sql_roots`", "sql" in wanted and not update_baseline), + ("[[tool.bdt.dead_code.apps]]", "routes" in wanted), + ) + if wanted_here + ) + raise ApiUsageError(f"nothing to check: configure {missing} in pyproject.toml") + findings: list[Finding] = [] + if run_sql: + findings.extend(check_sql_roots(table, repo_root=repo_root)) + if run_routes: + findings.extend(api_usage.run(table, repo_root=repo_root, update_baseline=update_baseline)) + return findings diff --git a/bmsdna/devtools/lint_findings.py b/bmsdna/devtools/lint_findings.py index bc2b437..67d1b63 100644 --- a/bmsdna/devtools/lint_findings.py +++ b/bmsdna/devtools/lint_findings.py @@ -1,5 +1,5 @@ -"""Shared `Finding` type + rendering for every `bdt lint` rule module (lint_sql, -lint_models, lint_tooling) -- kept separate from lint.py so each rule module +"""Shared `Finding` type + rendering for every `bdt lint`/`bdt dead-code` rule module (lint_sql, +lint_models, lint_typescript, lint_sql_files, lint_tooling, api_usage) -- kept separate from lint.py so each rule module only depends on this, not on the orchestrator (which depends on all of them). """ diff --git a/bmsdna/devtools/lint_sql_files.py b/bmsdna/devtools/lint_sql_files.py new file mode 100644 index 0000000..3a1221c --- /dev/null +++ b/bmsdna/devtools/lint_sql_files.py @@ -0,0 +1,149 @@ +"""The `sql` check of `bdt dead-code` (rule `sql-file-unreferenced`): a `.sql` file under a configured root that no Python +code references is dead weight (a query left behind after its caller was deleted or renamed). + +Part of `bdt dead-code`; opt-in: only runs when `[tool.bdt.dead_code] sql_roots = ["backend/db/queries", ...]` is set, since which +folders hold *loadable* SQL (as opposed to schema/migration scripts that are applied, never loaded) +is a per-repo fact. A file counts as referenced when any of these holds (strongest first): + +- a call `load_sql("topic", "name")` (positional or `topic=`/`name=` keywords) with both arguments literal, + where `topic` is the file's parent directory name and `name` its stem (loader function names are configurable); +- a call `load_sql("topic", )` with a non-literal name, in a Python file that also contains the + stem as a string literal (covers `name = "a" if x else "b"` / lookup tables / f-string parts); +- the same with a non-literal topic (any file calling the loader dynamically may name any stem); +- a string literal that is a path whose trailing segments equal the file's repo-relative path + (`get_sql_with_prm_list("backend/api/sql/x.sql")`); +- the bare filename as a literal in a Python file under the SQL folder's parent directory + (`Path(__file__).parent.parent / "sql" / "x.sql"`). + +Anything else is reported. Heuristic by design -- it never executes code, so a SQL file reached +through a fully computed path is a false positive; list such files in `sql_unreferenced_ignore`. +A Python file that cannot be parsed is reported too (its references are unknown, so the result is not trustworthy). +""" + +from __future__ import annotations + +import ast +import re +from collections.abc import Iterable +from fnmatch import fnmatch +from pathlib import Path + +from .lint_findings import Finding + +RULE = "sql-file-unreferenced" +RULE_UNPARSEABLE = "sql-check-python-unparseable" +DEFAULT_LOADER_FUNCTIONS = ("load_sql",) + + +def _call_name(node: ast.Call) -> str | None: + func = node.func + if isinstance(func, ast.Name): + return func.id + if isinstance(func, ast.Attribute): + return func.attr + return None + + +def _literal(node: ast.expr | None) -> str | None: + return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else None + + +def _path_needle(literal: str) -> str | None: + """A `.sql` path literal as a segment-aligned suffix to compare repo-relative paths against, or None when it + is a bare filename (handled separately) or not a relative path.""" + needle = re.sub(r"^(?:\.{1,2}/)+", "", literal.replace("\\", "/")) + return needle if "/" in needle and not needle.startswith("/") else None + + +class _References: + def __init__(self, repo_root: Path, loader_functions: frozenset[str]) -> None: + self.repo_root = repo_root.resolve() + self.loader_functions = loader_functions + self.exact: set[tuple[str, str]] = set() # (topic, name) from fully literal loader calls + self.dynamic_by_topic: dict[str, set[str]] = {} # topic -> string literals of files with a dynamic-name call + self.dynamic_any_topic: set[str] = set() # string literals of files with a dynamic-topic call + self.path_needles: set[str] = set() + self.bare_by_file: list[tuple[Path, set[str]]] = [] # (python file, string literals ending in .sql) + self.unparseable: list[Path] = [] + + def scan(self, python_files: Iterable[Path]) -> None: + for path in python_files: + try: + tree = ast.parse(path.read_text(encoding="utf-8-sig")) + except SyntaxError, OSError, ValueError, RecursionError, MemoryError: + self.unparseable.append(path) + continue + literals = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant) and isinstance(n.value, str)} + sql_literals = {s for s in literals if s.endswith(".sql")} + self.path_needles.update(n for s in sql_literals if (n := _path_needle(s))) + if sql_literals: + self.bare_by_file.append((path.resolve(), sql_literals)) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and _call_name(node) in self.loader_functions and (node.args or node.keywords): + self._add_loader_call(node, literals) + + def _add_loader_call(self, node: ast.Call, literals: set[str]) -> None: + keywords = {k.arg: k.value for k in node.keywords if k.arg} + topic = _literal(node.args[0] if node.args else keywords.get("topic")) + name = _literal(node.args[1] if len(node.args) > 1 else keywords.get("name")) + if topic is not None and name is not None: + self.exact.add((topic, name)) + elif topic is not None: + self.dynamic_by_topic.setdefault(topic, set()).update(literals - {topic}) + else: + self.dynamic_any_topic |= literals + + def is_referenced(self, sql_file: Path) -> bool: + stem, topic = sql_file.stem, sql_file.parent.name + if (topic, stem) in self.exact: + return True + if stem in self.dynamic_by_topic.get(topic, ()) or stem in self.dynamic_any_topic: + return True + resolved = sql_file.resolve() + rel = resolved.relative_to(self.repo_root).as_posix() + if any(rel == needle or rel.endswith("/" + needle) for needle in self.path_needles): + return True + scope = resolved.parent.parent + if scope == self.repo_root: # a top-level SQL folder: "its parent" would be the whole repo, tests included + return False + return any(sql_file.name in lits and py.is_relative_to(scope) for py, lits in self.bare_by_file) + + +def check_unreferenced_sql_files( + *, + repo_root: Path, + sql_files: list[Path], + python_files: Iterable[Path], + loader_functions: Iterable[str] = DEFAULT_LOADER_FUNCTIONS, + ignore_globs: Iterable[str] = (), +) -> list[Finding]: + """One `sql-file-unreferenced` finding per file of `sql_files` (all inside `repo_root`) that no Python file in + `python_files` references, plus one finding per Python file that could not be parsed.""" + if not sql_files: + return [] + refs = _References(repo_root, frozenset(loader_functions)) + refs.scan(python_files) + ignores = list(ignore_globs) + findings = [ + Finding( + path, + 0, + RULE_UNPARSEABLE, + "could not be parsed, so references to .sql files in it are unknown -- `sql-file-unreferenced` results may be wrong.", + ) + for path in refs.unparseable + ] + for sql_file in sql_files: + rel = sql_file.resolve().relative_to(refs.repo_root).as_posix() + if any(fnmatch(rel, g) for g in ignores) or refs.is_referenced(sql_file): + continue + findings.append( + Finding( + sql_file, + 0, + RULE, + "no Python code references this SQL file (load_sql topic/name or path) -- delete it, or list it in " + "[tool.bdt.dead_code] sql_unreferenced_ignore if it is reached some other way.", + ) + ) + return findings diff --git a/skills/bmsdna-devtools/SKILL.md b/skills/bmsdna-devtools/SKILL.md index eacf889..30e0caa 100644 --- a/skills/bmsdna-devtools/SKILL.md +++ b/skills/bmsdna-devtools/SKILL.md @@ -4,10 +4,12 @@ description: > Use the `bdt` CLI (from the bmsdna-devtools package) instead of ad hoc git/az/gh commands or repo-local scripts for: checking PR build/check status, creating a PR, creating a git worktree, committing and pushing files (with pre-flight - checks), and querying Azure logs. `bdt pr *` works against both Azure DevOps + checks), querying Azure logs, and static checks (`bdt lint`: SQL/psycopg rules, pydantic placement, tooling; `bdt dead-code`: + unreferenced .sql files and backend routes nothing calls). `bdt pr *` works against both Azure DevOps and GitHub — it auto-detects which one from the `origin` remote. Trigger whenever the user asks to check a build/PR status, create a PR, make a - worktree, commit changes, or fetch/tail application logs in a repo that has + worktree, commit changes, fetch/tail application logs, lint a repo, or find dead API routes / + unused .sql files in a repo that has bmsdna-devtools installed (check for `bdt` on PATH, or `bmsdna-devtools` in pyproject.toml, before assuming it applies). --- @@ -95,6 +97,15 @@ programmatically from an agent loop — the schema is documented in `bdt commit --help`. Pass `--subrepo database` (repeatable) for repos that vendor a git submodule under that path. +## Static checks + +- `bdt lint [paths]` -- SQL/psycopg rules, pydantic-model placement, hand-wired HTTP in TypeScript, baseline tooling. +- `uv run bdt dead-code` -- dead code a linter can't see, configured under `[tool.bdt.dead_code]` in pyproject.toml: + `.sql` files no Python code loads (`sql_roots`) and FastAPI routes that neither non-generated frontend code nor a + `url_for(...)` call uses (`[[tool.bdt.dead_code.apps]]`: app or openapi file, frontends, excludes, optional `baseline`; + adopt with `--update-baseline`). `--only sql|routes` runs one check. The routes check imports the app, so run it with + the repo's own interpreter (`uv run`). Generated API-client code and tests never count as callers. + ## Application Insights logs ```bash diff --git a/tests/test_api_usage.py b/tests/test_api_usage.py new file mode 100644 index 0000000..46f6173 --- /dev/null +++ b/tests/test_api_usage.py @@ -0,0 +1,654 @@ +import json +import textwrap +import time +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from bmsdna.devtools import _api_dump +from bmsdna.devtools import api_usage as au +from bmsdna.devtools.bdt_config import load_bdt_table +from bmsdna.devtools.cli import app + +_SDK = """ +export const getThing = (options: Options) => + (options.client ?? client).get({ + url: "/api/things/{thing_id}", + ...options, + }); + +export const deleteThing = (options: Options) => + (options.client ?? client).delete({ + url: "/api/things/{thing_id}", + ...options, + }); + +export const listThings = (options?: Options) => + (options?.client ?? client).get({ url: "/api/things", ...options }); + +export const headThings = (options?: Options) => + (options?.client ?? client).head({ url: "/api/things", ...options }); +""" + + +def _write(root: Path, rel: str, text: str = "") -> Path: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(textwrap.dedent(text)) + return path + + +def _fe(root: Path, rel: str = "fe/src", *, exclude_globs: list[str] | None = None) -> au.Frontend: + return au.build_frontend(root / rel, repo_root=root, exclude_globs=exclude_globs or []) + + +def _op(method: str, path: str, *tags: str, mount: str = "") -> au.Operation: + return au.Operation(method, path, tuple(tags), mount) + + +def _evidence(op: au.Operation, *frontends: au.Frontend) -> str | None: + return au.call_evidence(op, list(frontends)) + + +# ------------------------------------------------------------------ path normalisation / template scanning + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("/api/x/{id}/y", "/api/x/{}/y"), + ("/api/x/${id}/y?a=${b}", "/api/x/{}/y"), + ("/api/x/{}{}", "/api/x/{}"), + ("/api/x${qs}", "/api/x"), # glued trailing expression is a query suffix + ("/api/x/${id}", "/api/x/{}"), # but a placeholder segment is a path parameter + ("/api/x/", "/api/x"), + ], +) +def test_normalize_path(raw: str, expected: str) -> None: + assert au.normalize_path(raw) == expected + + +def test_read_template_handles_nested_template_in_expression() -> None: + src = '`/api/m/${encodeURIComponent(id)}/att${mailbox ? `?mb=${encodeURIComponent(mailbox)}` : ""}` + rest' + text, end = au._read_template(src, 0, len(src)) + assert text == "/api/m/{}/att{}" + assert src[end:] == " + rest" + + +def test_hostile_input_is_bounded(tmp_path: Path) -> None: + deep = "`/${" * 5000 # would recurse ~5000 levels + lone = "'${" * 70_000 # quadratic if the optional group may run to end of file + unclosed = "`/${" * 300 + "x" * 200_000 + sdk_like = "export const x = " + ".get " * 30_000 + _write(tmp_path, "fe/src/a.ts", deep + "\n" + lone + "\n" + unclosed + "\n") + _write(tmp_path, "fe/src/lib/generated/sdk.gen.ts", sdk_like) + started = time.monotonic() + fe = _fe(tmp_path) + assert time.monotonic() - started < 5 + assert fe.usage.files == 1 + + +# ------------------------------------------------------------------ backend inventory + + +def test_ops_from_doc_prefixes_mount_and_keeps_tags() -> None: + doc = {"paths": {"/a/{id}": {"get": {"tags": ["t"]}, "post": {}, "parameters": []}}} + ops = [au.Operation.from_dict(d) for d in _api_dump.ops_from_doc(doc, "/api/sub")] + assert {(o.method, o.path, o.tags, o.mount) for o in ops} == { + ("GET", "/api/sub/a/{id}", ("t",), "/api/sub"), + ("POST", "/api/sub/a/{id}", (), "/api/sub"), + } + + +class _FakeMount: + def __init__(self, path: str, app: object) -> None: + self.path, self.app = path, app + + +class _FakeApp: + def __init__(self, paths: dict, routes: list | None = None) -> None: + self._paths, self.routes = paths, routes or [] + + def openapi(self) -> dict: + return {"paths": self._paths} + + +def test_collect_recurses_into_mounts_only() -> None: + sub = _FakeApp({"/items": {"get": {}}}) + static = object() # a StaticFiles-like mount: no openapi() + root = _FakeApp({"/health": {"get": {}}}, [_FakeMount("/api/sub/", sub), _FakeMount("/assets", static), object()]) + assert sorted(f"{o['method']} {o['path']}" for o in _api_dump.collect(root)) == ["GET /api/sub/items", "GET /health"] + + +def test_load_app_operations_runs_in_subprocess_with_cwd_on_path(tmp_path: Path) -> None: + _write( + tmp_path, + "svc/fakeapp.py", + """ + class A: + routes = [] + def openapi(self): + return {"paths": {"/api/ping": {"get": {"tags": ["x"]}}}} + app = A() + """, + ) + ops = au.load_app_operations("fakeapp:app", cwd=tmp_path / "svc") + assert [(o.key, o.tags) for o in ops] == [("GET /api/ping", ("x",))] + + +def test_load_app_operations_survives_repo_local_bmsdna_package(tmp_path: Path) -> None: + # CCMT2 ships its own top-level `bmsdna` package, which the app itself imports; it must neither break the + # dump (by shadowing bmsdna.devtools) nor be shadowed by it. + _write(tmp_path, "bmsdna/__init__.py") + _write(tmp_path, "bmsdna/links.py", "PREFIX = '/api/linked'\n") + _write( + tmp_path, + "linkedapp.py", + """ + from bmsdna.links import PREFIX + class A: + routes = [] + def openapi(self): + return {"paths": {PREFIX: {"get": {}}}} + app = A() + """, + ) + assert [o.key for o in au.load_app_operations("linkedapp:app", cwd=tmp_path)] == ["GET /api/linked"] + + +def test_load_app_operations_reports_import_errors(tmp_path: Path) -> None: + _write(tmp_path, "boom.py", "raise RuntimeError('missing env var FOO')\n") + with pytest.raises(au.ApiUsageError, match="missing env var FOO"): + au.load_app_operations("boom:app", cwd=tmp_path) + + +def test_load_app_operations_rejects_bad_spec_and_dir(tmp_path: Path) -> None: + with pytest.raises(au.ApiUsageError, match="module.path:attribute"): + au.load_app_operations("nocolon", cwd=tmp_path) + with pytest.raises(au.ApiUsageError, match="not a directory"): + au.load_app_operations("a:b", cwd=tmp_path / "missing") + + +def test_load_app_operations_passes_env(tmp_path: Path) -> None: + _write( + tmp_path, + "envapp.py", + """ + import os + class A: + routes = [] + def openapi(self): + return {"paths": {"/" + os.environ["ROUTE_NAME"]: {"get": {}}}} + app = A() + """, + ) + assert [o.path for o in au.load_app_operations("envapp:app", cwd=tmp_path, env={"ROUTE_NAME": "hello"})] == ["/hello"] + + +def test_load_openapi_file_rejects_non_object(tmp_path: Path) -> None: + _write(tmp_path, "o.json", "[]") + with pytest.raises(au.ApiUsageError, match="not a JSON object"): + au.load_openapi_file(tmp_path / "o.json") + + +# ------------------------------------------------------------------ frontend evidence + + +def test_sdk_function_and_react_query_helper_count_as_calls(tmp_path: Path) -> None: + _write(tmp_path, "fe/src/lib/generated/sdk.gen.ts", _SDK) + _write( + tmp_path, + "fe/src/page.tsx", + "import { listThingsOptions } from '@/lib/generated/@tanstack/react-query.gen';\nuseQuery(listThingsOptions());\n", + ) + _write(tmp_path, "fe/src/other.ts", "import { getThing } from './lib/generated';\ngetThing({ path: { thing_id: 1 } });\n") + fe = _fe(tmp_path) + assert set(fe.sdk) == {"getThing", "deleteThing", "listThings", "headThings"} + assert _evidence(_op("GET", "/api/things"), fe) == "sdk" + assert _evidence(_op("GET", "/api/things/{thing_id}"), fe) == "sdk" + # same path, different method: deleteThing / headThings are never referenced + assert _evidence(_op("DELETE", "/api/things/{thing_id}"), fe) is None + assert _evidence(_op("HEAD", "/api/things"), fe) is None + + +def test_react_query_suffix_does_not_confuse_two_sdk_functions(tmp_path: Path) -> None: + sdk = """ + export const getUser = (o) => (o.client ?? client).get({ url: "/api/user", ...o }); + export const getUserQuery = (o) => (o.client ?? client).get({ url: "/api/user-query", ...o }); + """ + _write(tmp_path, "fe/src/lib/generated/sdk.gen.ts", sdk) + _write(tmp_path, "fe/src/p.ts", "getUserQuery();\n") + fe = _fe(tmp_path) + assert _evidence(_op("GET", "/api/user-query"), fe) == "sdk" + assert _evidence(_op("GET", "/api/user"), fe) is None + + +def test_sdk_functions_are_scoped_per_frontend(tmp_path: Path) -> None: + a_sdk = 'export const listItems = (o) => (o.client ?? client).get({ url: "/api/a/items", ...o });\n' + b_sdk = 'export const listItems = (o) => (o.client ?? client).get({ url: "/api/b/items", ...o });\n' + _write(tmp_path, "apps/a/src/gen/sdk.gen.ts", a_sdk) + _write(tmp_path, "apps/a/src/page.ts", "listItems();\n") + _write(tmp_path, "apps/b/src/gen/sdk.gen.ts", b_sdk) + _write(tmp_path, "apps/b/src/page.ts", "const x = 1;\n") + a, b = _fe(tmp_path, "apps/a/src"), _fe(tmp_path, "apps/b/src") + assert _evidence(_op("GET", "/api/a/items"), a, b) == "sdk" + assert _evidence(_op("GET", "/api/b/items"), a, b) is None # app b generates listItems too, but never calls it + + +@pytest.mark.parametrize( + "name", + [ + "lib/generated/react-query.gen.ts", + "lib/api-types.generated.ts", + "lib/api-types-v2.ts", + "lib/openapi_schema.generated.ts", + "lib/types.d.ts", + "page.test.tsx", + "page.test.js", + "page.spec.jsx", + "client.gen.js", + "client.generated.mjs", + "Page.test.vue", + "__tests__/a.ts", + "e2e/a.ts", + "tests/a.js", + ], +) +def test_generated_code_and_tests_are_not_callers(tmp_path: Path, name: str) -> None: + _write(tmp_path, f"fe/src/{name}", 'fetch("/api/things/1"); export const x = deleteThing(); const p = "/api/things/{thing_id}";\n') + _write(tmp_path, "fe/src/lib/generated/sdk.gen.ts", _SDK) + fe = _fe(tmp_path) + assert fe.usage.files == 0 + assert _evidence(_op("DELETE", "/api/things/{thing_id}"), fe) is None + assert _evidence(_op("GET", "/api/things/{thing_id}"), fe) is None + + +def test_comments_are_not_calls(tmp_path: Path) -> None: + _write( + tmp_path, + "fe/src/a.ts", + """ + // fetch("/api/commented") + /* client.GET("/api/block") */ + const label = "it's"; + const jsx =

Don't

; + fetch("/api/real"); + """, + ) + fe = _fe(tmp_path) + assert _evidence(_op("GET", "/api/real"), fe) == "url" + assert _evidence(_op("GET", "/api/commented"), fe) is None + assert _evidence(_op("GET", "/api/block"), fe) is None + + +def test_openapi_fetch_literal_is_method_exact(tmp_path: Path) -> None: + _write( + tmp_path, + "fe/src/svc.ts", + """ + const { data } = await client.GET("/api/things/{thing_id}", { params: { path: { thing_id } } }); + await client + .POST( + "/api/things", + { body }, + ); + """, + ) + fe = _fe(tmp_path) + assert _evidence(_op("GET", "/api/things/{thing_id}"), fe) == "fetch" + assert _evidence(_op("POST", "/api/things"), fe) == "fetch" + assert _evidence(_op("DELETE", "/api/things/{thing_id}"), fe) is None + assert _evidence(_op("GET", "/api/things"), fe) is None + + +def test_handwritten_fetch_is_method_agnostic(tmp_path: Path) -> None: + _write(tmp_path, "fe/src/svc.ts", "await fetch(`/api/things/${id}`, { method: 'DELETE' });\n") + fe = _fe(tmp_path) + for method in ("GET", "DELETE"): + assert _evidence(_op(method, "/api/things/{thing_id}"), fe) == "url" + + +def test_handwritten_fetch_with_nested_template(tmp_path: Path) -> None: + _write( + tmp_path, + "fe/src/mail.ts", + """ + const url = `/api/offer-parser/mail/${encodeURIComponent(mail.id)}/attachment${mailbox ? `?mailbox=${encodeURIComponent(mailbox)}` : ""}`; + """, + ) + assert _evidence(_op("GET", "/api/offer-parser/mail/{message_id}/attachment"), _fe(tmp_path)) == "url" + + +def test_concatenated_and_multi_expression_urls(tmp_path: Path) -> None: + _write( + tmp_path, + "fe/src/c.ts", + """ + await fetch("/api/items/" + id); + await fetch("/api/items/" + id + "/details"); + await fetch(`${a}${b}/api/viaprefix/${id}`); + """, + ) + fe = _fe(tmp_path) + assert _evidence(_op("GET", "/api/items/{item_id}"), fe) == "url" + assert _evidence(_op("GET", "/api/viaprefix/{x}"), fe) == "url-sfx" # leading base-URL expressions are ignored + + +def test_suffix_match_for_base_url_relative_clients(tmp_path: Path) -> None: + _write(tmp_path, "fe/src/c.ts", "const r = await http.get(`/customers/${id}/sales`);\n") + fe = _fe(tmp_path) + assert _evidence(_op("GET", "/api/customers/{customer_id}/sales"), fe) == "url-sfx" + assert _evidence(_op("GET", "/api/customers/{customer_id}/other"), fe) is None + + +def test_literal_longer_than_route_is_not_evidence(tmp_path: Path) -> None: + _write(tmp_path, "fe/src/c.ts", 'router.push("/admin/items/list");\nconst root = "/";\nsplit("/");\n') + fe = _fe(tmp_path) + assert _evidence(_op("GET", "/items/list"), fe) is None + assert _evidence(_op("GET", "/"), fe) is None + + +def test_mount_prefix_is_stripped_for_sdk_urls(tmp_path: Path) -> None: + _write(tmp_path, "fe/src/gen/sdk.gen.ts", _SDK.replace('"/api/things"', '"/things"')) + _write(tmp_path, "fe/src/p.ts", "listThings();\n") + fe = _fe(tmp_path) + # a sub-app's generated client uses mount-relative urls ("/things"); the operation carries the mount ("/api/sub/things") + assert _evidence(_op("GET", "/api/sub/things", mount="/api/sub"), fe) == "sdk" + assert _evidence(_op("GET", "/things"), fe) == "sdk" + assert _evidence(_op("GET", "/api/sub/other", mount="/api/sub"), fe) is None + assert _evidence(_op("GET", "/api/sub/", mount="/api/sub"), fe) is None # mount root must not crash on the empty remainder + + +# ------------------------------------------------------------------ excludes + + +def test_is_excluded_applies_prefix_tag_and_glob() -> None: + kwargs = {"prefixes": ["/external_api"], "tags": ["agent"], "path_globs": ["/auth/*", "POST /api/svc/*"]} + assert au.is_excluded(_op("GET", "/external_api/a"), **kwargs) + assert au.is_excluded(_op("GET", "/api/agent-tool", "agent"), **kwargs) + assert au.is_excluded(_op("GET", "/auth/callback"), **kwargs) + assert au.is_excluded(_op("POST", "/api/svc/sync"), **kwargs) + assert not au.is_excluded(_op("GET", "/api/svc/sync"), **kwargs) + assert not au.is_excluded(_op("GET", "/api/dead"), **kwargs) + + +# ------------------------------------------------------------------ config / orchestration + + +def test_parse_config_validation() -> None: + with pytest.raises(au.ApiUsageError, match="no .*apps"): + au.parse_config({}) + with pytest.raises(au.ApiUsageError, match="exactly one"): + au.parse_config({"apps": [{"name": "x", "frontends": ["fe"]}]}) + with pytest.raises(au.ApiUsageError, match="exactly one"): + au.parse_config({"apps": [{"name": "x", "app": "a:b", "openapi": "o.json", "frontends": ["fe"]}]}) + with pytest.raises(au.ApiUsageError, match="frontends"): + au.parse_config({"apps": [{"name": "x", "app": "a:b"}]}) + + +@pytest.mark.parametrize("key", ["frontends", "exclude_prefixes", "exclude_tags", "exclude_paths", "exclude_frontend_globs"]) +def test_parse_config_rejects_strings_where_lists_are_required(key: str) -> None: + raw = {"name": "x", "app": "a:b", "frontends": ["fe"], key: "/internal"} + with pytest.raises(au.ApiUsageError, match=f"`{key}` must be a list of strings"): + au.parse_config({"apps": [raw]}) + + +def _project(root: Path, *, baseline: bool = False, extra: str = "") -> None: + _write( + root, + "openapi.json", + json.dumps({"paths": {"/api/used": {"get": {}}, "/api/dead": {"get": {"tags": ["t"]}}, "/external/x": {"get": {}}}}), + ) + _write(root, "fe/src/a.ts", "fetch('/api/used');\n") + baseline_line = 'baseline = "api-usage-baseline.txt"\n' if baseline else "" + _write( + root, + "pyproject.toml", + f""" + [project] + name = "x" + + [[tool.bdt.dead_code.apps]] + name = "main" + openapi = "openapi.json" + frontends = ["fe/src"] + exclude_prefixes = ["/external"] + {baseline_line}{extra} + """, + ) + + +def _config(root: Path) -> au.AppConfig: + return au.parse_config(load_bdt_table("dead_code", root))[0] + + +def test_check_app_reports_uncalled_routes(tmp_path: Path) -> None: + _project(tmp_path) + findings = au.check_app(_config(tmp_path), repo_root=tmp_path) + assert [(f.rule, f.message.split(" is never")[0]) for f in findings] == [("api-route-uncalled", "GET /api/dead [t]")] + + +def test_frontend_globs_expand_and_exclude_frontend_globs_apply(tmp_path: Path) -> None: + _write(tmp_path, "openapi.json", json.dumps({"paths": {"/api/a": {"get": {}}, "/api/b": {"get": {}}, "/api/c": {"get": {}}}})) + _write(tmp_path, "apps/one/src/x.ts", "fetch('/api/a');\n") + _write(tmp_path, "apps/two/src/x.ts", "fetch('/api/b');\n") + _write(tmp_path, "apps/two/src/legacy/old.ts", "fetch('/api/c');\n") + config = au.AppConfig(name="m", openapi="openapi.json", frontends=["apps/*/src"], exclude_frontend_globs=["apps/two/src/legacy/*"]) + assert [f.message.split(" ")[1] for f in au.check_app(config, repo_root=tmp_path)] == ["/api/c"] + + +def test_frontends_entry_must_match_and_be_relative(tmp_path: Path) -> None: + _project(tmp_path) + with pytest.raises(au.ApiUsageError, match="matches no directory"): + au.check_app(au.AppConfig(name="m", openapi="openapi.json", frontends=["nope/*/src"]), repo_root=tmp_path) + with pytest.raises(au.ApiUsageError, match="repo-relative"): + au.check_app(au.AppConfig(name="m", openapi="openapi.json", frontends=["/abs"]), repo_root=tmp_path) + + +def test_empty_inventory_is_an_error_not_a_clean_pass(tmp_path: Path) -> None: + _project(tmp_path) + _write(tmp_path, "openapi.json", "{}") + with pytest.raises(au.ApiUsageError, match="no operations found"): + au.check_app(_config(tmp_path), repo_root=tmp_path) + + +def test_baseline_ratchet(tmp_path: Path) -> None: + _project(tmp_path, baseline=True) + config = _config(tmp_path) + + assert [f.rule for f in au.check_app(config, repo_root=tmp_path)] == ["api-route-uncalled"] # no baseline file yet + assert au.check_app(config, repo_root=tmp_path, update_baseline=True) == [] + assert "GET /api/dead" in (tmp_path / "api-usage-baseline.txt").read_text() + assert au.check_app(config, repo_root=tmp_path) == [] # tolerated now + + # a new dead route is reported; a baseline line whose route vanished is reported as stale + _write(tmp_path, "openapi.json", json.dumps({"paths": {"/api/used": {"get": {}}, "/api/newdead": {"get": {}}}})) + findings = au.check_app(config, repo_root=tmp_path) + assert sorted((f.rule, "GET /api/newdead" in f.message, "GET /api/dead" in f.message) for f in findings) == [ + ("api-route-baseline-stale", False, True), + ("api-route-uncalled", True, False), + ] + assert "no longer a backend route" in next(f for f in findings if f.rule == "api-route-baseline-stale").message + + +@pytest.mark.parametrize( + ("change", "reason"), + [("called", "now used"), ("excluded", "now excluded")], +) +def test_baseline_stale_reason(tmp_path: Path, change: str, reason: str) -> None: + _project(tmp_path, baseline=True) + config = _config(tmp_path) + au.check_app(config, repo_root=tmp_path, update_baseline=True) + if change == "called": + _write(tmp_path, "fe/src/b.ts", "fetch('/api/dead');\n") + else: + config = au.AppConfig(**{**{f: getattr(config, f) for f in config.__slots__}, "exclude_tags": ["t"]}) + findings = au.check_app(config, repo_root=tmp_path) + assert [f.rule for f in findings] == ["api-route-baseline-stale"] + assert reason in findings[0].message + + +def test_update_baseline_validates_all_apps_before_writing(tmp_path: Path) -> None: + _project(tmp_path, baseline=True) + table = load_bdt_table("dead_code", tmp_path) + second = {"name": "second", "openapi": "openapi.json", "frontends": ["fe/src"]} # no baseline + with pytest.raises(au.ApiUsageError, match="missing for: second"): + au.run({"apps": [*table["apps"], second]}, repo_root=tmp_path, update_baseline=True) + assert not (tmp_path / "api-usage-baseline.txt").exists() + + shared = {**second, "baseline": "api-usage-baseline.txt"} + with pytest.raises(au.ApiUsageError, match="share the same"): + au.run({"apps": [*table["apps"], shared]}, repo_root=tmp_path, update_baseline=True) + assert not (tmp_path / "api-usage-baseline.txt").exists() + + +def test_update_baseline_creates_parent_directories(tmp_path: Path) -> None: + _project(tmp_path) + config = au.AppConfig(name="m", openapi="openapi.json", frontends=["fe/src"], baseline="deep/dir/bl.txt") + au.check_app(config, repo_root=tmp_path, update_baseline=True) + assert (tmp_path / "deep/dir/bl.txt").is_file() + + +# ------------------------------------------------------------------ routes referenced by name (url_for) + + +def _op_id(method: str, path: str, operation_id: str, *, mount: str = "") -> au.Operation: + return au.Operation(method, path, (), mount, operation_id) + + +def test_read_referenced_route_names(tmp_path: Path) -> None: + _write( + tmp_path, + "backend/auth.py", + """ + url = request.url_for("auth_callback") + other = app.url_path_for(name='login') + mounted = request.url_for("sub:inner") + dynamic = request.url_for(provider) + also = request.url_for( + "multi_line", + ) + nope = my_url_for("not_a_call") + """, + ) + _write(tmp_path, "backend/templates/page.html", "x") + _write(tmp_path, "backend/tests/test_auth.py", 'url_for("only_in_tests")\n') + _write(tmp_path, "backend/test_x.py", 'url_for("only_in_test_file")\n') + _write(tmp_path, "backend/node_modules/x.py", 'url_for("vendored")\n') + assert au.read_referenced_route_names(tmp_path / "backend") == {"auth_callback", "login", "inner", "multi_line", "from_template"} + assert au.read_referenced_route_names(tmp_path / "backend", ("my_url_for",)) == {"not_a_call"} + + +@pytest.mark.parametrize( + ("op", "names", "expected"), + [ + # FastAPI's default operationId: route name + path with non-word characters as `_`, + method + (_op_id("GET", "/auth/login", "login_auth_login_get"), {"login"}, True), + (_op_id("GET", "/auth/cb/{x}", "auth_cb_auth_cb__x__get"), {"auth_cb"}, True), + (_op_id("GET", "/api/sub/items", "list_items_items_get", mount="/api/sub"), {"list_items"}, True), + (_op_id("GET", "/auth/login", "login_auth_login_get"), {"logout"}, False), + (_op_id("GET", "/auth/login", "login_auth_login_get"), {"log"}, False), # a prefix of the name is not the name + (_op_id("POST", "/auth/login", "login_auth_login_get"), {"login"}, False), # method is part of the id + # custom generate_unique_id + (_op_id("GET", "/auth/login", "login"), {"login"}, True), + (_op_id("GET", "/auth/login", "auth-login"), {"login"}, True), + (_op_id("GET", "/auth/login", "auth-relogin"), {"login"}, False), + (_op("GET", "/auth/login"), {"login"}, False), # no operationId (older openapi file): nothing to go on + ], +) +def test_is_named_by(op: au.Operation, names: set[str], expected: bool) -> None: + assert au.is_named_by(op, names) is expected + + +def test_route_referenced_by_url_for_is_not_reported(tmp_path: Path) -> None: + _write( + tmp_path, + "openapi.json", + json.dumps( + { + "paths": { + "/auth/callback": {"get": {"operationId": "auth_callback_auth_callback_get"}}, + "/auth/dead": {"get": {"operationId": "dead_auth_dead_get"}}, + } + } + ), + ) + _write(tmp_path, "fe/src/a.ts", "export {};\n") + _write(tmp_path, "backend/auth.py", 'redirect_uri = request.url_for("auth_callback")\n') + _write( + tmp_path, + "pyproject.toml", + """ + [[tool.bdt.dead_code.apps]] + openapi = "openapi.json" + app_dir = "backend" + frontends = ["fe/src"] + """, + ) + findings = au.check_app(_config(tmp_path), repo_root=tmp_path) + assert [f.message.split(" ")[:2] for f in findings] == [["GET", "/auth/dead"]] + + +def test_url_for_functions_are_configurable(tmp_path: Path) -> None: + _project(tmp_path, extra='url_for_functions = ["reverse"]') + _write(tmp_path, "openapi.json", json.dumps({"paths": {"/api/x": {"get": {"operationId": "x_api_x_get"}}}})) + _write(tmp_path, "svc.py", 'reverse("x")\n') + assert au.check_app(_config(tmp_path), repo_root=tmp_path) == [] + _write(tmp_path, "svc.py", 'url_for("x")\n') # not one of the configured functions any more + assert len(au.check_app(_config(tmp_path), repo_root=tmp_path)) == 1 + + +# ------------------------------------------------------------------ CLI + + +def test_cli_exit_codes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + runner = CliRunner() + _project(tmp_path) + monkeypatch.chdir(tmp_path) + result = runner.invoke(app, ["dead-code"]) + assert result.exit_code == 1 + assert "GET /api/dead" in result.output + assert "bdt lint:" not in result.output # not the `bdt lint` summary line + + _write(tmp_path, "fe/src/b.ts", "fetch('/api/dead');\n") + clean = runner.invoke(app, ["dead-code"]) + assert clean.exit_code == 0 + assert "no dead code found" in clean.output + + _write(tmp_path, "pyproject.toml", "[project]\nname='x'\n") + broken = runner.invoke(app, ["dead-code"]) + assert broken.exit_code == 2 + assert "nothing to check" in broken.output + + _write(tmp_path, "pyproject.toml", "[project\n") + assert runner.invoke(app, ["dead-code"]).exit_code == 2 # invalid TOML is a setup error, not a traceback + + +def test_cli_runs_sql_and_routes_checks_together(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + runner = CliRunner() + _project(tmp_path, extra='\n[tool.bdt.dead_code]\nsql_roots = ["sql"]') + _write(tmp_path, "sql/q/orphan.sql", "select 1") + monkeypatch.chdir(tmp_path) + both = runner.invoke(app, ["dead-code"]) + assert both.exit_code == 1 + assert "orphan.sql" in both.output + assert "GET /api/dead" in both.output + assert "2 issue(s) found" in both.output + sql_only = runner.invoke(app, ["dead-code", "--only", "sql"]) + assert "orphan.sql" in sql_only.output + assert "GET /api/dead" not in sql_only.output + routes_only = runner.invoke(app, ["dead-code", "--only", "routes"]) + assert "GET /api/dead" in routes_only.output + assert "orphan.sql" not in routes_only.output + + +def test_cli_update_baseline(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + runner = CliRunner() + _project(tmp_path, baseline=True) + monkeypatch.chdir(tmp_path) + assert runner.invoke(app, ["dead-code"]).exit_code == 1 + updated = runner.invoke(app, ["dead-code", "--update-baseline"]) + assert updated.exit_code == 0 + assert "baseline(s) updated" in updated.output + assert runner.invoke(app, ["dead-code"]).exit_code == 0 diff --git a/tests/test_lint_sql_files.py b/tests/test_lint_sql_files.py new file mode 100644 index 0000000..f035ef8 --- /dev/null +++ b/tests/test_lint_sql_files.py @@ -0,0 +1,215 @@ +from pathlib import Path + +import pytest + +from bmsdna.devtools import dead_code +from bmsdna.devtools.bdt_config import load_bdt_table +from bmsdna.devtools.api_usage import ApiUsageError +from bmsdna.devtools.lint_sql_files import check_unreferenced_sql_files + + +def _write(root: Path, rel: str, text: str = "") -> Path: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + return path + + +def _check(root: Path, *, sql_root: str = "backend", **kwargs) -> list[str]: + sql_files = sorted((root / sql_root).rglob("*.sql")) + findings = check_unreferenced_sql_files(repo_root=root, sql_files=sql_files, python_files=sorted(root.rglob("*.py")), **kwargs) + return [f.path.relative_to(root).as_posix() for f in findings if f.rule == "sql-file-unreferenced"] + + +def test_literal_load_sql_call_references_file(tmp_path: Path) -> None: + _write(tmp_path, "backend/queries/customers/get_all.sql", "select 1") + _write(tmp_path, "backend/queries/customers/dead.sql", "select 2") + _write(tmp_path, "backend/repo.py", 'q = load_sql("customers", "get_all")\n') + assert _check(tmp_path) == ["backend/queries/customers/dead.sql"] + + +def test_same_stem_in_other_topic_does_not_count(tmp_path: Path) -> None: + _write(tmp_path, "backend/queries/a/get.sql", "select 1") + _write(tmp_path, "backend/queries/b/get.sql", "select 1") + _write(tmp_path, "backend/repo.py", 'load_sql("a", "get")\n') + assert _check(tmp_path) == ["backend/queries/b/get.sql"] + + +def test_dynamic_name_counts_stems_literal_in_same_file(tmp_path: Path) -> None: + _write(tmp_path, "backend/queries/rel/by_a.sql") + _write(tmp_path, "backend/queries/rel/by_b.sql") + _write(tmp_path, "backend/queries/rel/unused.sql") + _write( + tmp_path, + "backend/repo.py", + 'name = "by_a" if x else "by_b"\nload_sql("rel", name)\n', + ) + assert _check(tmp_path) == ["backend/queries/rel/unused.sql"] + + +def test_dynamic_name_ignores_literals_from_other_files(tmp_path: Path) -> None: + _write(tmp_path, "backend/queries/rel/by_a.sql") + _write(tmp_path, "backend/repo.py", 'load_sql("rel", name)\n') + _write(tmp_path, "backend/other.py", 'x = "by_a"\n') + assert _check(tmp_path) == ["backend/queries/rel/by_a.sql"] + + +def test_dynamic_topic_references_any_stem_named_in_file(tmp_path: Path) -> None: + _write(tmp_path, "backend/queries/t/x.sql") + _write(tmp_path, "backend/queries/t/y.sql") + _write(tmp_path, "backend/repo.py", 'load_sql(topic, "x")\n') + assert _check(tmp_path) == ["backend/queries/t/y.sql"] + + +def test_repo_relative_path_literal_references_file(tmp_path: Path) -> None: + _write(tmp_path, "backend/api/sql/print/conditions.sql") + _write(tmp_path, "backend/api/sql/print/other.sql") + _write( + tmp_path, + "backend/print.py", + 'get_sql_with_prm_list("backend/api/sql/print/conditions.sql")\n', + ) + assert _check(tmp_path) == ["backend/api/sql/print/other.sql"] + + +def test_bare_filename_only_counts_inside_sql_folders_parent(tmp_path: Path) -> None: + _write(tmp_path, "backend/app/dedup/sql/active.sql") + _write(tmp_path, "backend/app/dedup/sql/lonely.sql") + _write( + tmp_path, + "backend/app/dedup/svc.py", + '_SQL = Path(__file__).parent / "sql"\nq = (_SQL / "active.sql").read_text()\n', + ) + _write( + tmp_path, + "backend/app/elsewhere/other.py", + 'q = (D / "lonely.sql").read_text()\n', + ) + assert _check(tmp_path) == ["backend/app/dedup/sql/lonely.sql"] + + +def test_custom_loader_function_name(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/t/a.sql") + _write(tmp_path, "backend/repo.py", 'get_query("t", "a")\n') + assert _check(tmp_path) == ["backend/q/t/a.sql"] + assert _check(tmp_path, loader_functions=["get_query"]) == [] + + +def test_method_call_on_loader_object_is_recognised(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/t/a.sql") + _write(tmp_path, "backend/repo.py", 'loader.load_sql("t", "a")\n') + assert _check(tmp_path) == [] + + +def test_ignore_globs_suppress_finding(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/t/a.sql") + assert _check(tmp_path, ignore_globs=["backend/q/t/*.sql"]) == [] + + +def test_keyword_arguments_to_loader_are_recognised(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/a/b.sql") + _write(tmp_path, "backend/q/a/c.sql") + _write(tmp_path, "backend/repo.py", 'load_sql(topic="a", name="b")\n') + assert _check(tmp_path) == ["backend/q/a/c.sql"] + + +def test_unresolvable_loader_call_is_treated_as_dynamic(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/a/b.sql") + _write(tmp_path, "backend/repo.py", 'load_sql(**spec)\nx = "b"\n') + assert _check(tmp_path) == [] + + +def test_dynamic_name_does_not_count_the_topic_literal_itself(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/orders/orders.sql") + _write(tmp_path, "backend/repo.py", 'load_sql("orders", name)\n') + assert _check(tmp_path) == ["backend/q/orders/orders.sql"] + + +def test_path_literal_must_match_whole_trailing_segments(tmp_path: Path) -> None: + _write(tmp_path, "backend/xsql/a.sql") + _write(tmp_path, "backend/sql/a.sql") + _write(tmp_path, "backend/sql/init.sql") + _write(tmp_path, "backend/sql/reinit.sql") + _write(tmp_path, "backend/use.py", 'a = open("backend/sql/a.sql")\nb = open("./sql/init.sql")\nc = base + "/reinit.sql"\n') + assert _check(tmp_path) == ["backend/sql/reinit.sql", "backend/xsql/a.sql"] + + +def test_windows_separators_in_path_literals(tmp_path: Path) -> None: + _write(tmp_path, "backend/sql/a/b.sql") + _write(tmp_path, "backend/use.py", 'p = "backend\\\\sql\\\\a\\\\b.sql"\n') + assert _check(tmp_path) == [] + + +def test_bare_filename_in_unrelated_file_does_not_count_for_top_level_sql_folder(tmp_path: Path) -> None: + _write(tmp_path, "sql/a.sql") + _write(tmp_path, "tests/t.py", 'x = "a.sql"\n') + assert _check(tmp_path, sql_root="sql") == ["sql/a.sql"] + + +def test_unparseable_python_file_is_reported_not_silently_skipped(tmp_path: Path) -> None: + _write(tmp_path, "backend/q/t/a.sql") + _write(tmp_path, "backend/broken.py", 'load_sql("t", "a")\ndef (:\n') + findings = check_unreferenced_sql_files( + repo_root=tmp_path, sql_files=[tmp_path / "backend/q/t/a.sql"], python_files=sorted(tmp_path.rglob("*.py")) + ) + assert sorted(f.rule for f in findings) == ["sql-check-python-unparseable", "sql-file-unreferenced"] + + +def _project(root: Path, config: str) -> None: + _write(root, "backend/q/t/a.sql") + _write(root, "backend/q/t/dead.sql") + _write(root, "backend/repo.py", 'load_sql("t", "a")\n') + _write(root, "backend/other.py", "x = 1\n") + _write(root, "pyproject.toml", f"[project]\nname='x'\n[tool.bdt.dead_code]\n{config}\n") + + +def _run(root: Path, **kwargs) -> list: + return dead_code.run(load_bdt_table("dead_code", root), repo_root=root, **kwargs) + + +def test_sql_check_is_opt_in(tmp_path: Path) -> None: + _project(tmp_path, "") + with pytest.raises(ApiUsageError, match="nothing to check"): + _run(tmp_path) + + +def test_sql_check_scans_whole_repo_for_references(tmp_path: Path) -> None: + _project(tmp_path, "sql_roots=['backend/q']") + # the reference to a.sql lives in repo.py, not in the file that happens to be changed + findings = _run(tmp_path) + assert [f.path.name for f in findings] == ["dead.sql"] + assert findings[0].rule == "sql-file-unreferenced" + + +def test_sql_check_reads_loader_and_ignore_config_from_pyproject(tmp_path: Path) -> None: + _project(tmp_path, "sql_roots=['backend/q']\nsql_loader_functions=['get_query']\nsql_unreferenced_ignore=['backend/q/t/*.sql']") + assert _run(tmp_path) == [] # everything ignored + _project(tmp_path, "sql_roots=['backend/q']\nsql_loader_functions=['get_query']") + # load_sql is no longer the loader name, so both files are unreferenced + assert sorted(f.path.name for f in _run(tmp_path)) == ["a.sql", "dead.sql"] + + +def test_sql_check_accepts_bare_string_for_list_settings(tmp_path: Path) -> None: + _project(tmp_path, "sql_roots='backend/q'") + assert [f.path.name for f in _run(tmp_path)] == ["dead.sql"] + + +def test_sql_check_reports_missing_or_escaping_sql_root(tmp_path: Path) -> None: + _project(tmp_path, "sql_roots=['nope', '../outside']") + assert [f.rule for f in _run(tmp_path)] == ["lint-path-not-found", "lint-path-not-found"] + + +def test_sql_check_honours_exclude_dirs(tmp_path: Path) -> None: + _project(tmp_path, "sql_roots=['backend/q']\nexclude_dirs=['t']") + assert _run(tmp_path) == [] # the only SQL files sit in an excluded subdirectory + + +def test_only_narrows_the_run_and_rejects_unknown_checks(tmp_path: Path) -> None: + _project(tmp_path, "sql_roots=['backend/q']") + assert len(_run(tmp_path, only=["sql"])) == 1 + with pytest.raises(ApiUsageError, match=r"apps"): + _run(tmp_path, only=["routes"]) + with pytest.raises(ApiUsageError, match="unknown check"): + _run(tmp_path, only=["nope"]) + with pytest.raises(ApiUsageError, match="only applies to"): + _run(tmp_path, only=["sql"], update_baseline=True)