146 lines
4.5 KiB
Python
146 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.definition_graphs import (
|
|
DefinitionEdge,
|
|
DefinitionGraphConstraints,
|
|
DefinitionGraphLibrary,
|
|
DefinitionNode,
|
|
DefinitionNodeCountConstraint,
|
|
DefinitionNodeType,
|
|
DefinitionPort,
|
|
validate_definition_graph,
|
|
)
|
|
|
|
|
|
def library(*, allow_cycles: bool) -> DefinitionGraphLibrary:
|
|
return DefinitionGraphLibrary(
|
|
id="test",
|
|
version="0.1.0",
|
|
category_labels={"control": "Control"},
|
|
node_types=(
|
|
DefinitionNodeType(
|
|
type="start",
|
|
category="control",
|
|
label="Start",
|
|
description="Start",
|
|
icon="play",
|
|
),
|
|
DefinitionNodeType(
|
|
type="step",
|
|
category="control",
|
|
label="Step",
|
|
description="Step",
|
|
icon="square",
|
|
input_ports=(DefinitionPort(id="input", label="Input"),),
|
|
),
|
|
DefinitionNodeType(
|
|
type="end",
|
|
category="control",
|
|
label="End",
|
|
description="End",
|
|
icon="circle-stop",
|
|
input_ports=(DefinitionPort(id="input", label="Input"),),
|
|
output_ports=(),
|
|
),
|
|
),
|
|
constraints=DefinitionGraphConstraints(
|
|
allow_cycles=allow_cycles,
|
|
node_counts=(
|
|
DefinitionNodeCountConstraint(
|
|
code="graph.start_count",
|
|
label="start",
|
|
minimum=1,
|
|
maximum=1,
|
|
node_types=("start",),
|
|
),
|
|
DefinitionNodeCountConstraint(
|
|
code="graph.end_count",
|
|
label="end",
|
|
minimum=1,
|
|
node_types=("end",),
|
|
),
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
class DefinitionGraphTests(unittest.TestCase):
|
|
def test_library_driven_ports_and_counts_validate_a_definition(self) -> None:
|
|
diagnostics = validate_definition_graph(
|
|
library(allow_cycles=False),
|
|
nodes=(
|
|
DefinitionNode(id="start", type="start"),
|
|
DefinitionNode(id="step", type="step"),
|
|
DefinitionNode(id="end", type="end"),
|
|
),
|
|
edges=(
|
|
DefinitionEdge(id="edge-1", source="start", target="step"),
|
|
DefinitionEdge(id="edge-2", source="step", target="end"),
|
|
),
|
|
)
|
|
|
|
self.assertEqual((), diagnostics)
|
|
|
|
def test_constraints_change_cycle_semantics_without_changing_graph_shape(self) -> None:
|
|
nodes = (
|
|
DefinitionNode(id="start", type="start"),
|
|
DefinitionNode(id="step", type="step"),
|
|
DefinitionNode(id="retry", type="step"),
|
|
DefinitionNode(id="end", type="end"),
|
|
)
|
|
edges = (
|
|
DefinitionEdge(id="edge-1", source="start", target="step"),
|
|
DefinitionEdge(id="edge-2", source="step", target="retry"),
|
|
DefinitionEdge(id="edge-3", source="retry", target="step"),
|
|
DefinitionEdge(id="edge-4", source="retry", target="end"),
|
|
)
|
|
|
|
dag_codes = {
|
|
item.code
|
|
for item in validate_definition_graph(
|
|
library(allow_cycles=False),
|
|
nodes=nodes,
|
|
edges=edges,
|
|
)
|
|
}
|
|
workflow_codes = {
|
|
item.code
|
|
for item in validate_definition_graph(
|
|
library(allow_cycles=True),
|
|
nodes=nodes,
|
|
edges=edges,
|
|
)
|
|
}
|
|
|
|
self.assertIn("graph.cycle", dag_codes)
|
|
self.assertNotIn("graph.cycle", workflow_codes)
|
|
|
|
def test_unknown_ports_and_missing_required_inputs_are_explicit(self) -> None:
|
|
diagnostics = validate_definition_graph(
|
|
library(allow_cycles=False),
|
|
nodes=(
|
|
DefinitionNode(id="start", type="start"),
|
|
DefinitionNode(id="end", type="end"),
|
|
),
|
|
edges=(
|
|
DefinitionEdge(
|
|
id="edge",
|
|
source="start",
|
|
target="end",
|
|
target_port="missing",
|
|
),
|
|
),
|
|
)
|
|
|
|
self.assertTrue(
|
|
{"edge.unknown_target_port", "node.input_required"}.issubset(
|
|
{item.code for item in diagnostics}
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|