GovOPlaN/govoplan-workflow
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Implement native BPMN workflows and guided modes

Browse Source
This commit is contained in:
zemion
2026-07-31 02:48:57 +02:00
parent c505e81006
commit f4974b4949
40 changed files with 8203 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_service.py
+202
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import unittest
from pathlib import Path
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
@@ -11,6 +12,7 @@ from govoplan_workflow.backend.db.models import (
WorkflowDefinitionRevision,
)
from govoplan_workflow.backend.schemas import (
BpmnRevisionInput,
WorkflowDefinitionCreateRequest,
WorkflowDefinitionUpdateRequest,
WorkflowEdge,
@@ -19,6 +21,7 @@ from govoplan_workflow.backend.schemas import (
WorkflowPosition,
)
from govoplan_workflow.backend.service import (
WorkflowBpmnValidationError,
WorkflowConflictError,
WorkflowNotFoundError,
activate_definition,
@@ -29,6 +32,18 @@ from govoplan_workflow.backend.service import (
list_definitions,
update_definition,
)
from govoplan_workflow.backend.bpmn_adapters import (
INTERCHANGE_ADAPTER_ID,
NATIVE_LINEAR_ADAPTER_ID,
)
from govoplan_workflow.backend.bpmn import inspect_bpmn_xml
from govoplan_workflow.backend.bpmn_graph import NATIVE_BPMN_ADAPTER_ID
try:
from test_bpmn import BPMN, NATIVE_BPMN
except ModuleNotFoundError as exc:
if exc.name != "test_bpmn":
raise
from tests.test_bpmn import BPMN, NATIVE_BPMN
def sample_graph(*, title: str = "Review request") -> WorkflowGraph:
@@ -174,6 +189,70 @@ class WorkflowServiceTests(unittest.TestCase):
),
)
def test_execution_mode_and_view_pin_are_immutable_revision_content(
self,
) -> None:
definition = create_definition(
self.session,
tenant_id="tenant-1",
actor_id="user-1",
payload=WorkflowDefinitionCreateRequest(
name="Guided review",
graph=sample_graph(),
execution_mode="guided",
view_id="view-1",
view_revision_id="view-revision-1",
),
)
updated = update_definition(
self.session,
tenant_id="tenant-1",
definition_id=definition.id,
actor_id="user-2",
payload=WorkflowDefinitionUpdateRequest(
name="Guided review",
graph=sample_graph(),
expected_revision=1,
execution_mode="hybrid",
view_id="view-1",
view_revision_id="view-revision-1",
),
)
revisions = list_definition_revisions(
self.session,
definition=updated,
)
self.assertEqual(2, updated.current_revision)
self.assertEqual("hybrid", revisions[0].execution_mode)
self.assertEqual("guided", revisions[1].execution_mode)
self.assertEqual("view-revision-1", revisions[1].view_revision_id)
self.assertNotEqual(revisions[0].content_hash, revisions[1].content_hash)
def test_automated_mode_rejects_human_handoff_paths(self) -> None:
definition = create_definition(
self.session,
tenant_id="tenant-1",
actor_id="user-1",
payload=WorkflowDefinitionCreateRequest(
name="Invalid automation",
graph=sample_graph(),
execution_mode="automated",
allow_automation=True,
),
)
with self.assertRaisesRegex(
WorkflowConflictError,
"human handoff paths",
):
activate_definition(
self.session,
tenant_id="tenant-1",
definition_id=definition.id,
actor_id="user-1",
)
def test_stale_update_and_cross_tenant_access_are_rejected(self) -> None:
definition = self._create()
with self.assertRaises(WorkflowConflictError):
@@ -220,6 +299,129 @@ class WorkflowServiceTests(unittest.TestCase):
),
)
def test_bpmn_xml_and_adapter_are_pinned_to_immutable_revisions(self) -> None:
definition = create_definition(
self.session,
tenant_id="tenant-1",
actor_id="user-1",
payload=WorkflowDefinitionCreateRequest(
name="BPMN review",
graph=sample_graph(),
bpmn=BpmnRevisionInput(
xml=NATIVE_BPMN,
adapter_id=NATIVE_LINEAR_ADAPTER_ID,
adapter_version="1.0.0",
),
),
)
self.session.commit()
first = list_definition_revisions(
self.session,
definition=definition,
)[0]
self.assertTrue(inspect_bpmn_xml(first.bpmn_xml or "").valid_xml)
self.assertEqual(NATIVE_BPMN_ADAPTER_ID, first.bpmn_adapter_id)
self.assertEqual("1.0.0", first.bpmn_adapter_version)
self.assertEqual("native_graph", first.bpmn_runtime_kind)
self.assertEqual(
"Review request",
first.graph["nodes"][1]["config"]["title"],
)
changed_xml = NATIVE_BPMN.replace(
"Review request",
"Review corrected request",
)
updated = update_definition(
self.session,
tenant_id="tenant-1",
definition_id=definition.id,
actor_id="user-2",
payload=WorkflowDefinitionUpdateRequest(
name="BPMN review",
graph=sample_graph(),
bpmn=BpmnRevisionInput(
xml=changed_xml,
adapter_id=NATIVE_LINEAR_ADAPTER_ID,
),
expected_revision=1,
),
)
activate_definition(
self.session,
tenant_id="tenant-1",
definition_id=definition.id,
actor_id="user-2",
revision=2,
)
self.session.commit()
revisions = list_definition_revisions(
self.session,
definition=updated,
)
self.assertEqual(2, updated.current_revision)
self.assertEqual(2, updated.active_revision)
self.assertIn("Review corrected request", revisions[0].bpmn_xml or "")
self.assertNotEqual(revisions[0].content_hash, revisions[1].content_hash)
def test_model_only_bpmn_revision_fails_closed_on_activation(self) -> None:
definition = create_definition(
self.session,
tenant_id="tenant-1",
actor_id="user-1",
payload=WorkflowDefinitionCreateRequest(
name="Interchange model",
graph=sample_graph(),
bpmn=BpmnRevisionInput(
xml=BPMN,
adapter_id=INTERCHANGE_ADAPTER_ID,
),
),
)
self.session.commit()
with self.assertRaisesRegex(
WorkflowBpmnValidationError,
"exclusive gateway",
):
activate_definition(
self.session,
tenant_id="tenant-1",
definition_id=definition.id,
actor_id="user-1",
)
def test_interchange_revision_preserves_extension_xml_exactly(self) -> None:
xml = (
Path(__file__).parent
/ "fixtures"
/ "bpmn"
/ "process.bpmn"
).read_text(encoding="utf-8")
definition = create_definition(
self.session,
tenant_id="tenant-1",
actor_id="user-1",
payload=WorkflowDefinitionCreateRequest(
name="Extended interchange model",
graph=sample_graph(),
bpmn=BpmnRevisionInput(
xml=xml,
adapter_id=INTERCHANGE_ADAPTER_ID,
adapter_version="1.0.0",
),
),
)
revision = list_definition_revisions(
self.session,
definition=definition,
)[0]
self.assertTrue(inspect_bpmn_xml(revision.bpmn_xml or "").valid_xml)
self.assertIn("fixture revision=\"1\"", revision.bpmn_xml or "")
if __name__ == "__main__":
unittest.main()