67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
# Tests for Phase 3: context.extractContent node, port types, executor dispatch.
|
|
|
|
import pytest
|
|
from modules.features.graphicalEditor.nodeDefinitions import STATIC_NODE_TYPES
|
|
from modules.features.graphicalEditor.portTypes import PORT_TYPE_CATALOG
|
|
from modules.workflows.automation2.udmUpstreamShapes import (
|
|
_coerceConsolidateResultInput,
|
|
_coerceUdmDocumentInput,
|
|
_coerceUdmNodeListInput,
|
|
)
|
|
|
|
|
|
def test_context_extractContent_node_exists():
|
|
ids = [n["id"] for n in STATIC_NODE_TYPES]
|
|
assert "context.extractContent" in ids
|
|
|
|
|
|
def test_context_extractContent_node_shape():
|
|
node = next(n for n in STATIC_NODE_TYPES if n["id"] == "context.extractContent")
|
|
assert node["category"] == "context"
|
|
assert node["meta"]["usesAi"] is False
|
|
assert node["_method"] == "context"
|
|
assert node["_action"] == "extractContent"
|
|
assert node["outputPorts"][0]["schema"] == "ActionResult"
|
|
assert "DocumentList" in node["inputPorts"][0]["accepts"]
|
|
assert "LoopItem" in node["inputPorts"][0]["accepts"]
|
|
names = [p["name"] for p in node["parameters"]]
|
|
assert names == ["documentList"]
|
|
|
|
|
|
def test_udm_port_types_registered():
|
|
assert "UdmDocument" in PORT_TYPE_CATALOG
|
|
assert "UdmNodeList" in PORT_TYPE_CATALOG
|
|
assert "ConsolidateResult" in PORT_TYPE_CATALOG
|
|
|
|
|
|
def test_coerceUdmDocument_from_direct():
|
|
upstream = {"id": "d1", "sourceType": "pdf", "sourcePath": "/a.pdf", "children": []}
|
|
result = _coerceUdmDocumentInput(upstream)
|
|
assert result["sourceType"] == "pdf"
|
|
|
|
|
|
def test_coerceUdmDocument_from_nested():
|
|
upstream = {"udm": {"id": "d1", "sourceType": "pdf", "sourcePath": "/a.pdf", "children": []}, "other": 1}
|
|
result = _coerceUdmDocumentInput(upstream)
|
|
assert result["sourceType"] == "pdf"
|
|
|
|
|
|
def test_coerceUdmNodeList():
|
|
upstream = {"nodes": [{"id": "n1"}, {"id": "n2"}], "count": 2}
|
|
result = _coerceUdmNodeListInput(upstream)
|
|
assert result["count"] == 2
|
|
assert len(result["nodes"]) == 2
|
|
|
|
|
|
def test_coerceConsolidateResult():
|
|
upstream = {"result": {"headers": [], "rows": []}, "mode": "table", "count": 3}
|
|
result = _coerceConsolidateResultInput(upstream)
|
|
assert result["mode"] == "table"
|
|
assert result["count"] == 3
|
|
|
|
|
|
def test_getExecutor_dispatches_context():
|
|
from modules.workflows.automation2.executionEngine import _getExecutor
|
|
from modules.workflows.automation2.executors import ActionNodeExecutor
|
|
executor = _getExecutor("context.extractContent", None)
|
|
assert isinstance(executor, ActionNodeExecutor)
|