notandClaude Opus 4.8
2bb18f4a51
Fails against a stack with Objecttypen up but no objecttype seeded yet: "no objecttype named 'RegisterRecord'". Green comes with the seeded registerrecord-init one-shot in the next commit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""S-18c (#141): prove the RegisterRecord objecttype is registered + published in the Objecttypen API.
|
|
|
|
Assert the objecttype named "RegisterRecord" exists, has a **published** version, and that version's
|
|
jsonSchema carries the public-safe fields (id, status, reference) — i.e. registerrecord-init ran and
|
|
seeded the schema S-19 will write records against. Stdlib only so it runs in a bare python:3-slim
|
|
container on the compose network.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = os.environ["OBJECTTYPEN"] # http://<ip>:8000
|
|
TOKEN = os.environ["OBJECTTYPEN_TOKEN"]
|
|
TIMEOUT = int(os.environ.get("REGISTERRECORD_TIMEOUT", "60"))
|
|
NAME = "RegisterRecord"
|
|
EXPECTED_FIELDS = {"id", "status", "reference"}
|
|
|
|
|
|
def get(path):
|
|
req = urllib.request.Request(f"{BASE}{path}", headers={"Authorization": f"Token {TOKEN}"})
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def check():
|
|
"""Return (ok, detail). Raises on transport errors so the caller can retry."""
|
|
ots = get("/api/v2/objecttypes").get("results", [])
|
|
match = next((o for o in ots if o.get("name") == NAME), None)
|
|
if not match:
|
|
return False, f"no objecttype named {NAME!r} (have: {[o.get('name') for o in ots]})"
|
|
if not match.get("versions"):
|
|
return False, f"{NAME} exists but has no versions"
|
|
# The versions list holds URLs; fetch each to find a published one.
|
|
for ver_url in match["versions"]:
|
|
ver = get(ver_url[len(BASE):] if ver_url.startswith(BASE) else ver_url)
|
|
if ver.get("status") != "published":
|
|
continue
|
|
props = set((ver.get("jsonSchema") or {}).get("properties", {}))
|
|
if not EXPECTED_FIELDS <= props:
|
|
return False, f"published version missing fields: {EXPECTED_FIELDS - props}"
|
|
return True, f"{NAME} v{ver.get('version')} published, fields={sorted(props)}"
|
|
return False, f"{NAME} has versions but none are published"
|
|
|
|
|
|
def main():
|
|
deadline = time.time() + TIMEOUT
|
|
detail = "no attempt"
|
|
while time.time() < deadline:
|
|
try:
|
|
ok, detail = check()
|
|
if ok:
|
|
print(f"OK — {detail}")
|
|
return 0
|
|
except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
|
|
detail = f"transport: {e}"
|
|
time.sleep(3)
|
|
print(f"FAIL — {detail}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|