"""12B1 live proof: Booking rejects invalid reservation intervals. Run from Core: python scripts/probe_booking_interval_12b1.py Requires local pc-gateway at http://127.0.0.1:8801. """ from __future__ import annotations import asyncio import json import sys import tomllib from datetime import UTC, datetime, timedelta from pathlib import Path from uuid import uuid4 import httpx ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) BASE = "http://127.0.0.1:8801" def weekday_slot(): now = datetime.now(UTC).replace(minute=0, second=0, microsecond=0) start = now + timedelta(days=1) while start.weekday() >= 5: start += timedelta(days=1) start = start.replace(hour=10) later = start + timedelta(minutes=30) return start.isoformat().replace("+00:00", "Z"), later.isoformat().replace("+00:00", "Z") async def main(): config = tomllib.loads((ROOT.parent / "P2HSCloud/apps/pc-gateway/wrangler.toml").read_text()) token = config["vars"]["INTERNAL_CONTROL_TOKEN"] workspace, project = uuid4(), uuid4() observations = [] def check(name, actual, expected): observations.append({ "check": name, "actual": actual, "expected": expected, "pass": actual == expected, }) first_slot, second_slot = weekday_slot() async with httpx.AsyncClient(base_url=BASE, timeout=15) as http: async def req(method, path, body=None): kwargs = {"headers": {"Authorization": f"Bearer {token}"}} if body is not None: kwargs["json"] = body return await http.request(method, path, **kwargs) async def control(path, body): (await req("PUT", "/v1/cloud/control/" + path, body)).raise_for_status() await control(f"workspaces/{workspace}", { "workspace_id": str(workspace), "name": "12B1 booking interval probe", "status": "active", "cloud_enabled": True, "suspended": False, }) await control(f"projects/{project}", { "project_id": str(project), "workspace_id": str(workspace), "name": "12B1 probe", "status": "active", "cloud_enabled": True, "suspended": False, "allowed_request_types": ["booking.reserve", "booking.cancel"], "origins": [], }) resource_key = "consult" created = await req("POST", f"/v1/projects/{project}/bookings/resources", { "resource_key": resource_key, "name": "Consult", "timezone": "UTC", "duration_minutes": 30, "buffer_minutes": 0, }) check("upsert resource", created.status_code, 201) created.raise_for_status() reverse = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": second_slot, "ends_at": first_slot, "idempotency_key": "12b1-reverse", }) reverse_body = reverse.json() check("reverse interval status", reverse.status_code, 400) check("reverse interval error", reverse_body.get("error"), "bad_request") check("reverse interval not persisted", "reservation" not in reverse_body, True) if reverse.status_code == 201: rid = reverse_body.get("reservation", {}).get("id") fetched = await req("GET", f"/v1/projects/{project}/bookings/{rid}") check("reverse interval leaked into store", fetched.status_code, 404) zero = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "ends_at": first_slot, "idempotency_key": "12b1-zero", }) zero_body = zero.json() check("zero interval status", zero.status_code, 400) check("zero interval error", zero_body.get("error"), "bad_request") check("zero interval not persisted", "reservation" not in zero_body, True) bad_start = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": "not-a-date", "ends_at": second_slot, "idempotency_key": "12b1-bad-start", }) start_body = {} try: start_body = bad_start.json() except Exception: start_body = {"raw": bad_start.text[:200]} check("invalid start status", bad_start.status_code, 400) check("invalid start not 500", bad_start.status_code != 500, True) check("invalid start error", start_body.get("error"), "bad_request") check("invalid start not persisted", "reservation" not in start_body, True) bad_end = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "ends_at": "definitely-not-an-instant", "idempotency_key": "12b1-bad-end", }) end_body = {} try: end_body = bad_end.json() except Exception: end_body = {"raw": bad_end.text[:200]} check("invalid end status", bad_end.status_code, 400) check("invalid end not 500", bad_end.status_code != 500, True) check("invalid end error", end_body.get("error"), "bad_request") check("invalid end not persisted", "reservation" not in end_body, True) valid = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "ends_at": second_slot, "customer_name": "Pat", "customer_email": "pat@example.invalid", "idempotency_key": "12b1-valid", }) check("valid reservation", valid.status_code, 201) reservation_id = None if valid.status_code == 201: reservation_id = (valid.json().get("reservation") or {}).get("id") check("valid reservation persisted", bool(reservation_id), True) overlap = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "ends_at": second_slot, "customer_name": "Other", "customer_email": "other@example.invalid", "idempotency_key": "12b1-overlap", }) check("overlap conflict", overlap.status_code, 409) check("overlap error", overlap.json().get("error"), "slot_taken") if reservation_id: fetched = await req("GET", f"/v1/projects/{project}/bookings/{reservation_id}") check("valid still readable", fetched.status_code, 200) failed = [item for item in observations if not item["pass"]] report = { "phase": "12B1", "probe": "booking_invalid_interval", "passed": len(observations) - len(failed), "failed": len(failed), "observations": observations, } print(json.dumps(report, indent=2)) if failed: raise SystemExit(1) if __name__ == "__main__": asyncio.run(main())