98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_workflow.backend.node_library import WORKFLOW_GRAPH_LIBRARY
|
|
from govoplan_workflow.backend.schemas import WorkflowEdge, WorkflowGraph, WorkflowNode
|
|
from govoplan_workflow.backend.validation import validate_workflow_graph
|
|
|
|
|
|
class WorkflowNodeLibraryTests(unittest.TestCase):
|
|
def test_valid_workflow_graph(self) -> None:
|
|
graph = WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(id="start", type="workflow.start.manual"),
|
|
WorkflowNode(
|
|
id="work",
|
|
type="workflow.activity",
|
|
config={"title": "Check submission"},
|
|
),
|
|
WorkflowNode(id="done", type="workflow.end.completed"),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(id="e1", source="start", target="work"),
|
|
WorkflowEdge(id="e2", source="work", target="done"),
|
|
],
|
|
)
|
|
|
|
self.assertEqual(validate_workflow_graph(graph), ())
|
|
|
|
def test_correction_loop_is_allowed(self) -> None:
|
|
graph = WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(id="start", type="workflow.start.manual"),
|
|
WorkflowNode(
|
|
id="work",
|
|
type="workflow.activity",
|
|
config={"title": "Prepare"},
|
|
),
|
|
WorkflowNode(
|
|
id="review",
|
|
type="workflow.review",
|
|
config={"title": "Review"},
|
|
),
|
|
WorkflowNode(id="done", type="workflow.end.completed"),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(id="e1", source="start", target="work"),
|
|
WorkflowEdge(id="e2", source="work", target="review"),
|
|
WorkflowEdge(
|
|
id="e3",
|
|
source="review",
|
|
source_port="changes",
|
|
target="work",
|
|
),
|
|
WorkflowEdge(
|
|
id="e4",
|
|
source="review",
|
|
source_port="approved",
|
|
target="done",
|
|
),
|
|
],
|
|
)
|
|
|
|
self.assertNotIn(
|
|
"graph.cycle",
|
|
{item.code for item in validate_workflow_graph(graph)},
|
|
)
|
|
|
|
def test_constraints_and_required_configuration_are_reported(self) -> None:
|
|
graph = WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(id="start-1", type="workflow.start.manual"),
|
|
WorkflowNode(
|
|
id="start-2",
|
|
type="workflow.start.event",
|
|
config={"event_type": ""},
|
|
),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(id="e1", source="start-1", target="start-2"),
|
|
],
|
|
)
|
|
|
|
diagnostics = validate_workflow_graph(graph)
|
|
codes = {item.code for item in diagnostics}
|
|
self.assertIn("graph.trigger_count", codes)
|
|
self.assertIn("graph.outcome_count", codes)
|
|
self.assertIn("node.config_required", codes)
|
|
self.assertIn("node.outgoing_required", codes)
|
|
|
|
def test_library_has_domain_specific_cycle_policy(self) -> None:
|
|
self.assertTrue(WORKFLOW_GRAPH_LIBRARY.constraints.allow_cycles)
|
|
self.assertEqual(WORKFLOW_GRAPH_LIBRARY.id, "workflow")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|