File size: 6,555 Bytes
193db9d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
import pytest
from workflows.errors import CyclicDependencyError, UnknownVariableError, WorkflowError
from workflows.utils import _create_variable_step_mapping, create_dependency_graph, topological_sort
# Dummy classes to simulate Workflow, Step, and Field
class DummyField:
def __init__(self, name, type="str", variable=None):
self.name = name
self.type = type
# For input fields, variable property is needed
self.variable = variable if variable is not None else name
class DummyStep:
def __init__(self, input_fields, output_fields):
self.input_fields = input_fields
self.output_fields = output_fields
class DummyWorkflow:
def __init__(self, steps):
# steps is a dict with key as step_id and value as DummyStep
self.steps = steps
# Tests for _create_variable_step_mapping
def test_create_variable_step_mapping_success():
# Create a workflow with two steps producing unique output variables
step_a = DummyStep(input_fields=[], output_fields=[DummyField("out1")])
step_b = DummyStep(input_fields=[], output_fields=[DummyField("out2")])
workflow = DummyWorkflow({"A": step_a, "B": step_b})
mapping = _create_variable_step_mapping(workflow)
assert mapping == {"A.out1": "A", "B.out2": "B"}
def test_create_variable_step_mapping_duplicate():
# Create a workflow where two steps produce an output with same name
step_a = DummyStep(input_fields=[], output_fields=[DummyField("out"), DummyField("out")])
workflow = DummyWorkflow({"A": step_a})
with pytest.raises(WorkflowError):
_create_variable_step_mapping(workflow)
def test_create_variable_step_mapping_empty():
"""Test _create_variable_step_mapping with an empty workflow should return an empty mapping."""
workflow = DummyWorkflow({})
mapping = _create_variable_step_mapping(workflow)
assert mapping == {}
def test_create_variable_step_mapping_multiple_outputs():
"""Test a workflow where a single step produces multiple outputs with unique names."""
step = DummyStep(input_fields=[], output_fields=[DummyField("out1"), DummyField("out2")])
workflow = DummyWorkflow({"A": step})
mapping = _create_variable_step_mapping(workflow)
assert mapping == {"A.out1": "A", "A.out2": "A"}
# Tests for create_dependency_graph
def test_create_dependency_graph_success_with_dependency():
# Step A produces 'A.out', which is used as input in step B
step_a = DummyStep(input_fields=[], output_fields=[DummyField("out")])
# For input_fields, explicitly set variable to reference A.out
step_b = DummyStep(input_fields=[DummyField("dummy", variable="A.out")], output_fields=[DummyField("result")])
workflow = DummyWorkflow({"A": step_a, "B": step_b})
# No external input provided for A.out so dependency must be created
deps = create_dependency_graph(workflow, input_values={})
# Step B depends on step A
assert deps["B"] == {"A"}
# Step A has no dependencies
assert deps["A"] == set()
def test_create_dependency_graph_success_with_external_input():
# Step B expects an input, but it is provided externally
step_b = DummyStep(
input_fields=[DummyField("param", variable="external_param")], output_fields=[DummyField("result")]
)
workflow = DummyWorkflow({"B": step_b})
# Provide external input for external_param
deps = create_dependency_graph(workflow, input_values={"external_param": 42})
# With external input, no dependency is needed
assert deps["B"] == set()
def test_create_dependency_graph_unknown_variable():
# Step B expects an input that is neither produced by any step nor provided externally
step_b = DummyStep(
input_fields=[DummyField("param", variable="non_existent")], output_fields=[DummyField("result")]
)
workflow = DummyWorkflow({"B": step_b})
with pytest.raises(UnknownVariableError):
_ = create_dependency_graph(workflow, input_values={})
def test_create_dependency_graph_complex():
"""Test create_dependency_graph on a more complex workflow with multiple dependencies."""
# Step A produces A.out, Step B uses A.out, Step C uses B.out, and Step D uses both A.out and B.out
step_a = DummyStep(input_fields=[], output_fields=[DummyField("out")])
step_b = DummyStep(input_fields=[DummyField("inp", variable="A.out")], output_fields=[DummyField("out")])
step_c = DummyStep(input_fields=[DummyField("inp", variable="B.out")], output_fields=[DummyField("result")])
step_d = DummyStep(
input_fields=[DummyField("inp1", variable="A.out"), DummyField("inp2", variable="B.out")],
output_fields=[DummyField("final")],
)
workflow = DummyWorkflow({"A": step_a, "B": step_b, "C": step_c, "D": step_d})
# Provide external input for "B.out" so that step B's output isn't expected to come from a step
# However, to simulate dependency, assume external input is not provided for the dependencies used in step C and D
# Therefore, workflow must resolve A.out for step B, and then step B produces B.out for steps C and D.
# Let's not provide any external input, so both dependencies are created.
deps = create_dependency_graph(workflow, input_values={})
# Expected dependencies:
# B depends on A
# C depends on B
# D depends on both A and B
assert deps["B"] == {"A"}
assert deps["C"] == {"B"}
assert deps["D"] == {"A", "B"}
# Tests for topological_sort
def test_topological_sort_success():
# Create a simple dependency graph: A -> B -> C
deps = {"A": set(), "B": {"A"}, "C": {"B"}}
order = topological_sort(deps)
# Check that order satisfies dependencies: A before B, B before C
assert order.index("A") < order.index("B") < order.index("C")
def test_topological_sort_cycle():
# Create a cyclic dependency: A -> B and B -> A
deps = {"A": {"B"}, "B": {"A"}}
with pytest.raises(CyclicDependencyError):
_ = topological_sort(deps)
def test_topological_sort_single_node():
"""Test topological_sort on a graph with a single node and no dependencies."""
deps = {"A": set()}
order = topological_sort(deps)
assert order == ["A"]
def test_topological_sort_disconnected():
"""Test topological_sort on a graph with disconnected nodes (no dependencies among them)."""
deps = {"A": set(), "B": set(), "C": set()}
order = topological_sort(deps)
# The order can be in any permutation, but must contain all nodes
assert set(order) == {"A", "B", "C"}
|