59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
"""Guardrails for Trustee ``getTemplateWorkflows`` graphs (new instance bootstrap)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from modules.features.trustee.mainTrustee import getTemplateWorkflows
|
|
|
|
|
|
def _receiptTemplateGraph():
|
|
templates = getTemplateWorkflows() or []
|
|
t = next((w for w in templates if w.get("id") == "trustee-receipt-import"), None)
|
|
assert t is not None, "template trustee-receipt-import must exist"
|
|
return t.get("graph") or {}
|
|
|
|
|
|
def _materializeInstance(graph: dict, instanceId: str) -> dict:
|
|
raw = json.dumps(graph)
|
|
raw = raw.replace("{{featureInstanceId}}", instanceId)
|
|
return json.loads(raw)
|
|
|
|
|
|
class TestTrusteeTemplateReceiptImport:
|
|
"""The receipt-import chain must use explicit DataRefs (Pick-not-Push).
|
|
|
|
Empty ``documentList: []`` is not auto-wired by ``materializeConnectionRefs``
|
|
(that helper only materializes empty ``userConnection`` references).
|
|
"""
|
|
|
|
def test_receiptImportWiresDocumentListRefs(self):
|
|
g = _receiptTemplateGraph()
|
|
inst = "00000000-0000-0000-0000-000000000001"
|
|
g = _materializeInstance(g, inst)
|
|
|
|
byId = {n["id"]: n for n in g.get("nodes", []) if isinstance(n, dict) and n.get("id")}
|
|
|
|
proc = byId.get("process")
|
|
sync = byId.get("sync")
|
|
assert proc and sync
|
|
|
|
dlp = (proc.get("parameters") or {}).get("documentList")
|
|
dls = (sync.get("parameters") or {}).get("documentList")
|
|
|
|
assert isinstance(dlp, dict) and dlp.get("type") == "ref"
|
|
assert dlp.get("nodeId") == "extract"
|
|
assert dlp.get("path") == ["documents"]
|
|
|
|
assert isinstance(dls, dict) and dls.get("type") == "ref"
|
|
assert dls.get("nodeId") == "process"
|
|
assert dls.get("path") == ["documents"]
|
|
|
|
def test_receiptImportFeatureInstanceIdSubstituted(self):
|
|
g = _receiptTemplateGraph()
|
|
inst = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
|
|
g = _materializeInstance(g, inst)
|
|
for n in g.get("nodes", []):
|
|
p = n.get("parameters") or {}
|
|
if "featureInstanceId" in p:
|
|
assert p["featureInstanceId"] == inst
|