"""Admin page and fault injection checks (T1-T11). T12 (admin off without the token) needs a restart and is manual.

Configuration comes from environment variables (see README.md in this folder):
  TABBY_SIM_BASE   simulator base URL            (default http://localhost:4030/tabby-sim)
  TABBY_SIM_KEY    Authorization header value    (default "Bearer sk_test_000000000", the simulator's test key)
  TABBY_SIM_ADMIN_TOKEN   the admin password the server was started with (required)
  TABBY_SIM_RECEIVER_LOG  file the receiver's output is redirected to (default receiver.log)
  TABBY_SIM_RECEIVER      receiver base URL (default http://localhost:9199)
  TABBY_SIM_DB            database name, read through the mysql client (credentials from ~/.my.cnf)
"""
import os
import sys

DB = os.environ.get("TABBY_SIM_DB") or sys.exit("set TABBY_SIM_DB: the database name, e.g. TABBY_SIM_DB=mydb")
ADMIN_PW = os.environ.get("TABBY_SIM_ADMIN_TOKEN") or sys.exit("set TABBY_SIM_ADMIN_TOKEN: the admin password")
import base64, json, re, subprocess, sys, threading, time, urllib.error, urllib.parse, urllib.request, uuid

BASE = os.environ.get("TABBY_SIM_BASE", "http://localhost:4030/tabby-sim")
KEY = os.environ.get("TABBY_SIM_KEY", "Bearer sk_test_000000000")
BASIC = "Basic " + base64.b64encode(f"admin:{ADMIN_PW}".encode()).decode()
RECEIVER = os.environ.get("TABBY_SIM_RECEIVER", "http://localhost:9199")
RECEIVER_LOG = os.environ.get("TABBY_SIM_RECEIVER_LOG", "receiver.log")
HOOK = RECEIVER + "/ok"
RUN = str(int(time.time()))[-5:]
WORKER = "DaProductApp.TabbySim.ExpiryWorker"
BAD = {"status": "error", "errorType": "internal", "error": "Internal Server Error"}
LOCK = {"status": "error", "errorType": "conflict", "error": "could not acquire lock: payment is locked"}
DISPUTED = {"status": "error", "error": "payment is disputed"}
REFUND_LOCK = {"status": "error", "error": "could not acquire lock: payment is locked"}


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *a, **k):
        return None


opener = urllib.request.build_opener(NoRedirect)


def call(method, path, body=None, form=None, bearer=True, basic=None, merchant=None, headers=None):
    data, hdrs = None, dict(headers or {})
    if body is not None:
        data = json.dumps(body).encode()
        hdrs["content-type"] = "application/json"
    if form is not None:
        data = urllib.parse.urlencode(form).encode()
    if bearer and basic is None:
        hdrs["Authorization"] = KEY
    if basic is not None:
        hdrs["Authorization"] = basic
    if merchant is not None:
        hdrs["X-Merchant-Code"] = merchant
    req = urllib.request.Request(BASE + path, data=data, method=method, headers=hdrs)
    try:
        r = opener.open(req, timeout=30)
    except urllib.error.HTTPError as e:
        r = e
    text = r.read().decode()
    status = r.status if hasattr(r, "status") else r.code
    try:
        parsed = json.loads(text)
    except ValueError:
        parsed = text
    return status, parsed, r.headers


def admin_get(path):
    return call("GET", path, bearer=False, basic=BASIC)


def admin_post(path, token, fields):
    return call("POST", path, form={"_token": token, **fields}, bearer=False, basic=BASIC)


def form_token():
    st, html, _ = admin_get("/admin")
    assert st == 200, st
    return re.search(r'name="_token" value="([^"]*)"', html).group(1)


def sql(query):
    out = subprocess.run(["mysql", DB, "-N", "-B", "-e", query], capture_output=True, text=True)
    if out.returncode != 0:
        raise RuntimeError(out.stderr)
    return [line.split("\t") for line in out.stdout.splitlines() if line]


results = []


def check(name, ok, detail=""):
    results.append(bool(ok))
    print(("PASS  " if ok else "FAIL  ") + name + (f"   [{detail}]" if (detail and not ok) else ""), flush=True)


def info(msg):
    print("      " + msg, flush=True)


def wait_until(fn, timeout=15, every=0.4):
    end = time.time() + timeout
    val = fn()
    while not val and time.time() < end:
        time.sleep(every)
        val = fn()
    return val


def location_msg(headers):
    loc = headers.get("location", "")
    q = urllib.parse.parse_qs(urllib.parse.urlparse(loc).query)
    return (q.get("kind", [""])[0], q.get("msg", [""])[0], loc)


def register(merchant):
    return call("POST", "/api/v1/webhooks", body={"url": HOOK, "header": {"title": "X-Sign", "value": "secret-7"}}, merchant=merchant)


def payment(merchant, approve=True, phone="500000001"):
    body = {"merchant_code": merchant, "payment": {"amount": "100.00", "currency": "AED", "buyer": {"phone": phone, "email": "otp.success@tabby.ai"},
            "order": {"reference_id": "T-step7", "items": [{"title": "Phone", "quantity": 1, "unit_price": "100.00"}]}},
            "merchant_urls": {"success": "https://example.com/ok", "cancel": "https://example.com/cancel", "failure": "https://example.com/fail"}}
    st, d, _ = call("POST", "/api/v2/checkout", body=body)
    assert st == 200, (st, d)
    sid, pid = d["id"], d["payment"]["id"]
    if approve:
        call("POST", f"/hpp/{sid}", form={"action": "approve", "otp": "8888"}, bearer=False)
    return sid, pid


def get(pid):
    return call("GET", f"/api/v2/payments/{pid}")


def capture(pid, amount, ref):
    return call("POST", f"/api/v2/payments/{pid}/captures", body={"amount": amount, "reference_id": ref})


def refund(pid, amount, ref):
    return call("POST", f"/api/v2/payments/{pid}/refunds", body={"amount": amount, "reference_id": ref})


def arm(token, endpoint, kind, payment_id="", count="1"):
    return admin_post("/admin/faults", token, {"endpoint": endpoint, "kind": kind, "payment_id": payment_id, "count": count})


def deliveries(pid):
    rows = sql(f"SELECT id, event, attempt, state, IFNULL(http_status,'-'), CAST(request_body AS CHAR) FROM tabby_sim_webhook_deliveries WHERE payment_id='{pid}' ORDER BY id")
    return [{"id": int(r[0]), "event": r[1], "attempt": int(r[2]), "state": r[3], "http": r[4], "body": json.loads(r[5]) if r[5] != "NULL" else None} for r in rows]


def lines(pid):
    return [l for l in open(RECEIVER_LOG).read().splitlines() if pid in l]


def line_body(line):
    return json.loads(line.split("| ", 2)[2])


def faults_left():
    return int(sql("SELECT COUNT(*) FROM tabby_sim_faults WHERE remaining > 0")[0][0])


def main():
    sql("DELETE FROM tabby_sim_faults")   # start clean (this table only holds test faults)
    print(f"== Step 7 HTTP checks, started {time.strftime('%H:%M:%S')} UTC-ish local", flush=True)

    # ------------------------------------------------------------------ T1 protection
    print("\n-- T1 admin protection", flush=True)
    st, body, h = call("GET", "/admin", bearer=False)
    check("T1 no password -> 401 with Basic challenge", st == 401 and "Basic" in h.get("www-authenticate", ""), f"{st} {h.get('www-authenticate')}")
    st, _, _ = call("GET", "/admin", bearer=False, basic="Basic " + base64.b64encode(b"admin:wrong").decode())
    check("T1 wrong password -> 401", st == 401)
    st, html, h = admin_get("/admin")
    check("T1 right password -> 200 HTML dashboard", st == 200 and "Tabby simulator admin" in html and "Make the next call fail" in html and "Armed faults" in html, str(st))
    check("T1 the admin token itself is never shown on the page", ADMIN_PW not in html)
    check("T1 settings table says the token is set", "TABBY_SIM_ADMIN_TOKEN" in html and ">set<" in html)
    check("T1 no-store cache header", "no-store" in h.get("cache-control", ""), h.get("cache-control", ""))
    st, txt, _ = call("POST", "/admin/faults", form={"endpoint": "capture", "kind": "conflict_lock"}, bearer=False, basic=BASIC)
    check("T1 POST without form token -> 400", st == 400 and "Invalid or expired form token" in str(txt), f"{st} {txt}")
    st, txt, _ = call("POST", "/admin/faults", form={"_token": "garbage", "endpoint": "capture", "kind": "conflict_lock"}, bearer=False, basic=BASIC)
    check("T1 POST with a forged token -> 400", st == 400)
    token = form_token()
    st, txt, _ = call("POST", "/admin/faults", form={"_token": token + "x", "endpoint": "capture", "kind": "conflict_lock"}, bearer=False, basic=BASIC)
    check("T1 POST with a tampered token -> 400", st == 400)
    st, _, _ = call("POST", "/admin/faults", form={"_token": token, "endpoint": "capture", "kind": "conflict_lock"}, bearer=False)
    check("T1 POST with a valid token but no password -> 401 (not 400)", st == 401, str(st))
    check("T1 none of those armed anything", faults_left() == 0)
    st, _, _ = call("GET", "/admin", bearer=True)  # the API bearer key must not open the admin page
    check("T1 the API bearer key does not open the admin page", st == 401, str(st))

    # ------------------------------------------------------------------ T2 409 lock
    print("\n-- T2 409 lock on capture", flush=True)
    sid, pid = payment("adm2" + RUN)
    st, _, h = arm(token, "capture", "conflict_lock", pid)
    kind, msg, loc = location_msg(h)
    check("T2 arm -> 302 ok redirect to the admin page", st == 302 and kind == "ok" and loc.startswith("/tabby-sim/admin?") and "conflict_lock" in msg, f"{st} {loc}")
    st, html, _ = admin_get("/admin")
    check("T2 the armed fault is listed on the dashboard with the payment id", pid in html and "could not acquire lock" in html)
    st, d, _ = capture(pid, "100.00", "cap-1")
    check("T2 capture -> 409 with the documented lock body (errorType conflict)", st == 409 and d == LOCK, f"{st} {d}")
    st, d, _ = get(pid)
    check("T2 nothing was captured", d["captures"] == [] and d["status"] == "AUTHORIZED")
    st, d, _ = capture(pid, "100.00", "cap-1")
    check("T2 the same capture again -> 200, CLOSED (the fault is used up)", st == 200 and d["status"] == "CLOSED" and len(d["captures"]) == 1, f"{st}")
    check("T2 the used fault is gone from the table", faults_left() == 0)

    # ------------------------------------------------------------------ T3 409 disputed
    print("\n-- T3 409 payment is disputed on refund", flush=True)
    sid, pid = payment("adm3" + RUN)
    capture(pid, "100.00", "cap-1")
    arm(token, "refund", "conflict_disputed", pid)
    st, d, _ = refund(pid, "30.00", "ref-1")
    check("T3 refund -> 409 with EXACTLY {status, error} and NO errorType", st == 409 and d == DISPUTED and "errorType" not in d, f"{st} {d}")
    st, d, _ = get(pid)
    check("T3 no refund was recorded", d["refunds"] == [])
    st, d, _ = refund(pid, "30.00", "ref-1")
    check("T3 the same refund again -> 200", st == 200 and len(d["refunds"]) == 1)
    # the lock fault also works on refund, and a capture fault does not hit a refund
    arm(token, "refund", "conflict_lock", pid)
    st, d, _ = capture(pid, "1.00", "x")   # CLOSED: normal 400, the refund fault must not fire here
    check("T3 a refund fault does not fire on a capture call (normal 400)", st == 400 and d.get("error") == "could not capture not authorized payments", f"{st} {d}")
    st, d, _ = refund(pid, "10.00", "ref-2")
    check("T3 409 lock on refund: the error codes page refund body, NO errorType", st == 409 and d == REFUND_LOCK and "errorType" not in d, f"{st} {d}")
    st, d, _ = refund(pid, "10.00", "ref-2")
    check("T3 ...then the refund works", st == 200 and len(d["refunds"]) == 2)

    # ------------------------------------------------------------------ T4 500 before/after
    print("\n-- T4 500 before / after the change, idempotent retry", flush=True)
    sid, pid = payment("adm4" + RUN)
    arm(token, "capture", "internal_before", pid)
    st, d, _ = capture(pid, "100.00", "cap-1")
    check("T4 500 'before': documented internal body", st == 500 and d == BAD, f"{st} {d}")
    st, d, _ = get(pid)
    check("T4 ...nothing applied (AUTHORIZED, no captures)", d["status"] == "AUTHORIZED" and d["captures"] == [])
    arm(token, "capture", "internal_after", pid)
    st, d, _ = capture(pid, "100.00", "cap-1")
    check("T4 500 'after': the same body", st == 500 and d == BAD, f"{st} {d}")
    st, d, _ = get(pid)
    check("T4 ...but the capture WAS applied (CLOSED, 1 capture)", d["status"] == "CLOSED" and len(d["captures"]) == 1, f"{d['status']} {len(d['captures'])}")
    st, d, _ = capture(pid, "100.00", "cap-1")
    check("T4 retry with the same reference_id -> 200, still ONE capture (idempotent)", st == 200 and len(d["captures"]) == 1, f"{st} {len(d['captures'])}")
    sid, pid2 = payment("adm4" + RUN)
    arm(token, "close", "internal_after", pid2)
    st, d, _ = call("POST", f"/api/v2/payments/{pid2}/close")
    check("T4 500 'after' on close: 500...", st == 500 and d == BAD)
    check("T4 ...and the close WAS applied", get(pid2)[1]["status"] == "CLOSED")
    sid, pid3 = payment("adm4" + RUN)
    arm(token, "refund", "internal_after", pid3)
    capture(pid3, "100.00", "c")
    st, d, _ = refund(pid3, "20.00", "r-1")
    check("T4 500 'after' on refund: 500 and the refund IS recorded", st == 500 and len(get(pid3)[1]["refunds"]) == 1)
    # every endpoint accepts a 500 'before'
    sid, pid4 = payment("adm4" + RUN)
    for endpoint, call_fn in [
        ("retrieve", lambda: get(pid4)),
        ("update", lambda: call("PUT", f"/api/v2/payments/{pid4}", body={"order": {"reference_id": "X"}})),
        ("close", lambda: call("POST", f"/api/v2/payments/{pid4}/close")),
        ("list", lambda: call("GET", "/api/v2/payments?limit=1")),
    ]:
        arm(token, endpoint, "internal_before", "" if endpoint == "list" else pid4)
        st, d, _ = call_fn()
        check(f"T4 500 'before' on {endpoint}", st == 500 and d == BAD, f"{st} {d}")
    check("T4 a 500 on update/close changed nothing (still AUTHORIZED, reference T-step7)", get(pid4)[1]["status"] == "AUTHORIZED" and get(pid4)[1]["order"]["reference_id"] == "T-step7")

    # ------------------------------------------------------------------ T5 fault before expire-on-read
    print("\n-- T5 a fault fires before expire-on-read", flush=True)
    sid, pid = payment("adm5" + RUN, approve=False)
    sql(f"DELETE FROM oban_jobs WHERE worker='{WORKER}' AND state IN ('available','scheduled','retryable')")
    arm(token, "retrieve", "internal_before", pid)
    sql(f"UPDATE tabby_sim_payments SET expires_at = UTC_TIMESTAMP() - INTERVAL 5 MINUTE WHERE id='{pid}'")
    st, d, _ = get(pid)
    check("T5 retrieve of a DUE payment -> 500", st == 500 and d == BAD, f"{st} {d}")
    r = sql(f"SELECT status, is_expired FROM tabby_sim_payments WHERE id='{pid}'")[0]
    check("T5 ...and the payment is still CREATED, not expired", r == ["CREATED", "0"], str(r))
    st, d, _ = get(pid)
    check("T5 the next retrieve (fault used up) expires it: 200 EXPIRED", st == 200 and d["status"] == "EXPIRED", f"{st}")

    # ------------------------------------------------------------------ T6 webhook endpoints + checkout
    print("\n-- T6 500 on webhook endpoints and checkout", flush=True)
    m6 = "adm6" + RUN
    arm(token, "webhook_register", "internal_before")
    st, d, _ = call("POST", "/api/v1/webhooks", body={"url": HOOK}, merchant=m6)
    check("T6 POST /api/v1/webhooks -> 500", st == 500 and d == BAD, f"{st} {d}")
    st, d, _ = call("GET", "/api/v1/webhooks", merchant=m6)
    check("T6 ...nothing was registered", st == 200 and d is None, f"{st} {d}")
    st, d, _ = call("POST", "/api/v1/webhooks", body={"url": HOOK}, merchant=m6)
    check("T6 the same call again -> 200", st == 200 and d["url"] == HOOK)
    arm(token, "webhook_list", "internal_before")
    st, d, _ = call("GET", "/api/v1/webhooks", merchant=m6)
    check("T6 GET /api/v1/webhooks -> 500", st == 500 and d == BAD)
    st, d, _ = call("GET", "/api/v1/webhooks", merchant=m6)
    check("T6 ...then 200 with the list", st == 200 and len(d) == 1)
    arm(token, "checkout", "internal_before")
    before = int(sql("SELECT COUNT(*) FROM tabby_sim_payments")[0][0])
    st, d, _ = call("POST", "/api/v2/checkout", body={"merchant_code": m6, "payment": {"amount": "10.00", "currency": "AED", "buyer": {"phone": "500000001", "email": "a@b.c"}}})
    check("T6 checkout -> 500 and no payment created", st == 500 and d == BAD and int(sql("SELECT COUNT(*) FROM tabby_sim_payments")[0][0]) == before, f"{st}")
    st, d, _ = call("POST", "/api/v2/checkout", body={"merchant_code": m6, "payment": {"amount": "10.00", "currency": "AED", "buyer": {"phone": "500000001", "email": "a@b.c"}}})
    check("T6 checkout again -> 200", st == 200)
    st, d, _ = call("POST", "/api/v1/webhooks", body={"url": HOOK}, merchant=m6, bearer=False)
    check("T6 a fault never answers before authentication (no key -> 401)", st == 401, str(st))

    # ------------------------------------------------------------------ T7 resend + delay
    print("\n-- T7 resend (duplicate) and delay", flush=True)
    m7 = "adm7" + RUN
    register(m7)
    sid, pid = payment(m7)
    wait_until(lambda: len(lines(pid)) >= 1)
    d0 = wait_until(lambda: deliveries(pid))
    check("T7 first delivery: authorize, success, HTTP 200", len(d0) == 1 and d0[0]["event"] == "authorize" and d0[0]["state"] == "success" and d0[0]["http"] == "200", str(d0))
    did = d0[0]["id"]
    st, _, h = admin_post("/admin/webhooks/resend", token, {"delivery_id": str(did), "delay": "0"})
    kind, msg, _ = location_msg(h)
    check("T7 resend -> 302 ok", st == 302 and kind == "ok" and "queued again" in msg, f"{st} {kind} {msg}")
    ok = wait_until(lambda: len(lines(pid)) >= 2)
    check("T7 the receiver got the SAME payload a second time", ok and line_body(lines(pid)[0]) == line_body(lines(pid)[1]), f"lines={len(lines(pid))}")
    rows = wait_until(lambda: len(deliveries(pid)) >= 2 and deliveries(pid))
    check("T7 the delivery log has a second row for the event (attempt 1, success)", rows and [r["event"] for r in rows] == ["authorize", "authorize"] and rows[1]["state"] == "success", str(rows))
    t0 = time.time()
    st, _, h = admin_post("/admin/webhooks/resend", token, {"delivery_id": str(did), "delay": "8"})
    kind, msg, _ = location_msg(h)
    check("T7 resend with a delay -> 302 ok, message mentions 8 s", st == 302 and kind == "ok" and "8 s" in msg, msg)
    time.sleep(3)
    check("T7 ...not delivered after 3 s", len(lines(pid)) == 2, str(len(lines(pid))))
    ok = wait_until(lambda: len(lines(pid)) >= 3, timeout=20)
    waited = time.time() - t0
    check("T7 ...delivered after about 8 s", ok and 6 <= waited <= 16, f"{waited:.1f}s")
    for bad, why in [({"delivery_id": "999999999", "delay": "0"}, "unknown delivery"), ({"delivery_id": str(did), "delay": "-5"}, "negative delay"), ({"delivery_id": str(did), "delay": "99999"}, "delay too large"), ({"delivery_id": "abc", "delay": "0"}, "bad id")]:
        st, _, h = admin_post("/admin/webhooks/resend", token, bad)
        kind, msg, _ = location_msg(h)
        check(f"T7 resend refused: {why}", st == 302 and kind == "error", f"{st} {kind} {msg}")

    # ------------------------------------------------------------------ T8 out of order
    print("\n-- T8 out of order", flush=True)
    m8 = "adm8" + RUN
    register(m8)
    sid, pid = payment(m8)
    capture(pid, "60.00", "cap-1")
    capture(pid, "40.00", "cap-2")
    wait_until(lambda: len(lines(pid)) >= 4)
    # authorize, capture 60, capture 40 (still 'authorized': a full capture is authorized-with-capture ...), then 'closed'
    check("T8 four webhooks arrived in order: authorized, authorized, authorized, closed", [line_body(l)["status"] for l in lines(pid)] == ["authorized", "authorized", "authorized", "closed"], str([line_body(l)["status"] for l in lines(pid)]))
    n_before = len(lines(pid))
    st, _, h = admin_post(f"/admin/payments/{pid}/out-of-order", token, {"delay": "0"})
    kind, msg, _ = location_msg(h)
    check("T8 out-of-order -> 302 ok, '4 webhook(s) queued newest first'", st == 302 and kind == "ok" and msg.startswith("4 webhook(s) queued newest first"), msg)
    wait_until(lambda: len(lines(pid)) >= n_before + 4, timeout=20)
    new = lines(pid)[n_before:]
    seq = [(line_body(l)["status"], len(line_body(l)["captures"])) for l in new]
    check("T8 they arrive newest first: closed (2 captures), authorized (2), authorized (1), authorized (0)", seq == [("closed", 2), ("authorized", 2), ("authorized", 1), ("authorized", 0)], str(seq))
    times = [l.split(" ")[0] for l in new]
    info("arrival times: " + ", ".join(times))
    sid, pid1 = payment(m8)
    wait_until(lambda: len(lines(pid1)) >= 1)
    st, _, h = admin_post(f"/admin/payments/{pid1}/out-of-order", token, {"delay": "0"})
    kind, msg, _ = location_msg(h)
    check("T8 a payment with only one event is refused ('needs at least two')", st == 302 and kind == "error" and "at least two" in msg, msg)
    st, _, h = admin_post(f"/admin/payments/not-a-uuid/out-of-order", token, {"delay": "0"})
    check("T8 a bad payment id is refused", location_msg(h)[0] == "error")

    # ------------------------------------------------------------------ T9 force status
    print("\n-- T9 force status", flush=True)
    m9 = "adm9" + RUN
    register(m9)
    sid, pid = payment(m9, approve=False)
    st, _, h = admin_post(f"/admin/payments/{pid}/force-status", token, {"status": "CLOSED", "webhook": "true"})
    kind, msg, _ = location_msg(h)
    check("T9 CREATED -> CLOSED refused (not a documented move)", st == 302 and kind == "error" and "not a documented move" in msg and get(pid)[1]["status"] == "CREATED", msg)
    st, _, h = admin_post(f"/admin/payments/{pid}/force-status", token, {"status": "AUTHORIZED", "webhook": "true"})
    check("T9 CREATED -> AUTHORIZED ok", location_msg(h)[0] == "ok" and get(pid)[1]["status"] == "AUTHORIZED")
    ok = wait_until(lambda: len(lines(pid)) >= 1)
    check("T9 ...with an 'authorized' webhook", ok and line_body(lines(pid)[0])["status"] == "authorized")
    st, _, h = admin_post(f"/admin/payments/{pid}/force-status", token, {"status": "CLOSED", "webhook": "true"})
    check("T9 AUTHORIZED -> CLOSED ok", location_msg(h)[0] == "ok" and get(pid)[1]["status"] == "CLOSED")
    ok = wait_until(lambda: len(lines(pid)) >= 2)
    b = line_body(lines(pid)[1]) if ok else {}
    check("T9 ...with a 'closed' webhook carrying closed_at", b.get("status") == "closed" and b.get("closed_at") and b.get("captures") == [], str(b))
    for to in ("AUTHORIZED", "EXPIRED", "REJECTED"):
        st, _, h = admin_post(f"/admin/payments/{pid}/force-status", token, {"status": to, "webhook": "true"})
        check(f"T9 nothing moves out of CLOSED (to {to} refused)", location_msg(h)[0] == "error")
    # rejected up front: expire, no webhook (SEND_EXPIRED is on)
    sid, pre = payment(m9, approve=False, phone="500000002")
    st, _, h = admin_post(f"/admin/payments/{pre}/force-status", token, {"status": "EXPIRED", "webhook": "true"})
    time.sleep(4)
    r = sql(f"SELECT status, is_expired FROM tabby_sim_payments WHERE id='{pre}'")[0]
    check("T9 forced EXPIRED on a checkout rejected up front: EXPIRED, is_expired 1", r == ["EXPIRED", "1"], str(r))
    check("T9 ...and NO webhook (expiry webhook rules)", len(deliveries(pre)) == 0 and len(lines(pre)) == 0)
    sid, exp = payment(m9, approve=False)
    admin_post(f"/admin/payments/{exp}/force-status", token, {"status": "EXPIRED", "webhook": "true"})
    ok = wait_until(lambda: len(lines(exp)) >= 1)
    b = line_body(lines(exp)[0]) if ok else {}
    check("T9 forced EXPIRED on a normal payment: 'expired' webhook with is_expired true and expired_at", b.get("status") == "expired" and b.get("is_expired") is True and b.get("expired_at"), str(b))
    sid, nohook = payment(m9, approve=False)
    admin_post(f"/admin/payments/{nohook}/force-status", token, {"status": "AUTHORIZED"})   # checkbox absent = off
    time.sleep(4)
    check("T9 'send the webhook' off: status changes, no webhook", get(nohook)[1]["status"] == "AUTHORIZED" and len(deliveries(nohook)) == 0)
    sid, rej = payment(m9, approve=False)
    admin_post(f"/admin/payments/{rej}/force-status", token, {"status": "REJECTED", "webhook": "true"})
    ok = wait_until(lambda: len(lines(rej)) >= 1)
    check("T9 forced REJECTED: 'rejected' webhook", ok and line_body(lines(rej)[0])["status"] == "rejected")
    st, _, h = admin_post(f"/admin/payments/{'0' * 8}-0000-4000-8000-{'0' * 12}/force-status", token, {"status": "EXPIRED"})
    check("T9 unknown payment refused", location_msg(h)[0] == "error")
    st, _, h = admin_post(f"/admin/payments/{pid}/force-status", token, {"status": "CREATED"})
    check("T9 a status outside the four targets refused", location_msg(h)[0] == "error")

    # ------------------------------------------------------------------ T10 update webhook
    print("\n-- T10 webhook on update", flush=True)
    m10 = "adm10" + RUN
    register(m10)
    sid, pid = payment(m10)
    wait_until(lambda: len(lines(pid)) >= 1)
    st, d, _ = call("PUT", f"/api/v2/payments/{pid}", body={"order": {"reference_id": "NEW-REF-1"}})
    check("T10 PUT -> 200 with the new reference", st == 200 and d["order"]["reference_id"] == "NEW-REF-1")
    ok = wait_until(lambda: len(lines(pid)) >= 2)
    b = line_body(lines(pid)[1]) if ok else {}
    check("T10 'update' webhook: status as before (authorized), new order.reference_id", b.get("status") == "authorized" and b.get("order") == {"reference_id": "NEW-REF-1"}, str(b))
    ev = [r["event"] for r in deliveries(pid)]
    check("T10 delivery log events: authorize, update", ev == ["authorize", "update"], str(ev))
    n = len(lines(pid))
    call("PUT", f"/api/v2/payments/{pid}", body={})
    time.sleep(4)
    check("T10 PUT with nothing to update sends nothing", len(lines(pid)) == n)
    capture(pid, "100.00", "c")
    wait_until(lambda: len(lines(pid)) >= n + 1)
    call("PUT", f"/api/v2/payments/{pid}", body={"order": {"reference_id": "NEW-REF-2"}})
    ok = wait_until(lambda: len(lines(pid)) >= n + 2)
    b = line_body(lines(pid)[-1]) if ok else {}
    check("T10 update on a CLOSED payment: webhook status 'closed', new reference", b.get("status") == "closed" and b.get("order") == {"reference_id": "NEW-REF-2"}, str(b))
    sid, created = payment(m10, approve=False)
    st, d, _ = call("PUT", f"/api/v2/payments/{created}", body={"order": {"reference_id": "X"}})
    time.sleep(3)
    check("T10 update on a CREATED payment: 400 and no webhook", st == 400 and len(deliveries(created)) == 0)

    # ------------------------------------------------------------------ T11 faults list / remove / refused
    print("\n-- T11 fault list, removal, refused combinations", flush=True)
    ghost = str(uuid.uuid4())
    st, _, h = arm(token, "capture", "conflict_lock", ghost, "2")
    fid = sql("SELECT MAX(id) FROM tabby_sim_faults")[0][0]
    st, html, _ = admin_get("/admin")
    check("T11 the fault is listed with its payment id and 2 uses left", ghost in html and ">2<" in html)
    st, _, h = admin_post(f"/admin/faults/{fid}/delete", token, {})
    kind, msg, _ = location_msg(h)
    check("T11 remove -> 302 ok 'Fault N removed.'", st == 302 and kind == "ok" and "removed" in msg, msg)
    check("T11 ...gone from the table", int(sql(f"SELECT COUNT(*) FROM tabby_sim_faults WHERE id={fid}")[0][0]) == 0)
    st, _, h = admin_post(f"/admin/faults/{fid}/delete", token, {})
    check("T11 removing it again says 'already gone'", "already gone" in location_msg(h)[1], location_msg(h)[1])
    for fields, why in [
        ({"endpoint": "close", "kind": "conflict_lock", "count": "1"}, "lock on close"),
        ({"endpoint": "capture", "kind": "conflict_disputed", "count": "1"}, "disputed on capture"),
        ({"endpoint": "update", "kind": "internal_after", "count": "1"}, "500 after on update"),
        ({"endpoint": "webhook_register", "kind": "internal_after", "count": "1"}, "500 after on webhooks"),
        ({"endpoint": "checkout", "kind": "internal_before", "payment_id": ghost, "count": "1"}, "payment id on checkout"),
        ({"endpoint": "capture", "kind": "conflict_lock", "payment_id": "not-a-uuid", "count": "1"}, "bad payment id"),
        ({"endpoint": "capture", "kind": "conflict_lock", "count": "0"}, "count 0"),
        ({"endpoint": "capture", "kind": "conflict_lock", "count": "101"}, "count 101"),
        ({"endpoint": "bogus", "kind": "conflict_lock", "count": "1"}, "unknown endpoint"),
        ({"endpoint": "capture", "kind": "bogus", "count": "1"}, "unknown kind"),
    ]:
        st, _, h = admin_post("/admin/faults", token, fields)
        check(f"T11 refused: {why}", st == 302 and location_msg(h)[0] == "error", f"{st} {location_msg(h)}")
    check("T11 none of those armed anything", faults_left() == 0)

    # ------------------------------------------------------------------ extra: atomic use under parallel calls, pages, escaping
    print("\n-- extra: atomic faults under parallel calls, pages, escaping", flush=True)
    sid, pid = payment("adm11" + RUN)
    arm(token, "capture", "conflict_lock", pid, "3")
    out = []
    barrier = threading.Barrier(8)

    def go(i):
        barrier.wait()
        out.append(capture(pid, "10.00", f"par-{i}")[0])

    ts = [threading.Thread(target=go, args=(i,)) for i in range(8)]
    [t.start() for t in ts]; [t.join() for t in ts]
    check("extra a fault with 3 uses and 8 parallel captures: exactly 3 get the 409", out.count(409) == 3 and out.count(200) == 5, str(sorted(out)))
    caps = get(pid)[1]["captures"]
    total = sum(float(c["amount"]) for c in caps)
    check("extra ...5 captures of 10.00 applied, nothing double-counted, total 50 never above the payment's 100", len(caps) == 5 and total == 50.0 and total <= 100.0, f"{len(caps)} captures, total {total}")
    check("extra ...the fault row is gone", faults_left() == 0)
    evil = "<b>x</b>" + RUN
    sid, pe = payment(evil, approve=False)
    st, html, _ = admin_get("/admin")
    check("extra HTML is escaped: a merchant code with markup shows as text", "&lt;b&gt;x&lt;/b&gt;" in html and "<b>x</b>" not in html)
    st, html, h = admin_get(f"/admin/payments/{pe}")
    check("extra payment page: 200, force form, out-of-order form, JSON, escaped", st == 200 and "force-status" in html and "out-of-order" in html and "Payment object" in html and "<b>x</b>" not in html, str(st))
    st, _, h = admin_get(f"/admin/payments/{str(uuid.uuid4())}")
    check("extra unknown payment page -> redirect with an error", st == 302 and location_msg(h)[0] == "error")
    st, _, h = admin_get("/admin/payments/not-a-uuid")
    check("extra bad id page -> redirect with an error", st == 302 and location_msg(h)[0] == "error")
    st, html, _ = admin_get("/admin?kind=error&msg=%3Cscript%3Ealert(1)%3C/script%3E")
    check("extra a notice from the query string is escaped", "<script>alert(1)" not in html and "&lt;script&gt;" in html)
    st, html, _ = admin_get(f"/admin/payments/{pid}")
    check("extra a final-state payment page has no force form", "final state" in html and "/force-status" not in html)
    st, _, _ = call("GET", "/admin", bearer=False, basic=BASIC, headers={})
    check("extra the dashboard lists deliveries and webhooks sections", "Latest webhook deliveries" in admin_get("/admin")[1] and "Registered webhooks" in admin_get("/admin")[1])

    # ------------------------------------------------------------------ cleanup
    sql("DELETE FROM tabby_sim_faults")
    check("cleanup: no armed fault left behind", faults_left() == 0)
    print(f"\n{sum(results)}/{len(results)} passed", flush=True)


if __name__ == "__main__":
    main()