"""Deterministic offline conformance examples, NOT independently operated interop."""
from copy import deepcopy
from hashlib import sha256
import json
from pathlib import Path

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

import reference as c

NOW = 1800000000
ISSUER = "https://attester.example/issuer"
SENDER = "https://attester.example/senders/example-company"
PROMO = "https://types.example/promotional"
TRANS = "https://types.example/transactional"
CONTROL = "https://attester.example/policies/oidc-01-v1"
ANCHOR = "https://profiles.example/TEST-ONLY-c01-anchor"
RECIPIENT = "Alice+offers@example.net"
CONSENT_ID = "urn:uuid:11111111-1111-4111-8111-111111111111"
KIDS = {kind: "TEST-ONLY-" + kind for kind in c.ROLES}
# Publicly reproducible test keys. Never use for real attestations or trust.
KEYS = {kind: Ed25519PrivateKey.from_private_bytes(sha256(
    ("XCIM C01 PUBLIC TEST KEY " + kind).encode()).digest()) for kind in c.ROLES}


def digest(text):
    return c.b64(sha256(text.encode()).digest())


def envelope(obj):
    return c.sign(obj, KEYS[obj["kind"]], KIDS[obj["kind"]])


def unpack(env):
    return c.strict_json(c.unb64(env["payload"]), canonical=True)


def policy():
    return {
        "test_only": True, "issuer": ISSUER, "third_party_admitted": True,
        "not_before": NOW - 10000, "expires_at": NOW + 20000,
        "clock_uncertainty": 0,
        "keys": {KIDS[kind]: {"public_key": c.b64(key.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw)),
            "roles": [c.ROLES[kind]], "not_before": NOW - 10000,
            "not_after": NOW + 20000, "revoked": False,
            "historical_accepted": True} for kind, key in KEYS.items()},
        "sender_epoch_floors": {SENDER: 1},
        "control_policies": {CONTROL: {"max_observation_age": 300}},
        "revocation_methods": {"recipient": ["recipient-session-v1"],
            "sender": ["sender-surrender-v1"], "attester": ["protective-action-v1"]},
        "log_ids": ["https://attester.example/logs/consent"],
        "anchor_profiles": {ANCHOR: {"context": {"network_id": "TEST-ONLY-local",
            "genesis_hash": "TEST-ONLY-genesis", "anchor_id": "TEST-ONLY-anchor"}}}}


def query():
    return {"canonical_recipient": RECIPIENT, "issuer": ISSUER,
        "sender_id": SENDER, "sender_epoch": 1, "type_id": PROMO,
        "type_version": 1, "evaluation_time": NOW}


def fixture(event_count=0):
    sender = {"profile": c.PROFILE, "kind": "sender", "issuer": ISSUER,
        "sender_id": SENDER, "sender_epoch": 1, "name": "Example Company",
        "scope": "Example Company retail service only",
        "entity_evidence_commitment": digest("TEST ONLY private entity evidence"),
        "issued_at": NOW - 1000, "expires_at": NOW + 10000}
    grant = {"profile": c.PROFILE, "kind": "consent", "issuer": ISSUER,
        "consent_id": CONSENT_ID, "sender_id": SENDER, "sender_epoch": 1,
        "sender_binding_commitment": c.commitment(sender), "address_profile": "ascii-idna15-1",
        "recipient_salt": digest("TEST ONLY deterministic salt"), "recipient_commitment": digest("placeholder"),
        "types": [{"type_id": PROMO, "version": 1, "title": "Promotional",
                   "description": "Offers and promotions for the stated service."},
                  {"type_id": TRANS, "version": 1, "title": "Transactional",
                   "description": "Order and service updates for the stated service."}],
        "notice": {"language": "en", "text": "I explicitly permit the identified company to send these two types of email."},
        "control": {"method": "oidc-01", "policy_id": CONTROL,
            "observed_at": NOW - 910, "evidence_commitment": digest("TEST ONLY observed session evidence")},
        "decision": "explicit_consent", "accepted_at": NOW - 900,
        "issued_at": NOW - 890, "expires_at": NOW + 5000}
    grant["recipient_commitment"] = c.recipient_commitment(grant, RECIPIENT)
    gh = c.commitment(grant)
    events, head = [], gh
    for idx in range(event_count):
        event = {"profile": c.PROFILE, "kind": "revocation", "issuer": ISSUER,
            "event_id": f"urn:uuid:{idx + 2:08d}-2222-4222-8222-222222222222",
            "consent_id": CONSENT_ID, "consent_commitment": gh, "sequence": idx + 1,
            "previous_event_commitment": head, "type_ids": [[PROMO], [TRANS]][idx],
            "actor": "recipient", "method": "recipient-session-v1",
            "evidence_commitment": digest("TEST ONLY withdrawal " + str(idx)),
            "requested_at": NOW - 90 + idx, "effective_at": NOW - 80 + idx,
            "issued_at": NOW - 70 + idx}
        events.append(event)
        head = c.commitment(event)
    status = {"profile": c.PROFILE, "kind": "status", "issuer": ISSUER,
        "consent_id": CONSENT_ID, "consent_commitment": gh,
        "sender_id": SENDER, "sender_epoch": 1, "sequence": len(events),
        "event_head": head, "active_type_ids": [PROMO, TRANS][event_count:],
        "this_update": NOW - 10, "next_update": NOW + 290}
    batch = {"profile": c.PROFILE, "kind": "anchor_batch", "issuer": ISSUER,
        "log_id": "https://attester.example/logs/consent", "sequence": 1,
        "previous_batch_commitment": None, "created_at": NOW - 100,
        "leaves": [gh]}
    wrapper = {"profile_id": ANCHOR, "network_id": "TEST-ONLY-local",
        "genesis_hash": "TEST-ONLY-genesis", "anchor_id": "TEST-ONLY-anchor",
        "block_hash": "TEST-ONLY-block", "transaction_hash": "TEST-ONLY-transaction",
        "record_index": 0, "batch_commitment": c.commitment(batch),
        "proof": {"test_only": "no-consensus-proof"}}
    return {"profile": c.PROFILE, "sender": envelope(sender), "consent": envelope(grant),
        "revocations": [envelope(e) for e in events], "status": envelope(status),
        "consent_anchor": {"batch": envelope(batch), "evidence": wrapper},
        "revocation_anchors": []}


def change(ev, slot, key, value):
    obj = unpack(ev[slot])
    obj[key] = value
    ev[slot] = envelope(obj)


def change_event(ev, key, value, idx=0):
    obj = unpack(ev["revocations"][idx])
    obj[key] = value
    ev["revocations"][idx] = envelope(obj)


def change_batch(ev, key, value):
    obj = unpack(ev["consent_anchor"]["batch"])
    obj[key] = value
    ev["consent_anchor"]["batch"] = envelope(obj)
    ev["consent_anchor"]["evidence"]["batch_commitment"] = c.commitment(obj)


def main():
    vectors, results = [], []
    adapters = {ANCHOR: c.fixture_anchor_adapter}
    def case(name, expected, mutate=None, events=0, qchange=None, pchange=None,
             reason=None, raw=None, floors=None, expected_checks=None):
        ev, q, p = fixture(events), query(), policy()
        if mutate:
            mutate(ev)
        if qchange:
            q.update(qchange)
        if pchange:
            pchange(p)
        data = raw if raw is not None else c.jcs(ev)
        actual = c.verify(data, q, p, adapters, floors)
        passed = actual["outcome"] == expected and (reason is None or actual["reason"] == reason)
        if expected_checks is not None:
            passed = passed and actual["consent_checks_result"] == expected_checks
        passed = passed and actual["can_assert_xcim_pass"] is False
        results.append({"id": name, "expected": expected, "expected_reason": reason,
                        "passed": passed, "actual": actual})
        vectors.append({"id": name, "evidence_base64url": c.b64(data), "query": q,
            "policy": p, "expected_outcome": expected, "expected_reason": reason,
            "stateful_floor_test": floors is not None})
        return actual

    case("consent-query-without-email", "INDETERMINATE", reason="ANCHOR_TEST_ONLY", expected_checks="CONSENT_CURRENT")
    case("transactional-requires-and-has-explicit-consent", "INDETERMINATE", qchange={"type_id": TRANS}, expected_checks="CONSENT_CURRENT")
    floors = c.FloorStore()
    case("same-query-first", "INDETERMINATE", floors=floors, expected_checks="CONSENT_CURRENT")
    before = deepcopy(floors.records)
    case("same-query-repeat-not-consumed", "INDETERMINATE", floors=floors, expected_checks="CONSENT_CURRENT")
    assert before == floors.records
    case("domain-case-and-outer-space", "INDETERMINATE", qchange={"canonical_recipient": " \tAlice+offers@EXAMPLE.NET\t "}, expected_checks="CONSENT_CURRENT")
    for name, changes in [
        ("wrong-recipient", {"canonical_recipient": "Bob@example.net"}),
        ("local-part-case-preserved", {"canonical_recipient": "alice+offers@example.net"}),
        ("plus-part-preserved", {"canonical_recipient": "Alice@example.net"}),
        ("different-company", {"sender_id": "https://attester.example/senders/another-company"}),
        ("different-sender-epoch", {"sender_epoch": 2}),
        ("different-issuer-query", {"issuer": "https://other.example/issuer"}),
        ("different-type", {"type_id": "https://types.example/other"}),
        ("different-type-version", {"type_version": 2}),
    ]:
        case(name, "NO_MATCH", qchange=changes)
    case("no-message-input-accepted", "INVALID_EVIDENCE", qchange={"email": "From: x@example.net"}, reason="QUERY_FIELDS")
    for address in ['"Alice"@example.net', "Alíce@example.net", "Alice..x@example.net", "Alice@example.net."]:
        case("unsupported-address-" + str(len(vectors)), "INVALID_EVIDENCE", qchange={"canonical_recipient": address}, reason="ADDRESS_SYNTAX")
    case("partial-withdrawal", "CONSENT_REVOKED", events=1)
    case("partial-withdrawal-retains-other-type", "INDETERMINATE", events=1,
         qchange={"type_id": TRANS}, expected_checks="CONSENT_CURRENT")
    case("all-types-withdrawn", "CONSENT_REVOKED", events=2, qchange={"type_id": TRANS})
    def add_revocation_anchor(e, invalid=False):
        event = unpack(e["revocations"][0])
        previous = unpack(e["consent_anchor"]["batch"])
        batch = deepcopy(previous)
        batch.update(sequence=2, previous_batch_commitment=c.commitment(previous),
                     created_at=NOW-60, leaves=[c.commitment(event)])
        wrapper = deepcopy(e["consent_anchor"]["evidence"])
        wrapper.update(batch_commitment=c.commitment(batch), record_index=1)
        if invalid:
            wrapper["proof"] = {"finalized": True}
        e["revocation_anchors"] = [{"event_id": event["event_id"],
            "batch": envelope(batch), "evidence": wrapper}]
    actual = case("withdrawal-anchor-presence-is-not-verification", "CONSENT_REVOKED", events=1,
                  mutate=add_revocation_anchor)
    assert len(actual["pending_revocation_anchors"]) == 1
    actual = case("bad-withdrawal-anchor-cannot-restore-consent", "CONSENT_REVOKED", events=1,
                  mutate=lambda e: add_revocation_anchor(e, True))
    assert len(actual["pending_revocation_anchors"]) == 1
    floor = c.FloorStore()
    case("record-authenticated-revocation", "CONSENT_REVOKED", events=1, floors=floor)
    case("old-active-snapshot-cannot-restore", "CONSENT_REVOKED", floors=floor)
    case("old-sequence-for-unremoved-type", "INDETERMINATE", floors=floor,
         qchange={"type_id": TRANS}, reason="OLDER_THAN_KNOWN_FLOOR")
    def fork_history(e):
        change_event(e, "evidence_commitment", digest("different authenticated withdrawal evidence"))
        change(e, "status", "event_head", c.commitment(unpack(e["revocations"][0])))
    case("signed-history-fork", "INDETERMINATE", events=1, floors=floor,
         mutate=fork_history, qchange={"type_id": TRANS}, reason="CONFLICTED_HISTORY")
    case("history-conflict-remains-latched", "INDETERMINATE", events=1, floors=floor,
         qchange={"type_id": TRANS}, reason="CONFLICTED_HISTORY")
    def renew_status(e):
        obj = unpack(e["status"])
        obj.update(this_update=NOW, next_update=NOW+300)
        e["status"] = envelope(obj)
    case("snapshot-renewal-does-not-change-grant", "INDETERMINATE", floors=floors,
         mutate=renew_status, expected_checks="CONSENT_CURRENT")
    type_floor = c.FloorStore()
    case("record-authenticated-type-definition", "INDETERMINATE", floors=type_floor)
    def new_grant_changed_type(e):
        grant = unpack(e["consent"])
        grant["consent_id"] = "urn:uuid:99999999-9999-4999-8999-999999999999"
        grant["types"][0]["description"] = "Any unsolicited advertising from unrelated companies."
        grant["recipient_commitment"] = c.recipient_commitment(grant, RECIPIENT)
        e["consent"] = envelope(grant)
        gh = c.commitment(grant)
        status = unpack(e["status"])
        status.update(consent_id=grant["consent_id"], consent_commitment=gh, event_head=gh)
        e["status"] = envelope(status)
        change_batch(e, "leaves", [gh])
    case("same-type-version-changed-meaning", "INDETERMINATE", floors=type_floor,
         mutate=new_grant_changed_type, reason="TYPE_DEFINITION_CONFLICT")
    case("type-definition-conflict-remains-latched", "INDETERMINATE", floors=type_floor,
         reason="TYPE_DEFINITION_CONFLICT")
    known_key = c.jcs([ISSUER, SENDER, 1, PROMO, 1]).decode()
    case("policy-known-type-definition-conflict", "INDETERMINATE",
         pchange=lambda p: p.update(type_definitions={known_key: {"title": "Different", "description": "Different"}}),
         reason="TYPE_DEFINITION_CONFLICT")
    case("expired-grant", "CONSENT_EXPIRED", qchange={"evaluation_time": NOW + 5000})
    case("expiration-exclusive-with-clock-uncertainty", "INDETERMINATE", qchange={"evaluation_time": NOW + 4995},
         pchange=lambda p: p.update(clock_uncertainty=5), reason="STATUS_FRESHNESS")
    case("stale-status", "INDETERMINATE", qchange={"evaluation_time": NOW + 290}, reason="STATUS_FRESHNESS")
    case("future-status", "INDETERMINATE", qchange={"evaluation_time": NOW - 11}, reason="STATUS_FRESHNESS")
    case("uncertainty-upper-bound", "INDETERMINATE", pchange=lambda p: p.update(clock_uncertainty=10), expected_checks="CONSENT_CURRENT")
    case("clock-uncertainty-too-large", "INVALID_EVIDENCE", pchange=lambda p: p.update(clock_uncertainty=61), reason="CLOCK_POLICY")
    case("policy-expired", "INDETERMINATE", pchange=lambda p: p.update(expires_at=NOW), reason="POLICY_TIME")
    case("third-party-not-admitted", "INVALID_EVIDENCE", pchange=lambda p: p.update(third_party_admitted=False), reason="ISSUER_NOT_ADMITTED")
    case("sender-epoch-floor", "INVALID_EVIDENCE", pchange=lambda p: p["sender_epoch_floors"].update({SENDER: 2}), reason="SENDER_EPOCH")
    case("unapproved-control-policy", "INVALID_EVIDENCE", pchange=lambda p: p.update(control_policies={}), reason="CONTROL_POLICY")
    case("control-observation-too-old", "INVALID_EVIDENCE", pchange=lambda p: p["control_policies"][CONTROL].update(max_observation_age=5), reason="CONTROL_AGE")
    case("role-confusion", "INVALID_EVIDENCE", pchange=lambda p: p["keys"][KIDS["consent"]].update(roles=["sender_binding"]), reason="KEY_ROLE")
    case("revoked-attestation-key", "INVALID_EVIDENCE", pchange=lambda p: p["keys"][KIDS["consent"]].update(revoked=True), reason="KEY_ROLE")
    case("signature-outside-key-window", "INVALID_EVIDENCE", pchange=lambda p: p["keys"][KIDS["consent"]].update(not_before=NOW), reason="KEY_TIME")
    case("routine-retired-historical-key", "INDETERMINATE", pchange=lambda p: p["keys"][KIDS["consent"]].update(not_after=NOW-500), expected_checks="CONSENT_CURRENT")
    case("status-key-retired", "INDETERMINATE", pchange=lambda p: p["keys"][KIDS["status"]].update(not_after=NOW), reason="STATUS_KEY_TIME")
    case("bad-signature", "INVALID_EVIDENCE", mutate=lambda e: e["consent"].update(signature=c.b64(bytes(64))), reason="SIGNATURE")
    case("signer-spoofed-issuer", "INVALID_EVIDENCE", mutate=lambda e: change(e, "consent", "issuer", "https://evil.example/issuer"), reason="ISSUER")
    case("extra-consent-property", "INVALID_EVIDENCE", mutate=lambda e: change(e, "consent", "message_id", "irrelevant"), reason="SCHEMA")
    case("implicit-transaction-relationship", "INVALID_EVIDENCE", mutate=lambda e: change(e, "consent", "decision", "transaction_relationship"), reason="SCHEMA")
    case("unknown-version", "UNSUPPORTED_PROFILE", mutate=lambda e: e.update(profile="xcim-consent-99"))
    case("unknown-payload-profile", "UNSUPPORTED_PROFILE", mutate=lambda e: change(e, "consent", "profile", "xcim-consent-99"))
    case("sender-binding-substitution", "INVALID_EVIDENCE", mutate=lambda e: change(e, "sender", "scope", "Another company and all affiliates"), reason="SENDER_BINDING")
    case("sender-uri-userinfo", "INVALID_EVIDENCE", mutate=lambda e: change(e, "sender", "sender_id", "https://evil@attester.example/company"), reason="IDENTIFIER_URI")
    case("sender-uri-query", "INVALID_EVIDENCE", mutate=lambda e: change(e, "sender", "sender_id", SENDER + "?tenant=2"), reason="IDENTIFIER_URI")
    case("integer-boolean-rejected", "INVALID_EVIDENCE", mutate=lambda e: change(e, "sender", "sender_epoch", True), reason="SCHEMA")
    case("empty-type-selection", "INVALID_EVIDENCE", mutate=lambda e: change(e, "consent", "types", []), reason="SCHEMA")
    case("types-unsorted", "INVALID_EVIDENCE", mutate=lambda e: change(e, "consent", "types", list(reversed(unpack(e["consent"])["types"]))), reason="TYPES_ORDER")
    case("status-active-set-fabricated", "INVALID_EVIDENCE", mutate=lambda e: change(e, "status", "active_type_ids", [TRANS]), reason="STATUS_PROJECTION")
    case("status-sequence-gap", "INVALID_EVIDENCE", mutate=lambda e: change(e, "status", "sequence", 1), reason="STATUS_PROJECTION")
    case("status-validity-too-long", "INVALID_EVIDENCE", mutate=lambda e: change(e, "status", "next_update", NOW+291), reason="STATUS_INTERVAL")
    case("revocation-chain-substitution", "INVALID_EVIDENCE", events=1,
         mutate=lambda e: change_event(e, "previous_event_commitment", digest("wrong")), reason="EVENT_CHAIN")
    case("revocation-sequence-gap", "INVALID_EVIDENCE", events=1,
         mutate=lambda e: change_event(e, "sequence", 2), reason="EVENT_CHAIN")
    case("revocation-actor-spoof", "INVALID_EVIDENCE", events=1,
         mutate=lambda e: change_event(e, "actor", "sender"), reason="ACTOR_METHOD")
    case("revocation-future-effective", "INVALID_EVIDENCE", events=1,
         mutate=lambda e: change_event(e, "effective_at", NOW+1), reason="EVENT_TIME")
    case("revocation-unknown-type", "INVALID_EVIDENCE", events=1,
         mutate=lambda e: change_event(e, "type_ids", ["https://types.example/other"]), reason="EVENT_TYPES")
    case("revocation-duplicate-removal", "INVALID_EVIDENCE", events=2,
         mutate=lambda e: change_event(e, "type_ids", [PROMO], 1), reason="EVENT_TYPES")
    case("revocation-regrant-forbidden", "INVALID_EVIDENCE", events=1,
         mutate=lambda e: change_event(e, "action", "reactivate"), reason="SCHEMA")
    case("grant-anchor-unavailable", "INDETERMINATE", mutate=lambda e: e["consent_anchor"]["evidence"].update(proof={"test_only": "unavailable"}), reason="ANCHOR_UNAVAILABLE")
    case("issuer-finalized-flag-not-proof", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(proof={"finalized": True}), reason="ANCHOR_PROOF")
    case("proof-names-are-profile-specific", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(proof={"issuer": 1}), reason="ANCHOR_PROOF")
    case("proof-must-be-object", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(proof=[]), reason="SCHEMA")
    case("anchor-identifiers-ascii-only", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(network_id="réseau"), reason="SCHEMA")
    case("anchor-identifier-length-bound", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(network_id="x" * 513), reason="SCHEMA")
    case("unknown-anchor-profile", "UNSUPPORTED_PROFILE", mutate=lambda e: e["consent_anchor"]["evidence"].update(profile_id="https://profiles.example/unknown"), reason="ANCHOR_PROFILE")
    case("wrong-network", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(network_id="another"), reason="ANCHOR_PROOF")
    case("wrong-genesis", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(genesis_hash="another"), reason="ANCHOR_PROOF")
    case("wrong-anchor-authority", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(anchor_id="another"), reason="ANCHOR_PROOF")
    case("missing-required-anchor-field", "INVALID_EVIDENCE", mutate=lambda e: e.pop("consent_anchor"), reason="SCHEMA")
    case("anchor-digest-substitution", "INVALID_EVIDENCE", mutate=lambda e: e["consent_anchor"]["evidence"].update(batch_commitment=digest("wrong")), reason="ANCHOR_COMMITMENT")
    case("anchor-nonmember", "INVALID_EVIDENCE", mutate=lambda e: change_batch(e, "leaves", [digest("another-object")]), reason="ANCHOR_MEMBER")
    case("anchor-predates-grant", "INVALID_EVIDENCE", mutate=lambda e: change_batch(e, "created_at", NOW-900), reason="ANCHOR_MEMBER")
    case("batch-reset-with-previous-hash", "INVALID_EVIDENCE", mutate=lambda e: change_batch(e, "previous_batch_commitment", digest("old")), reason="BATCH_PREVIOUS")
    case("batch-over-4096-leaves", "INVALID_EVIDENCE", mutate=lambda e: change_batch(e, "leaves", sorted(digest("leaf"+str(i)) for i in range(4097))), reason="SCHEMA")
    case("batch-large-envelope-valid", "INDETERMINATE", mutate=lambda e: change_batch(e, "leaves", sorted(
         [c.commitment(unpack(e["consent"]))] + [digest("leaf"+str(i)) for i in range(4095)])), expected_checks="CONSENT_CURRENT")
    case("unadmitted-log", "INVALID_EVIDENCE", mutate=lambda e: change_batch(e, "log_id", "https://evil.example/log"), reason="LOG_AUTHORITY")
    def noncanonical(e):
        obj = unpack(e["consent"])
        e["consent"]["payload"] = c.b64(json.dumps(obj, indent=2).encode())
    case("noncanonical-signed-payload", "INVALID_EVIDENCE", mutate=noncanonical, reason="JSON_NONCANONICAL")
    def wrong_header(e):
        header = c.strict_json(c.unb64(e["consent"]["protected"]))
        header["jku"] = "https://evil.example/key"
        e["consent"]["protected"] = c.b64(c.jcs(header))
    case("key-url-injection", "INVALID_EVIDENCE", mutate=wrong_header, reason="SCHEMA")
    case("padded-base64url", "INVALID_EVIDENCE", mutate=lambda e: e["consent"].update(payload=e["consent"]["payload"] + "="), reason="SCHEMA")
    for name, raw, reason in [
        ("duplicate-json-member", b'{"profile":"xcim-consent-01","profile":"xcim-consent-01"}', "JSON_DUPLICATE"),
        ("floating-number", b'{"x":1.0}', "JSON_INTEGER"),
        ("negative-zero", b'{"x":-0}', "JSON_INTEGER"),
        ("unsafe-integer", b'{"x":9007199254740992}', "JSON_INTEGER"),
        ("invalid-unicode-scalar", b'{"x":"\\ud800"}', "JSON_UNICODE"),
        ("depth-limit", b'[' * 17 + b'0' + b']' * 17, "JSON_DEPTH"),
        ("complete-evidence-size-limit", b' ' * (2*1024*1024+1), "JSON_LIMIT"),
    ]:
        case(name, "INVALID_EVIDENCE", raw=raw, reason=reason)
    # Invalid signed evidence must not poison authenticated local floors.
    floor = c.FloorStore()
    case("invalid-signature-cannot-change-floors", "INVALID_EVIDENCE", events=1, floors=floor,
         mutate=lambda e: e["revocations"][0].update(signature=c.b64(bytes(64))), reason="SIGNATURE")
    assert floor.records == {}
    case("invalid-active-anchor-cannot-change-floors", "INVALID_EVIDENCE", floors=floor,
         mutate=lambda e: e["consent_anchor"]["evidence"].update(proof={"finalized": True}), reason="ANCHOR_PROOF")
    assert floor.records == {}
    # Canonicalization assertions are independent of all fixture object signatures.
    assert c.canonical_address("Alice@bücher.example") == "Alice@xn--bcher-kva.example"
    assert c.canonical_address("Alice@faß.de") == "Alice@xn--fa-hia.de"
    assert c.jcs({"\U0001f600": 1, "\ue000": 2}) == '{"😀":1,"":2}'.encode()
    assert c.commitment({"kind": "consent", "x": 1}) != c.commitment({"kind": "sender", "x": 1})

    out = c.ROOT / "vectors"
    out.mkdir(exist_ok=True)
    (c.ROOT / "schemas").mkdir(exist_ok=True)
    (c.ROOT / "schemas" / "consent.schema.json").write_text(json.dumps(c.SCHEMA, indent=2) + "\n", encoding="utf-8")
    (out / "positive-consent-test-only.json").write_text(json.dumps(fixture(), indent=2) + "\n", encoding="utf-8")
    (out / "partial-revocation-test-only.json").write_text(json.dumps(fixture(1), indent=2) + "\n", encoding="utf-8")
    (out / "verifier-policy-TEST-ONLY.json").write_text(json.dumps(policy(), indent=2) + "\n", encoding="utf-8")
    (out / "query.json").write_text(json.dumps(query(), indent=2) + "\n", encoding="utf-8")
    # Oversize/depth raw data need not be duplicated in a public fixture bundle;
    # deterministic construction is retained in this source and the result report.
    (out / "cases.json").write_text(json.dumps([v for v in vectors if len(v["evidence_base64url"]) < 100000], indent=2) + "\n", encoding="utf-8")
    report = {"profile": c.PROFILE, "test_only": True, "can_assert_xcim_pass": False,
        "total": len(results), "passed": sum(r["passed"] for r in results),
        "failed": sum(not r["passed"] for r in results),
        "crypto": "Real Ed25519 signatures with public reproducible TEST ONLY keys",
        "anchor": "VERIFIED_TEST_ONLY equality stub; zero consensus or network verification",
        "limits": ["No live OIDC transaction or provider authority proof", "No blockchain consensus/finality verification",
            "No durable floor storage or crash-recovery implementation", "No independent implementation interoperation",
            "No production sender enrollment or actor authentication service", "No network retrieval/discovery implementation"],
        "results": results}
    (c.ROOT / "test-report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({k: report[k] for k in ("total", "passed", "failed", "can_assert_xcim_pass")}))
    for failure in (r for r in results if not r["passed"]):
        print(json.dumps(failure))
    assert report["failed"] == 0


if __name__ == "__main__":
    main()