#!/usr/bin/env python3
"""Metrics Exchange v1 conformance script (contract section 7.9).

Standard library only (urllib is the HTTP client). Run it against any provider:

  python3 conformance.py --base https://alpha.vibeback.io/metrics-exchange/v1 \
      --client-id mxc_... --client-secret mxs_... [--expect-provider HOST] [--check-rate-limit] [--wait-expiry]

The client should tick at least one metric (ideally every metric, with every breakdown).
--check-rate-limit locks this client's token door for up to a minute; run it last or on a spare client.
Checks a receiver cannot reproduce from outside (CLIENT_DISABLED, ACCOUNT_SUSPENDED, SCOPE_DISABLED,
UNAVAILABLE) are reported as SKIP; the provider's own test suite covers them.
Exit status: 0 when every check passed, 1 otherwise.
"""
from __future__ import annotations

import argparse
import csv
import io
import json
import os
import re
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import date, datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo
import xml.etree.ElementTree as ET

CONTRACT = 'metrics-exchange/1'
SHORTCUTS = ['today', 'yesterday', 'this_week', 'last_7_days', 'this_month', 'last_month']
KINDS = {'count', 'sum', 'distinct', 'ratio', 'gauge'}
UNITS = {'count', 'currency', 'percent', 'seconds', 'bytes'}
DEC = re.compile(r'^-?\d+(\.\d+)?$')
ID = re.compile(r'^[a-z][a-z0-9_]*$')


class Resp:
    def __init__(self, status: int, headers: dict, body: bytes):
        self.status = status
        self.headers = {k.lower(): v for k, v in headers.items()}
        self.body = body

    @property
    def ctype(self) -> str:
        return self.headers.get('content-type', '')

    def json(self):
        return json.loads(self.body.decode('utf-8'))


class Http:
    def __init__(self, base: str):
        self.base = base.rstrip('/')

    def call(self, method, path, query=None, form=None, body=None, token=None, accept=None) -> Resp:
        url = self.base + path + ('?' + urllib.parse.urlencode(query, doseq=True) if query else '')
        headers = {'User-Agent': 'metrics-exchange-conformance/1'}
        data = None
        if form is not None:
            data = urllib.parse.urlencode(form).encode()
            headers['Content-Type'] = 'application/x-www-form-urlencoded'
        elif body is not None:
            data = body if isinstance(body, bytes) else json.dumps(body).encode()
            headers['Content-Type'] = 'application/json'
        if token:
            headers['Authorization'] = f'Bearer {token}'
        if accept:
            headers['Accept'] = accept
        req = urllib.request.Request(url, data=data, method=method, headers=headers)
        try:
            with urllib.request.urlopen(req, timeout=120) as r:
                return Resp(r.status, dict(r.headers), r.read())
        except urllib.error.HTTPError as e:
            return Resp(e.code, dict(e.headers), e.read())


class Report:
    def __init__(self):
        self.passed = self.failed = self.skipped = 0

    def check(self, name: str, ok: bool, why: str = '') -> bool:
        if ok:
            self.passed += 1
            print(f'PASS  {name}')
        else:
            self.failed += 1
            print(f'FAIL  {name}' + (f' -- {why}' if why else ''))
        return ok

    def skip(self, name: str, why: str) -> None:
        self.skipped += 1
        print(f'SKIP  {name} -- {why}')

    def finish(self) -> int:
        print(f'\n{self.passed} passed, {self.failed} failed, {self.skipped} skipped')
        return 1 if self.failed else 0


def iso(s: str) -> datetime:
    return datetime.fromisoformat(s)


def problem(rep: Report, label: str, r: Resp, code: str, status: int) -> dict:
    body = {}
    try:
        body = r.json()
    except Exception:
        pass
    ok = (r.status == status and r.ctype.startswith('application/problem+json') and body.get('code') == code
          and body.get('status') == status and str(body.get('type', '')).endswith('/' + code)
          and all(k in body for k in ('title', 'detail', 'request_id')))
    rep.check(f'error {code} ({label})', ok, f'got {r.status} {r.ctype} {r.body[:200]!r}')
    return body


def is_zero(v: str) -> bool:
    return DEC.match(v) is not None and Decimal(v) == 0


def contiguous(points: list, period: dict) -> bool:
    if not points:
        return False
    if points[0]['start'] != period['from'] or points[-1]['end'] != period['to']:
        return False
    return all(points[i]['end'] == points[i + 1]['start'] for i in range(len(points) - 1))


def values(doc: dict) -> list:
    return [(m['id'], [(p['start'], p['value'], json.dumps(p['breakdown'])) for p in m['points']]) for m in doc['metrics']]


def check_sample(rep: Report, m: dict) -> None:
    sp = m.get('sample_point') or {}
    ok = (ID.match(m.get('id', '')) and isinstance(m.get('version'), int) and m['version'] >= 1
          and m.get('kind') in KINDS and m.get('unit') in UNITS and isinstance(m.get('label'), str)
          and isinstance(m.get('description'), str) and isinstance(m.get('breakdowns'), list)
          and re.match(r'^\d{4}-\d{2}-\d{2}$', m.get('available_from') or '') is not None
          and all(k in sp for k in ('start', 'end', 'value', 'final', 'breakdown'))
          and sp['final'] is False and sp['breakdown'] is None and is_zero(sp['value'])
          and iso(sp['start']) < iso(sp['end']))
    if m.get('kind') == 'ratio':
        ok = ok and sp.get('numerator') == '0' and sp.get('denominator') == '0'
    if m.get('unit') == 'currency':
        ok = ok and isinstance(sp.get('currency'), str)
    rep.check(f"catalogue metric {m.get('id')} is well formed with a valid sample_point", bool(ok), json.dumps(m)[:300])


def resolve(root: dict, ref: str) -> dict:
    node = root
    for part in ref.lstrip('#/').split('/'):
        node = node[part]
    return node


def validate(schema: dict, value, root: dict, path: str = '$') -> list:
    """The JSON-Schema subset openapi.json uses: $ref, type (incl. null), const, enum, pattern, properties, required,
    additionalProperties false, items, minimum, maximum, anyOf. Returns a list of error strings."""
    if '$ref' in schema:
        return validate(resolve(root, schema['$ref']), value, root, path)
    if 'anyOf' in schema:
        return [] if any(not validate(s, value, root, path) for s in schema['anyOf']) else [f'{path}: matches no anyOf branch']
    errs = []
    types = schema.get('type')
    if types is not None:
        types = types if isinstance(types, list) else [types]
        py = {'string': str, 'integer': int, 'boolean': bool, 'object': dict, 'array': list, 'null': type(None)}
        ok = any((t == 'integer' and isinstance(value, int) and not isinstance(value, bool)) or
                 (t != 'integer' and isinstance(value, py[t]) and not (t != 'boolean' and isinstance(value, bool))) for t in types)
        if not ok:
            return [f'{path}: expected {types}, got {type(value).__name__}']
    if 'const' in schema and value != schema['const']:
        errs.append(f'{path}: expected {schema["const"]!r}')
    if 'enum' in schema and value not in schema['enum']:
        errs.append(f'{path}: {value!r} not in enum')
    if 'pattern' in schema and isinstance(value, str) and not re.search(schema['pattern'], value):
        errs.append(f'{path}: {value!r} does not match {schema["pattern"]}')
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if 'minimum' in schema and value < schema['minimum']:
            errs.append(f'{path}: below minimum')
        if 'maximum' in schema and value > schema['maximum']:
            errs.append(f'{path}: above maximum')
    if isinstance(value, dict):
        for k in schema.get('required', []):
            if k not in value:
                errs.append(f'{path}: missing {k}')
        props = schema.get('properties', {})
        for k, v in value.items():
            if k in props:
                errs += validate(props[k], v, root, f'{path}.{k}')
            elif schema.get('additionalProperties') is False:
                errs.append(f'{path}: unexpected field {k}')
    if isinstance(value, list) and 'items' in schema:
        for i, v in enumerate(value):
            errs += validate(schema['items'], v, root, f'{path}[{i}]')
    return errs


def xml_points(xml_bytes: bytes) -> tuple:
    # Standard library only (no defusedxml): refuse any DTD or entity declaration before parsing, which rules out
    # external-entity and entity-expansion attacks. A contract document never carries either.
    if b'<!DOCTYPE' in xml_bytes or b'<!ENTITY' in xml_bytes:
        raise ValueError('XML with a DTD or entity declaration is refused')
    root = ET.fromstring(xml_bytes)
    head = {c.tag: c for c in root}
    out = []
    for m in head['metrics']:
        pts = []
        for p in m:
            kids = {c.tag: c for c in p}
            bd = kids.get('breakdown')
            pts.append({
                'start': p.get('start'), 'end': p.get('end'), 'final': p.get('final') == 'true', 'value': kids['value'].text,
                **{k: kids[k].text for k in ('currency', 'numerator', 'denominator') if k in kids},
                **({'unconverted_count': int(kids['unconverted_count'].text)} if 'unconverted_count' in kids else {}),
                'breakdown': {'name': bd.get('name'), 'value': bd.get('value')} if bd is not None else None,
            })
        out.append((m.get('id'), int(m.get('version')), m.get('label'), m.get('unit'), m.get('kind'), pts))
    return root.get('contract'), head['query_id'].text, out


def json_points(doc: dict) -> list:
    return [(m['id'], m['version'], m['label'], m['unit'], m['kind'], m['points']) for m in doc['metrics']]


def check_formats(rep: Report, http: Http, token: str, xsd_validator: str, query: dict) -> None:
    j = http.call('GET', '/data', token=token, query={**query, 'format': 'json'})
    x = http.call('GET', '/data', token=token, query={**query, 'format': 'xml'})
    c = http.call('GET', '/data', token=token, query={**query, 'format': 'csv'})
    a = http.call('GET', '/data', token=token, query=query, accept='text/csv')
    if not rep.check('all three formats answer 200', j.status == x.status == c.status == 200, f'{j.status} {x.status} {c.status}'):
        return
    doc = j.json()
    rep.check('content types', j.ctype.startswith('application/json') and x.ctype.startswith('application/xml') and c.ctype.startswith('text/csv'))
    rep.check('Accept: text/csv without format= answers CSV', a.status == 200 and a.ctype.startswith('text/csv'))
    rep.check('XML starts with the declaration', x.body.startswith(b'<?xml version="1.0" encoding="UTF-8"?>'))
    try:
        contract, qid, xp = xml_points(x.body)
        rep.check('XML equals the JSON answer (same snapshot, same points)', contract == '1' and qid == doc['query_id'] and xp == json_points(doc))
    except (ValueError, ET.ParseError) as e:
        rep.check('XML is a well-formed contract document', False, str(e))
    with tempfile.TemporaryDirectory() as d:
        xsd_path, xml_path = os.path.join(d, 'schema.xsd'), os.path.join(d, 'doc.xml')
        s = http.call('GET', '/schema.xsd', token=token)
        rep.check('schema.xsd is served behind the bearer check', s.status == 200 and http.call('GET', '/schema.xsd').status == 401)
        open(xsd_path, 'wb').write(s.body)
        open(xml_path, 'wb').write(x.body)
        try:
            out = subprocess.run([xsd_validator, '--noout', '--schema', xsd_path, xml_path], capture_output=True, text=True)
            rep.check('XML validates against /schema.xsd', out.returncode == 0, out.stderr[:300])
        except FileNotFoundError:
            rep.check('XML validates against /schema.xsd', False, f'{xsd_validator} not found (install libxml2-utils)')
    text = c.body.decode('utf-8')
    rep.check('CSV: UTF-8, LF, no BOM', not text.startswith('') and '\r\n' not in text and text.endswith('\n'))
    rows = list(csv.reader(io.StringIO(text)))
    header = 'query_id,dry_run,metric_id,metric_version,label,unit,kind,start,end,value,currency,numerator,denominator,final,breakdown_name,breakdown_value,timezone,data_as_of'.split(',')
    rep.check('CSV header: the 18 contract columns in order', rows[0] == header, str(rows[0]))
    expected = [[doc['query_id'], str(doc['dry_run']).lower(), m['id'], str(m['version']), m['label'], m['unit'], m['kind'], p['start'], p['end'],
                 p['value'], p.get('currency', ''), p.get('numerator', ''), p.get('denominator', ''), str(p['final']).lower(),
                 (p['breakdown'] or {}).get('name', ''), (p['breakdown'] or {}).get('value', ''), doc['timezone'], doc['data_as_of']]
                for m in doc['metrics'] for p in m['points']]
    rep.check('CSV rows equal the JSON points', rows[1:] == expected, f'{len(rows) - 1} rows vs {len(expected)} points')
    o = http.call('GET', '/openapi.json', token=token)
    if rep.check('openapi.json is served (3.1)', o.status == 200 and o.json().get('openapi') == '3.1.0'):
        spec = o.json()
        schema = spec['components']['schemas']
        errs = validate(schema['DataDocument'], doc, spec)
        rep.check('the JSON answer validates against openapi.json', not errs, '; '.join(errs[:5]))
        h = http.call('GET', '/health', token=token).json()
        rep.check('/health validates against openapi.json', not validate(schema['Health'], h, spec))
        rep.check('/time validates against openapi.json', not validate(schema['Time'], http.call('GET', '/time', token=token).json(), spec))
        rep.check('/catalogue validates against openapi.json', not validate(schema['Catalogue'], http.call('GET', '/catalogue', token=token).json(), spec))
        p = http.call('GET', '/data', token=token, query={'metrics': '__nope__', 'range': 'today'}).json()
        rep.check('a Problem validates against openapi.json', not validate(schema['Problem'], p, spec))


def check_paging(rep: Report, http: Http, token: str, query: dict) -> None:
    pages, total, cursor, first_cursor, headers_ok = 0, 0, None, None, True
    while True:
        r = http.call('GET', '/data', token=token, query={**query, **({'page': cursor} if cursor else {})})
        if r.status != 200:
            rep.check('paging: every page answers 200', False, f'{r.status} {r.body[:200]!r}')
            return
        d = r.json()
        pages += 1
        n = sum(len(m['points']) for m in d['metrics'])
        total += n
        nxt = d['next_page']
        headers_ok &= n <= 10000 and (r.headers.get('x-next-page') == nxt if nxt else 'x-next-page' not in r.headers)
        headers_ok &= ('link' in r.headers) == bool(nxt)
        if not nxt:
            break
        cursor = nxt
        first_cursor = first_cursor or nxt
    rep.check(f'paging: followed {pages} pages ({total} points); X-Next-Page/Link match next_page; <= 10,000 points per page',
              pages > 1 and headers_ok)
    problem(rep, 'garbage cursor', http.call('GET', '/data', token=token, query={**query, 'page': 'not-a-cursor'}), 'INVALID_RANGE', 400)
    if first_cursor:
        other = {**query, 'group': 'month' if query.get('group') != 'month' else 'week', 'page': first_cursor}
        problem(rep, 'cursor used with a different query', http.call('GET', '/data', token=token, query=other), 'INVALID_RANGE', 400)


def main() -> int:
    ap = argparse.ArgumentParser(description='Metrics Exchange v1 conformance')
    ap.add_argument('--base', required=True)
    ap.add_argument('--client-id', required=True)
    ap.add_argument('--client-secret', default=os.environ.get('MX_CLIENT_SECRET'),
                    help='or set MX_CLIENT_SECRET (keeps the secret out of ps and shell history)')
    ap.add_argument('--expect-provider', help='provider host to expect (default: the host of --base)')
    ap.add_argument('--check-rate-limit', action='store_true')
    ap.add_argument('--check-address-limit', action='store_true',
                    help='61 token calls from this address with unknown client ids must hit 429 (locks this address for a minute)')
    ap.add_argument('--wait-expiry', action='store_true', help='wait for a token to expire (minimum 5 minutes)')
    ap.add_argument('--formats', action='store_true', help='check XML, CSV, schema.xsd and openapi.json')
    ap.add_argument('--xsd-validator', default='xmllint')
    ap.add_argument('--paging-query', help="a query string known to exceed 10,000 points, e.g. 'metrics=website_sessions&breakdown=country&group=hour&from=2026-08-12&to=2026-10-12'")
    a = ap.parse_args()
    if not a.client_secret:
        ap.error('--client-secret or MX_CLIENT_SECRET is required')
    http, rep = Http(a.base), Report()
    provider = a.expect_provider or urllib.parse.urlparse(a.base).netloc
    creds = {'grant_type': 'client_credentials', 'client_id': a.client_id, 'client_secret': a.client_secret}

    # ── 1. Token ──────────────────────────────────────────────────────────────
    r = http.call('POST', '/token', form=creds)
    if not rep.check('token issued (form body)', r.status == 200, f'{r.status} {r.body[:300]!r}'):
        return rep.finish()
    t = r.json()
    token = t['access_token']
    rep.check('token shape (Bearer, metrics:read, 300 <= expires_in <= 86400, opaque mx_at_)',
              t.get('token_type') == 'Bearer' and t.get('scope') == 'metrics:read' and isinstance(t.get('expires_in'), int)
              and 300 <= t['expires_in'] <= 86400 and token.startswith('mx_at_'), str(t))
    rep.check('token issued (JSON body)', http.call('POST', '/token', body=creds).status == 200)
    problem(rep, 'wrong secret', http.call('POST', '/token', form={**creds, 'client_secret': a.client_secret + 'x'}), 'INVALID_CLIENT', 401)
    problem(rep, 'wrong grant_type', http.call('POST', '/token', form={**creds, 'grant_type': 'password'}), 'INVALID_CLIENT', 401)
    problem(rep, 'no bearer token', http.call('GET', '/health'), 'TOKEN_EXPIRED', 401)
    problem(rep, 'unknown token', http.call('GET', '/health', token='mx_at_not_a_token'), 'TOKEN_EXPIRED', 401)

    # ── 2. Health ─────────────────────────────────────────────────────────────
    r = http.call('GET', '/health', token=token)
    h = r.json() if r.status == 200 else {}
    rep.check('health 200 with X-Request-Id', r.status == 200 and 'x-request-id' in r.headers, f'{r.status}')
    rep.check('health shape', h.get('status') in ('ok', 'degraded') and h.get('contract') == CONTRACT
              and h.get('provider') == provider and h.get('server_time', '').endswith('Z'), json.dumps(h)[:300])
    cl = h.get('client', {})
    rep.check('health client block', all(k in cl for k in ('name', 'scope', 'metrics_enabled', 'formats', 'freshness_seconds',
                                                         'token_lifetime_seconds', 'timezone'))
              and cl.get('token_lifetime_seconds') == t['expires_in'] and 60 <= cl.get('freshness_seconds', 0) <= 3600, json.dumps(cl))

    # ── 3. Time ───────────────────────────────────────────────────────────────
    tm = http.call('GET', '/time', token=token).json()
    tz = tm['timezone']
    zone = ZoneInfo(tz)
    sc = tm['shortcuts']

    def in_zone(s: str) -> bool:
        d = iso(s)
        return d.utcoffset() == d.astimezone(zone).utcoffset()

    def day_start(s: str) -> bool:
        d = iso(s)
        return (d - timedelta(seconds=1)).astimezone(zone).date() != d.astimezone(zone).date()

    rep.check('time: every shortcut present', set(sc) == set(SHORTCUTS), str(list(sc)))
    rep.check('time: windows in the stated zone, on day boundaries, from < to',
              all(in_zone(sc[k]['from']) and in_zone(sc[k]['to']) and day_start(sc[k]['from']) and day_start(sc[k]['to'])
                  and iso(sc[k]['from']) < iso(sc[k]['to']) for k in SHORTCUTS))
    today = sc['today']
    rep.check('time: windows contiguous (yesterday/today, last_7_days ends today, last_month ends this_month)',
              sc['yesterday']['to'] == today['from'] and sc['last_7_days']['to'] == today['from']
              and sc['last_month']['to'] == sc['this_month']['from']
              and iso(sc['this_week']['from']) <= iso(today['from']) < iso(sc['this_week']['to'])
              and iso(sc['this_month']['from']) <= iso(today['from']) < iso(sc['this_month']['to']))
    rep.check('time: week starts on Monday', tm.get('week_starts_on') == 'monday' and iso(sc['this_week']['from']).astimezone(zone).weekday() == 0)
    rep.check('time: server_time_utc is UTC and server_time is in the zone', tm['server_time_utc'].endswith('Z') and in_zone(tm['server_time']))

    # ── 4. Catalogue ──────────────────────────────────────────────────────────
    cat = http.call('GET', '/catalogue', token=token).json()
    metrics = cat.get('metrics', [])
    rep.check('catalogue shape', cat.get('contract') == CONTRACT and cat.get('provider') == provider
              and cat.get('limits', {}).get('max_points') == 10000 and len(metrics) == cl.get('metrics_enabled')
              and cat.get('currency_mode') in ('single', 'per_currency'), json.dumps({k: v for k, v in cat.items() if k != 'metrics'})[:300])
    if not metrics:
        rep.check('catalogue lists at least one metric', False, 'tick at least one metric for this client')
        return rep.finish()
    for m in metrics:
        check_sample(rep, m)
    ids = [m['id'] for m in metrics]
    by_id = {m['id']: m for m in metrics}

    # ── 5. Dry run ────────────────────────────────────────────────────────────
    r = http.call('GET', '/data', token=token, query={'metrics': '*', 'range': 'last_7_days', 'group': 'day', 'dry_run': 'true'})
    d = r.json() if r.status == 200 else {}
    rep.check('dry run 200 and flagged', r.status == 200 and d.get('dry_run') is True, f'{r.status} {r.body[:200]!r}')
    if d:
        rep.check('dry run: same metrics, kinds and versions as the catalogue',
                  [m['id'] for m in d['metrics']] == ids
                  and all(dm['kind'] == by_id[dm['id']]['kind'] and dm['unit'] == by_id[dm['id']]['unit']
                          and dm['version'] == by_id[dm['id']]['version'] for dm in d['metrics']))
        rep.check('dry run: period equals /time last_7_days',
                  d['period']['from'] == sc['last_7_days']['from'] and d['period']['to'] == sc['last_7_days']['to']
                  and d['period']['range'] == 'last_7_days' and d['period']['group'] == 'day')
        rep.check('dry run: 7 zero points per metric, contiguous over the period',
                  all(len(m['points']) == 7 and all(is_zero(p['value']) for p in m['points']) and contiguous(m['points'], d['period'])
                      for m in d['metrics']))

    # ── 6. Real data ──────────────────────────────────────────────────────────
    r = http.call('GET', '/data', token=token, query={'metrics': '*', 'range': 'yesterday'})
    y = r.json() if r.status == 200 else {}
    rep.check('data 200, not a dry run, query_id', r.status == 200 and y.get('dry_run') is False and str(y.get('query_id', '')).startswith('mxq_'), f'{r.status} {r.body[:300]!r}')
    if y:
        rep.check('data echoes /time yesterday', y['period']['from'] == sc['yesterday']['from'] and y['period']['to'] == sc['yesterday']['to']
                  and y['period']['range'] == 'yesterday' and y['period']['group'] is None)
        pts = [(m, p) for m in y['metrics'] for p in m['points']]
        rep.check('data: decimal-string values; a closed period is final', all(DEC.match(p['value']) and p['final'] is True for _, p in pts))
        rep.check('data: ratio points carry numerator and denominator', all('numerator' in p and 'denominator' in p for m, p in pts if m['kind'] == 'ratio'))
        rep.check('data: money points carry a currency', all(isinstance(p.get('currency'), str) for m, p in pts if m['unit'] == 'currency'))
        rep.check('data: scope matches /health', y.get('scope') == cl.get('scope'))

    # ── 7. date= equals from/to; POST /data ───────────────────────────────────
    yday = iso(sc['yesterday']['from']).astimezone(zone).date().isoformat()
    tday = iso(today['from']).astimezone(zone).date().isoformat()
    d1 = http.call('GET', '/data', token=token, query={'metrics': '*', 'date': yday}).json()
    d2 = http.call('GET', '/data', token=token, query={'metrics': '*', 'from': yday, 'to': tday}).json()
    rep.check('date= equals the matching from/to (and echoes range null)',
              d1['period']['from'] == d2['period']['from'] and d1['period']['to'] == d2['period']['to']
              and d1['period']['range'] is None and values(d1) == values(d2))
    r = http.call('POST', '/data', token=token, body={'metrics': ids[:2], 'range': 'yesterday'})
    rep.check('POST /data with a JSON body', r.status == 200 and [m['id'] for m in r.json()['metrics']] == ids[:2], f'{r.status}')

    # ── 8. Errors ─────────────────────────────────────────────────────────────
    q = lambda **kw: http.call('GET', '/data', token=token, query=kw)
    problem(rep, 'unknown metric', q(metrics='__nope__', range='today'), 'METRIC_NOT_ALLOWED', 403)
    problem(rep, 'missing metrics', q(range='today'), 'METRIC_NOT_ALLOWED', 403)
    problem(rep, '* combined with ids', q(metrics='*,' + ids[0], range='today'), 'METRIC_NOT_ALLOWED', 403)
    problem(rep, 'unknown breakdown', q(metrics=ids[0], range='today', breakdown='__nope__'), 'BREAKDOWN_NOT_ALLOWED', 403)
    problem(rep, 'unknown range', q(metrics=ids[0], range='fortnight'), 'INVALID_RANGE', 400)
    problem(rep, 'from >= to', q(metrics=ids[0], **{'from': tday, 'to': tday}), 'INVALID_RANGE', 400)
    problem(rep, 'impossible date', q(metrics=ids[0], date='2026-02-30'), 'INVALID_RANGE', 400)
    problem(rep, 'range and date together', q(metrics=ids[0], range='today', date=tday), 'INVALID_RANGE', 400)
    m0 = metrics[0]
    first = date.fromisoformat(m0['available_from'])
    problem(rep, 'before available_from', q(metrics=m0['id'], **{'from': (first - timedelta(days=1)).isoformat(), 'to': (first + timedelta(days=1)).isoformat()}),
            'RANGE_BEFORE_AVAILABLE', 400)
    start = datetime.combine(first, datetime.min.time())
    body = problem(rep, 'point cap', q(metrics=m0['id'], group='hour', **{'from': start.strftime('%Y-%m-%dT%H:%M'),
                                                                          'to': (start + timedelta(hours=10001)).strftime('%Y-%m-%dT%H:%M')}),
                   'TOO_MANY_POINTS', 400)
    rep.check('TOO_MANY_POINTS suggests the next coarser group', body.get('suggested_group') == 'day', str(body))
    if cat['timezone']['override_allowed']:
        problem(rep, 'unknown timezone', q(metrics=ids[0], range='today', timezone='Mars/Base'), 'INVALID_TIMEZONE', 400)
    else:
        problem(rep, 'timezone while overrides are off', q(metrics=ids[0], range='today', timezone=tz), 'INVALID_TIMEZONE', 400)
    problem(rep, 'unknown format', q(metrics=ids[0], range='today', format='yaml'), 'FORMAT_NOT_ALLOWED', 406)
    problem(rep, 'unknown path', http.call('GET', '/__nope__', token=token), 'NOT_FOUND', 404)
    problem(rep, 'wrong method', http.call('DELETE', '/health', token=token), 'NOT_FOUND', 404)
    r = q(metrics='__nope__', range='today', format='csv')
    rep.check('errors are Problem Details JSON even when CSV is asked for', r.ctype.startswith('application/problem+json'), r.ctype)

    # ── 9. Revocation ─────────────────────────────────────────────────────────
    t2 = http.call('POST', '/token', form=creds).json()['access_token']
    rr = http.call('POST', '/token/revoke', form={'token': t2, 'client_id': a.client_id, 'client_secret': a.client_secret})
    rep.check('revoke answers 200 {}', rr.status == 200 and rr.json() == {}, f'{rr.status} {rr.body!r}')
    problem(rep, 'revoked token', http.call('GET', '/health', token=t2), 'TOKEN_EXPIRED', 401)
    rep.check('other tokens keep working', http.call('GET', '/health', token=token).status == 200)
    ru = http.call('POST', '/token/revoke', form={'token': 'mx_at_unknown', 'client_id': a.client_id, 'client_secret': a.client_secret})
    rep.check('revoke of an unknown token answers 200 {}', ru.status == 200 and ru.json() == {})

    # ── 10. Suppression ───────────────────────────────────────────────────────
    tested = 0
    for m in metrics:
        for b in m['breakdowns']:
            r = q(metrics=m['id'], range='last_7_days', group='day', breakdown=b)
            if r.status != 200:
                rep.check(f'breakdown {m["id"]}/{b} answers', False, f'{r.status} {r.body[:200]!r}')
                continue
            tested += 1
            pts = r.json()['metrics'][0]['points']
            # Generic, for any provider: a ratio group below 5 in its denominator must be Other. (A provider whose
            # denominator is not a record count — VibeBack's csat is 5 x ratings — pins its own rule in its own tests.)
            size = lambda p: Decimal(p['denominator'] if m['kind'] == 'ratio' else p['value'])
            rep.check(f'suppression {m["id"]}/{b}: no group under 5 except Other; every point names the breakdown',
                      all(p['breakdown'] and p['breakdown']['name'] == b and (p['breakdown']['value'] == 'Other' or size(p) >= 5) for p in pts))
    if not tested:
        rep.skip('suppression', 'no ticked metric has a breakdown')

    # ── 11. Formats, spec files and paging ────────────────────────────────────
    if a.formats:
        check_formats(rep, http, token, a.xsd_validator, {'metrics': '*', 'range': 'last_7_days', 'group': 'day'})
    else:
        rep.skip('formats', 'pass --formats (needs a client with json, xml and csv enabled)')
    if a.paging_query:
        check_paging(rep, http, token, dict(urllib.parse.parse_qsl(a.paging_query)))
    else:
        rep.skip('paging', 'pass --paging-query with a query that exceeds 10,000 points')

    # ── 12. Expiry (opt-in, slow) ─────────────────────────────────────────────
    if a.wait_expiry:
        time.sleep(t['expires_in'] + 5)
        problem(rep, 'expired token', http.call('GET', '/health', token=token), 'TOKEN_EXPIRED', 401)
    else:
        rep.skip('token expiry', 'pass --wait-expiry to wait expires_in seconds')

    # ── 13. Not reproducible from outside ─────────────────────────────────────
    for code in ('CLIENT_DISABLED', 'ACCOUNT_SUSPENDED', 'SCOPE_DISABLED', 'UNAVAILABLE'):
        rep.skip(f'error {code}', "needs a change on the provider side; the provider's own tests cover it")

    # ── 14. Rate limit (opt-in, last: it locks this client's token door for up to a minute) ──
    if a.check_rate_limit:
        limited = None
        for _ in range(25):
            r = http.call('POST', '/token', form={**creds, 'client_secret': 'wrong'})
            if r.status == 429:
                limited = r
                break
        if limited is None:
            rep.check('error RATE_LIMITED (token door)', False, 'no 429 within 25 calls')
        else:
            problem(rep, 'token door', limited, 'RATE_LIMITED', 429)
            rep.check('RATE_LIMITED carries Retry-After', limited.headers.get('retry-after', '').isdigit())
    else:
        rep.skip('error RATE_LIMITED', 'pass --check-rate-limit')

    # ── 15. Per-address token budget (opt-in, very last: locks this ADDRESS out of /token for up to a minute) ──
    # §13: 60 calls a minute per address across /token and /token/revoke, whatever the client id. Unknown client ids
    # keep every per-caller budget fresh, so only the per-address limit can stop the 61st call.
    if a.check_address_limit:
        # The budget is per clock minute and we cannot see where the provider's minute starts: 130 calls guarantee
        # that one window holds more than 60, however a boundary splits them.
        limited, sent = None, 0
        for k in range(130):
            r = http.call('POST', '/token', form={'grant_type': 'client_credentials', 'client_id': f'mxc_conformance_{k:03d}', 'client_secret': 'x'})
            sent += 1
            if r.status == 429:
                limited = r
                break
        if limited is None:
            rep.check('per-address token budget', False, 'no 429 within 130 calls from one address')
        else:
            problem(rep, 'per-address token budget', limited, 'RATE_LIMITED', 429)
            rep.check('per-address 429 carries Retry-After', limited.headers.get('retry-after', '').isdigit(), f'429 on call {sent}')
    else:
        rep.skip('per-address token budget', 'pass --check-address-limit (locks this address out of /token for a minute)')

    return rep.finish()


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