from __future__ import annotations from pathlib import Path from unittest import TestCase _REPO_ROOT = Path(__file__).resolve().parents[2] class TestOrderStatusRouterContracts(TestCase): """Source-code contracts for the order status PATCH endpoint.""" def _read_router_source(self) -> str: path = _REPO_ROOT / "backend/app/routers/orders.py" return path.read_text(encoding="utf-8") # ── PATCH endpoint contracts ───────────────────────────────────── def test_patch_status_endpoint_exists(self) -> None: source = self._read_router_source() self.assertIn("@router.patch(", source) self.assertIn('"/{order_id}/status"', source) def test_patch_endpoint_returns_202(self) -> None: source = self._read_router_source() patch_idx = source.index("@router.patch(") patch_block = source[patch_idx : patch_idx + 300] self.assertIn("HTTP_202_ACCEPTED", patch_block) def test_patch_endpoint_uses_order_status_update_schema(self) -> None: source = self._read_router_source() self.assertIn("OrderStatusUpdate", source) self.assertIn("data: OrderStatusUpdate", source) def test_patch_endpoint_uses_order_status_response(self) -> None: source = self._read_router_source() self.assertIn("OrderStatusResponse", source) def test_patch_endpoint_requires_tenant_auth(self) -> None: source = self._read_router_source() patch_idx = source.index("async def update_order_status") patch_block = source[patch_idx : patch_idx + 400] self.assertIn("get_current_restaurant", patch_block) def test_patch_endpoint_verifies_order_ownership(self) -> None: source = self._read_router_source() patch_idx = source.index("async def update_order_status") patch_block = source[patch_idx : patch_idx + 600] self.assertIn("Order.restaurant_id == restaurant.id", patch_block) def test_patch_endpoint_returns_404_for_missing_order(self) -> None: source = self._read_router_source() patch_idx = source.index("async def update_order_status") patch_block = source[patch_idx : patch_idx + 800] self.assertIn("HTTP_404_NOT_FOUND", patch_block) def test_patch_endpoint_dispatches_to_restate(self) -> None: source = self._read_router_source() patch_idx = source.index("async def update_order_status") patch_block = source[patch_idx : patch_idx + 1200] self.assertIn('"OrderObject"', patch_block) self.assertIn('"update_status"', patch_block) def test_patch_endpoint_invalidates_stats_cache(self) -> None: source = self._read_router_source() patch_idx = source.index("async def update_order_status") patch_block = source[patch_idx : patch_idx + 1200] self.assertIn("invalidate_restaurant_stats_cache", patch_block) class TestOrderStatusSchemaContracts(TestCase): """Source-code contracts for order status models.""" def _read_model_source(self) -> str: path = _REPO_ROOT / "backend/app/models/order.py" return path.read_text(encoding="utf-8") def test_order_status_enum_exists(self) -> None: source = self._read_model_source() self.assertIn("class OrderStatus", source) def test_order_status_enum_has_required_values(self) -> None: source = self._read_model_source() for status in ("pending", "preparing", "ready", "completed", "cancelled"): self.assertIn(f'"{status}"', source) def test_order_status_update_model_exists(self) -> None: source = self._read_model_source() self.assertIn("class OrderStatusUpdate", source) self.assertIn("new_status", source) def test_order_status_update_has_validator(self) -> None: source = self._read_model_source() self.assertIn("field_validator", source) self.assertIn('"new_status"', source) def test_order_status_response_model_exists(self) -> None: source = self._read_model_source() self.assertIn("class OrderStatusResponse", source) def test_order_status_response_has_required_fields(self) -> None: source = self._read_model_source() idx = source.index("class OrderStatusResponse") block = source[idx : idx + 300] self.assertIn("id:", block) self.assertIn("status:", block) self.assertIn("restaurant_id:", block) class TestOrderObjectTransitionContracts(TestCase): """Source-code contracts for Restate OrderObject transitions.""" def _read_handler_source(self) -> str: path = _REPO_ROOT / "restate_services/objects/order.py" return path.read_text(encoding="utf-8") def test_transition_map_has_dashboard_lifecycle(self) -> None: source = self._read_handler_source() self.assertIn('"pending"', source) self.assertIn('"preparing"', source) self.assertIn('"ready"', source) self.assertIn('"completed"', source) self.assertIn('"cancelled"', source) def test_pending_can_transition_to_preparing(self) -> None: source = self._read_handler_source() idx = source.index('"pending":') block = source[idx : idx + 100] self.assertIn('"preparing"', block) def test_preparing_can_transition_to_ready(self) -> None: source = self._read_handler_source() idx = source.index('"preparing":') block = source[idx : idx + 100] self.assertIn('"ready"', block) def test_ready_can_transition_to_completed(self) -> None: source = self._read_handler_source() idx = source.index('"ready":') block = source[idx : idx + 100] self.assertIn('"completed"', block) def test_completed_is_terminal(self) -> None: source = self._read_handler_source() self.assertIn("_TERMINAL_STATUSES", source) self.assertIn('"completed"', source) def test_cancelled_is_terminal(self) -> None: source = self._read_handler_source() self.assertIn('"cancelled"', source) def test_cancellation_allowed_from_non_terminal(self) -> None: source = self._read_handler_source() for status_key in ('"pending":', '"preparing":', '"ready":'): idx = source.index(status_key) block = source[idx : idx + 100] self.assertIn('"cancelled"', block) def test_terminal_state_error_message(self) -> None: source = self._read_handler_source() self.assertIn("already", source) self.assertIn("cannot be changed", source) def test_skipped_state_error_message(self) -> None: source = self._read_handler_source() self.assertIn("Cannot skip from", source) def test_update_status_handler_registered(self) -> None: source = self._read_handler_source() self.assertIn('@order_object.handler("update_status")', source) self.assertIn("async def update_status(", source)