from __future__ import annotations from pathlib import Path from unittest import TestCase _REPO_ROOT = Path(__file__).resolve().parents[2] class TestReservationApiContracts(TestCase): def _read_router_source(self) -> str: path = _REPO_ROOT / "backend/app/routers/reservations.py" return path.read_text(encoding="utf-8") # ── PATCH endpoint contracts ─────────────────────────────────── def test_patch_endpoint_exists(self) -> None: source = self._read_router_source() self.assertIn("@router.patch(", source) self.assertIn('"/{reservation_id}"', source) def test_patch_endpoint_uses_restate_proxy_update(self) -> None: source = self._read_router_source() self.assertIn('"ReservationObject"', source) self.assertIn('"update"', 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 + 200] self.assertIn("HTTP_202_ACCEPTED", patch_block) def test_patch_endpoint_uses_reservation_patch_schema(self) -> None: source = self._read_router_source() self.assertIn("ReservationPatch", source) self.assertIn("data: ReservationPatch", source) def test_patch_endpoint_invalidates_stats_cache(self) -> None: source = self._read_router_source() self.assertIn("invalidate_restaurant_stats_cache", source) def test_patch_endpoint_refreshes_from_db(self) -> None: """PATCH should re-read from DB to return committed state.""" source = self._read_router_source() patch_idx = source.index("async def patch_reservation") patch_block = source[patch_idx:] next_fn = patch_block.find("async def ", 10) if next_fn > 0: patch_block = patch_block[:next_fn] self.assertIn("select(Reservation)", patch_block) self.assertIn("scalar_one_or_none", patch_block) def test_patch_endpoint_exclude_unset(self) -> None: """PATCH should only send fields that were actually provided.""" source = self._read_router_source() self.assertIn("exclude_unset=True", source) # ── Approve endpoint contracts ──────────────────────────────── def test_approve_endpoint_exists(self) -> None: source = self._read_router_source() self.assertIn('"/{reservation_id}/approve"', source) self.assertIn("async def approve_reservation", source) def test_approve_endpoint_returns_202(self) -> None: source = self._read_router_source() approve_idx = source.index('"/{reservation_id}/approve"') approve_block = source[approve_idx - 200 : approve_idx + 200] self.assertIn("HTTP_202_ACCEPTED", approve_block) def test_approve_endpoint_uses_restate_approve_handler(self) -> None: source = self._read_router_source() approve_idx = source.index("async def approve_reservation") approve_block = source[approve_idx : approve_idx + 800] self.assertIn('"approve"', approve_block) self.assertIn('"ReservationObject"', approve_block) def test_approve_endpoint_returns_action_response(self) -> None: source = self._read_router_source() approve_idx = source.index('"/{reservation_id}/approve"') approve_block = source[approve_idx - 50 : approve_idx + 500] self.assertIn("ReservationActionResponse", approve_block) def test_approve_endpoint_requires_tenant_auth(self) -> None: source = self._read_router_source() approve_idx = source.index("async def approve_reservation") approve_block = source[approve_idx : approve_idx + 300] self.assertIn("get_current_restaurant", approve_block) # ── Reject endpoint contracts ───────────────────────────────── def test_reject_endpoint_exists(self) -> None: source = self._read_router_source() self.assertIn('"/{reservation_id}/reject"', source) self.assertIn("async def reject_reservation", source) def test_reject_endpoint_returns_202(self) -> None: source = self._read_router_source() reject_idx = source.index('"/{reservation_id}/reject"') reject_block = source[reject_idx - 200 : reject_idx + 200] self.assertIn("HTTP_202_ACCEPTED", reject_block) def test_reject_endpoint_uses_restate_reject_handler(self) -> None: source = self._read_router_source() reject_idx = source.index("async def reject_reservation") reject_block = source[reject_idx : reject_idx + 800] self.assertIn('"reject"', reject_block) self.assertIn('"ReservationObject"', reject_block) def test_reject_endpoint_returns_action_response(self) -> None: source = self._read_router_source() reject_idx = source.index('"/{reservation_id}/reject"') reject_block = source[reject_idx - 50 : reject_idx + 500] self.assertIn("ReservationActionResponse", reject_block) def test_reject_endpoint_requires_tenant_auth(self) -> None: source = self._read_router_source() reject_idx = source.index("async def reject_reservation") reject_block = source[reject_idx : reject_idx + 300] self.assertIn("get_current_restaurant", reject_block) # ── Schema contracts ────────────────────────────────────────── def test_reservation_patch_schema_exists(self) -> None: source = self._read_router_source() self.assertIn("from app.models.reservation import", source) self.assertIn("ReservationPatch", source) def test_reservation_action_response_schema_exists(self) -> None: source = self._read_router_source() self.assertIn("ReservationActionResponse", source) def test_reservation_patch_model_is_defined(self) -> None: """ReservationPatch model should be importable from the models module.""" model_path = _REPO_ROOT / "backend/app/models/reservation.py" source = model_path.read_text(encoding="utf-8") self.assertIn("class ReservationPatch", source) def test_reservation_action_response_model_is_defined(self) -> None: """ReservationActionResponse model should be importable from the models module.""" model_path = _REPO_ROOT / "backend/app/models/reservation.py" source = model_path.read_text(encoding="utf-8") self.assertIn("class ReservationActionResponse", source) def test_action_response_has_id_and_status(self) -> None: """ReservationActionResponse must have id and status fields.""" model_path = _REPO_ROOT / "backend/app/models/reservation.py" source = model_path.read_text(encoding="utf-8") idx = source.index("class ReservationActionResponse") block = source[idx : idx + 300] self.assertIn("id:", block) self.assertIn("status:", block) # ── Restate handler registration contracts ──────────────────── def test_restate_approve_handler_registered(self) -> None: handler_path = _REPO_ROOT / "restate_services/objects/reservation.py" source = handler_path.read_text(encoding="utf-8") self.assertIn('@reservation_object.handler("approve")', source) self.assertIn("async def approve(", source) def test_restate_reject_handler_registered(self) -> None: handler_path = _REPO_ROOT / "restate_services/objects/reservation.py" source = handler_path.read_text(encoding="utf-8") self.assertIn('@reservation_object.handler("reject")', source) self.assertIn("async def reject(", source) def test_approve_handler_enforces_pending_status(self) -> None: handler_path = _REPO_ROOT / "restate_services/objects/reservation.py" source = handler_path.read_text(encoding="utf-8") approve_idx = source.index("async def approve(") approve_block = source[approve_idx : approve_idx + 1600] self.assertIn('!= "pending"', approve_block) self.assertIn("status_code=409", approve_block) def test_reject_handler_enforces_pending_status(self) -> None: handler_path = _REPO_ROOT / "restate_services/objects/reservation.py" source = handler_path.read_text(encoding="utf-8") reject_idx = source.index("async def reject(") reject_block = source[reject_idx : reject_idx + 1600] self.assertIn('!= "pending"', reject_block) self.assertIn("status_code=409", reject_block)