"""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 a setup route 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()) == {''} 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'] == '' assert scrubbed['badge'] == '' assert scrubbed['checkedoutby'] == '' assert scrubbed['hostname'] == '' assert scrubbed['fqdn'] == '' assert scrubbed['ipaddress'] == '' assert scrubbed['servicetag'] == '' 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'] == '' assert scrubbed['detail'] == '' assert scrubbed['path'] == '' assert scrubbed['where'] == '' assert scrubbed['digest'] == '' assert scrubbed['bag'] == [''] 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'] == '' 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'] == '' 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(' 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': 'core-setup', 'endpoints': [{ 'method': 'GET', 'path': '/api/setup/claim', 'auth': 'setup-code (single-use setup code in ?code=)', 'params': 'not documented', 'purpose': 'Trade a setup code for this browser session.', 'example': 'curl http://localhost:5001/api/setup/claim', 'queryparams': [{'name': 'code', 'type': 'string', 'required': False}], 'responseexample': {'status': 'success', 'data': None}, }], }]) operation = built['paths']['/api/setup/claim']['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': 'scripts/site_imports', 'route': '/api/assets', }) assert scrubbed['shareroot'] == '' assert scrubbed['checkout'] == '' # A path that names no machine is the useful half, and it stays. assert scrubbed['relative'] == 'scripts/site_imports' 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/', 'auth': 'permission:applications.delete', 'purpose': 'Delete it.', 'readsbody': True, 'requestexample': {'confirm': 'OK'}}, {'method': 'DELETE', 'path': '/api/assets/', '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']