Source code for ewokscore.tests.test_graph_inputs
from typing import Union
from copy import deepcopy
from pydantic import Field
from ..task import Task
from ..model import BaseInputModel
from ..graph import inputs
from ..graph.taskgraph import TaskGraph
from ..bindings import load_graph
from ..bindings import execute_graph
from ..missing_data import MISSING_DATA
[docs]
def test_shorten_task_identifiers():
task_identifiers = ["a.b.c", "a.b.d", "a.b.e"]
shortmap = inputs._shorten_task_identifiers(task_identifiers)
expected = {"a.b.c": "c", "a.b.d": "d", "a.b.e": "e"}
assert shortmap == expected
task_identifiers = ["a.b.c", "a.b.d", "a.bb.c"]
shortmap = inputs._shorten_task_identifiers(task_identifiers)
expected = {"a.b.c": "b.c", "a.b.d": "d", "a.bb.c": "bb.c"}
assert shortmap == expected
task_identifiers = ["a.b.c", "a.b.c", "a.bb.c"]
shortmap = inputs._shorten_task_identifiers(task_identifiers)
expected = {"a.b.c": "b.c", "a.bb.c": "bb.c"}
assert shortmap == expected
[docs]
def test_graph_inputs():
graph = create_graph()
node_inputs = inputs.graph_inputs(graph)
expected = [
inputs.NodeInput(
id="task1",
label=None,
task_identifier="ClassExample",
name="a",
value=1,
required=True,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task1",
label=None,
task_identifier="ClassExample",
name="b",
value=MISSING_DATA,
required=True,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task1",
label=None,
task_identifier="ClassExample",
name="c",
value=MISSING_DATA,
required=False,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task1",
label=None,
task_identifier="ClassExample",
name="d",
value=MISSING_DATA,
required=False,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task2",
label="task2 label",
task_identifier="ClassExampleWithModel",
name="c",
value=4,
required=False,
description="parameter c",
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task2",
label="task2 label",
task_identifier="ClassExampleWithModel",
name="d",
value=-2,
required=False,
description=None,
examples=[100, "word"],
import_error=None,
),
inputs.NodeInput(
id="task3",
label=None,
task_identifier="method_example",
name="e",
value=5,
required=True,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task3",
label=None,
task_identifier="method_example",
name="f",
value=-3,
required=False,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task3",
label=None,
task_identifier="method_example",
name="g",
value=MISSING_DATA,
required=True,
description=None,
examples=None,
import_error=None,
),
inputs.NodeInput(
id="task4",
label=None,
task_identifier="NonExistingClass",
name="guess",
value=999,
required=True,
description=None,
examples=None,
import_error=ModuleNotFoundError("No module named 'does'"),
),
]
def sort(node_input):
return node_input.id, node_input.name
node_inputs = sorted(node_inputs, key=sort)
# Patch exceptions for comparison
for node_input in node_inputs:
if node_input.import_error:
node_input.import_error = repr(node_input.import_error)
for node_input in expected:
if node_input.import_error:
node_input.import_error = repr(node_input.import_error)
assert node_inputs == expected
[docs]
def test_graph_inputs_as_table():
graph = create_graph()
column_names, rows, metadata, footnotes = inputs.graph_inputs_as_table(graph)
expected_column_names = [
"Name",
"Value",
"Description",
"Examples",
"Task identifier",
"Id",
"Label",
]
expected_rows = [
["b⁽*⁾", "<MISSING_DATA>", "", "", "ClassExample", "task1", ""],
["g⁽*⁾", "<MISSING_DATA>", "", "", "method_example", "task3", ""],
["a", "1", "", "", "ClassExample", "task1", ""],
["d", "<MISSING_DATA>", "", "", "ClassExample", "task1", ""],
["c", "<MISSING_DATA>", "", "", "ClassExample", "task1", ""],
["c", "4", "parameter c", "", "ClassExampleWithModel", "task2", "task2 label"],
[
"d",
"-2",
"",
"• 100\n• 'word'",
"ClassExampleWithModel",
"task2",
"task2 label",
],
["e", "5", "", "", "method_example", "task3", ""],
["f", "-3", "", "", "method_example", "task3", ""],
["guess⁽†⁾", "999", "", "", "NonExistingClass", "task4", ""],
]
expected_metadata = {"id": "testgraph", "description": "Test graph inputs"}
expected_footnotes = [
"⁽*⁾ Value is required for execution.",
"⁽†⁾ Information from workflow only (task cannot be imported).",
]
nrequiredmissing = 2
assert sorted(rows[:nrequiredmissing]) == sorted(expected_rows[:nrequiredmissing])
assert sorted(rows[nrequiredmissing:]) == sorted(expected_rows[nrequiredmissing:])
assert column_names == expected_column_names
assert metadata == expected_metadata
assert footnotes == expected_footnotes
[docs]
class ClassExample(
Task,
input_names=["a", "b"],
optional_input_names=["c", "d"],
output_names=["a", "b", "c", "d"],
):
[docs]
def run(self):
self.outputs.a = self.inputs.a
self.outputs.b = self.inputs.b
self.outputs.c = self.get_input_value("c", -1)
self.outputs.d = self.get_input_value("d", -2)
[docs]
class InputModel(BaseInputModel):
a: Union[int, str] = Field(...)
a: Union[int, str] = Field(...)
c: Union[int, str] = Field(-1, description="parameter c")
d: Union[int, str] = Field(-2, examples=[100, "word"])
[docs]
class ClassExampleWithModel(
Task,
input_model=InputModel,
output_names=["a", "b", "c", "d"],
):
[docs]
def run(self):
self.outputs.a = self.inputs.a
self.outputs.b = self.inputs.b
self.outputs.c = self.inputs.c
self.outputs.d = self.inputs.d
[docs]
def method_example(a, b, e, g, c=-1, d=-2, f=-3):
return {"a": a, "b": b, "c": c, "d": d, "e": e, "f": f, "g": g}
[docs]
def create_graph() -> TaskGraph:
graph = {"id": "testgraph", "schema_version": "1.1", "label": "Test graph inputs"}
nodes = [
{
"id": "task1",
"default_inputs": [{"name": "a", "value": 1}],
"task_type": "class",
"task_identifier": f"{__name__}.ClassExample",
},
{
"id": "task2",
"label": "task2 label",
"default_inputs": [{"name": "b", "value": 3}, {"name": "c", "value": 4}],
"task_type": "class",
"task_identifier": f"{__name__}.ClassExampleWithModel",
},
{
"id": "task3",
"default_inputs": [{"name": "e", "value": 5}],
"task_type": "method",
"task_identifier": f"{__name__}.method_example",
},
]
links = [
{
"source": "task1",
"target": "task2",
"data_mapping": [{"source_output": "b", "target_input": "a"}],
},
{
"source": "task2",
"target": "task3",
"map_all_data": True,
},
]
graph_dict = {"graph": graph, "links": links, "nodes": nodes}
result = execute_graph(
deepcopy(graph_dict),
merge_outputs=False,
inputs=[
{"id": "task1", "name": "b", "value": 2},
{"id": "task3", "name": "e", "value": 5},
{"id": "task3", "name": "g", "value": 6},
],
outputs=[{"all": True}],
)
expected = {
"task1": {"a": 1, "b": 2, "c": -1, "d": -2},
"task2": {"a": 2, "b": 3, "c": 4, "d": -2},
"task3": {
"return_value": {"a": 2, "b": 3, "c": 4, "d": -2, "e": 5, "g": 6, "f": -3}
},
}
assert result == expected
nodes.append(
{
"id": "task4",
"default_inputs": [{"name": "guess", "value": 999}],
"task_type": "class",
"task_identifier": "does.not.exists.NonExistingClass",
}
)
task_graph = load_graph(graph_dict)
return task_graph