73 lines
3.4 KiB
Python
73 lines
3.4 KiB
Python
from decimal import Decimal
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
|
|
from apps.accounts.models import Customer
|
|
from apps.core.errors import ApiError
|
|
from apps.orders.services import OrderService
|
|
from apps.pricing.models import PricingRule
|
|
from apps.reservations.models import Facility, Reservation, Slot
|
|
from apps.workflows.engine import WorkflowEngine
|
|
from apps.workflows.models import Workflow, WorkflowStep, WorkflowTransition, WorkflowVersion
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
def make_active_workflow():
|
|
wf = Workflow.objects.create(name="W", slug="w-order")
|
|
v = WorkflowVersion.objects.create(workflow=wf, version=1)
|
|
s1 = WorkflowStep.objects.create(version=v, key="start", name="Start", order=1, is_start=True)
|
|
wash = WorkflowStep.objects.create(version=v, key="washing", name="Washing", order=2, required=True)
|
|
end = WorkflowStep.objects.create(version=v, key="end", name="End", order=3, is_terminal=True)
|
|
WorkflowTransition.objects.create(version=v, from_step=s1, to_step=wash)
|
|
WorkflowTransition.objects.create(version=v, from_step=wash, to_step=end)
|
|
WorkflowEngine().publish(wf, v)
|
|
PricingRule.objects.create(step=wash, name="Wash", rule_type="per_kg", amount=Decimal("5000"))
|
|
return wf, v
|
|
|
|
|
|
class OrderServiceTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="c", password="***")
|
|
self.customer = Customer.objects.create(user=self.user)
|
|
self.facility = Facility.objects.create(name="F")
|
|
self.slot = Slot.objects.create(
|
|
facility=self.facility, date=timezone.localdate(), start_time="08:00", end_time="08:30", capacity=5
|
|
)
|
|
self.reservation = Reservation.objects.create(customer=self.customer, slot=self.slot, status="confirmed")
|
|
make_active_workflow()
|
|
self.service = OrderService()
|
|
|
|
def test_full_lifecycle(self):
|
|
order = self.service.create_from_reservation(self.reservation)
|
|
order = self.service.record_weighing(order, "1000", "80")
|
|
self.assertEqual(order.net_weight_kg, Decimal("920"))
|
|
order = self.service.select_steps(order, ["washing"])
|
|
self.assertEqual(order.status, "awaiting_confirmation")
|
|
self.assertEqual(order.price_snapshot["total"], "4600000.00")
|
|
order = self.service.confirm(order)
|
|
self.assertEqual(order.status, "confirmed")
|
|
self.assertIsNotNone(order.confirmed_at)
|
|
|
|
def test_required_step_missing(self):
|
|
order = self.service.create_from_reservation(self.reservation)
|
|
order = self.service.record_weighing(order, "1000", "0")
|
|
with self.assertRaises(ApiError) as ctx:
|
|
self.service.select_steps(order, ["end"])
|
|
self.assertEqual(ctx.exception.code, "REQUIRED_STEP_MISSING")
|
|
|
|
def test_weighing_immutable(self):
|
|
order = self.service.create_from_reservation(self.reservation)
|
|
order = self.service.record_weighing(order, "1000", "80")
|
|
with self.assertRaises(ApiError) as ctx:
|
|
self.service.record_weighing(order, "900", "0")
|
|
self.assertEqual(ctx.exception.code, "WEIGHING_IMMUTABLE")
|
|
|
|
def test_duplicate_order_for_reservation(self):
|
|
self.service.create_from_reservation(self.reservation)
|
|
with self.assertRaises(ApiError) as ctx:
|
|
self.service.create_from_reservation(self.reservation)
|
|
self.assertEqual(ctx.exception.code, "ORDER_EXISTS")
|