Port Round 2 to Python (OrderBook): 14 tests total green, START state preserved
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
from order_book_domain.order_book import Order, OrderBook, OrderStatus
|
||||
|
||||
__all__ = ["Order", "OrderBook", "OrderStatus"]
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from enum import Enum, auto
|
||||
|
||||
|
||||
class OrderStatus(Enum):
|
||||
PENDING = auto()
|
||||
WAITING_FOR_PAYMENT = auto()
|
||||
FULFILLING = auto()
|
||||
DELIVERED = auto()
|
||||
CANCELLED = auto()
|
||||
|
||||
|
||||
class Order:
|
||||
# Constructed by OrderBook only.
|
||||
def __init__(self, id: int, customer: str) -> None:
|
||||
self.id = id
|
||||
self.customer = customer
|
||||
self.status = OrderStatus.PENDING
|
||||
self.finalized_at: datetime | None = None
|
||||
|
||||
|
||||
class OrderBook:
|
||||
def __init__(self, payment_window: timedelta | None = None) -> None:
|
||||
self.payment_window = payment_window or timedelta(days=14)
|
||||
self._orders_by_id: dict[int, Order] = {}
|
||||
self._next_id = 1
|
||||
|
||||
@property
|
||||
def orders(self) -> list[Order]:
|
||||
return list(self._orders_by_id.values())
|
||||
|
||||
def open_order(self, customer: str) -> Order:
|
||||
order = Order(self._next_id, customer)
|
||||
self._next_id += 1
|
||||
self._orders_by_id[order.id] = order
|
||||
return order
|
||||
|
||||
def get_order(self, order_id: int) -> Order:
|
||||
try:
|
||||
return self._orders_by_id[order_id]
|
||||
except KeyError:
|
||||
raise ValueError(f"Unknown order id {order_id}.") from None
|
||||
|
||||
def finalize_order(self, order_id: int, finalized_at: datetime) -> None:
|
||||
order = self.get_order(order_id)
|
||||
_require_status(order, OrderStatus.PENDING, "finalize")
|
||||
order.status = OrderStatus.WAITING_FOR_PAYMENT
|
||||
order.finalized_at = finalized_at
|
||||
|
||||
def pay_order(self, order_id: int, paid_at: datetime) -> None:
|
||||
order = self.get_order(order_id)
|
||||
_require_status(order, OrderStatus.WAITING_FOR_PAYMENT, "pay")
|
||||
if paid_at - order.finalized_at > self.payment_window:
|
||||
order.status = OrderStatus.CANCELLED
|
||||
raise ValueError(
|
||||
f"Payment for order {order_id} arrived after the payment window; the order was cancelled."
|
||||
)
|
||||
order.status = OrderStatus.FULFILLING
|
||||
|
||||
def ship_order(self, order_id: int) -> None:
|
||||
order = self.get_order(order_id)
|
||||
_require_status(order, OrderStatus.FULFILLING, "ship")
|
||||
order.status = OrderStatus.DELIVERED
|
||||
|
||||
def cancel_order(self, order_id: int) -> None:
|
||||
order = self.get_order(order_id)
|
||||
if order.status not in (OrderStatus.PENDING, OrderStatus.WAITING_FOR_PAYMENT):
|
||||
raise ValueError(f"Cannot cancel order {order_id} in status {order.status}.")
|
||||
order.status = OrderStatus.CANCELLED
|
||||
|
||||
|
||||
def _require_status(order: Order, expected: OrderStatus, action: str) -> None:
|
||||
if order.status is not expected:
|
||||
raise ValueError(
|
||||
f"Cannot {action} order {order.id} in status {order.status} (expected {expected})."
|
||||
)
|
||||
@@ -0,0 +1,104 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from order_book_domain import OrderBook, OrderStatus
|
||||
|
||||
DAY0 = datetime(2025, 6, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def at(day: int) -> datetime:
|
||||
return DAY0 + timedelta(days=day)
|
||||
|
||||
|
||||
def test_finalize_then_pay_moves_order_through_two_states():
|
||||
book = OrderBook()
|
||||
order = book.open_order("anna")
|
||||
|
||||
book.finalize_order(order.id, at(1))
|
||||
assert book.get_order(order.id).status is OrderStatus.WAITING_FOR_PAYMENT
|
||||
|
||||
book.pay_order(order.id, at(2))
|
||||
assert book.get_order(order.id).status is OrderStatus.FULFILLING
|
||||
|
||||
|
||||
def test_pay_then_ship_delivers_the_order():
|
||||
book = OrderBook()
|
||||
order = book.open_order("anna")
|
||||
|
||||
book.finalize_order(order.id, at(1))
|
||||
assert book.get_order(order.id).status is OrderStatus.WAITING_FOR_PAYMENT
|
||||
|
||||
book.pay_order(order.id, at(2))
|
||||
book.ship_order(order.id)
|
||||
assert book.get_order(order.id).status is OrderStatus.DELIVERED
|
||||
|
||||
|
||||
def test_pay_within_window_then_late_pay_on_second_order_cancels_only_that_order():
|
||||
book = OrderBook()
|
||||
first = book.open_order("anna")
|
||||
second = book.open_order("bram")
|
||||
|
||||
book.finalize_order(first.id, at(1))
|
||||
book.finalize_order(second.id, at(1))
|
||||
assert len(book.orders) == 2
|
||||
|
||||
book.pay_order(first.id, at(5))
|
||||
assert book.get_order(first.id).status is OrderStatus.FULFILLING
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
book.pay_order(second.id, at(30))
|
||||
assert book.get_order(second.id).status is OrderStatus.CANCELLED
|
||||
|
||||
|
||||
def test_pay_before_finalize_throws_then_finalize_still_works():
|
||||
book = OrderBook()
|
||||
order = book.open_order("anna")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
book.pay_order(order.id, at(1))
|
||||
assert book.get_order(order.id).status is OrderStatus.PENDING
|
||||
|
||||
book.finalize_order(order.id, at(2))
|
||||
assert book.get_order(order.id).status is OrderStatus.WAITING_FOR_PAYMENT
|
||||
|
||||
|
||||
def test_cancel_pending_order_then_pay_after_cancel_throws_again():
|
||||
book = OrderBook()
|
||||
order = book.open_order("anna")
|
||||
|
||||
book.cancel_order(order.id)
|
||||
assert book.get_order(order.id).status is OrderStatus.CANCELLED
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
book.pay_order(order.id, at(1))
|
||||
assert book.get_order(order.id).status is OrderStatus.CANCELLED
|
||||
|
||||
|
||||
def test_shipped_order_cannot_be_cancelled():
|
||||
book = OrderBook()
|
||||
order = book.open_order("anna")
|
||||
|
||||
book.finalize_order(order.id, at(1))
|
||||
book.pay_order(order.id, at(2))
|
||||
book.ship_order(order.id)
|
||||
assert book.get_order(order.id).status is OrderStatus.DELIVERED
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
book.cancel_order(order.id)
|
||||
assert book.get_order(order.id).status is OrderStatus.DELIVERED
|
||||
|
||||
|
||||
def test_two_customers_interleave_transitions_independently():
|
||||
book = OrderBook()
|
||||
annas = book.open_order("anna")
|
||||
brams = book.open_order("bram")
|
||||
|
||||
book.finalize_order(brams.id, at(1))
|
||||
assert book.get_order(annas.id).status is OrderStatus.PENDING
|
||||
assert book.get_order(brams.id).status is OrderStatus.WAITING_FOR_PAYMENT
|
||||
|
||||
book.pay_order(brams.id, at(2))
|
||||
book.cancel_order(annas.id)
|
||||
assert book.get_order(annas.id).status is OrderStatus.CANCELLED
|
||||
assert book.get_order(brams.id).status is OrderStatus.FULFILLING
|
||||
Reference in New Issue
Block a user