76 lines
3.4 KiB
Python
76 lines
3.4 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
|
|
from apps.accounts.models import Customer
|
|
from apps.orders.models import ProcessingOrder
|
|
from apps.workflows.engine import WorkflowEngine
|
|
from apps.workflows.models import Workflow, WorkflowStep, WorkflowTransition, WorkflowVersion
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
def make_workflow(slug="test"):
|
|
wf = Workflow.objects.create(name="Test", slug=slug)
|
|
v = WorkflowVersion.objects.create(workflow=wf, version=1, status="draft")
|
|
start = WorkflowStep.objects.create(version=v, key="start", name="Start", order=1, is_start=True)
|
|
mid = WorkflowStep.objects.create(version=v, key="mid", name="Mid", order=2)
|
|
end = WorkflowStep.objects.create(version=v, key="end", name="End", order=3, is_terminal=True)
|
|
WorkflowTransition.objects.create(version=v, from_step=start, to_step=mid)
|
|
WorkflowTransition.objects.create(version=v, from_step=mid, to_step=end)
|
|
return wf, v, start, mid, end
|
|
|
|
|
|
class WorkflowEngineTests(TestCase):
|
|
def test_publish_valid_version(self):
|
|
wf, v, *_ = make_workflow()
|
|
WorkflowEngine().publish(wf, v)
|
|
wf.refresh_from_db()
|
|
v.refresh_from_db()
|
|
self.assertEqual(wf.status, "active")
|
|
self.assertEqual(v.status, "active")
|
|
self.assertEqual(wf.active_version, v)
|
|
|
|
def test_publish_rejects_missing_terminal(self):
|
|
wf = Workflow.objects.create(name="T2", slug="t2")
|
|
v = WorkflowVersion.objects.create(workflow=wf, version=1)
|
|
WorkflowStep.objects.create(version=v, key="start", name="Start", is_start=True)
|
|
WorkflowStep.objects.create(version=v, key="mid", name="Mid")
|
|
with self.assertRaises(Exception) as ctx:
|
|
WorkflowEngine().publish(wf, v)
|
|
errors = ctx.exception.details.get("errors", [])
|
|
self.assertTrue(any("terminal" in e for e in errors), errors)
|
|
|
|
def test_available_next_steps_respect_conditions(self):
|
|
wf, v, start, mid, end = make_workflow()
|
|
WorkflowTransition.objects.create(
|
|
version=v, from_step=start, to_step=end,
|
|
conditions=[{"type": "step_selected", "step": "mid"}],
|
|
)
|
|
engine = WorkflowEngine()
|
|
keys = {s.key for s in engine.available_next_steps(v, start, {"selected_step_keys": []})}
|
|
self.assertIn("mid", keys)
|
|
self.assertNotIn("end", keys)
|
|
keys2 = {s.key for s in engine.available_next_steps(v, start, {"selected_step_keys": ["mid"]})}
|
|
self.assertIn("end", keys2)
|
|
|
|
|
|
class WorkflowVersioningTests(TestCase):
|
|
def test_order_keeps_its_version_after_new_publish(self):
|
|
wf, v, *_ = make_workflow()
|
|
WorkflowEngine().publish(wf, v)
|
|
user = User.objects.create_user(username="cust", password="x")
|
|
customer = Customer.objects.create(user=user)
|
|
order = ProcessingOrder.objects.create(customer=customer, workflow_version=v)
|
|
|
|
# Publish a second version.
|
|
v2 = WorkflowVersion.objects.create(workflow=wf, version=2, status="draft")
|
|
s2 = WorkflowStep.objects.create(version=v2, key="start", name="Start", is_start=True)
|
|
e2 = WorkflowStep.objects.create(version=v2, key="end", name="End", is_terminal=True)
|
|
WorkflowTransition.objects.create(version=v2, from_step=s2, to_step=e2)
|
|
WorkflowEngine().publish(wf, v2)
|
|
|
|
wf.refresh_from_db()
|
|
order.refresh_from_db()
|
|
self.assertEqual(wf.active_version, v2)
|
|
self.assertEqual(order.workflow_version, v)
|