patrickandClaude Sonnet 5
4d6acd19aa
CI / Web-Barrierefreiheitspruefung (axe, Tastatur, Reflow, Textabstaende) (push) Canceled after 0s
CI / Core-Modul (Build/Vet/Test) (push) Canceled after 0s
CI / DMS Security-Test-Gate (AC5, Merge-Gate) (push) Canceled after 0s
CI / DMS-Modul (Lint/Build/Test/Coverage/Artefakt) (push) Canceled after 0s
CI / DMS-Modul Supply-Chain-Scan (govulncheck) (push) Canceled after 0s
CI / Web (dms-browser) (push) Canceled after 0s
CI / Web (pdf-viewer) (push) Canceled after 0s
CI / Web (shl) (push) Canceled after 0s
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
411 lines
21 KiB
Python
Executable File
411 lines
21 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""FDN-19 Smoke-Test gegen das laufende Gesamtsystem (nur Standardbibliothek + psql).
|
|
|
|
fdn19_smoke.py smoke caller|token Lebenszyklus-Smoke
|
|
fdn19_smoke.py batch <n> n Textdokumente hochladen (fuer den Wiederanlauf-Test)
|
|
fdn19_smoke.py verify-batch <n> wartet auf Indexierung, prueft Anzahl und Dubletten
|
|
|
|
Umgebung: FDN19_API, FDN19_ROOT (Basis-URL ohne Praefix), FDN19_STUB, FDN19_DSN, FDN19_TENANT,
|
|
FDN19_CORPUS (Verzeichnis), FDN19_TERM, FDN19_TOOL (smoke-tool), FDN19_MIGRATIONS.
|
|
Alle Daten sind synthetisch.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
API = os.environ["FDN19_API"]
|
|
ROOT = os.environ["FDN19_ROOT"]
|
|
STUB = os.environ["FDN19_STUB"]
|
|
DSN = os.environ["FDN19_DSN"]
|
|
TENANT = os.environ["FDN19_TENANT"]
|
|
CORPUS = os.environ["FDN19_CORPUS"]
|
|
TERM = os.environ["FDN19_TERM"]
|
|
TOOL = os.environ["FDN19_TOOL"]
|
|
MIGRATIONS = os.environ.get("FDN19_MIGRATIONS", "")
|
|
CORE_ACCOUNT = os.environ.get("FDN19_CORE_ACCOUNT", "")
|
|
CORE_ACCOUNT_B = os.environ.get("FDN19_CORE_ACCOUNT_B", "")
|
|
CORE_PASSWORD = os.environ.get("FDN19_CORE_PASSWORD", "")
|
|
CORE_USERS = [u for u in os.environ.get("FDN19_CORE_USERS", "").split(",") if u]
|
|
|
|
results = []
|
|
MODE_CORE = False
|
|
|
|
|
|
def psql(sql):
|
|
out = subprocess.run(["psql", DSN, "-qAtX", "-v", "ON_ERROR_STOP=1", "-c", sql], capture_output=True, text=True)
|
|
if out.returncode != 0:
|
|
raise RuntimeError("psql: " + out.stderr.strip())
|
|
return out.stdout.strip()
|
|
|
|
|
|
def http(method, url, body=None, headers=None, timeout=60):
|
|
data = body if isinstance(body, (bytes, type(None))) else json.dumps(body).encode()
|
|
h = dict(headers or {})
|
|
if isinstance(body, (dict, list)):
|
|
h.setdefault("Content-Type", "application/json")
|
|
req = urllib.request.Request(url, data=data, method=method, headers=h)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return r.status, r.read()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()
|
|
|
|
|
|
CORE_EMAIL = {}
|
|
|
|
|
|
def core_module_token(account, email):
|
|
"""Login am echten Core-Kontodienst und Modul-Token per POST /auth/module-token (nur fuer Modus core)."""
|
|
req = urllib.request.Request(account + "/auth/login", data=json.dumps({"email": email, "password": CORE_PASSWORD}).encode(),
|
|
method="POST", headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
cookie = [c for c in r.headers.get_all("Set-Cookie") if c.startswith("nexarch_session=")][0].split(";")[0]
|
|
req = urllib.request.Request(account + "/auth/module-token", data=b"", method="POST", headers={"Cookie": cookie})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.loads(r.read())["token"]
|
|
|
|
|
|
class Actor:
|
|
def __init__(self, mode, user_id, tenant=TENANT):
|
|
self.mode, self.id, self.tenant = mode, user_id, tenant
|
|
self.token = None
|
|
if mode == "token":
|
|
self.token = self.mint(user_id, tenant)
|
|
elif mode == "core":
|
|
self.token = core_module_token(CORE_ACCOUNT, CORE_EMAIL[user_id])
|
|
|
|
@staticmethod
|
|
def mint(sub, tenant, ttl=3600):
|
|
q = urllib.parse.urlencode({"sub": sub, "tenant": tenant, "ttl": ttl})
|
|
st, body = http("POST", STUB + "/token?" + q)
|
|
assert st == 200, "token: %s" % st
|
|
return body.decode()
|
|
|
|
def call(self, method, path, body=None, headers=None, params=None):
|
|
params = dict(params or {})
|
|
h = dict(headers or {})
|
|
if self.mode in ("token", "core"):
|
|
h["Authorization"] = "Bearer " + self.token
|
|
else:
|
|
params["caller"] = self.id
|
|
url = API + path + ("?" + urllib.parse.urlencode(params) if params else "")
|
|
st, raw = http(method, url, body, h)
|
|
try:
|
|
return st, json.loads(raw) if raw else None
|
|
except ValueError:
|
|
return st, raw
|
|
|
|
def upload(self, filename, mime, content):
|
|
st, sess = self.call("POST", "/uploads", {"filename": filename, "mimeType": mime, "totalSize": len(content)})
|
|
assert st == 201, "upload start %s %s" % (st, sess)
|
|
sid = sess["sessionId"]
|
|
st, _ = self.call("PUT", "/uploads/%s/chunk" % sid, content, {"Content-Type": "application/octet-stream"}, {"offset": 0})
|
|
assert st == 200, "chunk %s" % st
|
|
st, done = self.call("POST", "/uploads/%s/complete" % sid)
|
|
assert st == 200, "complete %s %s" % (st, done)
|
|
return done["documentId"]
|
|
|
|
def search(self, term, **extra):
|
|
p = {"q": term}
|
|
p.update(extra)
|
|
return self.call("GET", "/search", params=p)
|
|
|
|
def wait_search(self, term, doc_id, timeout):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
st, body = self.search(term)
|
|
if st == 200 and any(h["document_id"] == doc_id for h in body["hits"]):
|
|
return True
|
|
time.sleep(2)
|
|
return False
|
|
|
|
|
|
def info(msg):
|
|
print("INFO %s" % msg, flush=True)
|
|
|
|
|
|
def check(name, ok, detail=""):
|
|
results.append((name, bool(ok), detail))
|
|
print("%-4s %-62s %s" % ("PASS" if ok else "FAIL", name, detail), flush=True)
|
|
|
|
|
|
def guard(name, fn):
|
|
try:
|
|
fn()
|
|
except Exception as e: # noqa: BLE001 - jeder Fehler ist ein Testbefund
|
|
check(name, False, "Ausnahme: %s" % e)
|
|
|
|
|
|
def new_user(label):
|
|
if MODE_CORE:
|
|
email = CORE_USERS[len(CORE_EMAIL) % len(CORE_USERS)]
|
|
uid = psql("SELECT id FROM users WHERE email = '%s'" % email).split("\n")[0]
|
|
CORE_EMAIL[uid] = email
|
|
return uid
|
|
return psql("INSERT INTO users (email, name) VALUES ('%s-%d@example.test', '%s') RETURNING id" % (label, time.time_ns(), label)).split("\n")[0]
|
|
|
|
|
|
def read(name):
|
|
with open(os.path.join(CORPUS, name), "rb") as f:
|
|
return f.read()
|
|
|
|
|
|
def smoke(mode):
|
|
global MODE_CORE
|
|
MODE_CORE = mode == "core"
|
|
a_id, b_id = new_user("smoke-a"), new_user("smoke-b")
|
|
a, b = Actor(mode, a_id), Actor(mode, b_id)
|
|
ctx = {}
|
|
|
|
def s01():
|
|
st, body = http("GET", ROOT + "/healthz")
|
|
check("S01 healthz ok", st == 200 and b'"status":"ok"' in body)
|
|
st, body = http("GET", ROOT + "/metrics")
|
|
check("S01 metrics erreichbar", st == 200 and b"dms_http_requests_total" in body)
|
|
guard("S01", s01)
|
|
|
|
def s02():
|
|
st, body = http("GET", API + "/openapi.json")
|
|
n = len(json.loads(body)["paths"]) if st == 200 else 0
|
|
check("S02 openapi ohne Anmeldung abrufbar", st == 200 and n >= 40, "%d Pfade" % n)
|
|
guard("S02", s02)
|
|
|
|
def s03():
|
|
applied = int(psql("SELECT count(*) FROM dms_schema_migrations"))
|
|
want = len([f for f in os.listdir(MIGRATIONS) if f.endswith(".up.sql")]) if MIGRATIONS else applied
|
|
check("S03 alle Migrationen angewendet", applied == want, "%d von %d" % (applied, want))
|
|
guard("S03", s03)
|
|
|
|
def s04():
|
|
st, _ = http("GET", API + "/tags")
|
|
check("S04 ohne Identitaet 401", st == 401, "status %s" % st)
|
|
guard("S04", s04)
|
|
|
|
def s05():
|
|
ctx["text"] = a.upload("smokebeleg.pdf", "application/pdf", read("text.pdf"))
|
|
check("S05 Upload Text-PDF (Dokument + Revision)", True, ctx["text"])
|
|
ok = a.wait_search(TERM, ctx["text"], 120)
|
|
check("S06 Text-PDF ueber Worker im Suchindex auffindbar", ok)
|
|
guard("S05", s05)
|
|
|
|
def s07():
|
|
ctx["scan"] = a.upload("lieferschein-scan.png", "image/png", read("scan.png"))
|
|
ok = a.wait_search("Kaffeemuehle", ctx["scan"], 240)
|
|
check("S07 Scan per OCR erkannt und ueber Suchindex auffindbar", ok, "Begriff steht nur im Bild")
|
|
guard("S07", s07)
|
|
|
|
def einvoice(doc_key, label, filename, mime, fname, must_validate):
|
|
ctx[doc_key] = a.upload(filename, mime, read(fname))
|
|
end, row = time.time() + 180, ""
|
|
while time.time() < end and not row:
|
|
row = psql("SELECT r.status || '/' || COALESCE((SELECT c.status FROM revision_einvoice_conformance c WHERE c.revision_id = r.revision_id ORDER BY c.checked_at DESC LIMIT 1), 'keine') FROM revision_einvoice r JOIN file_revisions fr ON fr.id = r.revision_id WHERE fr.document_id = '%s'" % ctx[doc_key])
|
|
if not row or row.endswith("/keine"):
|
|
row = ""
|
|
time.sleep(3)
|
|
read_st, conf = (row.split("/") + ["", ""])[:2]
|
|
check("%s E-Rechnung gelesen" % label, read_st == "valid", row or "kein Ergebnis")
|
|
if must_validate:
|
|
check("%s KoSIT-Pruefung durchgefuehrt und konform" % label, conf in ("conformant", "warnings"), "Ergebnis: %s" % conf)
|
|
else:
|
|
check("%s Konformitaetsergebnis je Revision abgelegt" % label, conf in ("conformant", "warnings", "non_conformant", "not_checked"), "Ergebnis: %s" % conf)
|
|
n = psql("SELECT count(*) FROM document_metadata_values v JOIN metadata_fields f ON f.id = v.field_id WHERE v.document_id = '%s' AND f.field_key LIKE 'einvoice_%%'" % ctx[doc_key])
|
|
check("%s Kerndaten als Metadatenfelder abgelegt" % label, int(n or 0) >= 5, "%s Felder" % n)
|
|
guard("S08", lambda: einvoice("ubl", "S08 XRechnung UBL (KoSIT-Beispiel)", "rechnung-ubl.xml", "application/xml", "kosit-ubl.xml", True))
|
|
guard("S09", lambda: einvoice("zf", "S09 ZUGFeRD (synthetisch)", "rechnung-zugferd.pdf", "application/pdf", "zugferd.pdf", False))
|
|
|
|
def s09b():
|
|
plain = a.upload("klartext.txt", "text/plain", ("Klartextbeleg Zinnoberrot Wasserturm %d" % time.time_ns()).encode())
|
|
time.sleep(8)
|
|
st, body = a.search("Wasserturm")
|
|
found = st == 200 and any(h["document_id"] == plain for h in body["hits"])
|
|
info("F5 Volltext von text/plain-Dokumenten im Suchindex: %s (nur Titel wird indexiert, wenn nein)" % ("ja" if found else "nein"))
|
|
guard("S09b", s09b)
|
|
|
|
def s10():
|
|
st, body = a.search(TERM, facets="created_by,document_type,tag")
|
|
check("S10 Suche mit Facetten", st == 200 and "facets" in body and body["total"] >= 1, "total=%s" % (body or {}).get("total"))
|
|
st, body = a.search(TERM, document_type="gibt-es-nicht")
|
|
check("S10 Filter auf unbekannten Dokumenttyp liefert nichts", st == 200 and body["total"] == 0, "status %s" % st)
|
|
guard("S10", s10)
|
|
|
|
def s11():
|
|
st, sv = a.call("POST", "/saved-searches", {"name": "Smoke", "query": "q=" + TERM, "shared": False})
|
|
check("S11 gespeicherte Suche angelegt", st == 201, "status %s" % st)
|
|
st, res = a.call("GET", "/saved-searches/%s/results" % sv["id"])
|
|
check("S11 gespeicherte Suche liefert aktuelle Treffer", st == 200 and res.get("total", 0) >= 1, "status %s" % st)
|
|
st, _ = b.call("GET", "/saved-searches/%s/results" % sv["id"])
|
|
check("S11 private Suche fuer Fremde nicht abrufbar", st in (403, 404), "status %s" % st)
|
|
guard("S11", s11)
|
|
|
|
def s12():
|
|
st, d = a.call("GET", "/documents/%s/download-url" % ctx["text"])
|
|
check("S12 Download-URL angefordert", st == 200 and "url" in d, "status %s" % st)
|
|
url = d["url"]
|
|
check("S12 URL traegt den Routenpraefix", url.startswith("/api/dms/v1/download?"), url[:40])
|
|
url = ROOT + url
|
|
st, raw = http("GET", url)
|
|
check("S12 Inhalt ueber signierte URL byteidentisch", st == 200 and hashlib.sha256(raw).hexdigest() == hashlib.sha256(read("text.pdf")).hexdigest(), "status %s" % st)
|
|
parts = urllib.parse.urlsplit(url)
|
|
q = dict(urllib.parse.parse_qsl(parts.query))
|
|
|
|
def variant(**changes):
|
|
nq = dict(q)
|
|
nq.update(changes)
|
|
return urllib.parse.urlunsplit(parts._replace(query=urllib.parse.urlencode(nq)))
|
|
|
|
flip = q["sig"][:-1] + ("A" if q["sig"][-1] != "A" else "B")
|
|
for label, u in (("Signatur veraendert", variant(sig=flip)),
|
|
("Ablaufzeit verlaengert", variant(expiry=str(int(q["expiry"]) + 86400))),
|
|
("andere Revision", variant(revision_id="00000000-0000-0000-0000-000000000000")),
|
|
("andere Dokument-ID", variant(document_id="00000000-0000-0000-0000-000000000000"))):
|
|
st, _ = http("GET", u)
|
|
check("S12 manipulierte URL abgelehnt: %s" % label, st in (400, 401, 403, 404), "status %s" % st)
|
|
st, _ = http("GET", variant(expiry="1", sig=q["sig"]))
|
|
check("S12 abgelaufene URL abgelehnt", st in (400, 401, 403, 410), "status %s" % st)
|
|
q2 = dict(q)
|
|
q2["filename"] = "anderer-name.pdf"
|
|
st, _ = http("GET", urllib.parse.urlunsplit(parts._replace(query=urllib.parse.urlencode(q2))))
|
|
check("S12 Dateiname ist nicht signiert (nur Anzeige, Inhalt bleibt derselbe)", st == 200, "status %s" % st)
|
|
guard("S12", s12)
|
|
|
|
def s13():
|
|
tag = psql("INSERT INTO tags (name) VALUES ('smoke-%d') RETURNING id" % time.time_ns()).split("\n")[0]
|
|
st, _ = a.call("POST", "/documents/%s/tags" % ctx["text"], {"tag_id": tag})
|
|
st2, tags = a.call("GET", "/documents/%s/tags" % ctx["text"])
|
|
check("S13 Tag zugewiesen und lesbar", st in (200, 201, 204) and st2 == 200 and any(t.get("id") == tag or t.get("ID") == tag for t in tags), "status %s/%s" % (st, st2))
|
|
st, _ = a.call("PUT", "/documents/%s/metadata" % ctx["text"], {})
|
|
check("S13 Metadaten-Endpunkt antwortet", st in (200, 204, 400, 422), "status %s" % st)
|
|
guard("S13", s13)
|
|
|
|
def s14():
|
|
st, res = a.call("POST", "/trash", {"document_ids": [ctx["text"]]})
|
|
check("S14 Dokument in den Papierkorb (POST /trash)", st == 200 and ctx["text"] in (res or {}).get("trashed_document_ids", []), "status %s" % st)
|
|
st, lst = a.call("GET", "/trash")
|
|
check("S14 Papierkorb listet Dokument", st == 200 and ctx["text"] in json.dumps(lst), "status %s" % st)
|
|
end, gone = time.time() + 60, False
|
|
while time.time() < end and not gone:
|
|
st, body = a.search(TERM)
|
|
gone = st == 200 and not any(h["document_id"] == ctx["text"] for h in body["hits"])
|
|
time.sleep(2)
|
|
check("S14 Papierkorb-Dokument verschwindet aus der Suche", gone)
|
|
st, _ = a.call("POST", "/trash/restore", {"document_ids": [ctx["text"]]})
|
|
check("S14 Wiederherstellen", st == 200, "status %s" % st)
|
|
check("S14 wiederhergestelltes Dokument wieder auffindbar", a.wait_search(TERM, ctx["text"], 60))
|
|
guard("S14", s14)
|
|
|
|
def s15():
|
|
a.call("POST", "/trash", {"document_ids": [ctx["text"]]})
|
|
st, body = a.call("POST", "/trash/purge", {"document_ids": [ctx["text"]]})
|
|
still = psql("SELECT count(*) FROM documents WHERE id = '%s'" % ctx["text"])
|
|
denied = psql("SELECT count(*) FROM audit_log WHERE action = 'document_purge_denied'")
|
|
keyfiles = psql("SELECT count(*) FROM file_revisions WHERE document_id = '%s'" % ctx["text"])
|
|
check("S15 endgueltiges Loeschen ohne Archive fail-closed verweigert", still == "1" and keyfiles == "1", "status %s, Dokument vorhanden=%s" % (st, still))
|
|
check("S15 Ablehnung im Audit-Log (document_purge_denied)", int(denied or 0) >= 1, "%s Eintraege" % denied)
|
|
a.call("POST", "/trash/restore", {"document_ids": [ctx["text"]]})
|
|
guard("S15", s15)
|
|
|
|
def s16():
|
|
st, body = b.search(TERM)
|
|
check("S16 fremder Nutzer sieht in der Suche nichts", st == 200 and body["total"] == 0 or not body["hits"], "hits=%s" % len((body or {}).get("hits", [])))
|
|
st, _ = b.call("GET", "/documents/%s/content" % ctx["scan"])
|
|
check("S16 fremder Nutzer erhaelt keinen Inhalt", st == 403, "status %s" % st)
|
|
guard("S16", s16)
|
|
|
|
def s17():
|
|
out = subprocess.run([TOOL, "verify-chain"], capture_output=True, text=True, env=os.environ)
|
|
check("S17 Audit-Kette gueltig", out.returncode == 0, out.stdout.strip())
|
|
guard("S17", s17)
|
|
|
|
def s18():
|
|
st, body = a.call("GET", "/cmis/browser")
|
|
check("S18 CMIS-Repositoryinfo abrufbar", st == 200 and TENANT in json.dumps(body), "status %s" % st)
|
|
guard("S18", s18)
|
|
|
|
if mode == "token":
|
|
def s19():
|
|
st, _ = http("GET", API + "/tags", headers={"Authorization": "Bearer " + Actor.mint(a_id, "globex")})
|
|
check("S19 Token anderer Mandant 401", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/tags", headers={"Authorization": "Bearer " + Actor.mint(a_id, TENANT, -3600)})
|
|
check("S19 abgelaufenes Token 401", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/tags", headers={"Authorization": "Bearer " + a.token[:-3] + "AAA"})
|
|
check("S19 manipuliertes Token 401", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/tags?caller=" + a_id)
|
|
check("S19 caller-Parameter ohne Token wirkungslos (401)", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/documents/%s/tags?caller=%s" % (ctx["scan"], a_id), headers={"Authorization": "Bearer " + b.token})
|
|
check("S19 gefaelschter caller neben fremdem Token bleibt wirkungslos (403)", st == 403, "status %s" % st)
|
|
guard("S19", s19)
|
|
|
|
if mode == "core":
|
|
def s19c():
|
|
st, _ = http("GET", API + "/tags", headers={"Authorization": "Bearer " + core_module_token(CORE_ACCOUNT_B, CORE_USERS[0])})
|
|
check("S19 echtes Token eines anderen Mandanten (zweiter Kontodienst) 401", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/tags", headers={"Authorization": "Bearer " + a.token[:-3] + "AAA"})
|
|
check("S19 manipuliertes Token 401", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/tags?caller=" + a_id)
|
|
check("S19 caller-Parameter ohne Token wirkungslos (401)", st == 401, "status %s" % st)
|
|
st, _ = http("GET", API + "/documents/%s/tags?caller=%s" % (ctx["scan"], a_id), headers={"Authorization": "Bearer " + b.token})
|
|
check("S19 gefaelschter caller neben fremdem Token bleibt wirkungslos (403)", st == 403, "status %s" % st)
|
|
info("abgelaufenes Token: durch e2e-core-dms.sh abgedeckt (TTL 5 s); hier nicht erzeugbar")
|
|
guard("S19c", s19c)
|
|
|
|
|
|
def mini_pdf(lines):
|
|
content = "BT /F1 12 Tf 50 700 Td\n" + "".join("(%s) Tj 0 -18 Td\n" % l for l in lines) + "ET"
|
|
return ("%%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n"
|
|
"3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 595 842]/Contents 4 0 R/Resources<</Font<</F1 5 0 R>>>>>>endobj\n"
|
|
"4 0 obj<</Length %d>>stream\n%s\nendstream endobj\n5 0 obj<</Type/Font/Subtype/Type1/BaseFont/Helvetica>>endobj\n"
|
|
"xref\n0 6\n0000000000 65535 f \ntrailer<</Size 6/Root 1 0 R>>\nstartxref\n0\n%%%%EOF" % (len(content), content)).encode()
|
|
|
|
|
|
def batch(n):
|
|
actor_id = psql("SELECT id FROM users ORDER BY created_at LIMIT 1")
|
|
a = Actor("caller", actor_id)
|
|
for i in range(int(n)):
|
|
lines = ["Wiederanlaufbeleg %d" % i, "Lauf %d" % time.time_ns(), "Musterhandel Nordlicht"]
|
|
print(a.upload("batch-%d.pdf" % i, "application/pdf", mini_pdf(lines)))
|
|
|
|
|
|
def verify_batch(n):
|
|
actor_id = psql("SELECT id FROM users ORDER BY created_at LIMIT 1")
|
|
a = Actor("caller", actor_id)
|
|
start = time.time()
|
|
end = start + 480
|
|
open_jobs = "?"
|
|
while time.time() < end:
|
|
open_jobs = psql("SELECT count(*) FROM processing_jobs WHERE status NOT IN ('succeeded', 'dead_letter')")
|
|
st, body = a.search("Wiederanlaufbeleg")
|
|
if open_jobs == "0" and st == 200 and body["total"] >= int(n):
|
|
break
|
|
time.sleep(3)
|
|
waited = int(time.time() - start)
|
|
docs = psql("SELECT count(*) FROM documents WHERE title LIKE 'batch-%.pdf'")
|
|
dup = psql("SELECT count(*) FROM (SELECT d.id FROM file_revisions fr JOIN documents d ON d.id = fr.document_id WHERE d.title LIKE 'batch-%.pdf' GROUP BY d.id HAVING count(*) > 1) x")
|
|
same = psql("SELECT count(*) FROM (SELECT checksum_sha256 FROM file_revisions fr JOIN documents d ON d.id = fr.document_id WHERE d.title LIKE 'batch-%.pdf' GROUP BY checksum_sha256 HAVING count(*) > 1) x")
|
|
dead = psql("SELECT count(*) FROM processing_jobs WHERE status = 'dead_letter'")
|
|
check("W1 alle %s Dokumente vorhanden, keine Dubletten" % n, docs == str(n) and dup == "0" and same == "0", "Dokumente=%s Dokumente mit mehreren Revisionen=%s gleiche Pruefsummen=%s" % (docs, dup, same))
|
|
check("W2 alle Dokumente nach hartem Neustart im Suchindex", st == 200 and body["total"] >= int(n), "total=%s" % (body or {}).get("total"))
|
|
check("W3 keine offenen Jobs, kein Dead-Letter", open_jobs == "0" and dead == "0", "offen=%s dead_letter=%s" % (open_jobs, dead))
|
|
info("Wiederanlauf-Dauer bis alle Jobs erledigt waren: %d s" % waited)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1]
|
|
if cmd == "smoke":
|
|
smoke(sys.argv[2])
|
|
elif cmd == "batch":
|
|
batch(sys.argv[2])
|
|
elif cmd == "verify-batch":
|
|
verify_batch(sys.argv[2])
|
|
else:
|
|
sys.exit("unbekannt: " + cmd)
|
|
failed = [r for r in results if not r[1]]
|
|
if results:
|
|
print("\nGESAMT: %d Pruefungen, %d bestanden, %d fehlgeschlagen" % (len(results), len(results) - len(failed), len(failed)))
|
|
sys.exit(1 if failed else 0)
|