"""Tests for the integration fabric.

Run against your own work:      pytest
Run against the solution:       LAB_MODULE=solution pytest
"""

from __future__ import annotations

import importlib
import os
from typing import Any, Dict, List, Mapping

import pytest

lab = importlib.import_module(os.environ.get("LAB_MODULE", "lab"))

MessageIdentifier = lab.MessageIdentifier
NS = lab.NS
Pain001Parser = lab.Pain001Parser
Severity = lab.Severity
to_minor = lab.to_minor
from_minor = lab.from_minor
currency_exponent = lab.currency_exponent
valid_iban = lab.valid_iban
valid_bic = lab.valid_bic
Finality = lab.Finality
Rail = lab.Rail
RAILS = lab.RAILS
choose_rail = lab.choose_rail
DualWriteStore = lab.DualWriteStore
Outbox = lab.Outbox
OutboxRelay = lab.OutboxRelay
MessageLog = lab.MessageLog
ConsumerGroup = lab.ConsumerGroup
IdempotentConsumer = lab.IdempotentConsumer
Compatibility = lab.Compatibility
Field = lab.Field
Schema = lab.Schema
check_backward = lab.check_backward
check_forward = lab.check_forward
check_compatibility = lab.check_compatibility
DEPLOY_ORDER = lab.DEPLOY_ORDER
SchemaRegistry = lab.SchemaRegistry
QualityRule = lab.QualityRule
DataProductContract = lab.DataProductContract
DataProduct = lab.DataProduct
BreakType = lab.BreakType
reconcile = lab.reconcile


# ======================================================================================
# helpers
# ======================================================================================


def clock(start: int = 0, step: int = 1):
    state = {"t": start - step}

    def now() -> int:
        state["t"] += step
        return state["t"]

    return now


def frozen(value: int = 0):
    return lambda: value


VALID_IBAN = "AE070331234567890123456"
VALID_IBAN_2 = "AE460090000000123456789"


def pain001(*, msg_id: str = "M1", nb_of_txs: str = "1", ctrl_sum: str = "100.00",
            exec_date: str = "2026-02-12", debtor_iban: str = VALID_IBAN,
            transactions: str = None) -> str:
    txns = transactions if transactions is not None else txn()
    ctrl = f"<CtrlSum>{ctrl_sum}</CtrlSum>" if ctrl_sum else ""
    return f"""<?xml version="1.0"?>
<Document xmlns="{NS}">
  <CstmrCdtTrfInitn>
    <GrpHdr>
      <MsgId>{msg_id}</MsgId>
      <CreDtTm>2026-02-11T09:15:00</CreDtTm>
      <NbOfTxs>{nb_of_txs}</NbOfTxs>
      {ctrl}
      <InitgPty><Nm>Falcon Trading LLC</Nm></InitgPty>
    </GrpHdr>
    <PmtInf>
      <PmtInfId>P1</PmtInfId>
      <ReqdExctnDt>{exec_date}</ReqdExctnDt>
      <Dbtr><Nm>Falcon Trading LLC</Nm><PstlAdr><Ctry>AE</Ctry></PstlAdr></Dbtr>
      <DbtrAcct><Id><IBAN>{debtor_iban}</IBAN></Id></DbtrAcct>
      <DbtrAgt><FinInstnId><BICFI>NBADAEAA</BICFI></FinInstnId></DbtrAgt>
      {txns}
    </PmtInf>
  </CstmrCdtTrfInitn>
</Document>"""


def txn(*, e2e: str = "E1", amount: str = "100.00", ccy: str = "AED",
        creditor: str = "Zenith Supplies FZE", iban: str = VALID_IBAN_2,
        omit_ccy: bool = False, omit_account: bool = False) -> str:
    amt = (f'<InstdAmt>{amount}</InstdAmt>' if omit_ccy
           else f'<InstdAmt Ccy="{ccy}">{amount}</InstdAmt>')
    acct = "" if omit_account else f"<CdtrAcct><Id><IBAN>{iban}</IBAN></Id></CdtrAcct>"
    return f"""<CdtTrfTxInf>
        <PmtId><EndToEndId>{e2e}</EndToEndId></PmtId>
        <Amt>{amt}</Amt>
        <Cdtr><Nm>{creditor}</Nm></Cdtr>
        {acct}
      </CdtTrfTxInf>"""


def parse(xml: str):
    return Pain001Parser().parse(xml)


def codes(rejections) -> List[str]:
    return [r.code for r in rejections]


def elements(rejections) -> List[str]:
    return [r.element for r in rejections]


# ======================================================================================
# 1. message identifiers
# ======================================================================================


def test_a_message_identifier_parses_into_four_parts():
    mid = MessageIdentifier.parse("pain.001.001.09")
    assert (mid.business_area, mid.message_number, mid.variant, mid.version) == \
        ("pain", "001", "001", "09")


def test_a_message_identifier_round_trips():
    assert str(MessageIdentifier.parse("pacs.008.001.10")) == "pacs.008.001.10"


def test_the_business_area_names_the_family():
    assert MessageIdentifier.parse("pain.001.001.09").family == "Payments Initiation"
    assert "Clearing" in MessageIdentifier.parse("pacs.008.001.10").family


def test_a_known_message_has_a_description():
    assert MessageIdentifier.parse("camt.053.001.08").description == \
        "BankToCustomerStatement"


def test_an_unknown_message_number_does_not_raise():
    assert MessageIdentifier.parse("pain.999.001.01").description == "unknown message"


def test_a_malformed_identifier_raises():
    for bad in ("pain.001", "pain.001.001", "PAIN.001.001.09", "pain.1.1.9"):
        with pytest.raises(ValueError):
            MessageIdentifier.parse(bad)


# ======================================================================================
# 2. amounts and minor units
# ======================================================================================


def test_most_currencies_have_two_minor_digits():
    assert currency_exponent("AED") == currency_exponent("USD") == 2


def test_jpy_has_no_minor_digits():
    assert currency_exponent("JPY") == 0


def test_kwd_has_three_minor_digits():
    assert currency_exponent("KWD") == 3


def test_an_unknown_currency_defaults_to_two():
    assert currency_exponent("ZZZ") == 2


def test_currency_lookup_is_case_insensitive():
    assert currency_exponent("jpy") == 0


def test_a_two_digit_amount_converts():
    assert to_minor("100.00", "AED") == 10_000


def test_a_jpy_amount_is_not_multiplied_by_a_hundred():
    assert to_minor("100", "JPY") == 100


def test_a_kwd_amount_uses_three_digits():
    assert to_minor("100.000", "KWD") == 100_000


def test_the_conversion_avoids_binary_float_error():
    assert to_minor("1.15", "USD") == 115          # 1.15 * 100 == 114.99999999999999


def test_extra_precision_is_refused_not_rounded():
    with pytest.raises(ValueError):
        to_minor("100.001", "AED")


def test_a_non_numeric_amount_raises():
    with pytest.raises(ValueError):
        to_minor("one hundred", "AED")


def test_minor_units_round_trip():
    for amount, ccy in (("100.00", "AED"), ("100", "JPY"), ("0.010", "KWD")):
        assert from_minor(to_minor(amount, ccy), ccy) == amount


# ======================================================================================
# 3. identifier checksums
# ======================================================================================


def test_a_valid_iban_passes():
    assert valid_iban(VALID_IBAN)


def test_an_iban_with_one_digit_changed_fails():
    assert not valid_iban("AE070331234567890123457")


def test_iban_validation_ignores_spaces():
    assert valid_iban("AE07 0331 2345 6789 0123 456")


def test_something_too_short_is_not_an_iban():
    assert not valid_iban("AE07")


def test_a_valid_eight_character_bic_passes():
    assert valid_bic("NBADAEAA")


def test_a_valid_eleven_character_bic_passes():
    assert valid_bic("NBADAEAAXXX")


def test_a_bic_of_the_wrong_length_fails():
    assert not valid_bic("NBADAE")
    assert not valid_bic("NBADAEAAXX")


def test_a_bic_with_digits_in_the_institution_code_fails():
    assert not valid_bic("NB4DAEAA")


# ======================================================================================
# 4. pain.001 parsing
# ======================================================================================


def test_a_valid_message_parses_with_no_rejections():
    instruction, rejections = parse(pain001())
    assert rejections == [] and instruction is not None


def test_a_parsed_message_carries_its_transactions():
    instruction, _ = parse(pain001())
    assert instruction.number_of_transactions == 1
    assert instruction.transfers[0].amount_minor == 10_000


def test_a_parsed_message_carries_the_debtor():
    instruction, _ = parse(pain001())
    assert instruction.debtor.name == "Falcon Trading LLC"
    assert instruction.debtor.iban == VALID_IBAN


def test_malformed_xml_is_rejected_not_raised():
    instruction, rejections = parse("<Document>")
    assert instruction is None and codes(rejections) == ["FF01"]


def test_the_wrong_namespace_is_rejected():
    instruction, rejections = parse(pain001().replace(NS, "urn:wrong"))
    assert instruction is None and codes(rejections) == ["FF02"]


def test_a_missing_group_header_is_rejected():
    xml = pain001()
    start = xml.index("<GrpHdr>")
    end = xml.index("</GrpHdr>") + len("</GrpHdr>")
    instruction, rejections = parse(xml[:start] + xml[end:])
    assert instruction is None and "GrpHdr" in rejections[0].element


def test_a_missing_message_id_names_the_element():
    instruction, rejections = parse(pain001(msg_id=""))
    assert "GrpHdr/MsgId" in elements(rejections)


def test_a_non_iso_execution_date_is_rejected():
    _, rejections = parse(pain001(exec_date="12/02/2026"))
    assert "FF03" in codes(rejections)


def test_an_invalid_debtor_iban_is_rejected():
    _, rejections = parse(pain001(debtor_iban="AE070331234567890123457"))
    assert "FF07" in codes(rejections)


def test_a_transaction_with_no_currency_attribute_is_rejected():
    _, rejections = parse(pain001(transactions=txn(omit_ccy=True)))
    assert "MS02" in codes(rejections)


def test_an_unsupported_currency_is_rejected():
    _, rejections = parse(pain001(transactions=txn(ccy="XYZ")))
    assert "FF04" in codes(rejections)


def test_a_negative_amount_is_rejected():
    _, rejections = parse(pain001(transactions=txn(amount="-5.00")))
    assert "AM01" in codes(rejections)


def test_a_zero_amount_is_rejected():
    _, rejections = parse(pain001(transactions=txn(amount="0.00")))
    assert "AM01" in codes(rejections)


def test_a_missing_creditor_account_is_rejected():
    _, rejections = parse(pain001(transactions=txn(omit_account=True)))
    assert any("CdtrAcct" in e for e in elements(rejections))


def test_a_missing_creditor_name_is_rejected():
    _, rejections = parse(pain001(transactions=txn(creditor="")))
    assert any("Cdtr/Nm" in e for e in elements(rejections))


def test_no_transactions_is_rejected():
    _, rejections = parse(pain001(nb_of_txs="0", transactions=""))
    assert any("CdtTrfTxInf" in e for e in elements(rejections))


def test_a_duplicate_end_to_end_id_is_rejected():
    two = txn(e2e="E1") + txn(e2e="E1")
    _, rejections = parse(pain001(nb_of_txs="2", ctrl_sum="200.00", transactions=two))
    assert "DU01" in codes(rejections)


def test_a_wrong_transaction_count_is_rejected():
    _, rejections = parse(pain001(nb_of_txs="5"))
    assert "CS01" in codes(rejections)


def test_a_wrong_control_sum_is_rejected():
    _, rejections = parse(pain001(ctrl_sum="999.00"))
    assert "CS02" in codes(rejections)


def test_a_correct_control_sum_over_two_transactions_passes():
    two = txn(e2e="E1", amount="60.00") + txn(e2e="E2", amount="40.00")
    instruction, rejections = parse(
        pain001(nb_of_txs="2", ctrl_sum="100.00", transactions=two))
    assert rejections == [] and instruction.actual_sum_minor == 10_000


def test_every_problem_is_reported_not_only_the_first():
    two = txn(e2e="E1", omit_ccy=True) + txn(e2e="E2", creditor="", ccy="XYZ")
    _, rejections = parse(pain001(nb_of_txs="9", exec_date="bad", transactions=two))
    assert len(rejections) >= 5


def test_rejections_are_sorted_so_the_report_is_stable():
    two = txn(e2e="E1", omit_ccy=True) + txn(e2e="E2", ccy="XYZ")
    _, rejections = parse(pain001(nb_of_txs="9", transactions=two))
    assert rejections == sorted(rejections, key=lambda r: (r.element, r.code))


def test_a_rejected_message_yields_no_instruction():
    instruction, _ = parse(pain001(transactions=txn(ccy="XYZ")))
    assert instruction is None


def test_the_transaction_limit_is_enforced():
    parser = Pain001Parser(max_transactions=2)
    three = "".join(txn(e2e=f"E{i}") for i in range(3))
    _, rejections = parser.parse(
        pain001(nb_of_txs="3", ctrl_sum="300.00", transactions=three))
    assert "LM01" in codes(rejections)


# ======================================================================================
# 5. rails and finality
# ======================================================================================


def test_an_instant_payment_is_final_immediately():
    rail = RAILS["instant"]
    assert rail.finality_at(600, 600) is Finality.FINAL


def test_a_batch_payment_before_cutoff_is_revocable():
    rail = RAILS["ach"]
    assert rail.finality_at(9 * 60, 10 * 60) is Finality.REVOCABLE


def test_a_batch_payment_after_cutoff_is_only_conditionally_revocable():
    rail = RAILS["ach"]
    assert rail.finality_at(9 * 60, 15 * 60) is Finality.CONDITIONALLY_REVOCABLE


def test_a_settled_payment_is_final():
    rail = RAILS["ach"]
    assert rail.finality_at(9 * 60, 9 * 60 + 2000) is Finality.FINAL


def test_a_rail_with_no_recall_goes_straight_to_final():
    rail = Rail("x", "X", "AED", cutoff_minute=600, settles_after_minutes=1000,
                supports_recall=False)
    assert rail.finality_at(500, 700) is Finality.FINAL


def test_the_fastest_eligible_rail_is_chosen():
    rail, _ = choose_rail(1_000_00, "AED", 10 * 60)
    assert rail == "instant"


def test_a_payment_above_a_rails_limit_moves_to_the_next_one():
    rail, why = choose_rail(250_000_00, "AED", 10 * 60)
    assert rail == "rtgs" and "limit" in why


def test_a_payment_after_every_cutoff_gets_no_rail():
    rail, why = choose_rail(250_000_00, "AED", 23 * 60)
    assert rail is None and "cut-off" in why


def test_an_unsupported_currency_gets_no_rail():
    rail, why = choose_rail(100_00, "ZZZ", 10 * 60)
    assert rail is None and "ZZZ" in why


def test_an_urgent_payment_will_not_take_a_slow_rail():
    rail, _ = choose_rail(10_000_00, "USD", 10 * 60, urgent=True)
    assert rail is None


def test_a_non_urgent_payment_will_take_the_slow_rail():
    rail, _ = choose_rail(10_000_00, "USD", 10 * 60, urgent=False)
    assert rail == "swift"


def test_the_reason_is_always_populated():
    for args in ((1_000_00, "AED", 600), (250_000_00, "AED", 23 * 60)):
        assert choose_rail(*args)[1]


# ======================================================================================
# 6. dual write and the outbox
# ======================================================================================


def test_a_dual_write_crash_loses_the_event():
    log = MessageLog(partitions=2)
    store = DualWriteStore(broker=log, fail_after_db=True)
    with pytest.raises(RuntimeError):
        store.save_and_publish("P1", {"status": "OK"}, "payments")
    assert "P1" in store.rows and log.all("payments") == []


def test_a_dual_write_without_a_crash_does_both():
    log = MessageLog(partitions=2)
    store = DualWriteStore(broker=log)
    store.save_and_publish("P1", {"status": "OK"}, "payments")
    assert "P1" in store.rows and len(log.all("payments")) == 1


def test_the_outbox_writes_the_row_and_the_event_together():
    outbox = Outbox(now=clock())
    outbox.transact("P1", {"status": "OK"}, event_type="Released")
    assert "P1" in outbox.rows and len(outbox.records()) == 1


def test_a_crash_before_commit_writes_neither():
    outbox = Outbox(now=clock())
    with pytest.raises(RuntimeError):
        outbox.transact("P1", {"status": "OK"}, event_type="Released",
                        fail_before_commit=True)
    assert outbox.rows == {} and outbox.records() == []


def test_outbox_sequence_numbers_start_at_one_and_increment():
    outbox = Outbox(now=clock())
    for i in range(3):
        outbox.transact(f"P{i}", {}, event_type="E")
    assert [r.seq for r in outbox.records()] == [1, 2, 3]


def test_a_failed_transaction_does_not_consume_a_sequence_number():
    outbox = Outbox(now=clock())
    outbox.transact("P1", {}, event_type="E")
    with pytest.raises(RuntimeError):
        outbox.transact("P2", {}, event_type="E", fail_before_commit=True)
    outbox.transact("P3", {}, event_type="E")
    assert [r.seq for r in outbox.records()] == [1, 2]


def test_the_payload_defaults_to_the_row():
    outbox = Outbox(now=clock())
    record = outbox.transact("P1", {"status": "OK"}, event_type="E")
    assert record.payload == {"status": "OK"}


def test_a_payload_can_differ_from_the_row():
    outbox = Outbox(now=clock())
    record = outbox.transact("P1", {"status": "OK"}, event_type="E",
                             payload={"id": "P1"})
    assert record.payload == {"id": "P1"}


def test_the_relay_publishes_unpublished_records():
    outbox = Outbox(now=clock())
    log = MessageLog(partitions=2)
    relay = OutboxRelay(outbox=outbox, log=log, topic="t")
    for i in range(3):
        outbox.transact(f"P{i}", {}, event_type="E")
    assert relay.run_once() == 3
    assert len(log.all("t")) == 3


def test_the_relay_marks_records_published():
    outbox = Outbox(now=clock())
    relay = OutboxRelay(outbox=outbox, log=MessageLog(partitions=2), topic="t")
    outbox.transact("P1", {}, event_type="E")
    relay.run_once()
    assert outbox.unpublished() == []


def test_a_second_relay_run_publishes_nothing_new():
    outbox = Outbox(now=clock())
    log = MessageLog(partitions=2)
    relay = OutboxRelay(outbox=outbox, log=log, topic="t")
    outbox.transact("P1", {}, event_type="E")
    relay.run_once()
    assert relay.run_once() == 0 and len(log.all("t")) == 1


def test_a_relay_crash_after_publishing_republishes_at_least_once():
    outbox = Outbox(now=clock())
    log = MessageLog(partitions=2)
    relay = OutboxRelay(outbox=outbox, log=log, topic="t")
    for i in range(3):
        outbox.transact(f"P{i}", {}, event_type="E")
    relay.run_once(fail_after_publish=True)
    relay.run_once()
    assert len(log.all("t")) == 4          # the first one went twice


def test_the_relay_publishes_in_sequence_order():
    outbox = Outbox(now=clock())
    log = MessageLog(partitions=1)
    relay = OutboxRelay(outbox=outbox, log=log, topic="t")
    for i in range(5):
        outbox.transact(f"P{i}", {}, event_type="E")
    relay.run_once()
    assert [m.value["seq"] for m in log.read("t", 0)] == [1, 2, 3, 4, 5]


# ======================================================================================
# 7. the partitioned log
# ======================================================================================


def test_a_log_needs_at_least_one_partition():
    with pytest.raises(ValueError):
        MessageLog(partitions=0)


def test_the_same_key_always_lands_in_the_same_partition():
    log = MessageLog(partitions=8)
    assert len({log.partition_for("ACC-A") for _ in range(20)}) == 1


def test_partitioning_is_derived_so_it_survives_a_restart():
    a, b = MessageLog(partitions=8), MessageLog(partitions=8)
    assert a.partition_for("ACC-A") == b.partition_for("ACC-A")


def test_partitions_are_within_range():
    log = MessageLog(partitions=4)
    assert all(0 <= log.partition_for(f"k{i}") < 4 for i in range(50))


def test_offsets_within_a_partition_start_at_zero_and_increment():
    log = MessageLog(partitions=1)
    for i in range(3):
        log.append("t", "k", {"i": i})
    assert [m.offset for m in log.read("t", 0)] == [0, 1, 2]


def test_order_is_preserved_within_a_partition():
    log = MessageLog(partitions=4)
    for i in range(10):
        log.append("t", "ACC-A", {"seq": i})
    home = log.partition_for("ACC-A")
    seqs = [m.value["seq"] for m in log.read("t", home) if m.key == "ACC-A"]
    assert seqs == sorted(seqs)


def test_two_keys_can_share_a_partition():
    log = MessageLog(partitions=1)
    log.append("t", "a", {})
    log.append("t", "b", {})
    assert len(log.read("t", 0)) == 2


def test_reading_from_an_offset_skips_earlier_messages():
    log = MessageLog(partitions=1)
    for i in range(5):
        log.append("t", "k", {"i": i})
    assert [m.value["i"] for m in log.read("t", 0, from_offset=3)] == [3, 4]


def test_a_read_limit_is_respected():
    log = MessageLog(partitions=1)
    for i in range(5):
        log.append("t", "k", {"i": i})
    assert len(log.read("t", 0, limit=2)) == 2


def test_an_empty_partition_reads_empty():
    assert MessageLog(partitions=2).read("t", 1) == []


def test_the_high_water_mark_is_the_message_count():
    log = MessageLog(partitions=1)
    for i in range(4):
        log.append("t", "k", {})
    assert log.high_water_mark("t", 0) == 4


def test_topics_are_independent():
    log = MessageLog(partitions=1)
    log.append("a", "k", {})
    assert log.all("b") == []


# ======================================================================================
# 8. consumer groups
# ======================================================================================


def loaded_log(messages: int = 6, partitions: int = 3) -> MessageLog:
    log = MessageLog(partitions=partitions)
    for i in range(messages):
        log.append("t", f"k{i}", {"i": i})
    return log


def test_a_new_group_starts_at_offset_zero():
    log = loaded_log()
    assert ConsumerGroup("g", log).committed("t", 0) == 0


def test_polling_returns_uncommitted_messages():
    log = loaded_log(messages=6)
    assert len(ConsumerGroup("g", log).poll("t", max_records=100)) == 6


def test_polling_respects_max_records():
    log = loaded_log(messages=6)
    assert len(ConsumerGroup("g", log).poll("t", max_records=2)) == 2


def test_committing_advances_past_the_message():
    log = MessageLog(partitions=1)
    for i in range(3):
        log.append("t", "k", {"i": i})
    group = ConsumerGroup("g", log)
    group.commit("t", 0, 1)
    assert [m.value["i"] for m in group.poll("t")] == [1, 2]


def test_committing_everything_leaves_nothing_to_poll():
    log = MessageLog(partitions=1)
    for i in range(3):
        log.append("t", "k", {})
    group = ConsumerGroup("g", log)
    group.commit("t", 0, 3)
    assert group.poll("t") == []


def test_commit_cannot_rewind():
    log = loaded_log()
    group = ConsumerGroup("g", log)
    group.commit("t", 0, 2)
    with pytest.raises(ValueError):
        group.commit("t", 0, 1)


def test_seek_can_rewind_deliberately():
    log = MessageLog(partitions=1)
    for i in range(3):
        log.append("t", "k", {})
    group = ConsumerGroup("g", log)
    group.commit("t", 0, 3)
    group.seek("t", 0, 0)
    assert len(group.poll("t")) == 3


def test_seek_clamps_at_zero():
    log = loaded_log()
    group = ConsumerGroup("g", log)
    group.seek("t", 0, -5)
    assert group.committed("t", 0) == 0


def test_lag_is_the_uncommitted_count():
    log = MessageLog(partitions=1)
    for i in range(5):
        log.append("t", "k", {})
    group = ConsumerGroup("g", log)
    group.commit("t", 0, 2)
    assert group.lag("t")[0] == 3


def test_total_lag_sums_the_partitions():
    log = loaded_log(messages=6, partitions=3)
    assert ConsumerGroup("g", log).total_lag("t") == 6


def test_two_groups_track_offsets_independently():
    log = MessageLog(partitions=1)
    for i in range(3):
        log.append("t", "k", {})
    a, b = ConsumerGroup("a", log), ConsumerGroup("b", log)
    a.commit("t", 0, 3)
    assert a.poll("t") == [] and len(b.poll("t")) == 3


# ======================================================================================
# 9. idempotent consumption
# ======================================================================================


def consumer_over(values: List[Mapping[str, Any]], *, partitions: int = 1):
    log = MessageLog(partitions=partitions)
    for i, value in enumerate(values):
        log.append("t", f"k{i}", value)
    effects: List[Any] = []
    group = ConsumerGroup("g", log)
    consumer = IdempotentConsumer(
        group, lambda m: effects.append(m.value["event_id"]),
        key_of=lambda m: m.value["event_id"])
    return log, group, consumer, effects


def test_each_distinct_event_produces_one_effect():
    _, _, consumer, effects = consumer_over(
        [{"event_id": "e1"}, {"event_id": "e2"}, {"event_id": "e3"}])
    consumer.consume("t")
    assert effects == ["e1", "e2", "e3"]


def test_a_duplicate_in_the_log_produces_one_effect():
    _, _, consumer, effects = consumer_over(
        [{"event_id": "e1"}, {"event_id": "e1"}, {"event_id": "e2"}])
    consumer.consume("t")
    assert effects == ["e1", "e2"] and consumer.duplicates == 1


def test_a_full_replay_produces_no_new_effects():
    log, group, consumer, effects = consumer_over(
        [{"event_id": "e1"}, {"event_id": "e2"}])
    consumer.consume("t")
    group.seek("t", 0, 0)
    consumer.consume("t")
    assert effects == ["e1", "e2"] and consumer.duplicates == 2


def test_replaying_five_times_still_produces_one_effect_each():
    log, group, consumer, effects = consumer_over([{"event_id": "e1"}])
    for _ in range(5):
        group.seek("t", 0, 0)
        consumer.consume("t")
    assert effects == ["e1"]


def test_a_crash_before_commit_redelivers_but_does_not_re_effect():
    log, group, consumer, effects = consumer_over(
        [{"event_id": "e1"}, {"event_id": "e2"}])
    consumer.consume("t", fail_before_commit=True)
    assert effects == ["e1"] and group.committed("t", 0) == 0
    consumer.consume("t")
    assert effects == ["e1", "e2"] and consumer.duplicates == 1


def test_the_processed_set_records_every_distinct_event():
    _, _, consumer, _ = consumer_over(
        [{"event_id": "e1"}, {"event_id": "e1"}, {"event_id": "e2"}])
    consumer.consume("t")
    assert consumer.processed == {"e1", "e2"}


def test_the_default_key_uses_the_message_content_not_the_offset():
    log = MessageLog(partitions=1)
    log.append("t", "P1", {"seq": 7})
    log.append("t", "P1", {"seq": 7})
    effects: List[Any] = []
    consumer = IdempotentConsumer(ConsumerGroup("g", log),
                                  lambda m: effects.append(m.value["seq"]))
    consumer.consume("t")
    assert effects == [7]


# ======================================================================================
# 10. schema compatibility
# ======================================================================================


V1 = [Field("id", "string"), Field("amount", "int"), Field("currency", "string")]


def schema(fields, version: int = 1) -> Schema:
    return Schema("s", version, tuple(fields))


def test_adding_an_optional_field_is_backward_compatible():
    new = V1 + [Field("rail", "string", required=False)]
    assert check_backward(schema(V1), schema(new, 2)) == []


def test_adding_a_required_field_without_a_default_breaks_backward():
    new = V1 + [Field("rail", "string")]
    assert check_backward(schema(V1), schema(new, 2))


def test_adding_a_required_field_with_a_default_is_backward_compatible():
    new = V1 + [Field("rail", "string", default="rtgs")]
    assert check_backward(schema(V1), schema(new, 2)) == []


def test_removing_a_field_is_backward_compatible():
    assert check_backward(schema(V1), schema(V1[:2], 2)) == []


def test_removing_a_required_field_breaks_forward():
    assert check_forward(schema(V1), schema(V1[:2], 2))


def test_adding_a_required_field_is_forward_compatible():
    new = V1 + [Field("rail", "string")]
    assert check_forward(schema(V1), schema(new, 2)) == []


def test_a_type_change_breaks_both_directions():
    new = [Field("id", "string"), Field("amount", "string"),
           Field("currency", "string")]
    assert check_backward(schema(V1), schema(new, 2))
    assert check_forward(schema(V1), schema(new, 2))


def test_full_compatibility_is_the_union_of_both():
    new = V1 + [Field("rail", "string")]
    problems = check_compatibility(schema(V1), schema(new, 2), Compatibility.FULL)
    assert problems == check_backward(schema(V1), schema(new, 2))


def test_compatibility_none_permits_anything():
    assert check_compatibility(schema(V1), schema([], 2), Compatibility.NONE) == []


def test_every_mode_has_a_stated_deploy_order():
    assert set(DEPLOY_ORDER) == set(Compatibility)


def test_backward_means_consumers_first():
    assert DEPLOY_ORDER[Compatibility.BACKWARD].startswith("consumers")


def test_forward_means_producers_first():
    assert DEPLOY_ORDER[Compatibility.FORWARD].startswith("producers")


def test_the_first_version_registers_unconditionally():
    registry = SchemaRegistry()
    assert registry.register("s", V1).version == 1


def test_versions_increment():
    registry = SchemaRegistry()
    registry.register("s", V1)
    assert registry.register("s", V1 + [Field("x", "string", required=False)]).version == 2


def test_the_registry_refuses_an_incompatible_schema():
    registry = SchemaRegistry(default_mode=Compatibility.BACKWARD)
    registry.register("s", V1)
    with pytest.raises(ValueError, match="backward"):
        registry.register("s", V1 + [Field("rail", "string")])


def test_a_refused_registration_does_not_create_a_version():
    registry = SchemaRegistry()
    registry.register("s", V1)
    with pytest.raises(ValueError):
        registry.register("s", V1 + [Field("rail", "string")])
    assert registry.latest("s").version == 1


def test_the_mode_can_be_set_per_subject():
    registry = SchemaRegistry(default_mode=Compatibility.BACKWARD)
    registry.set_mode("s", Compatibility.FORWARD)
    registry.register("s", V1)
    registry.register("s", V1 + [Field("rail", "string")])       # forward-ok
    assert registry.latest("s").version == 2


def test_an_older_version_can_be_read_back():
    registry = SchemaRegistry()
    registry.register("s", V1)
    registry.register("s", V1 + [Field("x", "string", required=False)])
    assert len(registry.version("s", 1).fields) == 3


# ======================================================================================
# 11. data products
# ======================================================================================


TRACE_SCHEMA = Schema("traces", 1, (
    Field("trace_id", "string"), Field("cost_micros", "int")))


def product(*, now=None, slo: int = 5, rules=()) -> DataProduct:
    contract = DataProductContract("p", owner="layla", schema=TRACE_SCHEMA,
                                   freshness_slo_ticks=slo, quality_rules=tuple(rules))
    return DataProduct(contract, now=now or frozen(0))


def test_a_conforming_product_passes():
    p = product()
    p.publish([{"trace_id": "t1", "cost_micros": 10}])
    assert p.check().passed


def test_a_missing_required_field_fails():
    p = product()
    p.publish([{"trace_id": "t1"}])
    result = p.check()
    assert not result.passed and result.schema_problems


def test_a_wrong_type_fails():
    p = product()
    p.publish([{"trace_id": "t1", "cost_micros": "ten"}])
    assert not p.check().passed


def test_a_boolean_is_not_an_integer():
    p = product()
    p.publish([{"trace_id": "t1", "cost_micros": True}])
    assert not p.check().passed


def test_a_failing_quality_rule_fails_the_contract():
    rule = QualityRule("non_empty", lambda rows: len(rows) > 0, "not empty")
    p = product(rules=[rule])
    p.publish([])
    result = p.check()
    assert not result.passed and "non_empty" in result.quality_failures


def test_a_passing_quality_rule_does_not_fail_the_contract():
    rule = QualityRule("non_empty", lambda rows: len(rows) > 0, "not empty")
    p = product(rules=[rule])
    p.publish([{"trace_id": "t1", "cost_micros": 1}])
    assert p.check().passed


def test_every_failing_rule_is_named():
    rules = [QualityRule("a", lambda rows: False, ""),
             QualityRule("b", lambda rows: False, "")]
    p = product(rules=rules)
    p.publish([{"trace_id": "t1", "cost_micros": 1}])
    assert set(p.check().quality_failures) == {"a", "b"}


def test_a_product_within_its_freshness_slo_is_fresh():
    now = clock(start=0)
    p = product(now=now, slo=5)
    p.publish([{"trace_id": "t1", "cost_micros": 1}])
    assert p.check().fresh


def test_a_product_past_its_freshness_slo_is_stale():
    now = clock(start=0)
    p = product(now=now, slo=2)
    p.publish([{"trace_id": "t1", "cost_micros": 1}])
    for _ in range(10):
        now()
    result = p.check()
    assert not result.fresh and not result.passed


def test_republishing_resets_the_freshness_clock():
    now = clock(start=0)
    p = product(now=now, slo=2)
    p.publish([{"trace_id": "t1", "cost_micros": 1}])
    for _ in range(10):
        now()
    p.publish([{"trace_id": "t2", "cost_micros": 1}])
    assert p.check().fresh


def test_the_report_names_the_product_and_the_problem():
    p = product()
    p.publish([{"trace_id": "t1"}])
    report = p.check().report()
    assert report.startswith("p:") and "cost_micros" in report


def test_a_passing_report_says_so():
    p = product()
    p.publish([{"trace_id": "t1", "cost_micros": 1}])
    assert "OK" in p.check().report()


def test_schema_problems_are_capped_so_a_report_stays_readable():
    p = product()
    p.publish([{"trace_id": f"t{i}"} for i in range(50)])
    assert len(p.check().schema_problems) <= 5


# ======================================================================================
# 12. reconciliation
# ======================================================================================


def test_identical_records_produce_no_breaks():
    assert reconcile({"a": 1}, {"a": 1}) == []


def test_a_key_only_in_a_is_missing_in_b():
    breaks = reconcile({"a": 1}, {})
    assert breaks[0].break_type is BreakType.MISSING_IN_B


def test_a_key_only_in_b_is_missing_in_a():
    breaks = reconcile({}, {"a": 1})
    assert breaks[0].break_type is BreakType.MISSING_IN_A


def test_differing_values_are_a_mismatch():
    breaks = reconcile({"a": 1}, {"a": 2})
    assert breaks[0].break_type is BreakType.VALUE_MISMATCH
    assert (breaks[0].a_value, breaks[0].b_value) == (1, 2)


def test_breaks_are_sorted_by_key():
    breaks = reconcile({"z": 1, "a": 1}, {"m": 1})
    assert [b.key for b in breaks] == ["a", "m", "z"]


def test_every_break_is_reported():
    assert len(reconcile({"a": 1, "b": 1}, {"b": 2, "c": 1})) == 3


def test_reconciling_empty_records_is_clean():
    assert reconcile({}, {}) == []


def test_a_break_renders_readably():
    assert "A=1 B=2" in str(reconcile({"a": 1}, {"a": 2})[0])