bruno: replace Python generator with Go golden-writer in the drift test

The generator moves into the Go toolchain per repo preference:
internal/docs/bruno_gen.go emits the OpenCollection mirrors (same output
contract — gate-verified example-request bodies, env-var params,
device-token headers, login/register/refresh token-capture scripts) and
TestBrunoMirrorsDocs grows a -update flag: verification by default,
regeneration on demand:

  go test ./internal/docs -run TestBrunoMirrorsDocs -update

Generator and checker are now the same code — they cannot drift apart.
scripts/gen_bruno_from_docs.py deleted; opencollection.yml README points
at the go-test command.

Also fixed vs the Python pass: mirror placement follows the docs tree
folder (RelPath), not the area frontmatter (several docs — highlights,
notes, ratings, users, progress — carry a different area); docs blocks
emit the collection's mapping form (content + type), not a bare scalar;
JSON body sanity is checked at generation time.

Idempotent: back-to-back -update runs are byte-identical.
make test green (9 pkgs).
This commit is contained in:
John O'Keefe committed 2026-10-24 18:47:56 -04:00
1 parent acd82fff0a
commit cfcd95b25e
195 files changed
+1479 -1073

No files matched your search

-253
View File
@@ -1,253 +0,0 @@
#!/usr/bin/env python3
"""Generate OpenCollection (Bruno) request YAMLs from the canonical API docs.
Truth flow: code -> docs/developer/api (canonical) -> bruno/ (executable
verification, never hand-authored). Every frontmatter'd endpoint doc
becomes one request file at bruno/<area>/<docname>.yml, with the
gate-verified example-request JSON as the body.
Re-run after doc changes: python3 scripts/gen_bruno_from_docs.py [--apply]
Without --apply, prints the plan only.
"""
import os, re, sys, glob
DOCS = "docs/developer/api"
BRUNO = "bruno"
KEEP_DIRS = {"NewDevDBSetup", "environments"} # hand-authored setup + env
ROOT_KEEP = {"opencollection.yml", "NewDB.sh"}
# Hand-authored edge-case verification (not docs mirrors) — kept as-is.
KEEP_SUBSTRINGS = ("/scenarios/", "/filters/", "/search/")
# Token-capture after-response scripts, keyed by doc filename — these chain
# the collection (later requests need {{token}} / {{refresh_token}}).
TOKEN_SCRIPT = """function onResponse(res) {
try {
const responseBody = res.getBody();
const token = responseBody.access_token;
const refreshToken = responseBody.refresh_token;
if (token) {
bru.setEnvVar("token", token, { persist: true });
console.log("Access token saved");
}
if (refreshToken) {
bru.setEnvVar("refresh_token", refreshToken, { persist: true });
console.log("Refresh token saved");
}
} catch (error) {
console.error("Token capture failed:", error.message);
}
}
onResponseResponsePlaceholder();"""
TOKEN_SCRIPT = TOKEN_SCRIPT.replace("onResponseResponsePlaceholder();", "onResponse();")
SCRIPT_DOCS = {"login", "register", "refresh_token"}
# Param-name -> env var (checked longest-first, then generic :id by area).
NAMED_PARAMS = {
"registration_id": "registration_id",
"deviceid": "device_id",
"libraryid": "ebook_library_id",
"collectionid": "collection_id",
"trackindex": "track_index",
"highlightid": "highlight_id",
"annotationid": "annotation_id",
"ruleid": "rule_id",
"aliasid": "alias_id",
"folderid": "library_folder_id",
"jobid": "job_id",
"bookid": "book_uuid",
"sha256": "book_sha256",
"noteid": "note_id",
"userid": "user_id",
"uuid": "book_uuid",
"token": "kobo_device_token",
}
# (area, subpath-contains) -> env for generic :id
ID_BY_CONTEXT = [
("media-items", "media_item_id"),
("collections", "collection_id"),
("libraries", "ebook_library_id"),
("devices", "device_id"),
("progress", "media_item_id"),
("stats/books", "media_item_id"),
("annotations", "annotation_id"),
]
NEW_ENV_VARS = [
"device_id", "rule_id", "annotation_id", "alias_id", "track_index",
"registration_id", "unlinked_book_id", "device_token",
]
def norm(p: str) -> str:
p = re.sub(r"/+", "/", p)
return re.sub(r":[^/{}]+", lambda m: "{" + m.group(0)[1:] + "}", p).rstrip("/")
def parse_doc(path: str):
src = open(path).read()
m = re.match(r"---\n(.*?)\n---\n", src, re.S)
if not m:
return None
fm = m.group(1)
method = re.search(r"method:\s*(\w+)", fm)
route = re.search(r"path:\s*(\S+)", fm)
auth = re.search(r"auth:\s*(\S+)", fm)
if not (method and route):
return None
body_src = src[m.end():]
title = re.search(r"^# (.+)$", body_src, re.M)
ex = re.search(r"```json example-request\n(.*?)\n```", body_src, re.S)
return {
"rel": os.path.relpath(path, DOCS),
"area": os.path.relpath(path, DOCS).split(os.sep)[0],
"name": os.path.splitext(os.path.basename(path))[0],
"method": method.group(1).upper(),
"path": route.group(1),
"auth": auth.group(1) if auth else "jwt",
"title": title.group(1).strip() if title else route.group(1),
"example": ex.group(1) if ex else None,
}
def env_for(area: str, path: str, param: str) -> str:
key = param.lstrip(":").lower()
if key in NAMED_PARAMS:
return NAMED_PARAMS[key]
if key == "id":
ctx = f"{area}/{path.split('/')[-2] if '/' in path else ''}"
for prefix, var in ID_BY_CONTEXT:
if ctx.startswith(prefix) or f"/api/{prefix}/" in path + "/":
return var
if path.startswith("/api/sync/unlinked-books"):
return "unlinked_book_id"
return "media_item_id" # best default for this API's :id grammar
return key # fallback: env var named after the param
def yaml_str_block(indent: int, text: str) -> str:
pad = " " * indent
lines = text.rstrip("\n").split("\n")
out = [f"{pad}|-"]
out += [pad + " " + ln if ln else "" for ln in lines]
return "\n".join(out)
def emit(doc: dict, seq: int) -> str:
path = norm(doc["path"])
path = re.sub(r"\{(\w+)\}", lambda m: "{{" + env_for(doc["area"], doc["path"], m.group(1)) + "}}", path)
url = "{{base_url}}" + path
out = [
"info:",
f" name: {doc['title']}",
" type: http",
f" seq: {seq}",
"",
"http:",
f" method: {doc['method']}",
f" url: '{url}'",
" auth: inherit",
]
if doc["example"]:
out += [" body:", " type: json", " data: " + yaml_str_block(4, doc["example"]).lstrip()]
if doc["auth"] == "device-token":
out += [
" headers:",
" - key: Authorization",
" value: Bearer {{device_token}}",
]
out += [
"settings:",
" encodeUrl: true",
" timeout: 0",
" followRedirects: true",
" maxRedirects: 5",
]
if doc["name"] in SCRIPT_DOCS:
out += [
"",
"runtime:",
" scripts:",
" - type: after-response",
" code: " + yaml_str_block(8, TOKEN_SCRIPT).lstrip(),
]
out += [
"",
"docs: " + yaml_str_block(0, (
f"# {doc['title']}\n\n"
f"`{doc['method']} {doc['path']}` — auth: {doc['auth']}.\n\n"
f"Canonical reference: `docs/developer/api/{doc['rel']}` "
f"(docs site → Developer → API). This request is GENERATED from "
f"that doc; edit the doc, not this file."
)),
"",
]
return "\n".join(out)
def main():
apply = "--apply" in sys.argv
docs = []
for path in sorted(glob.glob(f"{DOCS}/**/*.md", recursive=True)):
d = parse_doc(path)
if d:
docs.append(d)
print(f"{len(docs)} frontmatter'd endpoint docs")
# Plan writes
writes = {}
by_area = {}
for d in docs:
by_area.setdefault(d["area"], []).append(d)
for area, ds in sorted(by_area.items()):
ds.sort(key=lambda d: d["name"])
for i, d in enumerate(ds, 1):
writes[f"{BRUNO}/{area}/{d['name']}.yml"] = emit(d, i)
# Plan deletes: existing request ymls not regenerated (setup/env/root,
# and hand-authored scenario suites kept)
deletes = []
for path in glob.glob(f"{BRUNO}/**/*.yml", recursive=True):
rel = os.path.relpath(path, BRUNO)
top = rel.split(os.sep)[0]
if top in KEEP_DIRS or rel in ROOT_KEEP or os.path.basename(path) == "folder.yml":
continue
if any(s in path for s in KEEP_SUBSTRINGS):
continue
if path not in writes:
deletes.append(path)
print(f"write/update {len(writes)}, delete {len(deletes)}")
for p in sorted(deletes):
print(" DEL", p)
if not apply:
print("(dry run — pass --apply)")
return
import collections
for path, content in writes.items():
os.makedirs(os.path.dirname(path), exist_ok=True)
open(path, "w").write(content)
for p in deletes:
os.remove(p)
# prune empty dirs
for root, dirs, files in os.walk(BRUNO, topdown=False):
if root == BRUNO:
continue
if not files and not dirs and os.path.basename(root) not in KEEP_DIRS:
os.rmdir(root)
# extend env file with any new vars
envp = f"{BRUNO}/environments/Bookhoard.yml"
env = open(envp).read()
add = [v for v in NEW_ENV_VARS if f"name: {v}\n" not in env]
if add:
block = "".join(f" - secret: true\n name: {v}\n" for v in add)
env = env.replace("variables:\n", "variables:\n" + block, 1)
open(envp, "w").write(env)
print("env vars added:", ", ".join(add))
print("applied.")
if __name__ == "__main__":
main()