55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Smoke-check Flowable: the registratie process is deployed, and starting an
|
|
instance parks it on the OpenZaakAanmaken external task. Stdlib only.
|
|
"""
|
|
import base64, json, sys, time, urllib.error, urllib.request
|
|
|
|
BASE = "http://localhost:8090/flowable-rest/service"
|
|
AUTH = "Basic " + base64.b64encode(b"rest-admin:test").decode()
|
|
|
|
|
|
def call(method, path, payload=None):
|
|
data = json.dumps(payload).encode() if payload is not None else None
|
|
req = urllib.request.Request(BASE + path, data=data, method=method, headers={
|
|
"Authorization": AUTH, "Content-Type": "application/json", "Accept": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return r.status, json.loads(r.read() or "null")
|
|
|
|
|
|
def main():
|
|
# 1. process definition deployed? (wait for the async init-container deploy)
|
|
defs = {"total": 0}
|
|
for _ in range(40):
|
|
_, defs = call("GET", "/repository/process-definitions?key=registratie")
|
|
if defs["total"] >= 1:
|
|
break
|
|
time.sleep(3)
|
|
assert defs["total"] >= 1, "registratie process definition not deployed"
|
|
print(f"process definition 'registratie' deployed (total={defs['total']})")
|
|
|
|
# 2. start an instance
|
|
st, pi = call("POST", "/runtime/process-instances", {"processDefinitionKey": "registratie"})
|
|
assert st == 201, f"start failed: {st} {pi}"
|
|
pid = pi["id"]
|
|
assert pi.get("ended") is False, "instance ended immediately — external task not reached"
|
|
print(f"started instance {pid} (ended={pi.get('ended')})")
|
|
|
|
# 3. waiting on the external task?
|
|
_, ex = call("GET", f"/runtime/executions?processInstanceId={pid}")
|
|
activities = [e.get("activityId") for e in ex["data"]]
|
|
assert "OpenZaakAanmaken" in activities, f"not waiting at OpenZaakAanmaken: {activities}"
|
|
print(f"instance is waiting at the external task: {activities}")
|
|
|
|
# 4. cleanup
|
|
try:
|
|
call("DELETE", f"/runtime/process-instances/{pid}")
|
|
print("cleaned up instance")
|
|
except urllib.error.HTTPError:
|
|
pass
|
|
|
|
print("flowable smoke OK")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|