Skip to content

Instantly share code, notes, and snippets.

@isaquealves
Last active July 17, 2023 20:25
Show Gist options
  • Save isaquealves/a98f624b8216d65523c91a39c4771c2b to your computer and use it in GitHub Desktop.
Save isaquealves/a98f624b8216d65523c91a39c4771c2b to your computer and use it in GitHub Desktop.
Status rule
from dataclasses import dataclass
from datetime import datetime
import enum
from functools import reduce
from typing import List, Optional
import uuid
class StatusEnum(enum.Enum):
PENDING: str = "PENDING"
APPROVED: str = "APPROVED"
INVOICED: str = "INVOICED"
SHIPPED: str = "SHIPPED"
DELIVERED: str = "DELIVERED"
CANCELED: str = "CANCELED"
@dataclass
class Status:
name: str
previous: Optional["Status"]
immediate_next_valid: Optional[str] = None
is_final: bool = False
def __repr__(self) -> str:
return f"Status(name={self.name})"
canceled = Status(name=StatusEnum.CANCELED.value, previous=None, is_final=True)
pending = Status(name=StatusEnum.PENDING.value, previous=None, immediate_next_valid=StatusEnum.APPROVED.value)
approved = Status(name=StatusEnum.APPROVED.value, previous=pending, immediate_next_valid=StatusEnum.INVOICED.value)
invoiced = Status(name=StatusEnum.INVOICED.value, previous=approved, immediate_next_valid=StatusEnum.SHIPPED.value)
shipped = Status(name=StatusEnum.SHIPPED.value, previous=invoiced, immediate_next_valid=StatusEnum.DELIVERED.value)
delivered = Status(name=StatusEnum.DELIVERED.value, previous=shipped, immediate_next_valid=None, is_final=True)
class StatusesRule:
happy_steps = (pending, approved, invoiced, shipped, delivered)
sad_steps = (canceled,)
def find_by_name(self, name: str) -> Optional[Status]:
return reduce(lambda acc, x: x if x.name == name else acc, self.happy_steps + self.sad_steps, None)
def is_valid(self, current_status, next_status) -> bool:
return (
self.find_by_name(current_status).immediate_next_valid == next_status or # type: ignore
next_status in [step.name for step in self.sad_steps]
)
@dataclass
class StatusEvent:
name: str
date: datetime
status: Status
def __repr__(self):
return f"StatusEvent(name={self.name}, date={self.date}, status={self.status})"
class EventsStorage:
__data: List[StatusEvent]
def __init__(self):
self.__data = []
def add(self, event: StatusEvent):
self.__data.append(event)
def __repr__(self):
return f"Events(data={self.__data})"
def all(self):
return self.__data
class Order:
rule = StatusesRule()
events: EventsStorage
id: uuid.UUID
_status: str = None
def __init__(self, events: Events, status: Optional[str] = None) -> None:
self.id = uuid.uuid4()
self.events = events
self.status = StatusEnum.PENDING.value if not status else status
@property
def status(self) -> str:
return self._status or StatusEnum.PENDING.value
@status.setter
def status(self, value):
self.events.add(
StatusEvent(name="set status", date=datetime.now(), status=self.rule.find_by_name(value))
)
if not self._status:
self._status = value
if self.rule.is_valid(self._status, value):
self._status = value
class StatusService:
def __init__(self, order: Order):
self.order = order
def set_status(self, name):
self.order.status = name
from random import randint
from .status import Order, StatusesRule
from .status import StatusEnum
from .status import StatusService
from .status import EventsStorage
class TestStatusFlow:
def setup_class(cls):
rand_status = StatusesRule.happy_steps[randint(0, len(StatusesRule.happy_steps) - 1)].name
cls.order_one = Order(events=EventsStorage(), status=StatusEnum.PENDING.value)
cls.order_two = Order(events=EventsStorage(), status=StatusEnum.APPROVED.value)
cls.order_three = Order(events=EventsStorage(), status=rand_status)
def test_happy_flow(self):
service = StatusService(self.order_one)
service.set_status(StatusEnum.APPROVED.value)
service.set_status(StatusEnum.INVOICED.value)
service.set_status(StatusEnum.SHIPPED.value)
service.set_status(StatusEnum.DELIVERED.value)
service.set_status(StatusEnum.SHIPPED.value)
service.set_status(StatusEnum.SHIPPED.value)
assert self.order_one.status == StatusEnum.DELIVERED.value
assert len(service.order.events.all()) == 7
assert service.order.events.all()[0].status.name == StatusEnum.PENDING.value
assert service.order.events.all()[1].status.name == StatusEnum.APPROVED.value
assert service.order.events.all()[2].status.name == StatusEnum.INVOICED.value
assert service.order.events.all()[3].status.name == StatusEnum.SHIPPED.value
assert service.order.events.all()[4].status.name == StatusEnum.DELIVERED.value
assert service.order.events.all()[5].status.name == StatusEnum.SHIPPED.value
assert service.order.events.all()[6].status.name == StatusEnum.SHIPPED.value
def test_cancel_order_two_approved(self):
service = StatusService(self.order_two)
service.set_status(StatusEnum.CANCELED.value)
service.set_status(StatusEnum.INVOICED.value)
service.set_status(StatusEnum.SHIPPED.value)
service.set_status(StatusEnum.DELIVERED.value)
assert self.order_two.status == StatusEnum.CANCELED.value
assert len(service.order.events.all()) == 5
def test_cancel_order_two_invoiced(self):
self.order_two.status = StatusEnum.INVOICED.value
service = StatusService(self.order_two)
service.set_status(StatusEnum.CANCELED.value)
service.set_status(StatusEnum.SHIPPED.value)
service.set_status(StatusEnum.DELIVERED.value)
assert self.order_two.status == StatusEnum.CANCELED.value
assert len(service.order.events.all()) == 9
def test_cancel_order_cancel_three_rand_status(self):
service = StatusService(self.order_three)
service.set_status(StatusEnum.CANCELED.value)
service.set_status(StatusEnum.SHIPPED.value)
service.set_status(StatusEnum.DELIVERED.value)
assert self.order_three.status == StatusEnum.CANCELED.value
assert len(service.order.events.all()) == 4
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment