147 lines
5.2 KiB
Python
147 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_workflow_engine.backend.node_library import (
|
|
BPMN_NODE_TYPES,
|
|
WORKFLOW_GRAPH_LIBRARY,
|
|
)
|
|
from govoplan_workflow_engine.backend.schemas import WorkflowEdge, WorkflowGraph, WorkflowNode
|
|
from govoplan_workflow_engine.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")
|
|
self.assertEqual("1.0.0", WORKFLOW_GRAPH_LIBRARY.version)
|
|
activity = WORKFLOW_GRAPH_LIBRARY.node_type("workflow.activity")
|
|
self.assertIn(
|
|
"view_surface_ids",
|
|
{field.id for field in activity.config_fields},
|
|
)
|
|
|
|
def test_native_palette_uses_standard_bpmn_vocabulary(self) -> None:
|
|
node_types = {item.type for item in BPMN_NODE_TYPES}
|
|
|
|
self.assertIn("bpmn.startEvent", node_types)
|
|
self.assertIn("bpmn.userTask", node_types)
|
|
self.assertIn("bpmn.exclusiveGateway", node_types)
|
|
self.assertIn("bpmn.participant", node_types)
|
|
self.assertIn("bpmn.textAnnotation", node_types)
|
|
self.assertTrue(all(item.startswith("bpmn.") for item in node_types))
|
|
|
|
def test_bpmn_rejects_multiple_default_flows(self) -> None:
|
|
graph = WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(id="start", type="bpmn.startEvent"),
|
|
WorkflowNode(id="choice", type="bpmn.exclusiveGateway"),
|
|
WorkflowNode(id="end-a", type="bpmn.endEvent"),
|
|
WorkflowNode(id="end-b", type="bpmn.endEvent"),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(id="to-choice", source="start", target="choice"),
|
|
WorkflowEdge(
|
|
id="default-a",
|
|
source="choice",
|
|
target="end-a",
|
|
config={"default": True},
|
|
),
|
|
WorkflowEdge(
|
|
id="default-b",
|
|
source="choice",
|
|
target="end-b",
|
|
config={"default": True},
|
|
),
|
|
],
|
|
)
|
|
|
|
self.assertIn(
|
|
"bpmn.multiple_default_flows",
|
|
{item.code for item in validate_workflow_graph(graph)},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|