Merge remote-tracking branch 'origin/wj' into wilm

This commit is contained in:
benjamin.baker
2026-09-19 14:40:46 -04:00
58 changed files with 37493 additions and 6385 deletions

View File

@@ -183,3 +183,63 @@ def test_documented_auth_is_what_the_code_enforces():
'inventory is derived, so fix the code or teach the classifier in '
'scripts/gen_openapi.py, then re-run it with --refresh:\n %s'
% (len(disagreements), '\n '.join(disagreements[:15])))
# ---------------------------------------------------------------------------
# The third derived field: query parameters
# ---------------------------------------------------------------------------
#
# A Flask rule carries its path arguments and nothing else, so a route whose
# real interface is '?code=' or '?search=' documented no parameters at all -
# /api/usb/kiosk/setup among them, where '?code=' is the ONLY way to call it
# successfully. gen_openapi.py now reads them out of each view's AST and writes
# them into the inventory as `queryparams`.
#
# That field is DERIVED, like auth, and gated the same way and for the same
# reason: a refresh only rewrites, so without a comparison a hand-edit or a
# stale file publishes a query contract the handler does not implement. A
# parameter that is not really there is the worst kind of documentation defect,
# because a reader believes it, sends it, and is silently ignored.
def derived_queryparams():
"""{(METHOD, path): [parameter, ...]} as the sweep reads the code now."""
generator = _generator()
app = create_app('testing')
derived = {}
for rule, methods, endpoint, view in generator._served_routes(app):
parameters = generator._query_parameters(view)
for method in methods:
derived[(method, normalise(rule))] = parameters
return derived
def documented_queryparams():
"""{(METHOD, path): [parameter, ...]} as the inventory has it committed."""
documented = {}
for surface in json.loads(INVENTORY.read_text()):
for entry in surface.get('endpoints', []):
path = normalise(entry.get('path') or '')
for method in re.split(r'[/,|]', entry.get('method') or 'GET'):
method = method.strip().upper()
if method:
documented[(method, path)] = entry.get('queryparams') or []
return documented
def test_documented_query_parameters_are_what_the_handler_reads():
derived = derived_queryparams()
documented = documented_queryparams()
disagreements = []
for key in sorted(set(derived) & set(documented)):
if derived[key] != documented[key]:
disagreements.append(
'%s %s\n documented: %s\n read: %s'
% (key[0], key[1],
[item['name'] for item in documented[key]],
[item['name'] for item in derived[key]]))
assert not disagreements, (
'%d route(s) document query parameters that are not what the handler '
'reads. The field is derived, so regenerate with '
'scripts/gen_openapi.py --refresh:\n %s'
% (len(disagreements), '\n '.join(disagreements[:15])))

View File

@@ -0,0 +1,283 @@
"""The Swagger UI "try it" page at /api/docs/try.
Two things are being proved, and they are the two that would hurt on the
production box:
1. It is OFF unless a site turns it on, and off means the URL does not exist.
The page fires real requests at this server with the browser's own session,
so a disabled deployment must not serve it, advertise it, or confirm it is
a thing by answering anything other than 404.
2. Nothing it loads comes from the internet. Production is air-gapped: a CDN
reference, a webfont, or a source map pointing at a public host is not a
slow page, it is a page that never renders and a request that hangs.
These build their own Flask app around the docs blueprint rather than using the
session app from conftest. The blueprint touches no database, and a second
create_app() would rebind the SQLAlchemy extension and break whatever module
runs next (see the note on the gated_app fixture in test_public_read_surface).
It also lets one test see the flag on and the next see it off, which is the
whole behaviour under test.
"""
import os
import re
import pytest
from flask import Flask
from shopdb.core.api.docs import docs_bp
# Everything the try-it page is allowed to load, and the files behind them.
_CSS_PATH = '/api/docs/swagger-ui.css'
_JS_PATH = '/api/docs/swagger-ui-bundle.min.js'
_STATICDOCS = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
'shopdb', 'core', 'api', 'staticdocs')
_VENDORED = ('swagger-ui.css', 'swagger-ui-bundle.min.js')
def _docs_app(tryit_enabled):
"""A bare app carrying only the docs blueprint, mounted where the real app
mounts it (register_blueprints uses /api/<name>)."""
application = Flask(__name__)
application.config['API_DOCS_TRYIT_ENABLED'] = tryit_enabled
application.register_blueprint(docs_bp, url_prefix='/api/docs')
return application
@pytest.fixture
def enabled_client():
return _docs_app(True).test_client()
@pytest.fixture
def disabled_client():
return _docs_app(False).test_client()
def test_enabled_page_renders_swagger_over_the_published_spec(enabled_client):
response = enabled_client.get('/api/docs/try')
assert response.status_code == 200
page = response.get_data(as_text=True)
assert 'SwaggerUIBundle(' in page
assert 'id="swagger-ui"' in page
# One spec, two renderers: the try-it page must read the same document the
# Redoc page serves, not a second copy that could drift from it.
assert '/api/docs/openapi.json' in page
def test_the_page_references_no_other_origin_at_all(enabled_client):
"""Absolute AND protocol-relative.
The first version of this scanned for https?://, so
<script src="//cdn.jsdelivr.net/npm/x.js"> - the form a copy-paste from any
CDN's own docs produces - matched nothing and passed.
"""
page = enabled_client.get('/api/docs/try').get_data(as_text=True)
offenders = re.findall(r'(?:src|href)\s*=\s*[\'"]((?:https?:)?//[^\'"]+)', page)
assert not offenders, 'the try-it page loads from another origin: %s' % offenders
def test_the_redoc_page_survives_try_it_being_unmounted(app):
"""The banner linking to the try-it page is built with url_for, and url_for
raises BuildError when those routes are not registered - which is every
production box. An unconditional link therefore 500s /api/docs/ in exactly
the deployment nobody tests locally.
"""
# The testing config is the unmounted case, and it is the one available
# here: ProductionConfig refuses to build without a real SECRET_KEY, which
# is correct of it and makes it useless as a fixture.
page = app.test_client().get('/api/docs/')
assert page.status_code == 200, page.get_data(as_text=True)[:200]
body = page.get_data(as_text=True)
assert 'redoc' in body.lower()
assert '/api/docs/try' not in body
def test_enabled_page_loads_only_the_vendored_assets(enabled_client):
page = enabled_client.get('/api/docs/try').get_data(as_text=True)
assert 'href="%s"' % _CSS_PATH in page
assert 'src="%s"' % _JS_PATH in page
for path, expected_type in ((_CSS_PATH, 'text/css'),
(_JS_PATH, 'application/javascript')):
asset = enabled_client.get(path)
assert asset.status_code == 200, path
assert expected_type in asset.headers['Content-Type']
assert asset.headers.get('Content-Length', '0') != '0'
def test_the_page_names_no_external_host(enabled_client):
"""Every URL on the page has to be same-origin. On an air-gapped box an
absolute http(s) URL is a request that hangs until it times out."""
page = enabled_client.get('/api/docs/try').get_data(as_text=True)
external = re.findall(r'https?://[^\s"\'<>)]+', page)
assert not external, 'try-it page references external hosts: %s' % external
def test_the_spec_validator_is_turned_off(enabled_client):
"""Swagger UI's one outbound default: unset, it posts the spec to a public
validator to draw a badge. That is a hung request here, and it would be an
upload of the internal API surface anywhere it worked."""
page = enabled_client.get('/api/docs/try').get_data(as_text=True)
assert re.search(r'validatorUrl:\s*null', page), (
'validatorUrl is not disabled; Swagger UI will call the public '
'validator')
def test_disabled_deployment_has_no_such_route(disabled_client):
"""404, not a rendered page explaining that it is disabled. The route is
never bound, so the URL confirms nothing."""
for path in ('/api/docs/try', '/api/docs/try/', _CSS_PATH, _JS_PATH):
assert disabled_client.get(path).status_code == 404, path
def test_the_redoc_page_advertises_try_it_only_when_it_exists(
enabled_client, disabled_client):
enabled_page = enabled_client.get('/api/docs/').get_data(as_text=True)
assert '/api/docs/try' in enabled_page
disabled_page = disabled_client.get('/api/docs/').get_data(as_text=True)
assert '/api/docs/try' not in disabled_page
# The word too, not just the href: a note about a view that 404s is worse
# than no note.
assert 'try-it' not in disabled_page
def test_every_url_follows_the_mount_path(enabled_client):
"""Prod runs under /shopdb. url_for has to carry the prefix onto the spec
and both assets, or the page loads nothing and can validate nothing."""
page = enabled_client.get(
'/api/docs/try',
environ_overrides={'SCRIPT_NAME': '/shopdb'}).get_data(as_text=True)
assert '/shopdb/api/docs/openapi.json' in page
assert '/shopdb%s' % _CSS_PATH in page
assert '/shopdb%s' % _JS_PATH in page
# No unprefixed twin left behind by a hardcoded path.
assert 'href="%s"' % _CSS_PATH not in page
assert 'src="%s"' % _JS_PATH not in page
redoc_page = enabled_client.get(
'/api/docs/',
environ_overrides={'SCRIPT_NAME': '/shopdb'}).get_data(as_text=True)
assert '/shopdb/api/docs/try' in redoc_page
@pytest.mark.parametrize('filename', _VENDORED)
def test_a_vendored_asset_fetches_nothing_from_the_internet(filename):
"""The assets themselves, not just the page.
Scoped to what a BROWSER would actually request. The swagger-ui bundle
carries around a hundred http(s) strings that are prose - RFC and
json-schema citations, licence URLs, the example.com host in the spec's own
documentation - and none of them is fetched, so a blanket "no http in this
file" assertion would be a lie that fails on the next upstream bump. What
would really produce a request is a CDN or webfont reference, a stylesheet
url() pointing off-box, or a source map: hence these.
"""
body = open(os.path.join(_STATICDOCS, filename),
encoding='utf-8', errors='replace').read()
# NOT a host allowlist. The first version of this named five CDNs, so an
# upstream bump to esm.sh or cdn.skypack.dev walked straight through it, and
# the one real leak in this directory was cdn.redoc.ly - which was on no
# list. Match the SHAPES a browser fetches instead, whatever the host:
# a stylesheet import, a css url(), a script/link/img source.
fetching = [
(r'@import\s+(?:url\()?[\'"]?(?:https?:)?//', 'an @import from another origin'),
(r'url\(\s*[\'"]?(?:https?:)?//', 'a css url() on another origin'),
(r'(?:src|href)\s*=\s*[\'"](?:https?:)?//', 'an element loaded from another origin'),
(r'@font-face', 'a webfont, which is a fetch unless it is a data: URI'),
]
for pattern, what in fetching:
for match in re.finditer(pattern, body):
window = body[max(0, match.start() - 60):match.start() + 120]
# data: URIs are inline, not fetched, and this CSS is full of them.
if 'data:' in window[:len(window) // 2 + 40]:
continue
raise AssertionError(
'%s carries %s: ...%s...' % (filename, what, window.strip()[:140]))
# A map reference is a request too. The upstream .map files are not
# vendored (they are tens of megabytes and only devtools reads them), so
# any surviving pointer is a guaranteed 404 on a box that cannot reach the
# internet to try elsewhere.
assert 'sourceMappingURL' not in body, (
'%s still points at a source map that is not vendored' % filename)
# Stylesheet resources: swagger-ui inlines its images as data: URIs, and
# anything else would be an off-box fetch on first paint. CSS only - in the
# minified bundle "url(" is a function call, not a resource.
if filename.endswith('.css'):
for match in re.findall(r'url\(\s*["\']?([^"\')]+)', body):
assert match.startswith('data:'), (
'%s loads %s rather than an inline data: URI'
% (filename, match))
# ---------------------------------------------------------------------------
# The spec has to say where this instance actually lives
# ---------------------------------------------------------------------------
def test_the_spec_declares_the_mount_it_is_served_from(enabled_client):
"""Production runs under /shopdb, and the generated file cannot know that.
gen_openapi.py writes servers: [{"url": "/"}] - correct for a root install
and wrong for every subpath one - and wsgi.py only moves MOUNT_PATH into
SCRIPT_NAME at runtime. Served as written, both renderers built
https://host/api/... for an app mounted at /shopdb: a URL an operator copied
out of the docs page and got a 404 from, and the address Swagger's try-it
fired at.
"""
import json as _json
at_root = _json.loads(
enabled_client.get('/api/docs/openapi.json').get_data(as_text=True))
assert at_root['servers'] == [{'url': '/', 'description': 'this instance'}]
mounted = _json.loads(enabled_client.get(
'/api/docs/openapi.json',
environ_overrides={'SCRIPT_NAME': '/shopdb'}).get_data(as_text=True))
assert mounted['servers'] == [
{'url': '/shopdb', 'description': 'this instance'}]
# The paths themselves stay mount-free: the server prefix is what varies,
# and duplicating it into every path would double it under a subpath.
assert '/api/apitokens' in mounted['paths']
assert '/shopdb/api/apitokens' not in mounted['paths']
# ---------------------------------------------------------------------------
# The read-only page follows the OS theme
# ---------------------------------------------------------------------------
def test_the_redoc_page_follows_the_system_theme(client):
"""The app has a dark theme; /api/docs was the one page that stayed white.
Redoc ignores prefers-color-scheme and the <redoc spec-url> element takes no
theme, so the page has to go through Redoc.init - which is also why this
test pins the mechanism and not just the CSS. A future edit that puts the
element back would lose dark mode without changing a single colour.
"""
page = client.get('/api/docs/').get_data(as_text=True)
assert 'Redoc.init' in page
assert '<redoc spec-url' not in page, 'the element takes no theme'
assert '@media (prefers-color-scheme: dark)' in page
assert 'var DARK=' in page and 'var LIGHT=' in page
# Re-render on a live scheme change, with the fallback for the older
# MediaQueryList that a kiosk browser may still have.
assert 'addEventListener("change",draw)' in page
assert 'q.addListener(draw)' in page
def test_the_docs_page_still_loads_nothing_from_another_origin(client):
"""Theming is the kind of change that quietly introduces a font or a CDN."""
import re as _re
page = client.get('/api/docs/').get_data(as_text=True)
assert not _re.findall(r'(?:src|href)\s*=\s*[\'"](?:https?:)?//', page)

393
tests/test_docs_index.py Normal file
View File

@@ -0,0 +1,393 @@
"""docs/INDEX.md is the router, and a router nobody checks is a router that lies.
The corpus is larger than any reader - person or agent - can hold, so the only
thing that makes it usable is being able to find the minimum correct set of
pages for a task and skip the rest. That needs two things to stay true, and
neither survives on good intentions:
1. Every page declares what it is for and whether it is still current, in a
`docmeta` block a machine can filter on without opening the page.
2. docs/INDEX.md routes every current page, and quarantines the ones that are
superseded or are only a record.
The failure this gate exists to stop is the quiet one: a page is added, nothing
routes to it, and it rots unread - or worse, a page goes `historical` and keeps
being followed because it is still sitting in a task row.
A second pass over the whole corpus added three more, because the first three
turned out to be satisfiable by an index that was still useless:
3. Being NAMED somewhere in INDEX.md is not the same as being routed. A
current page that appears only in a paragraph, or only in the Archive
section's prose, is a page no task row sends anyone to. It must be in a
row.
4. The line counts have to be right, not roughly right. The cost column is
the reason to trust the index over opening pages at random, and a corpus
that is being cut shrinks pages constantly - so a stale number is the
normal failure, not an exotic one.
5. The two routers - INDEX.md and START-HERE.md - must not point at a file
that is not there. Pages get merged and renamed; a dead link in the router
is worse than a missing row, because the reader trusts it.
Line-count tolerance: INDEX.md prints what a row costs to read, so a reader can
see the price before paying it. That number is checked against the real files
within 3 percent or 10 lines, whichever is larger. That absorbs a reflowed
paragraph and nothing else: the tolerance used to be 10 percent or 25 lines,
which let a 52-line row claim 77 and let whole sections drift after an edit
pass. If a row is out by more than a reflow, the number is simply wrong and
recomputing it is one command.
"""
import re
from pathlib import Path
import pytest
REPO = Path(__file__).resolve().parents[1]
DOCS = REPO / 'docs'
INDEX = DOCS / 'INDEX.md'
START_HERE = DOCS / 'START-HERE.md'
# The two routers. Everything else in docs/ is a destination; these two are how
# a reader arrives, so they are exempt from "must sit in a task row" (a row
# pointing at the index the row is in would be noise) and their links are held
# to a stricter standard than anyone else's.
ROUTERS = (INDEX, START_HERE)
# Same skip rule as tests/test_docs_publishable.py: docs/ prose is stripped from
# the published mirror and lives in the wiki, so a checkout with no docs/*.md
# has nothing to route and this module is inapplicable. docs/openapi.json is
# tracked, so keying off the DIRECTORY would make this run on the mirror.
pytestmark = pytest.mark.skipif(
not any(DOCS.glob('*.md')),
reason='no docs/ prose in this checkout - it is published to the wiki instead')
PURPOSES = {'tutorial', 'howto', 'reference', 'explanation'}
AUDIENCES = {'operator', 'adopter', 'developer', 'integrator'}
STATUSES = {'current', 'legacy', 'historical'}
FIELDS = ('purpose', 'audience', 'status', 'summary', 'loadwith')
ARCHIVED = {'legacy', 'historical'}
SUMMARY_LIMIT = 120
# The block must be the FIRST thing in the file, before the H1. A GitHub wiki
# renders YAML front matter as visible junk, which is why this is an HTML
# comment, and a comment further down the page is one a reader trips over.
DOCMETA = re.compile(r'\A<!-- docmeta\n(.*?)\n-->\n', re.S)
LINK = re.compile(r'\[[^\]]*\]\(([^)#]+)(?:#[^)]*)?\)')
ARCHIVE_HEADING = re.compile(r'^##\s+Archive\b', re.M)
ARCHIVE_TOTAL = re.compile(r'they are ([\d,]+) lines')
ALL_LINKS = re.compile(r'\[[^\]]*\]\(([^)]+)\)')
def pages():
"""Every authored page in docs/, including adr/ and proposals/."""
return sorted(p for p in DOCS.rglob('*.md'))
def rel(path):
return path.relative_to(DOCS).as_posix()
def line_count(path):
return len(path.read_text(errors='replace').splitlines())
def parse_docmeta(path):
"""Return the docmeta fields, or None when the block is missing."""
match = DOCMETA.match(path.read_text(errors='replace'))
if not match:
return None
fields = {}
for line in match.group(1).splitlines():
key, sep, value = line.partition(':')
if not sep:
return None
fields[key.strip()] = value.strip()
return fields
def value(fields, key):
"""A field with the template's trailing '# comment' hint removed."""
return re.sub(r'\s+#.*$', '', fields.get(key, '')).strip()
def status_of(path):
fields = parse_docmeta(path)
return value(fields, 'status') if fields else None
def links_in(text):
"""Every in-repo .md target a page links to, as written."""
return [t for t in LINK.findall(text) if t.endswith('.md') and '://' not in t]
def all_targets_in(text):
"""Every relative link target on a page, whatever its extension.
links_in() only sees .md, which is right for routing questions and wrong
for "does this file exist": the index links llms.txt, and a router that
points at a file nobody kept is the failure this catches.
"""
out = []
for target in ALL_LINKS.findall(text):
target = target.split()[0] if target.split() else target
if '://' in target or target.startswith(('mailto:', '#')):
continue
out.append(target.split('#')[0])
return [t for t in out if t]
def resolve(page, target):
"""A link target on `page`, as a path under docs/, or None if it escapes."""
candidate = (page.parent / target).resolve()
try:
candidate.relative_to(DOCS.resolve())
except ValueError:
return None
return candidate
def index_sections():
"""INDEX.md split into (task side, archive side) at the Archive heading."""
text = INDEX.read_text()
match = ARCHIVE_HEADING.search(text)
assert match, 'docs/INDEX.md must carry a "## Archive" section heading'
return text[:match.start()], text[match.start():]
def table_rows(text):
"""(link targets, stated line total) for every table row that states one."""
rows = []
for line in text.splitlines():
line = line.strip()
if not line.startswith('|') or set(line) <= set('|- '):
continue
cells = [c.strip() for c in line.strip('|').split('|')]
if len(cells) < 2 or not cells[-1].isdigit():
continue
targets = links_in(line)
if targets:
rows.append((targets, int(cells[-1])))
return rows
def tolerance(real):
"""How far a stated line count may sit from the real one. See the docstring."""
return max(10, real * 0.03)
def test_there_is_something_to_check():
"""A path change that matched nothing would make this whole file vacuous."""
assert len(pages()) > 20, 'docs/ prose disappeared, or the glob stopped matching'
assert INDEX.is_file(), 'docs/INDEX.md is missing - there is no router'
assert len(table_rows(index_sections()[0])) > 20, 'docs/INDEX.md routes almost nothing'
@pytest.mark.parametrize('page', pages(), ids=rel)
def test_every_page_carries_a_docmeta_block(page):
fields = parse_docmeta(page)
assert fields is not None, (
'%s has no docmeta block. It must be the first thing in the file, '
'before the H1: an HTML comment opening "<!-- docmeta" and closing '
'with "-->", one "key: value" per line.' % rel(page))
missing = [f for f in FIELDS if f not in fields]
assert not missing, '%s docmeta is missing: %s' % (rel(page), ', '.join(missing))
unknown = [f for f in fields if f not in FIELDS]
assert not unknown, '%s docmeta has unknown fields: %s' % (rel(page), ', '.join(unknown))
@pytest.mark.parametrize('page', pages(), ids=rel)
def test_docmeta_uses_the_allowed_vocabularies(page):
fields = parse_docmeta(page)
if fields is None:
pytest.skip('no docmeta - reported by test_every_page_carries_a_docmeta_block')
purpose = value(fields, 'purpose')
assert purpose in PURPOSES, (
'%s: purpose "%s" is not one of %s. A page that is two of these is a '
'page that should be split.' % (rel(page), purpose, sorted(PURPOSES)))
audience = [a.strip() for a in value(fields, 'audience').split(',') if a.strip()]
assert audience, '%s: audience is empty' % rel(page)
bad = [a for a in audience if a not in AUDIENCES]
assert not bad, '%s: audience %s not in %s' % (rel(page), bad, sorted(AUDIENCES))
status = value(fields, 'status')
assert status in STATUSES, (
'%s: status "%s" is not one of %s' % (rel(page), status, sorted(STATUSES)))
@pytest.mark.parametrize('page', pages(), ids=rel)
def test_summary_answers_is_this_the_page_i_need(page):
fields = parse_docmeta(page)
if fields is None:
pytest.skip('no docmeta - reported by test_every_page_carries_a_docmeta_block')
summary = fields.get('summary', '').strip()
assert summary, '%s: summary is empty. It is what a router shows INSTEAD of the page.' % rel(page)
assert len(summary) <= SUMMARY_LIMIT, (
'%s: summary is %d chars, limit is %d. Say what question the page '
'answers, not what it contains.' % (rel(page), len(summary), SUMMARY_LIMIT))
@pytest.mark.parametrize('page', pages(), ids=rel)
def test_loadwith_targets_exist(page):
"""loadwith names the pages this one is not complete without."""
fields = parse_docmeta(page)
if fields is None:
pytest.skip('no docmeta - reported by test_every_page_carries_a_docmeta_block')
raw = value(fields, 'loadwith')
if not raw or raw.lower() == 'none':
return
for name in [n.strip() for n in raw.split(',') if n.strip()]:
target = resolve(page, name) or (DOCS / name)
if not target.is_file():
target = DOCS / name
assert target.is_file(), (
'%s: loadwith names "%s", which does not exist' % (rel(page), name))
@pytest.mark.parametrize(
'page', [p for p in pages() if p != INDEX], ids=lambda p: rel(p))
def test_every_page_is_routed_or_archived(page):
"""A page nothing routes to is a page nobody reads and nobody maintains."""
text = INDEX.read_text()
targets = {resolve(INDEX, t) for t in links_in(text)}
assert page in targets, (
'%s is in docs/ but docs/INDEX.md does not name it. Add a task row for '
'it, or - if it is superseded or only a record - mark it legacy or '
'historical in its docmeta and list it in the Archive section.' % rel(page))
@pytest.mark.parametrize('router', ROUTERS, ids=lambda p: rel(p))
def test_a_router_never_points_at_a_file_that_is_not_there(router):
"""INDEX.md and START-HERE.md are the two pages everyone is sent to first.
A dead link here is worse than a missing row: a missing row makes a reader
look further, a dead link makes them think the answer is gone. Pages get
merged, split and renamed, and both routers name files by hand, so this is
the check that catches it the same day.
Every relative target counts, not just .md - the index links llms.txt too -
and a target outside docs/ is dead by definition once these pages are
published to the wiki, which carries docs/ and nothing above it.
"""
broken = []
for target in sorted(set(all_targets_in(router.read_text()))):
page = resolve(router, target)
if page is None:
broken.append('%s (outside docs/)' % target)
elif not page.is_file():
broken.append(target)
assert not broken, (
'docs/%s links to files that do not exist: %s'
% (rel(router), ', '.join(broken)))
def test_every_current_page_is_reachable_from_a_task_row():
"""Named in the index is not the same as routed by it.
Nothing stops a page being mentioned in a paragraph, or slipped into the
Archive section's prose, and passing "INDEX.md names it". A reader works
the rows: they find the row that matches the job and read what it lists. A
current page that is in no row is a page that gets written, gets stale and
never gets opened - and the cost of keeping it is paid by everyone who
greps the corpus.
The routers themselves are exempt: they are where a reader starts.
"""
tasks, _ = index_sections()
routed = set()
for targets, _lines in table_rows(tasks):
for target in targets:
page = resolve(INDEX, target)
if page:
routed.add(page)
stranded = sorted(
rel(p) for p in pages()
if status_of(p) == 'current' and p not in routed and p not in ROUTERS)
assert not stranded, (
'these pages are current guidance but no task row in docs/INDEX.md '
'sends anyone to them: %s. Give each one a row saying what task it '
'answers, or - if it is superseded or only a record - mark it legacy '
'or historical and move it to the Archive section.' % ', '.join(stranded))
def test_task_rows_never_route_to_an_archived_page():
"""Historical and legacy pages belong in the Archive section, nowhere else."""
tasks, _ = index_sections()
offenders = []
for target in sorted(set(links_in(tasks))):
page = resolve(INDEX, target)
if page and page.is_file() and status_of(page) in ARCHIVED:
offenders.append('%s (%s)' % (target, status_of(page)))
assert not offenders, (
'docs/INDEX.md routes a task to a page that is not current guidance: %s'
% ', '.join(offenders))
def test_the_archive_section_holds_nothing_current():
_, archive = index_sections()
offenders = []
for target in sorted(set(links_in(archive))):
page = resolve(INDEX, target)
if page and page.is_file() and status_of(page) == 'current':
offenders.append(target)
assert not offenders, (
'docs/INDEX.md files these current pages under Archive, where readers '
'are told not to follow them: %s' % ', '.join(offenders))
def test_no_current_page_hides_behind_an_archived_one():
"""Walk out from INDEX.md without passing through legacy or historical pages.
A current page that can only be reached through a superseded one is a page
a careful reader will never open, because they were told to skip the page
that links it.
"""
seen = set()
queue = [INDEX]
while queue:
page = queue.pop()
if page in seen:
continue
seen.add(page)
if status_of(page) in ARCHIVED:
continue # a dead end on purpose: nothing may depend on it
for target in links_in(page.read_text(errors='replace')):
nxt = resolve(page, target)
if nxt and nxt.is_file():
queue.append(nxt)
stranded = sorted(
rel(p) for p in pages() if status_of(p) == 'current' and p not in seen)
assert not stranded, (
'these current pages are only reachable through an archived page: %s'
% ', '.join(stranded))
def test_index_line_totals_are_honest():
"""The cost column is the point of the index. A wrong one is worse than none."""
tasks, archive = index_sections()
offenders = []
for targets, stated in table_rows(tasks) + table_rows(archive):
real = 0
for target in targets:
page = resolve(INDEX, target)
if page and page.is_file():
real += line_count(page)
if abs(real - stated) > tolerance(real):
offenders.append('%s: says %d lines, is %d' % ('+'.join(targets), stated, real))
assert not offenders, (
'docs/INDEX.md misstates what a row costs to read:\n %s' % '\n '.join(offenders))
def test_the_archive_total_is_honest():
_, archive = index_sections()
match = ARCHIVE_TOTAL.search(archive)
assert match, 'the Archive section should say how many lines it adds up to'
stated = int(match.group(1).replace(',', ''))
real = sum(count for _, count in table_rows(archive))
assert abs(real - stated) <= tolerance(real), (
'the Archive section claims %d lines; its rows add up to %d' % (stated, real))

View File

@@ -34,14 +34,6 @@ DOCS = REPO / 'docs'
# published mirror - keying off the directory made this module run there against
# a prose-less tree and fail. A published checkout is one with no authored
# prose, matching the canonical definition in scripts/check-gates-ran.sh.
# KNOWN GAP, left deliberately for a decision rather than closed here: this
# keys on docs/*.md, so prose that lives only in docs/wiki-drafts/ does NOT
# turn the module on, and on gh-wj that is every page there is. Pointed at the
# drafts on 2026-09-18 the scrub fails on 27 lines - site name, internal git
# server, developer home paths - which is a question about what those drafts
# are for (internal wiki, or publication) and not a bug to fix by rewriting
# them. scripts/check-gates-ran.sh matches this definition exactly; change both
# together or the gate runner reports a skip it was told to expect.
pytestmark = pytest.mark.skipif(
not any(DOCS.glob('*.md')),
reason='no docs/ prose in this checkout - it is excluded from publication and lives in the wiki')

View File

@@ -0,0 +1,637 @@
"""The two derived halves of the published spec: query parameters and examples.
docs/openapi.json is served at /api/docs on production AND published to the
wiki, and it is what the MCP server builds its tools from. Two things were
missing from every operation in it, and both are now derived rather than
written:
QUERY PARAMETERS. A Flask rule carries its path arguments and nothing else,
so 0 of 573 operations published a query parameter - including
/api/usb/kiosk/setup, where '?code=' is the only way to call it successfully.
scripts/gen_openapi.py now reads them out of each view's AST.
EXAMPLES. 0 of 573 carried a machine-readable example. They are harvested
from the test suite, which already sends real bodies and asserts real
responses.
Harvesting is where this module earns its place. A harvested body is whatever a
test happened to build, and tests are written with real-looking material -
passwords, tokens, badge numbers, hostnames, addresses. None of that matters in
a test file and all of it matters on a public page, so redaction runs before
anything reaches the spec. `redact` is tested here directly, and the generated
spec is swept for the same material, because a scrubber nobody checks is a
scrubber that stops working quietly.
The other property tested here is that NOTHING IS INVENTED. A query argument
whose name the generator cannot resolve to a literal is not published at all:
an argument that is not really there is worse than a missing one, because a
reader believes it, sends it and is silently ignored.
"""
import importlib.util
import json
import textwrap
from pathlib import Path
import pytest
import re
REPO = Path(__file__).resolve().parents[1]
SPEC = REPO / 'docs' / 'openapi.json'
def _generator():
"""scripts/gen_openapi.py, loaded by path.
It is a script, not a package module. Loading it runs its imports and
nothing else - it builds no app and reads no database until it is asked to.
"""
path = REPO / 'scripts' / 'gen_openapi.py'
spec = importlib.util.spec_from_file_location('gen_openapi_examples', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.fixture(scope='module')
def generator():
return _generator()
@pytest.fixture(scope='module')
def spec():
if not SPEC.is_file():
pytest.skip('no generated spec in this checkout')
return json.loads(SPEC.read_text())
def operations(spec):
for path, item in spec['paths'].items():
for verb, operation in item.items():
yield path, verb, operation
def examples(spec):
"""Every published example, as (where, value)."""
for path, verb, operation in operations(spec):
body = operation.get('requestBody') or {}
media = (body.get('content') or {}).get('application/json') or {}
if 'example' in media:
yield '%s %s request' % (verb.upper(), path), media['example']
success = ((operation.get('responses') or {}).get('200') or {})
media = (success.get('content') or {}).get('application/json') or {}
if 'example' in media:
yield '%s %s response' % (verb.upper(), path), media['example']
# ---------------------------------------------------------------------------
# redact
# ---------------------------------------------------------------------------
def test_a_credential_is_redacted_by_its_key(generator):
"""The key is the reliable signal, whatever the value happens to be today.
A fixture that leaves passwordhash empty must not teach the scrubber that
passwordhash is safe.
"""
scrubbed = generator.redact({
'password': 'hunter2',
'passwordhash': '',
'apikey': 'abc',
'refresh_token': 'abc',
'X-API-Key': 'abc',
'signature': 'abc',
})
assert set(scrubbed.values()) == {'<redacted:secret>'}
def test_a_person_and_a_machine_are_redacted_by_their_key(generator):
"""Not secrets, and not ours to publish either. ADR-015 for the machine
half: one plant's server name is meaningless to every other site reading
these pages."""
scrubbed = generator.redact({
'username': 'jsmith',
'badge': '204411',
'checkedoutby': 'J Smith',
'hostname': 'WJSF0123',
'fqdn': 'wjsf0123.example.net',
'ipaddress': '10.4.2.19',
'servicetag': 'H3K9QP2',
})
assert scrubbed['username'] == '<redacted:person>'
assert scrubbed['badge'] == '<redacted:person>'
assert scrubbed['checkedoutby'] == '<redacted:person>'
assert scrubbed['hostname'] == '<redacted:host>'
assert scrubbed['fqdn'] == '<redacted:host>'
assert scrubbed['ipaddress'] == '<redacted:host>'
assert scrubbed['servicetag'] == '<redacted:host>'
def test_a_credential_with_no_telling_key_is_redacted_by_its_value(generator):
"""The half the key rules cannot reach: material in a free-text field, or
in a list, where nothing names it."""
scrubbed = generator.redact({
'note': 'mail bob.smith@example.com when it lands',
'detail': 'bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.aaaa.bbbb',
'path': '\\\\fileserver\\deploy\\agent.msi',
'where': 'reported from 10.11.12.13',
'digest': 'd41d8cd98f00b204e9800998ecf8427e',
'bag': ['carol@example.com'],
})
assert scrubbed['note'] == '<redacted:email>'
assert scrubbed['detail'] == '<redacted:token>'
assert scrubbed['path'] == '<redacted:sharepath>'
assert scrubbed['where'] == '<redacted:address>'
assert scrubbed['digest'] == '<redacted:digest>'
assert scrubbed['bag'] == ['<redacted:email>']
def test_ordinary_material_survives(generator):
"""Over-redaction has a cost too: an example scrubbed to nothing shows a
reader nothing. What is safe stays."""
payload = {
'assetnumber': 'A-10024',
'model': 'Dell Latitude 5420',
'isactive': True,
'quantity': 12,
'status': 'In Service',
'url': 'http://localhost:5001/api/assets',
}
assert generator.redact(payload) == payload
def test_a_numeric_id_is_not_a_credential(generator):
"""'tokenid': 3 is the row number, not the token.
Redacting it cost every token route the one field a caller needs to address
the row. The exemption is numeric-only, so a string session id is still
scrubbed.
"""
scrubbed = generator.redact({'tokenid': 3, 'apikeyid': 7,
'sessionid': 'a-long-opaque-string-value'})
assert scrubbed['tokenid'] == 3
assert scrubbed['apikeyid'] == 7
assert scrubbed['sessionid'] == '<redacted:secret>'
def test_redaction_reaches_every_level(generator):
"""A token two objects down is as published as one at the top."""
scrubbed = generator.redact(
{'data': {'items': [{'inner': {'password': 'hunter2'}}]}})
assert scrubbed['data']['items'][0]['inner']['password'] == '<redacted:secret>'
def test_redaction_is_idempotent(generator):
"""A harvest runs again over a spec that already carries markers, and the
markers must not be re-scrubbed into something else."""
once = generator.redact({'password': 'hunter2', 'model': 'Dell 5420'})
assert generator.redact(once) == once
def test_a_long_list_is_trimmed_and_says_so(generator):
"""Three rows show the shape, which is the whole job of an example - but a
reader must not conclude the endpoint returns three."""
scrubbed = generator.redact({'data': list(range(50))})
assert scrubbed['data'][:3] == [0, 1, 2]
assert 'more' in scrubbed['data'][-1]
def test_a_cyclic_depth_is_bounded(generator):
"""Harvested bodies come from a running app, not from this file. A deeply
nested one must not take the generator down with it."""
deep = current = {}
for _ in range(50):
current['child'] = {}
current = current['child']
generator.redact(deep)
# ---------------------------------------------------------------------------
# The published spec, swept for the same material
# ---------------------------------------------------------------------------
# Patterns held HERE, deliberately not imported from the generator. A gate that
# checks the scrubber with the scrubber's own patterns can only ever assert that
# redact() ran: it was green while the spec published a live bootstrap token and
# a live kiosk setup code, because both hid in a URL whose key ('scripturl',
# 'url') names no secret and whose value the opaque rule excludes by design.
# These must keep catching that class even if someone edits the generator.
INDEPENDENT_URL_CREDENTIAL = re.compile(
r'(?i)[?&](t|token|code|key|apikey|api_key|secret|sig|signature|password|'
r'auth|access_token|refresh_token)=([^&\s"\'<]{8,})')
# Product token prefixes. If a new one is minted, add it here the same day.
INDEPENDENT_TOKEN_PREFIX = re.compile(r'\b(sdbb_|sdbe_|shopdb_pat_)[A-Za-z0-9_-]{8,}')
def _looks_high_entropy(text):
"""A long mixed-case, digit-bearing run with no word structure.
Deliberately crude: it is a backstop for a secret shape nobody has thought
of yet, not a classifier. Anything it flags is either a secret or an example
that would read better with a shorter value.
"""
for chunk in re.split(r'[\s"\'<>]+', text):
if len(chunk) < 28 or '.' in chunk or '/' in chunk:
continue
has = lambda pattern: re.search(pattern, chunk)
if has(r'[a-z]') and has(r'[A-Z]') and has(r'\d'):
return chunk
return None
def test_no_published_example_hides_a_credential_in_a_url(spec):
"""The class the scrubber's own patterns cannot see.
Two real credentials reached docs/openapi.json this way on 2026-09-18, in a
file served at /api/docs on production and copied to a published wiki.
"""
offenders = []
for where, example in examples(spec):
blob = json.dumps(example)
found = INDEPENDENT_URL_CREDENTIAL.search(blob)
if found and not found.group(2).startswith('<redacted:'):
offenders.append('%s: %s=%s' % (where, found.group(1), found.group(2)[:16]))
prefixed = INDEPENDENT_TOKEN_PREFIX.search(blob)
if prefixed:
offenders.append('%s: token prefix %s' % (where, prefixed.group(0)[:20]))
entropy = _looks_high_entropy(blob)
if entropy:
offenders.append('%s: high-entropy %r' % (where, entropy[:20]))
assert not offenders, (
'Published examples carry what looks like a credential:\n '
+ '\n '.join(sorted(set(offenders))))
def test_no_published_example_carries_a_secret_a_person_or_a_machine(spec, generator):
"""The gate on the whole harvest, from the side a reader sees.
Checked against the scrubber's own patterns rather than a second list, so
the gate and the scrubber cannot drift apart: this fails if redaction stops
being applied, if it is applied to a stale copy, or if a hand-edit puts
real material back.
"""
offenders = []
for where, example in examples(spec):
for key, value in _walk(example):
if not isinstance(value, str) or value.startswith('<redacted:'):
continue
name = (key or '').lower()
# Only strings reach here, and the numeric-id exemption in redact()
# is numeric-only, so no key is exempt on this side.
if name and generator.SECRET_KEY_PATTERN.search(name):
offenders.append('%s: key %r' % (where, key))
elif name and generator.PERSON_KEY_PATTERN.search(name):
offenders.append('%s: key %r' % (where, key))
elif name and generator.HOST_KEY_PATTERN.search(name):
offenders.append('%s: key %r' % (where, key))
elif generator._redacted_string(value):
offenders.append('%s: value under %r' % (where, key))
assert not offenders, (
'%d published example value(s) look like a secret, a person or a '
'machine. docs/openapi.json is served on production and published to '
'the wiki; regenerate it with scripts/gen_openapi.py:\n %s'
% (len(offenders), '\n '.join(sorted(set(offenders))[:15])))
def _walk(value, key=''):
"""(key, value) for every leaf in a harvested example."""
if isinstance(value, dict):
for itemkey, itemvalue in value.items():
for found in _walk(itemvalue, str(itemkey)):
yield found
elif isinstance(value, list):
for item in value:
for found in _walk(item, key):
yield found
else:
yield key, value
def test_a_published_example_is_bounded(spec, generator):
"""Everybody downloads this document and every generated client parses it.
One 400-row list response costs all of them."""
offenders = [where for where, example in examples(spec)
if len(json.dumps(example)) > generator.EXAMPLE_MAX_CHARS]
assert not offenders, '\n '.join(offenders[:15])
def test_a_response_example_never_contradicts_the_schema_beside_it(spec):
"""An example and a `$ref` that disagree are worse than no example at all.
Nineteen routes here do not return the standard envelope and every one was
documented as if it did. A harvested transcript is evidence about which,
so the generator drops the envelope `$ref` on any operation whose real
response is a different shape. This is the gate on that: an operation
published with the envelope `$ref` and an example that is not an envelope
is the exact contradiction the harvest exists to remove.
"""
offenders = []
for path, verb, operation in operations(spec):
media = (((operation.get('responses') or {}).get('200') or {})
.get('content') or {}).get('application/json') or {}
if 'example' not in media:
continue
claims = 'SuccessEnvelope' in json.dumps(media.get('schema') or {})
isenvelope = (isinstance(media['example'], dict)
and media['example'].get('status') == 'success')
if claims != isenvelope:
offenders.append('%s %s: schema says envelope=%s, example is=%s'
% (verb.upper(), path, claims, isenvelope))
assert not offenders, '\n '.join(offenders[:15])
# ---------------------------------------------------------------------------
# Derived query parameters: what is emitted, and what is deliberately not
# ---------------------------------------------------------------------------
def _derive(tmp_path, source):
"""Run the generator's query sweep over one written-out module."""
module = tmp_path / 'sample_views.py'
module.write_text(textwrap.dedent(source))
return _generator()._query_index(str(module))
def test_a_literal_query_read_is_emitted(tmp_path):
index = _derive(tmp_path, '''
def view():
search = request.args.get('search')
active = request.args.get('active', 'true')
days = request.args.get('days', 30, type=int)
tags = request.args.getlist('tag')
page = request.args['page']
''')
found = {item['name']: item for item in index['view']}
assert sorted(found) == ['active', 'days', 'page', 'search', 'tag']
assert found['search'] == {'name': 'search', 'type': 'string', 'required': False}
assert found['active']['default'] == 'true'
assert found['days'] == {'name': 'days', 'type': 'integer',
'required': False, 'default': 30}
assert found['tag']['type'] == 'array'
# request.args['x'] raises without the argument. It is the one read in the
# language that proves the caller must send it.
assert found['page']['required'] is True
def test_a_name_that_is_not_a_literal_is_not_published(tmp_path):
"""The honesty rule, and the reason this is safe to derive at all.
A key built at runtime cannot be resolved, and a parameter that is not
really there is worse than a missing one: a reader believes it, sends it,
and is silently ignored. A module-level constant DOES resolve, because it
is a literal with a name.
"""
index = _derive(tmp_path, '''
TOKEN_ARG = 'bootstraptoken'
def view(field):
built = request.args.get('filter_' + field)
indirect = request.args.get(field)
formatted = request.args.get(f'{field}id')
named = request.args.get(TOKEN_ARG)
''')
assert [item['name'] for item in index['view']] == ['bootstraptoken']
def test_the_repos_own_argument_helpers_are_read(tmp_path):
"""get_pagination_params and intarg/intargs read arguments on a handler's
behalf, so a sweep that only looks for request.args misses the paging
contract of every list route in the product."""
index = _derive(tmp_path, '''
def view():
page, perpage = get_pagination_params(request)
typeid = intarg('typeid')
since = intarg('days', 30)
rest = intargs('locationid', 'businessunitid')
''')
found = {item['name']: item for item in index['view']}
assert sorted(found) == ['businessunitid', 'days', 'locationid', 'page',
'per_page', 'perpage', 'typeid']
assert found['page'] == {'name': 'page', 'type': 'integer',
'required': False, 'default': 1}
assert found['days']['default'] == 30
assert found['typeid']['type'] == 'integer'
def test_a_view_that_reads_nothing_publishes_nothing(tmp_path):
index = _derive(tmp_path, '''
def view(asset_id):
return success_response({'assetid': asset_id})
''')
assert index['view'] == []
# ---------------------------------------------------------------------------
# The generator without any of this
# ---------------------------------------------------------------------------
def test_a_spec_generates_with_no_harvest_and_no_query_parameters(generator):
"""Both halves are OPTIONAL. A published checkout has no tests to harvest
from, and an inventory written before this existed has no `queryparams`. A
generator that needs either is a generator that cannot be run where it
matters."""
built = generator.build([{
'surface': 'core-assets',
'endpoints': [{
'method': 'GET|POST',
'path': '/api/assets',
'auth': 'permission:assets.view',
'params': 'not documented',
'purpose': 'List assets.',
'example': 'curl http://localhost:5001/api/assets',
}],
}])
operation = built['paths']['/api/assets']['get']
assert 'parameters' not in operation
media = built['paths']['/api/assets']['post']['requestBody']['content']
assert 'example' not in media['application/json']
success = operation['responses']['200']['content']['application/json']
assert 'example' not in success
assert success['schema'] == {'$ref': '#/components/schemas/SuccessEnvelope'}
def test_a_query_parameter_survives_into_the_spec(generator):
"""The other direction: an inventory entry that has them publishes them."""
built = generator.build([{
'surface': 'plugin-usb',
'endpoints': [{
'method': 'GET',
'path': '/api/usb/kiosk/setup',
'auth': 'setup-code (single-use kiosk setup code in ?code=)',
'params': 'not documented',
'purpose': 'Trade a setup code for this browser kiosk cookie.',
'example': 'curl http://localhost:5001/api/usb/kiosk/setup',
'queryparams': [{'name': 'code', 'type': 'string', 'required': False}],
'responseexample': {'status': 'success', 'data': None},
}],
}])
operation = built['paths']['/api/usb/kiosk/setup']['get']
assert operation['parameters'] == [{
'name': 'code', 'in': 'query', 'required': False,
'schema': {'type': 'string'},
}]
success = operation['responses']['200']['content']['application/json']
assert success['example'] == {'status': 'success', 'data': None}
# A query filter that matches nothing is an empty list and a 200. Only a
# path parameter can be a record that is not there.
assert '404' not in operation['responses']
def test_a_per_request_value_is_pinned_but_keeps_its_shape(generator):
"""A transcript carries the instant it was taken and the id of the request
that produced it, and both differ on every run.
Published verbatim, re-harvesting rewrote 313 of 320 examples with no change
of meaning - an unreviewable diff over a committed, published file. An
example shows a reader the FORMAT of a field, not the moment one call
happened, so the format is kept exactly and the value is pinned.
"""
scrubbed = generator.redact({
'meta': {'timestamp': '2026-09-18T18:30:48.498252Z',
'requestid': '859aa86b'},
'createddate': '2026-09-18 08:15:00',
'expiresat': '2026-12-31T23:59:59+00:00',
'installedon': '2026-09-18',
})
# Microseconds, the space separator and the offset all survive: those are
# three different contracts and a reader is reading the example for which.
assert scrubbed['meta']['timestamp'] == '2026-01-01T00:00:00.000000Z'
assert scrubbed['meta']['requestid'] == '00000000'
assert scrubbed['createddate'] == '2026-01-01 00:00:00'
assert scrubbed['expiresat'] == '2026-01-01T00:00:00+00:00'
assert scrubbed['installedon'] == '2026-01-01'
def test_pinning_does_not_reach_ordinary_text(generator):
"""Only a value that IS a timestamp. A model number that happens to contain
digits is not one."""
payload = {'model': '5420-2026', 'note': 'due 2026-09-18 at the latest',
'version': '1.2.3'}
assert generator.redact(payload) == payload
def test_a_local_path_is_not_published(generator):
"""A path off the machine the harvest ran on says nothing to a reader and
is different on every run. A developer's home directory is also the exact
material tests/test_docs_publishable.py exists to keep off a public page."""
scrubbed = generator.redact({
'shareroot': '/tmp/pytest-of-someone/pytest-240/test_roundtrip0',
'checkout': '/home/someone/projects/shopdb-flask',
'relative': 'deploy/windows/installer',
'route': '/api/assets',
})
assert scrubbed['shareroot'] == '<redacted:path>'
assert scrubbed['checkout'] == '<redacted:path>'
# A path that names no machine is the useful half, and it stays.
assert scrubbed['relative'] == 'deploy/windows/installer'
assert scrubbed['route'] == '/api/assets'
def test_a_harvest_adds_and_never_clears(generator, tmp_path):
"""A targeted harvest has to be targeted.
Clearing on absence meant `--harvest tests/test_core` stripped the examples
off every route those tests do not touch, which is most of them - the
narrow command would have done the most damage. It also made the committed
file churn for nothing: run under xdist against a shared instance
directory, two full runs of the same tree reached 311 and 327 operations.
"""
inventory = tmp_path / 'api-inventory.json'
inventory.write_text(json.dumps([{
'surface': 'core-assets',
'endpoints': [
{'method': 'GET', 'path': '/api/assets',
'responseexample': {'status': 'success', 'data': ['kept']}},
{'method': 'POST', 'path': '/api/assets'},
],
}]))
original = generator.INVENTORY
generator.INVENTORY = str(inventory)
try:
generator.fold_examples({
('POST', '/api/assets'): {
'request': [{'assetnumber': 'A-1'}],
'response': [{'status': 'success', 'data': {'assetid': 1}}],
},
})
finally:
generator.INVENTORY = original
entries = json.loads(inventory.read_text())[0]['endpoints']
# Untouched by this harvest, and still carrying what it had.
assert entries[0]['responseexample'] == {'status': 'success', 'data': ['kept']}
assert entries[1]['requestexample'] == {'assetnumber': 'A-1'}
assert entries[1]['responseexample'] == {'status': 'success',
'data': {'assetid': 1}}
def test_picking_one_example_out_of_many_is_deterministic(generator):
"""The suite runs in parallel, so the order samples arrive in is not stable.
Largest-that-fits with a text tiebreak means two runs over the same samples
produce the same spec, which is what makes a regenerated docs/openapi.json
reviewable as a diff instead of a rewrite.
"""
samples = [
{'status': 'success', 'data': {'assetid': 1}},
{'status': 'success', 'data': {'assetid': 2, 'assetnumber': 'A-2'}},
{'status': 'success', 'data': None},
]
picked = generator._best_example(samples)
assert picked == generator._best_example(list(reversed(samples)))
assert picked['data']['assetnumber'] == 'A-2'
def test_an_example_that_cannot_be_made_to_fit_is_dropped(generator):
"""Silence beats a truncated JSON document: half an object is not an
example of anything, and it is not what the route returned."""
# A wide object, not a long string: strings and lists are already capped
# inside redact, so an object with hundreds of keys is the shape that can
# still come out over the limit.
huge = {'status': 'success',
'data': {'field%03d' % index: 'value' for index in range(500)}}
assert generator._best_example([huge]) is None
# And a smaller sample of the same operation still wins, rather than the
# whole operation losing its example to one oversized transcript.
small = {'status': 'success', 'data': {'assetid': 1}}
assert generator._best_example([huge, small]) == small
def test_a_delete_that_reads_a_body_says_so(generator):
"""DELETE /api/applications/{id} refuses without {"confirm": "OK"}.
The generator declared a body for POST, PUT and PATCH only, so that route
published a call that cannot succeed and a client built from the spec fails
on the day.
Evidence is the HANDLER reading a body (`readsbody`, derived from its AST),
not a harvested transcript. A transcript proves the test CLIENT sent
something, and the suite sends {} to routes that ignore it: on 2026-09-18
transcript-only evidence published "this route reads a JSON body" on DELETE
/api/endpoints/agentkey, whose handler never calls get_json. Both halves are
pinned below.
"""
entries = [
{'method': 'DELETE', 'path': '/api/applications/<int:app_id>',
'auth': 'permission:applications.delete', 'purpose': 'Delete it.',
'readsbody': True, 'requestexample': {'confirm': 'OK'}},
{'method': 'DELETE', 'path': '/api/assets/<int:assetid>',
'auth': 'permission:assets.delete', 'purpose': 'Delete it.'},
# A transcript with no handler evidence: the suite sent a body, the
# route ignores it, so the contract must stay silent.
{'method': 'DELETE', 'path': '/api/endpoints/agentkey',
'auth': 'permission:endpoints.manage', 'purpose': 'Revoke it.',
'requestexample': {}},
]
built = generator.build([{'surface': 'core-applications', 'endpoints': entries}])
withbody = built['paths']['/api/applications/{app_id}']['delete']
media = withbody['requestBody']['content']['application/json']
assert withbody['requestBody']['required'] is False
assert media['example'] == {'confirm': 'OK'}
# Transcript without handler evidence declares nothing.
ignored = built['paths']['/api/endpoints/agentkey']['delete']
assert 'requestBody' not in ignored
assert 'requestBody' not in built['paths']['/api/assets/{assetid}']['delete']

View File

@@ -189,3 +189,58 @@ def test_summaries_are_not_cut_mid_word(spec):
if body and not body[-1].isalnum() and body[-1] not in ')]"\'':
offenders.append('%s %s: %r' % (verb.upper(), path, summary[-40:]))
assert not offenders, '\n '.join(offenders[:15])
def test_every_parameter_is_well_formed(spec):
"""A parameter with no name, no location or no schema is not a parameter.
Query parameters are derived from each handler's AST now rather than left
out, which is 87 operations that could not be filtered, searched or paged
from this document before. Derived means a code change can produce a
malformed one, so the shape is checked here rather than assumed.
"""
offenders = []
for path, verb, operation in operations(spec):
for parameter in operation.get('parameters', []):
where = '%s %s: %r' % (verb.upper(), path, parameter.get('name'))
if not parameter.get('name'):
offenders.append('%s %s: a parameter with no name' % (verb.upper(), path))
if parameter.get('in') not in ('path', 'query', 'header', 'cookie'):
offenders.append(where + ' has no valid `in`')
if not isinstance(parameter.get('required'), bool):
offenders.append(where + ' does not say whether it is required')
if not parameter.get('schema'):
offenders.append(where + ' has no schema')
if parameter.get('in') == 'path' and not parameter.get('required'):
offenders.append(where + ' is in the path and optional, which '
'OpenAPI does not allow')
assert not offenders, '\n '.join(offenders[:15])
def test_no_parameter_is_declared_twice(spec):
"""(name, in) is the identity of a parameter. Two entries for one name is
ambiguous to a generated client, which picks one and drops the other."""
offenders = []
for path, verb, operation in operations(spec):
seen = set()
for parameter in operation.get('parameters', []):
key = (parameter.get('name'), parameter.get('in'))
if key in seen:
offenders.append('%s %s: %s in %s'
% (verb.upper(), path, key[0], key[1]))
seen.add(key)
assert not offenders, '\n '.join(offenders[:15])
def test_a_query_parameter_is_not_the_path_argument_under_another_name(spec):
"""`/api/assets/{assetid}?assetid=` is a contradiction: the same value in
two places, and a generated client has to guess which one the route reads."""
offenders = []
for path, verb, operation in operations(spec):
intemplate = set(re.findall(r'\{([^}]+)\}', path))
query = {p['name'] for p in operation.get('parameters', [])
if p.get('in') == 'query'}
clash = sorted(intemplate & query)
if clash:
offenders.append('%s %s: %s' % (verb.upper(), path, clash))
assert not offenders, '\n '.join(offenders[:15])