85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
"""Record/check persistent Mock IDs across a process/container restart."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("mode", choices=["record", "check"])
|
|
args = parser.parse_args()
|
|
base = os.getenv("FRAME_FLOW_URL", "http://127.0.0.1:18787").rstrip("/")
|
|
token = os.getenv("FRAME_FLOW_TOKEN", "")
|
|
if not token:
|
|
for line in Path(".env").read_text().splitlines():
|
|
if line.startswith("FRAME_FLOW_TOKEN="):
|
|
token = line.split("=", 1)[1]
|
|
|
|
|
|
def api(path, method="GET", payload=None):
|
|
request = Request(
|
|
base + path,
|
|
method=method,
|
|
data=json.dumps(payload).encode() if payload is not None else None,
|
|
headers={
|
|
"Authorization": "Bearer " + token,
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
with urlopen(request, timeout=10) as response:
|
|
body = response.read()
|
|
return json.loads(body) if body else None
|
|
|
|
|
|
assert api("/api/v1/settings")["mock"] is True, "Persistence test requires Mock mode"
|
|
record = Path("data/persistence-acceptance.json")
|
|
if args.mode == "record":
|
|
if record.exists():
|
|
raise SystemExit("An acceptance record already exists; run check first")
|
|
job = api(
|
|
"/api/v1/jobs",
|
|
"POST",
|
|
{
|
|
"name": "Persistence acceptance (paused)",
|
|
"enabled": False,
|
|
"schedule": {"type": "daily", "time": "07:00"},
|
|
"source": {"type": "template", "template": "quote"},
|
|
},
|
|
)
|
|
accepted = api(
|
|
"/api/v1/render",
|
|
"POST",
|
|
{
|
|
"type": "template",
|
|
"template": "quote",
|
|
"data": {"title": "PERSISTENCE TEST"},
|
|
},
|
|
)
|
|
for _ in range(20):
|
|
op = api("/api/v1/operations/" + accepted["operation_id"])
|
|
if op["status"] == "COMPLETED":
|
|
break
|
|
time.sleep(1)
|
|
assert op["status"] == "COMPLETED"
|
|
record.write_text(
|
|
json.dumps(
|
|
{
|
|
"job_id": job["id"],
|
|
"operation_id": op["operation_id"],
|
|
"asset_id": op["result"]["asset_id"],
|
|
}
|
|
)
|
|
)
|
|
print("Persistent Mock job, operation and image recorded")
|
|
else:
|
|
saved = json.loads(record.read_text())
|
|
assert any(j["id"] == saved["job_id"] for j in api("/api/v1/jobs"))
|
|
assert api("/api/v1/operations/" + saved["operation_id"])["status"] == "COMPLETED"
|
|
assert any(a["asset_id"] == saved["asset_id"] for a in api("/api/v1/assets"))
|
|
assert api("/ready")["status"] == "ready"
|
|
api("/api/v1/jobs/" + saved["job_id"], "DELETE")
|
|
record.unlink()
|
|
print("Restart persistence: 4 checks passed; acceptance job removed")
|