71 lines
2.4 KiB
Python
71 lines
2.4 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,
|
|
INPUT_EXTRACTORS,
|
|
_extractUdmDocument,
|
|
_extractUdmNodeList,
|
|
_extractConsolidateResult,
|
|
)
|
|
|
|
|
|
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"] == "UdmDocument"
|
|
assert "DocumentList" in node["inputPorts"][0]["accepts"]
|
|
|
|
|
|
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_udm_extractors_registered():
|
|
assert "UdmDocument" in INPUT_EXTRACTORS
|
|
assert "UdmNodeList" in INPUT_EXTRACTORS
|
|
assert "ConsolidateResult" in INPUT_EXTRACTORS
|
|
|
|
|
|
def test_extractUdmDocument_from_direct():
|
|
upstream = {"id": "d1", "sourceType": "pdf", "sourcePath": "/a.pdf", "children": []}
|
|
result = _extractUdmDocument(upstream)
|
|
assert result["sourceType"] == "pdf"
|
|
|
|
|
|
def test_extractUdmDocument_from_nested():
|
|
upstream = {"udm": {"id": "d1", "sourceType": "pdf", "sourcePath": "/a.pdf", "children": []}, "other": 1}
|
|
result = _extractUdmDocument(upstream)
|
|
assert result["sourceType"] == "pdf"
|
|
|
|
|
|
def test_extractUdmNodeList():
|
|
upstream = {"nodes": [{"id": "n1"}, {"id": "n2"}], "count": 2}
|
|
result = _extractUdmNodeList(upstream)
|
|
assert result["count"] == 2
|
|
assert len(result["nodes"]) == 2
|
|
|
|
|
|
def test_extractConsolidateResult():
|
|
upstream = {"result": {"headers": [], "rows": []}, "mode": "table", "count": 3}
|
|
result = _extractConsolidateResult(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)
|