Skip to content

Commit 77b3423

Browse files
test: add Prism-based OpenAPI mock verification
Signed-off-by: laurentketterle-hub <laurentketterle-hub@users.noreply.github.com>
1 parent 9f20a99 commit 77b3423

1 file changed

Lines changed: 61 additions & 0 deletions

File tree

scripts/prism_replay.py

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
#!/usr/bin/env python3
2+
# Replay a recorded request corpus against a Prism mock server.
3+
#
4+
# Usage:
5+
# python3 scripts/prism_replay.py [corpus.json] [base_url]
6+
#
7+
# Exits non-zero if any recorded request cannot be resolved by Prism.
8+
# Writes a JSON coverage report to prism-coverage.json.
9+
import json
10+
import sys
11+
import urllib.request
12+
import urllib.error
13+
14+
15+
def main():
16+
corpus_path = sys.argv[1] if len(sys.argv) > 1 else "tests/pact/prism-replay-corpus.json"
17+
base_url = (sys.argv[2] if len(sys.argv) > 2 else "http://localhost:4010").rstrip("/")
18+
19+
with open(corpus_path, "r", encoding="utf-8") as f:
20+
corpus = json.load(f)
21+
22+
report = {"total": len(corpus), "resolved": 0, "unresolved": [], "results": []}
23+
for entry in corpus:
24+
method = entry.get("method", "GET").upper()
25+
path = entry.get("path", "/")
26+
url = base_url + path
27+
req = urllib.request.Request(url, method=method)
28+
try:
29+
with urllib.request.urlopen(req, timeout=15) as resp:
30+
status = resp.status
31+
body = resp.read().decode("utf-8", "replace")
32+
except urllib.error.HTTPError as e:
33+
status = e.code
34+
body = e.read().decode("utf-8", "replace")
35+
except Exception as e:
36+
status = -1
37+
body = str(e)
38+
39+
prism_error = "stoplight.io/prism/errors" in body
40+
resolved = (status < 400) and not prism_error
41+
42+
item = {"method": method, "path": path, "status": status, "resolved": resolved}
43+
report["results"].append(item)
44+
if resolved:
45+
report["resolved"] += 1
46+
else:
47+
report["unresolved"].append({"method": method, "path": path, "status": status})
48+
49+
with open("prism-coverage.json", "w", encoding="utf-8") as f:
50+
json.dump(report, f, indent=2)
51+
52+
print("Prism replay coverage: %d/%d resolved" % (report["resolved"], report["total"]))
53+
for u in report["unresolved"]:
54+
print(" UNRESOLVED: %s %s (status %s)" % (u["method"], u["path"], u["status"]))
55+
56+
if report["unresolved"]:
57+
sys.exit(1)
58+
59+
60+
if __name__ == "__main__":
61+
main()

0 commit comments

Comments
 (0)