310 lines
11 KiB
Python
310 lines
11 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
"""
|
|
Phase-5 Schicht-4 — unit tests for ``materializeFeatureInstanceRefs`` and the
|
|
runtime envelope unwrap in ``graphUtils.resolveParameterReferences``.
|
|
|
|
Plan: ``wiki/c-work/1-plan/2026-04-typed-action-architecture.md`` (T11).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
import pytest
|
|
|
|
from modules.workflows.automation2.featureInstanceRefMigration import (
|
|
materializeFeatureInstanceRefs,
|
|
)
|
|
from modules.workflows.automation2.graphUtils import (
|
|
_isTypedRefEnvelope,
|
|
_unwrapTypedRef,
|
|
resolveParameterReferences,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Migration: raw UUID -> typed envelope
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMaterializeFeatureInstanceRefs:
|
|
def test_emptyGraphIsReturnedAsIs(self):
|
|
out = materializeFeatureInstanceRefs({})
|
|
assert out == {}
|
|
|
|
def test_nonDictInputIsPassthrough(self):
|
|
# Defensive: callers may pass a None / list by accident.
|
|
assert materializeFeatureInstanceRefs(None) is None
|
|
assert materializeFeatureInstanceRefs([]) == []
|
|
|
|
def test_graphWithoutFeatureInstanceIdIsUnchanged(self):
|
|
graph = {"nodes": [{"id": "n1", "type": "trigger.manual", "parameters": {}}]}
|
|
original = copy.deepcopy(graph)
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
assert out == original
|
|
|
|
def test_inputIsNotMutated(self):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": "abc-123"},
|
|
}
|
|
]
|
|
}
|
|
snapshot = copy.deepcopy(graph)
|
|
materializeFeatureInstanceRefs(graph)
|
|
assert graph == snapshot
|
|
|
|
def test_rawUuidIsConvertedToEnvelope(self):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": "abc-123"},
|
|
}
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
param = out["nodes"][0]["parameters"]["featureInstanceId"]
|
|
assert param == {
|
|
"$type": "FeatureInstanceRef",
|
|
"id": "abc-123",
|
|
"featureCode": "trustee",
|
|
}
|
|
|
|
def test_rawUuidPreservedWhitespaceIsTrimmed(self):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": " abc-123 "},
|
|
}
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
assert out["nodes"][0]["parameters"]["featureInstanceId"]["id"] == "abc-123"
|
|
|
|
def test_emptyStringIsLeftUntouched(self):
|
|
# Empty featureInstanceId is the editor placeholder for "not yet bound";
|
|
# the migration must NOT pretend an empty value is a real UUID.
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": ""},
|
|
}
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
assert out["nodes"][0]["parameters"]["featureInstanceId"] == ""
|
|
|
|
def test_alreadyTypedEnvelopeIsIdempotent(self):
|
|
envelope = {
|
|
"$type": "FeatureInstanceRef",
|
|
"id": "abc-123",
|
|
"featureCode": "trustee",
|
|
}
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": envelope},
|
|
}
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
assert out["nodes"][0]["parameters"]["featureInstanceId"] == envelope
|
|
|
|
def test_runMigrationTwiceProducesSameResult(self):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": "abc-123"},
|
|
}
|
|
]
|
|
}
|
|
once = materializeFeatureInstanceRefs(graph)
|
|
twice = materializeFeatureInstanceRefs(once)
|
|
assert once == twice
|
|
|
|
@pytest.mark.parametrize(
|
|
"nodeType,expectedFeatureCode",
|
|
[
|
|
("trustee.extractFromFiles", "trustee"),
|
|
("trustee.processDocuments", "trustee"),
|
|
("redmine.createIssue", "redmine"),
|
|
("clickup.createTask", "clickup"),
|
|
("sharepoint.listFiles", "sharepoint"),
|
|
("outlook.readEmails", "outlook"),
|
|
("email.searchEmail", "outlook"),
|
|
],
|
|
)
|
|
def test_featureCodeIsDerivedFromNodeTypePrefix(
|
|
self, nodeType, expectedFeatureCode
|
|
):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n",
|
|
"type": nodeType,
|
|
"parameters": {"featureInstanceId": "uuid-x"},
|
|
}
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
env = out["nodes"][0]["parameters"]["featureInstanceId"]
|
|
assert env["featureCode"] == expectedFeatureCode
|
|
|
|
def test_unknownNodeTypePrefixOmitsFeatureCode(self):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n",
|
|
"type": "weird.unknown.action",
|
|
"parameters": {"featureInstanceId": "uuid-x"},
|
|
}
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
env = out["nodes"][0]["parameters"]["featureInstanceId"]
|
|
assert env == {"$type": "FeatureInstanceRef", "id": "uuid-x"}
|
|
|
|
def test_multipleNodesAreAllMigrated(self):
|
|
graph = {
|
|
"nodes": [
|
|
{
|
|
"id": "n5",
|
|
"type": "trustee.extractFromFiles",
|
|
"parameters": {"featureInstanceId": "uuid-1"},
|
|
},
|
|
{
|
|
"id": "n6",
|
|
"type": "trustee.queryData",
|
|
"parameters": {"featureInstanceId": "uuid-2"},
|
|
},
|
|
{
|
|
"id": "n9",
|
|
"type": "trustee.processDocuments",
|
|
"parameters": {"featureInstanceId": "uuid-3"},
|
|
},
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
ids = [n["parameters"]["featureInstanceId"]["id"] for n in out["nodes"]]
|
|
assert ids == ["uuid-1", "uuid-2", "uuid-3"]
|
|
|
|
def test_nodesWithoutParametersAreSkipped(self):
|
|
graph = {
|
|
"nodes": [
|
|
{"id": "n1", "type": "trigger.manual"},
|
|
{"id": "n2", "type": "trustee.queryData"}, # no parameters key
|
|
{
|
|
"id": "n3",
|
|
"type": "trustee.processDocuments",
|
|
"parameters": None,
|
|
},
|
|
]
|
|
}
|
|
out = materializeFeatureInstanceRefs(graph)
|
|
assert out == graph
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Runtime envelope unwrap (graphUtils._unwrapTypedRef + resolveParameterReferences)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestIsTypedRefEnvelope:
|
|
def test_recognisesFeatureInstanceRef(self):
|
|
env = {"$type": "FeatureInstanceRef", "id": "abc"}
|
|
assert _isTypedRefEnvelope(env) is True
|
|
|
|
def test_recognisesConnectionRef(self):
|
|
env = {"$type": "ConnectionRef", "id": "abc"}
|
|
assert _isTypedRefEnvelope(env) is True
|
|
|
|
def test_rejectsRawDict(self):
|
|
assert _isTypedRefEnvelope({"id": "abc"}) is False
|
|
|
|
def test_rejectsUnknownType(self):
|
|
assert _isTypedRefEnvelope({"$type": "Foobar", "id": "abc"}) is False
|
|
|
|
def test_rejectsNonDict(self):
|
|
assert _isTypedRefEnvelope("abc") is False
|
|
assert _isTypedRefEnvelope(None) is False
|
|
assert _isTypedRefEnvelope(["abc"]) is False
|
|
|
|
|
|
class TestUnwrapTypedRef:
|
|
def test_unwrapsFeatureInstanceRefToId(self):
|
|
env = {"$type": "FeatureInstanceRef", "id": "uuid-x", "featureCode": "trustee"}
|
|
assert _unwrapTypedRef(env) == "uuid-x"
|
|
|
|
def test_unwrapsConnectionRefToId(self):
|
|
env = {"$type": "ConnectionRef", "id": "conn-y", "authority": "msft"}
|
|
assert _unwrapTypedRef(env) == "conn-y"
|
|
|
|
def test_unwrapsSharePointFileRefToFilePath(self):
|
|
env = {"$type": "SharePointFileRef", "filePath": "/Sites/X/file.pdf"}
|
|
assert _unwrapTypedRef(env) == "/Sites/X/file.pdf"
|
|
|
|
def test_passthroughForNonEnvelope(self):
|
|
assert _unwrapTypedRef("plain-string") == "plain-string"
|
|
assert _unwrapTypedRef({"id": "abc"}) == {"id": "abc"}
|
|
assert _unwrapTypedRef(None) is None
|
|
|
|
def test_returnsEnvelopeIfPrimaryFieldMissing(self):
|
|
# Defensive: malformed envelope without ``id`` falls back to itself
|
|
# rather than silently dropping data.
|
|
env = {"$type": "FeatureInstanceRef", "featureCode": "trustee"}
|
|
assert _unwrapTypedRef(env) == env
|
|
|
|
|
|
class TestResolveParameterReferencesUnwrap:
|
|
def test_typedEnvelopeAtTopLevelIsUnwrapped(self):
|
|
env = {"$type": "FeatureInstanceRef", "id": "uuid-z", "featureCode": "trustee"}
|
|
out = resolveParameterReferences(env, nodeOutputs={})
|
|
assert out == "uuid-z"
|
|
|
|
def test_typedEnvelopeNestedInDictIsUnwrapped(self):
|
|
params = {
|
|
"featureInstanceId": {
|
|
"$type": "FeatureInstanceRef",
|
|
"id": "uuid-z",
|
|
"featureCode": "trustee",
|
|
},
|
|
"mode": "lookup",
|
|
}
|
|
out = resolveParameterReferences(params, nodeOutputs={})
|
|
assert out == {"featureInstanceId": "uuid-z", "mode": "lookup"}
|
|
|
|
def test_typedEnvelopesInListAreUnwrappedElementwise(self):
|
|
params = [
|
|
{"$type": "FeatureInstanceRef", "id": "u1"},
|
|
{"$type": "FeatureInstanceRef", "id": "u2"},
|
|
"static",
|
|
]
|
|
out = resolveParameterReferences(params, nodeOutputs={})
|
|
assert out == ["u1", "u2", "static"]
|
|
|
|
def test_typedEnvelopeIsResolvedBeforeRefLookup(self):
|
|
# If a workflow somehow contains both shapes, the typed envelope wins;
|
|
# ref-resolution is for upstream-bound DataRefs which never carry
|
|
# ``$type`` at the top level.
|
|
env = {
|
|
"$type": "FeatureInstanceRef",
|
|
"id": "uuid-z",
|
|
# nonsensical ``type: ref`` shadow — must be ignored.
|
|
"type": "ref",
|
|
"nodeId": "nope",
|
|
"path": ["whatever"],
|
|
}
|
|
out = resolveParameterReferences(env, nodeOutputs={"nope": {"whatever": "x"}})
|
|
assert out == "uuid-z"
|