Status rule

status.py

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

test_status.py

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