66 lines
2.9 KiB
Python
Executable File
66 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Retry 5 POST relationships SENZA campo `meta` (was extra_forbidden 422). Chained graph DEP→BUILD→PR→TASK→REQ + TEST→PR."""
|
|
from __future__ import annotations
|
|
import json, subprocess, sys
|
|
BASE = "http://127.0.0.1:8000/api/v1"
|
|
H = ["-H","X-ASF-Tenant-Id: t-demo-003","-H","X-ASF-User-Id: u-mario","-H","Content-Type: application/json"]
|
|
|
|
def curl(method, path, body=None):
|
|
cmd = ["curl","-s","-m","15","-o","-","-w","\n%{http_code}","-X",method] + H
|
|
if body is not None: cmd += ["--data", json.dumps(body, ensure_ascii=False)]
|
|
cmd.append(BASE+path)
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
out = (r.stdout or "{}").strip()
|
|
if "\n" in out:
|
|
jp, cp = out.rsplit("\n", 1)
|
|
else:
|
|
jp, cp = "", out
|
|
try: code = int(cp)
|
|
except: code = 0
|
|
try: data = json.loads(jp or "null")
|
|
except Exception as exc: data = f"ERR {exc}: {jp[:200]!r}"
|
|
print(f" [{method:6s}{path:22s}] HTTP={code:3d}",
|
|
f"id={str(data.get('id',''))[:10] if isinstance(data,dict) else '?'}",
|
|
"rel=", (data.get("source_artifact_id","")[:8] if isinstance(data,dict) else "?"),
|
|
"→",(data.get("target_artifact_id","")[:8] if isinstance(data,dict) else "?"),
|
|
"type=",(data.get("relationship_type","?") if isinstance(data,dict) else "?"),
|
|
file=sys.stderr)
|
|
return code, data
|
|
|
|
def main():
|
|
_, arts = curl("GET", "/artifacts?limit=50")
|
|
assert isinstance(arts, list)
|
|
by_type = {}
|
|
for a in arts:
|
|
if isinstance(a,dict):
|
|
by_type.setdefault(a["type"], []).append(a)
|
|
req = by_type["requirement"][0]
|
|
task = by_type["task"][0]
|
|
pr = by_type["pull_request"][0]
|
|
test = by_type["test"][0]
|
|
build = by_type["build"][0]
|
|
deploy= by_type["deployment"][0]
|
|
print(f" req={str(req['id'])[:10]} task={str(task['id'])[:10]} pr={str(pr['id'])[:10]} test={str(test['id'])[:10]} build={str(build['id'])[:10]} deploy={str(deploy['id'])[:10]}", file=sys.stderr)
|
|
rels = [
|
|
(task["id"], req["id"], "implements"),
|
|
(pr["id"], task["id"], "generated_from"),
|
|
(build["id"], pr["id"], "generated_from"),
|
|
(test["id"], pr["id"], "verifies"),
|
|
(deploy["id"], build["id"], "deployed_as"),
|
|
]
|
|
oks=0
|
|
for s,t,k in rels:
|
|
code, d = curl("POST","/relationships", {"source_artifact_id": s, "target_artifact_id": t, "relationship_type": k})
|
|
if 200 <= code < 300: oks += 1
|
|
print(f"\n rels OK = {oks}/5", file=sys.stderr)
|
|
|
|
code, graph = curl("GET", "/traceability/graph?limit=200")
|
|
assert code == 200 and isinstance(graph, dict)
|
|
nodes = graph.get("nodes", []); edges = graph.get("edges", [])
|
|
print(json.dumps({"nodes":len(nodes),"edges":len(edges),
|
|
"node_types":[n.get("type") for n in nodes],
|
|
"edge_labels":[e.get("label") for e in edges]}))
|
|
return 0 if oks == 5 else 2
|
|
|
|
raise SystemExit(main())
|