gogo-telefon/tests/fake_partner.py

"""Fake partner booking-software API — reference implementation of Appendix B.

Used by the test suite to exercise the `partner_api` provider (auth, idempotency
replay, webhook signatures, polling fallback) AND intended to be handed to
partner IT teams as a working example of the contract they must implement.

Run standalone:  uvicorn tests.fake_partner:app --port 9100
(API key: "test-partner-key"; see PartnerState to preload slots.)
"""

from __future__ import annotations

import hashlib
import hmac
from dataclasses import dataclass, field
from datetime import datetime

import httpx
from fastapi import FastAPI, Header, Query, Request
from fastapi.responses import JSONResponse

API_KEY = "test-partner-key"
WEBHOOK_SECRET = "test-webhook-secret"


@dataclass
class PartnerState:
    """In-memory 'database' of the partner's booking software."""

    # service_id -> list of slot dicts {"start", "end", optionally staff}
    slots: dict[str, list[dict]] = field(default_factory=dict)
    # idempotency key -> stored response
    idempotency: dict[str, dict] = field(default_factory=dict)
    # partner_request_id -> request record (incl. "status")
    requests: dict[str, dict] = field(default_factory=dict)
    services: list[dict] = field(default_factory=list)
    seq: int = 0
    # availability latency injection for tests (seconds)
    availability_delay: float = 0.0

    def reset(self) -> None:
        self.slots.clear()
        self.idempotency.clear()
        self.requests.clear()
        self.services.clear()
        self.seq = 0
        self.availability_delay = 0.0


state = PartnerState()
app = FastAPI(title="Fake Partner Booking API (Appendix B reference)")


def _err(status: int, code: str, message: str) -> JSONResponse:
    # §B.0 error shape
    return JSONResponse({"error": {"code": code, "message": message}}, status_code=status)


def _check_auth(authorization: str | None) -> JSONResponse | None:
    if authorization != f"Bearer {API_KEY}":
        return _err(401, "unauthorized", "Missing or invalid API key")
    return None


@app.get("/availability")
async def availability(
    service_id: str = Query(...),
    home_visit: str = Query("false"),
    from_: str = Query(..., alias="from"),
    to: str = Query(...),
    authorization: str | None = Header(default=None),
):
    """§B.1 — ready-to-offer free slots; partner does all the slot math."""
    if err := _check_auth(authorization):
        return err
    if state.availability_delay:
        import asyncio

        await asyncio.sleep(state.availability_delay)
    date_from = datetime.fromisoformat(from_).date() if len(from_) > 10 else from_
    date_to = datetime.fromisoformat(to).date() if len(to) > 10 else to
    slots = [
        s
        for s in state.slots.get(service_id, [])
        if str(date_from) <= s["start"][:10] <= str(date_to)
    ]
    return {"service_id": service_id, "slots": slots[:20]}


@app.post("/booking-requests", status_code=201)
async def create_booking_request(
    request: Request,
    authorization: str | None = Header(default=None),
    idempotency_key: str | None = Header(default=None),
):
    """§B.2 — push a booking request; idempotent per Idempotency-Key."""
    if err := _check_auth(authorization):
        return err
    payload = await request.json()

    key = idempotency_key or payload.get("gogo_request_id", "")
    if key in state.idempotency:
        return state.idempotency[key]  # replay MUST return the original result

    if payload.get("dry_run"):
        return {"partner_request_id": None, "status": "dry_run_ok"}

    state.seq += 1
    partner_id = f"REQ-{state.seq:04d}"
    state.requests[partner_id] = {**payload, "status": "pending"}
    response = {"partner_request_id": partner_id, "status": "received"}
    state.idempotency[key] = response
    return response


@app.get("/booking-requests/{partner_request_id}")
async def booking_request_status(
    partner_request_id: str, authorization: str | None = Header(default=None)
):
    """§B.3 polling fallback."""
    if err := _check_auth(authorization):
        return err
    rec = state.requests.get(partner_request_id)
    if rec is None:
        return _err(404, "not_found", "Unknown partner_request_id")
    return {"status": rec["status"], "confirmed_slot": rec.get("confirmed_slot")}


@app.get("/services")
async def services(authorization: str | None = Header(default=None)):
    """§B.4 optional catalog endpoint."""
    if err := _check_auth(authorization):
        return err
    return {"services": state.services}


# -- helpers for driving the partner side in tests / demos -------------------


def sign_webhook(body: bytes, secret: str = WEBHOOK_SECRET) -> str:
    return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()


async def staff_resolve(
    gogo_base: httpx.AsyncClient,
    tenant_id: str,
    partner_request_id: str,
    outcome: str,
    confirmed_slot: dict | None = None,
    secret: str = WEBHOOK_SECRET,
) -> httpx.Response:
    """Simulate salon staff resolving a request in the partner software:
    updates local state and calls Gogo's outcome webhook (§B.3)."""
    import json as jsonlib

    rec = state.requests.get(partner_request_id)
    if rec is not None:
        rec["status"] = "resolved" if outcome == "resolved" else outcome
        if confirmed_slot:
            rec["confirmed_slot"] = confirmed_slot
    payload = {
        "gogo_request_id": rec["gogo_request_id"] if rec else None,
        "partner_request_id": partner_request_id,
        "outcome": outcome,
    }
    if confirmed_slot:
        payload["confirmed_slot"] = confirmed_slot
    body = jsonlib.dumps(payload).encode()
    return await gogo_base.post(
        f"/webhooks/partner/{tenant_id}",
        content=body,
        headers={
            "X-Gogo-Signature": sign_webhook(body, secret),
            "Content-Type": "application/json",
        },
    )