#!/usr/bin/env python3
"""Merchant Supplier API conformance suite (MA-444).

Standalone checker for a merchant implementation of
docs/suppliers/merchant-api-spec.md — the go-live gate referenced by the
spec's §10. Merchants (or their AI agents) run it against their own
staging until green; MeshHub runs the same suite before activation.

    python merchant_api_conformance.py \
        --base-url https://staging.merchant.example --api-key <key>

Only stdlib + httpx. Run it against STAGING: the booking checks create
and cancel real reservations (skip them with --skip-book).

Multi-category merchants (spec §9.2) pass their full declared set via
the repeatable --category flag to run the per-category probes, the
cross-category isolation check, and the undeclared-slug check. The
declarable category slugs are listed in spec §9.

Categories group into families (spec §9): groups of categories that
share one integration surface and offer shape. The offer-shape,
filter, and booking-selection checks dispatch by family, so pass
every slug you declare to --category — without it only the generic
core is checked. A merchant declaring categories from more than one
family gets those checks once per family. Expanding into a new family
(spec §9.3) means passing this suite with the new slugs added before
those categories can be activated.

Results come in two tiers: PASS/FAIL for required behavior, plus INFO
(optional-to-support behavior the merchant chose not to implement) and
SKIP (a check that could not be exercised against this inventory). Only
FAIL affects the exit code.
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from datetime import timedelta
from decimal import Decimal
from decimal import InvalidOperation

import httpx

DEFAULT_PATHS = {
    "ping": "api/supplier/v1/ping/",
    "search": "api/supplier/v1/search/",
    "price": "api/supplier/v1/price/",
    "book": "api/supplier/v1/book/",
    "status": "api/supplier/v1/booking/{supplier_order_id}/",
    "cancel": "api/supplier/v1/booking/{supplier_order_id}/cancel/",
    "partners": "api/supplier/v1/partners/",
}

ENVELOPE_CODES = {
    "AUTH_FAILED",
    "VALIDATION",
    "NOT_FOUND",
    "OFFER_EXPIRED",
    "SLOT_TAKEN",
    "RATE_LIMITED",
    "ERROR",
}
STATUS_VOCABULARY = {
    "pending",
    "confirmed",
    "paid",
    "cancelled",
    "failed",
    "unknown",
}
SUPPORTED_CURRENCIES = {"EUR", "USD", "GBP"}

# ── Category families (spec §9) ──────────────────────────────────
# Every declarable slug belongs to exactly one family; the family
# fixes the offer shape and which family-specific checks run. A
# future family (spec §9.3) adds its slug→family rows here plus one
# entry in each FAMILY_* dispatch table below the Checker class.
SALON_LIKE = "salon-like"
CAR_RENTAL = "car-rental"
HOTEL = "hotel"
TRAVEL = "travel"
SHIPPING = "shipping"

# The car-rental family's closed vocabulary (spec §9.4).
TRANSMISSIONS = ("manual", "automatic")
IATA_LENGTH = 3
COUNTRY_LENGTH = 2

FAMILY_OF: dict[str, str] = {
    "hair_salon": SALON_LIKE,
    "massage_center": SALON_LIKE,
    "health_clinic": SALON_LIKE,
    "beauty_salon": SALON_LIKE,
    "spa": SALON_LIKE,
    "fitness": SALON_LIKE,
    "dental_clinic": SALON_LIKE,
    "pet_service": SALON_LIKE,
    "nail_salon": SALON_LIKE,
    "tennis_court": SALON_LIKE,
    "child_care": SALON_LIKE,
    "training": SALON_LIKE,
    "housekeeping": SALON_LIKE,
    "pet_care": SALON_LIKE,
    "nursing": SALON_LIKE,
    "refurbishment": SALON_LIKE,
    "restaurant": SALON_LIKE,
    "car_repair": SALON_LIKE,
    # Appended after the salon slugs so the "first undeclared slug"
    # pick in check_declared_categories is unchanged for a salon
    # merchant.
    "car_rental": CAR_RENTAL,
    "hotel": HOTEL,
    "travel": TRAVEL,
    "shipping": SHIPPING,
}

# The declarable MeshHub category slugs (spec §9).
CATEGORY_SLUGS: tuple[str, ...] = tuple(FAMILY_OF)

HTTP_OK = 200
HTTP_CREATED = 201
HTTP_UNAUTHORIZED = 401
HTTP_NOT_FOUND = 404
HTTP_METHOD_NOT_ALLOWED = 405
HTTP_4XX = 400
HTTP_5XX = 500

MAX_OFFERS_CHECKED = 5
MAX_PAGES_WALKED = 10
DEFAULT_WINDOW_DAYS = 7

TEST_END_USER = {
    "first_name": "Conformance",
    "last_name": "Probe",
    "email": "conformance@meshhub.ai",
    "phone": "+37060000001",
}


class Checker:
    def __init__(
        self,
        base_url: str,
        api_key: str,
        endpoints: dict,
        emit=None,
    ):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.endpoints = endpoints
        self.client = httpx.Client(timeout=30.0)
        self.results: list[tuple[str, str, str]] = []
        # Where progress lines go. Defaults to stdout for the CLI; MeshHub
        # passes a sink so a server-side run does not print into the
        # worker log. `results` is the real output either way.
        self._emit = emit if emit is not None else print

    def say(self, line: str) -> None:
        self._emit(line)

    def url(self, op: str, **kwargs) -> str:
        template = self.endpoints.get(op) or (
            f"{self.base_url}/{DEFAULT_PATHS[op]}"
        )
        return template.format(**kwargs)

    def call(self, method: str, op: str, *, key=True, **kwargs):
        headers = {"X-Api-Key": self.api_key if key else "definitely-wrong"}
        url_kwargs = kwargs.pop("url_kwargs", {})
        return self.client.request(
            method, self.url(op, **url_kwargs), headers=headers, **kwargs
        )

    def record(self, name: str, status: str, detail: str = "") -> None:
        self.results.append((name, status, detail))
        line = f"  [{status}] {name}"
        if detail and status != "PASS":
            line += f" — {detail}"
        self.say(line)

    def check(self, name: str, ok: bool = False, detail: str = "") -> bool:  # noqa: FBT001, FBT002
        self.record(name, "PASS" if ok else "FAIL", detail)
        return ok

    def info(self, name: str, detail: str = "") -> None:
        """Optional-to-support behavior the merchant does not implement."""
        self.record(name, "INFO", detail)

    def skip(self, name: str, detail: str = "") -> None:
        """A check that could not be exercised against this inventory."""
        self.record(name, "SKIP", detail)

    # ── checks ───────────────────────────────────────────────────────

    def check_ping(self):
        self.say("ping/")
        r = self.call("GET", "ping")
        self.check(
            "ping returns 200",
            r.status_code == HTTP_OK,
            f"got {r.status_code}",
        )
        bad = self.call("GET", "ping", key=False)
        self.check(
            "invalid key returns 401",
            bad.status_code == HTTP_UNAUTHORIZED,
            f"got {bad.status_code}",
        )
        self.check_envelope(
            "invalid-key body is the error envelope",
            bad,
            expect_code="AUTH_FAILED",
        )

    def check_envelope(
        self, name: str, response, expect_code: str | None = None
    ):
        try:
            body = response.json()
        except ValueError:
            return self.check(name, ok=False, detail="body is not JSON")
        ok = isinstance(body, dict) and {"code", "message"} <= set(body)
        if ok and expect_code:
            ok = body["code"] == expect_code
        if ok:
            ok = body.get("code") in ENVELOPE_CODES
        return self.check(name, ok, f"body: {json.dumps(body)[:200]}")

    def check_search(
        self, query: dict, family: str | None = None
    ) -> list[dict]:
        self.say("search/")
        r = self.call("GET", "search", params=query)
        self.check(
            "search returns 200",
            r.status_code == HTTP_OK,
            f"got {r.status_code}",
        )
        try:
            body = r.json()
        except ValueError:
            self.check("search body is JSON", ok=False)
            return []
        keys = {"count", "page", "page_size", "results"}
        pagination_ok = isinstance(body, dict) and keys <= set(body)
        self.check(
            "pagination envelope (count/page/page_size/results)",
            pagination_ok,
        )
        offers = body.get("results") if pagination_ok else []
        if not offers:
            self.check(
                "search returned at least one offer",
                ok=False,
                detail="cannot exercise price/book — adjust --search-q/"
                "--search-city to hit live inventory",
            )
            return []
        for i, offer in enumerate(offers[:MAX_OFFERS_CHECKED]):
            self._check_offer(f"offer[{i}]", offer, family)
        return offers

    def _check_offer(self, ctx: str, offer: dict, family: str | None = None):
        offer_id = offer.get("offer_id")
        self.check(
            f"{ctx}.offer_id is a non-empty string",
            isinstance(offer_id, str) and bool(offer_id),
        )
        self.check(
            f"{ctx}.price is a decimal string",
            self._decimal_ok(offer.get("price")),
            f"got {offer.get('price')!r}",
        )
        self.check(
            f"{ctx}.price is greater than zero",
            self._positive_decimal(offer.get("price")),
            f"got {offer.get('price')!r} — spec 2.4 §7.2: a service with no "
            "stated price is not an offer (a price column defaulting to 0 is "
            "the usual cause); MeshHub drops these",
        )
        self.check(
            f"{ctx}.currency in EUR/USD/GBP",
            offer.get("currency") in SUPPORTED_CURRENCIES,
            f"got {offer.get('currency')!r}",
        )
        self.check(
            f"{ctx}.required_fields is a list",
            isinstance(offer.get("required_fields"), list),
        )
        # family=None (a mixed-family run) → generic core only.
        shape_check = FAMILY_SHAPE_CHECKS.get(family)
        if shape_check is not None:
            shape_check(self, ctx, offer)

    def _check_salon_like_shape(self, ctx: str, offer: dict):
        slots = offer.get("slots")
        slots_ok = isinstance(slots, list) and bool(slots)
        self.check(
            f"{ctx}.slots is a non-empty list",
            slots_ok,
            "spec 2.0: one offer = service + master + place, carrying its "
            f"free times in 'slots'; got {type(slots).__name__}",
        )
        if not slots_ok:
            return
        ts = slots[0].get("time_start") if isinstance(slots[0], dict) else None
        self.check(
            f"{ctx}.slots[0].time_start is ISO-8601 with offset",
            isinstance(ts, str) and ("+" in ts or ts.endswith("Z")),
            f"got {ts!r} (naive timestamps are not compliant, spec §4)",
        )

    @staticmethod
    def _decimal_ok(value) -> bool:
        if not isinstance(value, str):
            return False
        try:
            Decimal(value)
        except InvalidOperation:
            return False
        return True

    @staticmethod
    def _positive_decimal(value) -> bool:
        if not isinstance(value, str):
            return False
        try:
            return Decimal(value) > 0
        except InvalidOperation:
            return False

    def _check_car_rental_shape(self, ctx: str, offer: dict):
        """The car-rental family block (spec §9.4)."""
        vehicle = offer.get("vehicle")
        ok = isinstance(vehicle, dict) and bool(vehicle)
        self.check(
            f"{ctx}.vehicle is an object",
            ok,
            "spec §9.4: a car-rental offer is a vehicle, carried in the "
            f"'vehicle' block; got {type(vehicle).__name__}",
        )
        if not ok:
            return
        for field in ("name", "vehicle_class"):
            value = vehicle.get(field)
            self.check(
                f"{ctx}.vehicle.{field} is a non-empty string",
                isinstance(value, str) and bool(value.strip()),
                f"got {value!r}",
            )
        transmission = vehicle.get("transmission")
        if transmission is None:
            self.skip(
                f"{ctx}.vehicle.transmission in manual/automatic",
                "not stated",
            )
        else:
            self.check(
                f"{ctx}.vehicle.transmission in manual/automatic",
                transmission in TRANSMISSIONS,
                f"got {transmission!r}",
            )
        for field in ("seats", "doors", "small_bags", "large_bags"):
            value = vehicle.get(field)
            if value is None:
                continue
            self.check(
                f"{ctx}.vehicle.{field} is a non-negative integer",
                isinstance(value, int)
                and not isinstance(value, bool)
                and value >= 0,
                f"got {value!r}",
            )
        locations = vehicle.get("locations")
        self.check(
            f"{ctx}.vehicle.locations is a list of "
            "{city, pickup_point} objects",
            isinstance(locations, list)
            and all(
                isinstance(row, dict) and isinstance(row.get("city"), str)
                for row in locations
            ),
            "spec §9.4: MeshHub matches city and pickup_point against one "
            f"and the same location row; got {locations!r}",
        )
        self.check(
            f"{ctx}.price_unit is 'per_day'",
            offer.get("price_unit") == "per_day",
            "spec §9.4: a car-rental price is a DAILY rate and must say "
            f"so; got {offer.get('price_unit')!r}",
        )
        self.check(
            f"{ctx} carries no 'slots'",
            "slots" not in offer,
            "spec §9.4: car rental is not slot-based — the rental window "
            "is chosen at book time with extra.date_from/date_to",
        )

    def _run_car_rental_filter_suite(self, offers: list[dict], query: dict):
        """The search-filter checks tied to the car-rental offer shape."""
        self.say("search/ filters (mandatory support)")
        vehicle = offers[0].get("vehicle") or {}
        locations = vehicle.get("locations") or []
        city = next(
            (
                row.get("city")
                for row in locations
                if isinstance(row, dict) and row.get("city")
            ),
            None,
        )
        if city:
            found = self.check_search({"city": city}, CAR_RENTAL)
            cities = [
                row.get("city")
                for item in found
                for row in ((item.get("vehicle") or {}).get("locations") or [])
                if isinstance(row, dict)
            ]
            self.check(
                "city= returns only vehicles collectable there",
                bool(cities)
                and any(
                    isinstance(c, str) and c.casefold() == city.casefold()
                    for c in cities
                ),
                f"asked {city!r}, got {sorted({str(c) for c in cities})}",
            )
        else:
            self.skip(
                "city= returns only vehicles collectable there",
                "no offer listed a pickup location to search by",
            )

        klass = vehicle.get("vehicle_class")
        if klass:
            found = self.check_search({"vehicle_class": klass}, CAR_RENTAL)
            classes = [
                (item.get("vehicle") or {}).get("vehicle_class")
                for item in found
            ]
            self.check(
                "vehicle_class= returns only that class",
                bool(classes)
                and all(
                    isinstance(c, str) and c.casefold() == klass.casefold()
                    for c in classes
                ),
                f"asked {klass!r}, got {sorted({str(c) for c in classes})}",
            )
        self.check_count_bound(query)

    def _check_hotel_shape(self, ctx: str, offer: dict):
        """The hotel family block (spec §9.5)."""
        for name in ("hotel", "room"):
            block = offer.get(name)
            self.check(
                f"{ctx}.{name} is an object",
                isinstance(block, dict) and bool(block),
                f"spec §9.5: a hotel offer carries the '{name}' block; "
                f"got {type(block).__name__}",
            )
        hotel = offer.get("hotel") or {}
        room = offer.get("room") or {}
        for block, name, field in (
            (hotel, "hotel", "code"),
            (hotel, "hotel", "name"),
            (room, "room", "name"),
        ):
            value = block.get(field)
            self.check(
                f"{ctx}.{name}.{field} is present",
                value not in (None, ""),
                f"got {value!r}",
            )
        stay = offer.get("stay") or {}
        check_in = stay.get("check_in")
        check_out = stay.get("check_out")
        if not (check_in and check_out):
            self.skip(f"{ctx}.stay dates are ISO dates", "stay not stated")
            return
        parsed = []
        for label, value in (("check_in", check_in), ("check_out", check_out)):
            try:
                parsed.append(datetime.fromisoformat(str(value)[:10]).date())
            except ValueError:
                self.check(
                    f"{ctx}.stay.{label} is an ISO-8601 date",
                    ok=False,
                    detail=f"got {value!r}",
                )
                return
        self.check(
            f"{ctx}.stay.check_out is after check_in",
            parsed[1] > parsed[0],
            "spec §9.5: a stay is [check_in, check_out) and the departure "
            f"day is not a night; got {check_in!r} to {check_out!r}",
        )
        self.check(
            f"{ctx} carries no 'slots'",
            "slots" not in offer,
            "spec §9.5: a hotel offer is priced for the stay that was "
            "searched, not for a list of free times",
        )

    def _check_travel_shape(self, ctx: str, offer: dict):
        """The travel family block (spec §9.6)."""
        flight = offer.get("flight") or offer.get("itinerary")
        ok = isinstance(flight, dict) and bool(flight)
        self.check(
            f"{ctx}.flight is an object",
            ok,
            "spec §9.6: a travel offer carries the itinerary in the "
            f"'flight' block; got {type(flight).__name__}",
        )
        if not ok:
            return
        legs = flight.get("legs")
        self.check(
            f"{ctx}.flight.legs is a non-empty list",
            isinstance(legs, list) and bool(legs),
            f"got {type(legs).__name__}",
        )
        for label in ("origin_airport", "destination_airport"):
            code = flight.get(label)
            self.check(
                f"{ctx}.flight.{label} is a 3-letter IATA code",
                isinstance(code, str)
                and len(code) == IATA_LENGTH
                and code.isalpha()
                and code.isupper(),
                f"got {code!r}",
            )
        for label in ("departure_time", "arrival_time"):
            ts = flight.get(label)
            self.check(
                f"{ctx}.flight.{label} is ISO-8601 with offset",
                isinstance(ts, str) and ("+" in ts or ts.endswith("Z")),
                f"got {ts!r} (naive timestamps are not compliant, spec §4)",
            )
        self.check(
            f"{ctx} carries no 'slots'",
            "slots" not in offer,
            "spec §9.6: a travel offer is one priced itinerary, bounded "
            "by expires_at rather than by a list of free times",
        )

    def _check_shipping_shape(self, ctx: str, offer: dict):
        """The shipping family block (spec §9.7)."""
        shipment = offer.get("shipment")
        ok = isinstance(shipment, dict) and bool(shipment)
        self.check(
            f"{ctx}.shipment is an object",
            ok,
            "spec §9.7: a shipping offer carries the quoted parcel in "
            f"the 'shipment' block; got {type(shipment).__name__}",
        )
        if not ok:
            return
        for label in ("origin", "destination"):
            code = shipment.get(label)
            self.check(
                f"{ctx}.shipment.{label} is a 2-letter country code",
                isinstance(code, str)
                and len(code) == COUNTRY_LENGTH
                and code.isalpha(),
                f"got {code!r} (ISO-3166-1 alpha-2)",
            )
        self.check(
            f"{ctx}.shipment.weight_kg is a decimal string above zero",
            self._positive_decimal(shipment.get("weight_kg")),
            f"got {shipment.get('weight_kg')!r}",
        )
        self.check(
            f"{ctx} carries no 'slots'",
            "slots" not in offer,
            "spec §9.7: a shipping offer is a quote for the parcel that "
            "was searched, not a list of free times",
        )

    def _run_echo_filter_suite(
        self, offers: list[dict], query: dict, block: str, fields: tuple
    ):
        """The family filter check for a family whose search fixes the
        offer: whatever was asked for must come back stated.

        Shared by shipping and travel, which have nothing to re-search
        by — the route IS the query, so the only thing to verify is
        that the merchant quoted what it was asked about rather than
        something else.
        """
        self.say("search/ filters (mandatory support)")
        for field in fields:
            asked = query.get(field)
            if not asked:
                self.skip(f"{field}= is echoed on the offer", "not asked")
                continue
            got = [(item.get(block) or {}).get(field) for item in offers]
            self.check(
                f"{field}= is echoed on every offer",
                all(
                    isinstance(v, str)
                    and v.casefold() == str(asked).casefold()
                    for v in got
                ),
                f"asked {asked!r}, got {sorted({str(v) for v in got})}",
            )
        self.check_count_bound(query)

    def _run_hotel_filter_suite(self, offers: list[dict], query: dict):
        self.say("search/ filters (mandatory support)")
        for field in ("check_in", "check_out"):
            asked = query.get(field)
            if not asked:
                self.skip(f"{field}= is echoed on the offer", "not asked")
                continue
            got = [(item.get("stay") or {}).get(field) for item in offers]
            self.check(
                f"{field}= is the stay every offer is priced for",
                all(str(v)[:10] == str(asked)[:10] for v in got),
                f"asked {asked!r}, got {sorted({str(v) for v in got})}",
            )
        self.check_count_bound(query)

    def _run_travel_filter_suite(self, offers: list[dict], query: dict):
        self._run_echo_filter_suite(
            offers, query, "flight", ("origin_airport", "destination_airport")
        )

    def _run_shipping_filter_suite(self, offers: list[dict], query: dict):
        self._run_echo_filter_suite(
            offers, query, "shipment", ("origin", "destination")
        )

    def _appointment_at(self, offer: dict) -> str | None:
        """The first free time on an offer — what price/book select with."""
        slots = offer.get("slots") or []
        if slots and isinstance(slots[0], dict):
            return slots[0].get("time_start")
        return None

    def _booking_time(self, offer: dict, family: str | None) -> str | None:
        """The family's booking-time selector applied to the offer.

        family=None (a mixed-family run) has no selector: price/book
        then omit appointment_at and exercise the generic core only.
        """
        selector = FAMILY_BOOKING_SELECTORS.get(family)
        if selector is None:
            return None
        return selector(self, offer)

    def check_search_filters(self, offer: dict, family: str | None = None):
        """The mandatory filters must be honoured, not ignored (§7.2)."""
        self.say("search/ filters (mandatory support)")

        appointment_at = self._booking_time(offer, family)
        if appointment_at:
            day = appointment_at[:10]
            offers = self.check_search({"date": day}, family)
            starts = [
                slot.get("time_start", "")
                for item in offers
                for slot in (item.get("slots") or [])
            ]
            self.check(
                "date= returns only slots on that day",
                bool(starts) and all(s[:10] == day for s in starts),
                f"asked {day}, got days {sorted({s[:10] for s in starts})}",
            )

        master_id = (offer.get("master") or {}).get("id")
        if master_id is not None:
            offers = self.check_search({"master_id": master_id}, family)
            masters = [(o.get("master") or {}).get("id") for o in offers]
            self.check(
                "master_id= returns only that master's offers",
                bool(masters) and all(m == master_id for m in masters),
                f"asked {master_id!r}, got {sorted(set(map(str, masters)))}",
            )

    def _search_page(self, query: dict) -> dict | None:
        """Raw search call; the parsed envelope, or None if unusable."""
        r = self.call("GET", "search", params=query)
        if r.status_code != HTTP_OK:
            return None
        try:
            body = r.json()
        except ValueError:
            return None
        if not isinstance(body, dict):
            return None
        if not isinstance(body.get("results"), list):
            return None
        return body

    def check_master_name_filter(self, offers: list[dict], query: dict):
        """`master_name=` narrowing — optional to support (§7.2, 2.1)."""
        self.say("search/ master_name= (optional support)")
        name = "master_name= narrows to matching staff names"
        target = ((offers[0].get("master") or {}).get("name") or "").strip()
        if not target:
            self.skip(name, "no master.name on the sampled offer")
            return
        body = self._search_page({**query, "master_name": target})
        if body is None:
            self.check(name, ok=False, detail="no valid search envelope")
            return
        results = body["results"]
        if not results:
            self.check(
                name,
                ok=False,
                detail=f"master_name={target!r} returned no offers, "
                "though the name was taken from a live offer",
            )
            return
        names = [(o.get("master") or {}).get("name") or "" for o in results]
        if all(target.lower() in n.lower() for n in names):
            self.check(name, ok=True)
            return
        if {o.get("offer_id") for o in results} == {
            o.get("offer_id") for o in offers
        }:
            self.info(
                name,
                "merchant ignores master_name (allowed: MeshHub narrows "
                "by staff name itself, ignoring it only wastes payload)",
            )
            return
        self.check(
            name,
            ok=False,
            detail=f"asked {target!r}, got staff {sorted(set(names))[:5]}",
        )

    def check_count_bound(self, query: dict):
        """`count` MUST be >= offers returned across all pages (§7.2)."""
        self.say("search/ count upper bound")
        name = "count >= offers returned across walked pages"
        body = self._search_page({**query, "page": 1})
        if body is None:
            self.check(name, ok=False, detail="no valid search envelope")
            return
        count = body.get("count")
        if not isinstance(count, int):
            self.check(name, ok=False, detail=f"count is {count!r}")
            return
        page_size = body.get("page_size")
        if not isinstance(page_size, int) or page_size <= 0:
            page_size = max(len(body["results"]), 1)
        total = len(body["results"])
        pages = 1
        while len(body["results"]) >= page_size and pages < MAX_PAGES_WALKED:
            pages += 1
            body = self._search_page({**query, "page": pages})
            if body is None:
                break
            total += len(body["results"])
        self.check(
            name,
            count >= total,
            f"count={count} but {total} offers came back over {pages} "
            "page(s) — results and page exhaustion are authoritative",
        )

    def check_default_window(self, offers: list[dict]):
        """A no-date search stays within [today, today + 7d] (§7.2, 2.1)."""
        self.say("search/ no-date default window")
        name = (
            "no-date search slots fall within [today, today + "
            f"{DEFAULT_WINDOW_DAYS} days]"
        )
        window = timedelta(days=DEFAULT_WINDOW_DAYS)
        bad: list[str] = []
        seen = 0
        for offer in offers:
            for slot in offer.get("slots") or []:
                ts = slot.get("time_start") if isinstance(slot, dict) else None
                if not isinstance(ts, str):
                    continue
                try:
                    start = datetime.fromisoformat(ts)
                except ValueError:
                    continue
                if start.tzinfo is None:
                    continue  # naive timestamps already fail their own check
                seen += 1
                # "Today" in the merchant's own offset, per timestamp.
                today = datetime.now(start.tzinfo).date()
                if not today <= start.date() <= (today + window):
                    bad.append(ts)
        if not seen:
            self.skip(name, "no offset-aware slots to inspect")
            return
        self.check(name, not bad, f"out-of-window slots: {bad[:5]}")

    def _run_salon_like_filter_suite(self, offers: list[dict], query: dict):
        """The search-filter checks tied to the salon-like offer shape.

        check_count_bound is envelope-level and family-neutral; it
        runs here so the check order matches pre-family output, and a
        mixed-family run calls it directly instead (see main()).
        """
        self.check_search_filters(offers[0], SALON_LIKE)
        self.check_master_name_filter(offers, query)
        self.check_count_bound(query)
        self.check_default_window(offers)

    def check_declared_categories(self, declared: list[str], query: dict):
        """The `category` param and declared-set rules (§7.2, §9, 2.2)."""
        self.say("search/ category= (declared categories, spec §9)")
        per_slug: dict[str, set[str]] = {}
        for slug in declared:
            name = f"category={slug} (declared) returns offers"
            body = self._search_page({**query, "category": slug})
            if body is None:
                self.check(name, ok=False, detail="no valid search envelope")
                continue
            offers = body["results"]
            self.check(
                name,
                bool(offers),
                "a declared category must have live staging inventory "
                "(spec §10)",
            )
            per_slug[slug] = {
                o.get("offer_id") for o in offers if isinstance(o, dict)
            }
        if len(declared) > 1:
            self._check_category_isolation(declared, per_slug)
        self._check_undeclared_category(declared, query)

    def _check_category_isolation(
        self, declared: list[str], per_slug: dict[str, set[str]]
    ):
        """No offer may surface under two declared slugs (§9.2)."""
        name = "offer_ids are disjoint across declared categories"
        populated = [slug for slug in declared if per_slug.get(slug)]
        if len(populated) < 2:  # noqa: PLR2004
            self.skip(name, "fewer than two categories returned offers")
            return
        overlaps: list[str] = []
        for i, a in enumerate(populated):
            for b in populated[i + 1 :]:
                shared = per_slug[a] & per_slug[b]
                if shared:
                    ids = ", ".join(sorted(map(str, shared))[:3])
                    overlaps.append(f"{a}/{b}: {ids}")
        self.check(
            name,
            not overlaps,
            "each service belongs to exactly one category (spec §9.2); "
            f"overlapping: {overlaps[:3]}",
        )

    def _check_undeclared_category(self, declared: list[str], query: dict):
        """An undeclared slug gets empty results, not an error (§7.2)."""
        name = "undeclared category slug returns empty results"
        undeclared = next(
            (slug for slug in CATEGORY_SLUGS if slug not in declared), None
        )
        if undeclared is None:
            self.skip(name, "every known category slug is declared")
            return
        body = self._search_page({**query, "category": undeclared})
        if body is None:
            self.check(name, ok=False, detail="no valid search envelope")
            return
        results = body["results"]
        if not results:
            self.check(name, ok=True)
            return
        baseline = self._search_page(query)
        returned_ids = {
            o.get("offer_id") for o in results if isinstance(o, dict)
        }
        baseline_ids = {
            o.get("offer_id")
            for o in (baseline["results"] if baseline else [])
            if isinstance(o, dict)
        }
        if baseline is not None and returned_ids == baseline_ids:
            if len(declared) == 1:
                self.info(
                    name,
                    "merchant ignores category (allowed for a "
                    "single-category merchant: it only ever receives "
                    "its own slug, spec §7.2)",
                )
                return
            self.check(
                name,
                ok=False,
                detail="merchant ignores category= — multi-category "
                "merchants MUST honour it (spec §9.2)",
            )
            return
        self.check(
            name,
            ok=False,
            detail=f"category={undeclared} is not declared, yet "
            f"{len(results)} offer(s) came back — expected empty results",
        )

    def _book_extra(self, family: str | None) -> dict:
        """The family's own ``extra`` block for price/book (spec §9.x).

        Empty for a family whose offer already fixes everything, and for
        salon-like, which selects its slot with ``appointment_at``.
        """
        builder = FAMILY_BOOK_EXTRAS.get(family)
        return builder(self) if builder is not None else {}

    def _car_rental_book_extra(self) -> dict:
        """A rental window (spec §9.4). A car-rental offer is a daily
        rate, so there is nothing to book without dates. Two weeks out
        so a fleet with a short booking horizon still has the vehicle
        free; local dates, because a calendar date is what the spec
        asks for."""
        start = datetime.now().date() + timedelta(days=14)  # noqa: DTZ005
        return {
            "date_from": start.isoformat(),
            "date_to": (start + timedelta(days=3)).isoformat(),
        }

    def _check_car_rental_total(self, offer: dict, extra: dict, body: dict):
        """The book total is the offer's daily rate x the window's days
        (spec 2.9 §9.4): days = date_to - date_from in calendar days,
        minimum one. Within 5 %, which is MeshHub's own refusal line."""
        try:
            window = datetime.fromisoformat(
                extra["date_to"]
            ) - datetime.fromisoformat(extra["date_from"])
            days = max(1, window.days)
            expected = Decimal(str(offer["price"])) * days
            booked = Decimal(str(body.get("price")))
        except (KeyError, TypeError, ValueError, InvalidOperation):
            self.check(
                "book price is the window total (daily rate x days)",
                ok=False,
                detail=(
                    f"unreadable: offer.price={offer.get('price')!r}, "
                    f"book.price={body.get('price')!r}"
                ),
            )
            return
        drift = abs(booked - expected) / expected * 100 if expected else 0
        self.check(
            f"book price is the window total ({offer['price']} x {days} days)",
            drift <= 5,  # noqa: PLR2004 — spec §9.4's own figure
            f"spec §9.4: expected {expected} within 5 %, got {booked}",
        )

    def _hotel_book_extra(self) -> dict:
        """One adult guest and a holder (spec §9.5). The stay itself
        was fixed by the search, so there is nothing else to name."""
        return {
            "guests": [
                {
                    "type": "AD",
                    "name": TEST_END_USER["first_name"],
                    "surname": TEST_END_USER["last_name"],
                }
            ],
            "holder": {
                "name": TEST_END_USER["first_name"],
                "surname": TEST_END_USER["last_name"],
            },
        }

    def _travel_book_extra(self) -> dict:
        """One adult passenger (spec §9.6). The itinerary is the offer;
        who is flying is not."""
        return {
            "passengers": [
                {
                    "firstName": TEST_END_USER["first_name"],
                    "lastName": TEST_END_USER["last_name"],
                    "gender": "male",
                    "dateOfBirth": "1990-01-01",
                }
            ]
        }

    def _shipping_book_extra(self) -> dict:
        """Sender, recipient and one parcel (spec §9.7). A courier
        cannot collect a parcel without two addresses."""
        party = {
            "contact_name": (
                f"{TEST_END_USER['first_name']} {TEST_END_USER['last_name']}"
            ),
            "phone": TEST_END_USER.get("phone") or "+37060000000",
            "email_address": TEST_END_USER["email"],
            "address": "Gedimino pr. 1",
            "city": "Vilnius",
            "postcode": "01103",
            "country": "LT",
        }
        return {
            "shipper_details": party,
            "consignee_details": dict(party, contact_name="MeshHub Test"),
            "parcels": [
                {"weight": "1", "width": 10, "height": 10, "length": 10}
            ],
        }

    def check_price(self, offer: dict, family: str | None = None):
        self.say("price/")
        payload = {"offer_id": offer["offer_id"]}
        if appointment_at := self._booking_time(offer, family):
            payload["appointment_at"] = appointment_at
        if extra := self._book_extra(family):
            payload["extra"] = extra
        r = self.call("POST", "price", json=payload)
        if r.status_code == HTTP_NOT_FOUND:
            self.check(
                "price endpoint present (Tier 2, recommended)",
                ok=False,
                detail="404 — skipping",
            )
            return
        self.check(
            "price returns 200",
            r.status_code == HTTP_OK,
            f"got {r.status_code}",
        )
        try:
            body = r.json()
        except ValueError:
            self.check("price body is JSON", ok=False)
            return
        self.check(
            "price echoes offer_id",
            body.get("offer_id") == offer["offer_id"],
        )
        self.check(
            "price.price is a decimal string",
            self._decimal_ok(body.get("price")),
        )
        breakdown = body.get("breakdown")
        ok = isinstance(breakdown, list) and bool(breakdown)
        if ok:
            try:
                total = sum(Decimal(line["amount"]) for line in breakdown)
                ok = total == Decimal(body["price"])
            except (KeyError, TypeError, InvalidOperation):
                ok = False
        self.check("breakdown lines sum to price", bool(ok))

    def check_partners(self, offers: list | None = None):
        """``GET partners/`` — optional, and absence is not a failure.

        Spec §7.7. Nothing in MeshHub depends on this endpoint, so a 404
        or 405 records INFO rather than FAIL. When it *is* implemented,
        the ids have to be the ones search accepts as ``partner_id`` —
        a directory naming businesses you cannot then search for is
        worse than no directory.
        """
        self.say("partners/ (optional)")
        r = self.call("GET", "partners")
        if r.status_code in (HTTP_NOT_FOUND, HTTP_METHOD_NOT_ALLOWED):
            self.info(
                "partners endpoint present (Tier 3, optional)",
                f"{r.status_code} — not implemented, skipping",
            )
            return
        self.check(
            "partners returns 200",
            r.status_code == HTTP_OK,
            f"got {r.status_code}",
        )
        try:
            body = r.json()
        except ValueError:
            self.check("partners body is JSON", ok=False)
            return

        partners = body.get("partners")
        self.check("partners.partners is a list", isinstance(partners, list))
        if not isinstance(partners, list) or not partners:
            self.skip(
                "partner entries have id and name",
                "no partners returned",
            )
            return
        self.check(
            "every partner has a non-empty id and name",
            all(
                isinstance(p, dict)
                and p.get("id") not in (None, "")
                and p.get("name") not in (None, "")
                for p in partners
            ),
        )

        offer_partner_ids = {
            str((offer.get("partner") or {}).get("id"))
            for offer in (offers or [])
            if (offer.get("partner") or {}).get("id") is not None
        }
        if not offer_partner_ids:
            self.skip(
                "listed partners match the ids search returns",
                "no partner ids seen in search results",
            )
            return
        listed = {str(p.get("id")) for p in partners if isinstance(p, dict)}
        missing = offer_partner_ids - listed
        self.check(
            "listed partners match the ids search returns",
            not missing,
            f"search returned partner ids absent from partners/: "
            f"{sorted(missing)}",
        )

    def check_book_flow(self, offer: dict, family: str | None = None):
        self.say("book/ + status + cancel (idempotency)")
        payload = {
            "offer_id": offer["offer_id"],
            "client_ref": "mh-conformance",
            "end_user": TEST_END_USER,
        }
        # One offer spans several slots, so the time is what selects one —
        # and what the idempotency key must include (spec 2.0 §7.4).
        if appointment_at := self._booking_time(offer, family):
            payload["appointment_at"] = appointment_at
        # A family whose offer does not fix the booking outright carries
        # the rest in ``extra`` — the rental window, for car rental.
        # It is part of the idempotency key for the same reason
        # ``appointment_at`` is: the same vehicle for a different window
        # is a different order.
        if extra := self._book_extra(family):
            payload["extra"] = extra
        first = self.call("POST", "book", json=payload)
        if first.status_code not in (HTTP_OK, HTTP_CREATED):
            self.check(
                "book succeeds",
                ok=False,
                detail=f"got {first.status_code}: {first.text[:200]}",
            )
            return
        self.check(
            "first book returns 201 (created)",
            first.status_code == HTTP_CREATED,
            f"got {first.status_code}",
        )
        body = first.json()
        order_id = body.get("supplier_order_id")
        self.check(
            "book returns supplier_order_id", order_id not in (None, "")
        )
        self.check(
            "book status in vocabulary",
            body.get("status") in STATUS_VOCABULARY,
            f"got {body.get('status')!r}",
        )
        # A car-rental offer is a daily rate and the book total must be
        # that rate x days (spec 2.9 §9.4). MeshHub refuses a total more
        # than 5 % away, so a merchant that gets this wrong books nothing.
        if family == "car_rental" and "extra" in payload:
            self._check_car_rental_total(offer, payload["extra"], body)

        replay = self.call("POST", "book", json=payload)
        self.check(
            "replayed book returns 200 (not 201)",
            replay.status_code == HTTP_OK,
            f"got {replay.status_code}",
        )
        try:
            same = replay.json().get("supplier_order_id") == order_id
        except ValueError:
            same = False
        self.check("replayed book returns the SAME supplier_order_id", same)

        url_kwargs = {"supplier_order_id": order_id}
        r = self.call("GET", "status", url_kwargs=url_kwargs)
        self.check("status endpoint returns 200", r.status_code == HTTP_OK)
        self.check(
            "status value in vocabulary",
            r.status_code == HTTP_OK
            and r.json().get("status") in STATUS_VOCABULARY,
        )

        missing = self.call(
            "GET", "status", url_kwargs={"supplier_order_id": "999999999"}
        )
        self.check(
            "unknown order returns 404",
            missing.status_code == HTTP_NOT_FOUND,
            f"got {missing.status_code}",
        )
        self.check_envelope(
            "unknown-order body is the error envelope",
            missing,
            expect_code="NOT_FOUND",
        )

        c1 = self.call("POST", "cancel", url_kwargs=url_kwargs)
        self.check(
            "cancel returns 200",
            c1.status_code == HTTP_OK,
            f"got {c1.status_code}",
        )
        self.check(
            "cancel lands in cancelled",
            c1.status_code == HTTP_OK
            and c1.json().get("status") == "cancelled",
        )
        c2 = self.call("POST", "cancel", url_kwargs=url_kwargs)
        self.check(
            "repeat cancel is a 200 no-op",
            c2.status_code == HTTP_OK,
            f"got {c2.status_code}",
        )

    def check_stale_offer(self):
        self.say("error envelope on a garbage offer")
        r = self.call(
            "POST",
            "book",
            json={"offer_id": "tampered-token", "end_user": TEST_END_USER},
        )
        self.check(
            "garbage offer_id rejected (4xx)",
            HTTP_4XX <= r.status_code < HTTP_5XX,
            f"got {r.status_code}",
        )
        self.check_envelope("rejection body is the error envelope", r)

    def summary(self) -> int:
        tally = {"PASS": 0, "FAIL": 0, "INFO": 0, "SKIP": 0}
        for _, status, _ in self.results:
            tally[status] = tally.get(status, 0) + 1
        failed = [n for n, status, _ in self.results if status == "FAIL"]
        self.say(
            f"\n{len(self.results)} checks: "
            + ", ".join(f"{n} {s}" for s, n in tally.items() if n)
        )
        if failed:
            self.say("Failed:")
            for name in failed:
                self.say(f"  - {name}")
            return 1
        self.say("Conformance suite GREEN — ready for MeshHub activation.")
        return 0


# ── Family dispatch tables (spec §9) ─────────────────────────────
# One row per family in each table, referencing unbound Checker
# methods. A future family (spec §9.3) registers its slug→family
# rows in FAMILY_OF plus one entry per table here; a family absent
# from a table gets no family-specific behavior for that phase.
FAMILY_SHAPE_CHECKS = {
    SALON_LIKE: Checker._check_salon_like_shape,  # noqa: SLF001
    CAR_RENTAL: Checker._check_car_rental_shape,  # noqa: SLF001
    HOTEL: Checker._check_hotel_shape,  # noqa: SLF001
    TRAVEL: Checker._check_travel_shape,  # noqa: SLF001
    SHIPPING: Checker._check_shipping_shape,  # noqa: SLF001
}
FAMILY_BOOKING_SELECTORS = {
    SALON_LIKE: Checker._appointment_at,  # noqa: SLF001
}
FAMILY_FILTER_SUITES = {
    SALON_LIKE: Checker._run_salon_like_filter_suite,  # noqa: SLF001
    CAR_RENTAL: Checker._run_car_rental_filter_suite,  # noqa: SLF001
    HOTEL: Checker._run_hotel_filter_suite,  # noqa: SLF001
    TRAVEL: Checker._run_travel_filter_suite,  # noqa: SLF001
    SHIPPING: Checker._run_shipping_filter_suite,  # noqa: SLF001
}

# What ``price``/``book`` must carry beyond ``offer_id`` for a family
# whose booking selector is not a single timestamp. Without this the
# booking checks would fail for the suite's own reasons — a car-rental
# merchant rightly refuses a booking with no rental window.
FAMILY_BOOK_EXTRAS = {
    CAR_RENTAL: Checker._car_rental_book_extra,  # noqa: SLF001
    HOTEL: Checker._hotel_book_extra,  # noqa: SLF001
    TRAVEL: Checker._travel_book_extra,  # noqa: SLF001
    SHIPPING: Checker._shipping_book_extra,  # noqa: SLF001
}


def _parse_endpoints(
    parser: argparse.ArgumentParser, pairs: list[str]
) -> dict:
    endpoints = {}
    for pair in pairs:
        op, sep, url = pair.partition("=")
        if not sep or op not in DEFAULT_PATHS:
            parser.error(f"--endpoint OP must be one of {list(DEFAULT_PATHS)}")
        endpoints[op] = url
    return endpoints


def _parse_declared(
    parser: argparse.ArgumentParser, categories: list[str]
) -> list[str]:
    declared = list(dict.fromkeys(categories))
    for slug in declared:
        if slug not in CATEGORY_SLUGS:
            parser.error(
                f"--category {slug!r} is not a MeshHub category slug "
                f"(spec §9): {list(CATEGORY_SLUGS)}"
            )
    return declared


def _run_family_phases(
    checker: Checker,
    family: str | None,
    query: dict,
    *,
    skip_book: bool,
) -> list:
    """The phases that depend on the offer shape, for one family.

    Returns the offers found, so the caller can decide which family's
    offers the family-neutral checks should look at.
    """
    offers = checker.check_search(query, family)
    if not offers:
        return offers
    suite = FAMILY_FILTER_SUITES.get(family)
    if suite is not None:
        suite(checker, offers, query)
    else:
        # Envelope-level, family-neutral — runs in every run (the family
        # suites call it at its historical position).
        checker.check_count_bound(query)
    checker.check_price(offers[0], family)
    if skip_book:
        checker.skip("book/ + status + cancel checks", "--skip-book")
    else:
        checker.check_book_flow(offers[0], family)
    return offers


def run_suite(  # noqa: PLR0913 — entry point; the knobs are orthogonal
    base_url: str,
    api_key: str,
    endpoints: dict | None = None,
    declared: list | None = None,
    *,
    skip_book: bool = False,
    query: dict | None = None,
    emit=None,
) -> Checker:
    """Run every check and return the Checker holding the results.

    The single ordering of the suite, shared by the CLI below and by
    MeshHub's pre-activation run, so the two can never drift into
    checking different things — which is exactly what §10 of the spec
    used to promise and the code did not deliver (MA-573).

    A merchant declaring categories from more than one family gets the
    shape-dependent phases **once per family**, each scoped with an
    explicit ``category``. Running them once over the merged set would
    check whichever family answered first and silently pass the other —
    and since MeshHub feeds this every declared slug in one call, an
    aggregator would have activated with a family nobody ever checked.

    Returns the Checker rather than an exit code: a caller that wants
    per-check results reads ``.results`` (name, status, detail), and the
    CLI turns them into an exit code via ``.summary()``.
    """
    endpoints = endpoints or {}
    declared = declared or []
    query = dict(query or {})

    by_family: dict[str, list[str]] = {}
    for slug in declared:
        by_family.setdefault(FAMILY_OF[slug], []).append(slug)

    checker = Checker(base_url, api_key, endpoints, emit=emit)
    checker.check_ping()

    # ``partners/`` is the salon-like family's business directory (spec
    # §7.7) and no other family declares ``partner_id``. Checked once,
    # against salon-like offers when there are any, so a cross-family
    # merchant is not judged on whichever family happened to run last.
    partner_offers: list = []

    if not by_family:
        # No --category: there is no family to check against, and
        # assuming one would run salon-like checks against whatever the
        # merchant actually sells.
        checker.say(
            "  [INFO] no --category given — running the generic core "
            "only. Pass each slug you declare to --category so the "
            "family shape, filter and booking checks run (spec §9)."
        )
        partner_offers = _run_family_phases(
            checker, None, query, skip_book=skip_book
        )
        checker.say(
            "  [INFO] category= checks not run — multi-category "
            "merchants: pass each declared slug via --category (spec §9)"
        )

    for family, slugs in by_family.items():
        family_query = dict(query)
        if len(by_family) > 1:
            # Only when there is more than one family to separate: a
            # single-family merchant sends the search it always did.
            checker.say(f"── {family} family ({', '.join(slugs)}) ──")
            family_query["category"] = slugs[0]
        offers = _run_family_phases(
            checker, family, family_query, skip_book=skip_book
        )
        if offers and (family == SALON_LIKE or not partner_offers):
            partner_offers = offers

    if declared:
        checker.check_declared_categories(declared, query)
    checker.check_partners(partner_offers)
    checker.check_stale_offer()
    return checker


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", required=True)
    parser.add_argument("--api-key", required=True)
    parser.add_argument(
        "--endpoint",
        action="append",
        default=[],
        metavar="OP=URL",
        help=(
            "Per-operation URL override (repeatable), e.g. "
            "status=https://…/{supplier_order_id}"
        ),
    )
    parser.add_argument(
        "--category",
        action="append",
        default=[],
        metavar="SLUG",
        help=(
            "MeshHub category slug you declared (spec §9); repeatable — "
            "pass your full declared set"
        ),
    )
    parser.add_argument(
        "--search-q", default="", help="Free-text q for the search checks"
    )
    parser.add_argument(
        "--search-city", default="", help="City filter for search checks"
    )
    parser.add_argument(
        "--skip-book",
        action="store_true",
        help="Skip checks that create/cancel a real reservation",
    )
    args = parser.parse_args()

    query = {}
    if args.search_q:
        query["q"] = args.search_q
    if args.search_city:
        query["city"] = args.search_city

    checker = run_suite(
        args.base_url,
        args.api_key,
        endpoints=_parse_endpoints(parser, args.endpoint),
        declared=_parse_declared(parser, args.category),
        skip_book=args.skip_book,
        query=query,
    )
    return checker.summary()


if __name__ == "__main__":
    sys.exit(main())