recovery/tests/test_bindingdb_adapter.py

"""Focused tests for the BindingDB curated-articles adapter.

All fixtures are tiny synthetic ZIP archives built in tmp_path; no publisher
data, network access, or large files are involved.
"""

from __future__ import annotations

import hashlib
import json
import sys
import zipfile
from importlib import util as importlib_util
from pathlib import Path

import pytest
from mitointeract_recovery import bindingdb_adapter as adapter
from mitointeract_recovery.chemistry import canonicalize_smiles, stable_id
from mitointeract_recovery.source_measurement import (
    MeasurementRelation,
    MeasurementType,
    normalize_sequence,
)

SCRIPT_PATH = Path(__file__).parents[1] / "scripts" / "prepare_bindingdb_gold_kd.py"
_spec = importlib_util.spec_from_file_location("prepare_bindingdb_gold_kd", SCRIPT_PATH)
assert _spec is not None and _spec.loader is not None
prepare_script = importlib_util.module_from_spec(_spec)
_spec.loader.exec_module(prepare_script)

PAD_PREFIX = "PAD"
EXTRA_HEADER_FIELDS = [
    adapter.COL_PH,
    adapter.COL_TEMP,
    adapter.COL_CURATION,
    adapter.COL_ARTICLE_DOI,
    adapter.COL_ENTRY_DOI,
    adapter.COL_PMID,
    adapter.COL_PUBCHEM_AID,
    adapter.COL_PATENT,
    adapter.COL_DATE_PUBLICATION,
    adapter.COL_DATE_BINDINGDB,
    adapter.COL_NUM_CHAINS,
    adapter.COL_SEQUENCE_1,
    adapter.COL_PUBCHEM_CID,
    adapter.COL_PUBCHEM_SID,
    adapter.COL_CHEBI_ID,
    adapter.COL_CHEMBL_ID,
    adapter.COL_DRUGBANK_ID,
    adapter.COL_KEGG_ID,
    adapter.COL_ZINC_ID,
    "UniProt (SwissProt) Primary ID of Target Chain 1",
    "UniProt (TrEMBL) Primary ID of Target Chain 1",
]
PAD_FIELD_COUNT = (
    adapter.MAIN_EXPECTED_COLUMN_COUNT
    - len(adapter.MAIN_HEADER_PREFIX)
    - len(EXTRA_HEADER_FIELDS)
)

SEQ_SRC = "MGSNKSKPKDASQRRR"
SEQ_HSP90 = "MPEHHQTETQPMPAET"
SEQ_MUT = "MUTATEDKINASE"

SMILES_A = "COc1cc2c(Nc3ccc(Cl)cc3Cl)c(cnc2cc1O)C#N"
SMILES_B = "c1ccncc1"
SMILES_C = "CC(=O)Oc1ccccc1C(=O)O"

ROW_KI_EXACT = {
    "Ki (nM)": " 8.7",
    "IC50 (nM)": "",
    "Kd (nM)": "",
    "EC50 (nM)": "",
}
ROW_MULTI_TYPE = {
    "Ki (nM)": ">1000",
    "IC50 (nM)": "50",
    "Kd (nM)": "8.7",
    "EC50 (nM)": "~200",
}
ROW_GOLD = {
    "Ki (nM)": "5",
    "IC50 (nM)": "",
    "Kd (nM)": " 8.7",
    "EC50 (nM)": "",
}
ROW_MALFORMED = {
    "Ki (nM)": "abc",
    "IC50 (nM)": "10-20",
    "Kd (nM)": "-5",
    "EC50 (nM)": "0",
}
ROW_KD_CENSORED = {
    "Ki (nM)": "",
    "IC50 (nM)": "",
    "Kd (nM)": ">100",
    "EC50 (nM)": "",
}


def _base_row(
    rsid,
    *,
    smiles=SMILES_A,
    target_name="Proto-oncogene tyrosine-protein kinase Src",
    organism="Homo sapiens",
    chains="1",
    sequence=SEQ_SRC,
    sp="P12931",
    tr="",
    doi="10.1016/j.bmcl.2003.07.001",
    pmid="14552782",
    pub_date="11/3/2003",
    bdb_date="12/12/2017",
    curation="Curated from the literature by BindingDB",
    ph="7.4",
    temp="37.00 C",
    cells=None,
):
    row = {PAD_PREFIX: ""}
    row.update(
        {
            "BindingDB Reactant_set_id": str(rsid),
            "Ligand SMILES": smiles,
            "Ligand InChI": "InChI=1S/synthetic",
            "Ligand InChI Key": "SYNTHETICINCHIKEY-AAAA",
            "BindingDB MonomerID": "4521",
            "BindingDB Ligand Name": "synthetic ligand",
            "Target Name": target_name,
            "Target Source Organism According to Curator or DataSource": organism,
            "Ki (nM)": "",
            "IC50 (nM)": "",
            "Kd (nM)": "",
            "EC50 (nM)": "",
            "pH": ph,
            "Temp (C)": temp,
            "Curation/DataSource": curation,
            "Article DOI": doi,
            "BindingDB Entry DOI": "10.7270/Q27942VF",
            "PMID": pmid,
            "PubChem AID": "",
            "Patent Number": "",
            "Date of publication": pub_date,
            "Date in BindingDB": bdb_date,
            "Number of Protein Chains in Target (>1 implies a multichain complex)": chains,
            "BindingDB Target Chain Sequence 1": sequence,
            "PubChem CID": "5328914",
            "PubChem SID": "8034189",
            "ChEBI ID of Ligand": "",
            "ChEMBL ID of Ligand": "CHEMBL941",
            "DrugBank ID of Ligand": "",
            "KEGG ID of Ligand": "",
            "ZINC ID of Ligand": "",
            "UniProt (SwissProt) Primary ID of Target Chain 1": sp,
            "UniProt (TrEMBL) Primary ID of Target Chain 1": tr,
        }
    )
    if cells:
        row.update(cells)
    return row


MAIN_ROWS = [
    _base_row("1", cells=ROW_KI_EXACT),
    _base_row(
        "2",
        smiles=SMILES_B,
        target_name="Heat shock protein HSP 90-alpha",
        sequence=SEQ_HSP90,
        sp="P07900",
        doi="10.1000/xyz",
        pmid="99999999",
        pub_date="1/1/2020",
        cells=ROW_MULTI_TYPE,
    ),
    _base_row(
        "3",
        smiles=SMILES_C,
        target_name="Mutated Src kinase",
        sequence=SEQ_MUT,
        sp="",
        tr="A0A023GPI8",
        doi="",
        pmid="12345678",
        pub_date="5/5/2010",
        ph="",
        temp="",
        cells=ROW_GOLD,
    ),
    _base_row("4", cells=ROW_MALFORMED),
    _base_row("5", chains="2", cells=ROW_KI_EXACT),
    _base_row("6", sequence="   ", cells=ROW_KI_EXACT),
    _base_row("7", smiles="not_a_smiles", cells=ROW_KI_EXACT),
    _base_row("9999", cells=ROW_KI_EXACT),
    _base_row("10", cells=ROW_KD_CENSORED),
]

MAPPING_LINES = [
    ("1", "100_1"),
    ("2", "100_1"),
    ("2", "100_2"),
    ("2", "100_2"),  # duplicate line must collapse to one emission
    ("3", "200_1"),
    ("4", "300_1"),
    ("5", "100_1"),
    ("6", "100_1"),
    ("7", "100_1"),
    ("10", "404_1"),  # present in mapping, absent from assays export
    ("11", "100_1"),
]

ASSAY_ROWS = [
    ("100", "1", "Src kinase inhibition assay", "Hot天 kinase reaction at pH 7.4."),
    (
        "100",
        "2",
        "Src SPR direct binding",
        "Surface plasmon resonance, immobilized Src.",
    ),
    ("200", "1", "Displacement assay", "Radioligand displacement at 37 C."),
    ("300", "1", "", ""),
]


def main_header():
    return (
        list(adapter.MAIN_HEADER_PREFIX)
        + EXTRA_HEADER_FIELDS
        + [f"{PAD_PREFIX} {index}" for index in range(PAD_FIELD_COUNT)]
    )


def main_lines():
    lines = ["\t".join(main_header())]
    for row in MAIN_ROWS:
        lines.append("\t".join(row.get(name, "") for name in main_header()))
    return lines


def write_zip(path: Path, member: str, lines: list[str]) -> str:
    payload = ("\n".join(lines) + "\n").encode("utf-8")
    with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr(member, payload)
    return hashlib.sha256(path.read_bytes()).hexdigest()


def fixture_zips(tmp_path):
    paths = {
        "main_zip": tmp_path / "main.zip",
        "mapping_zip": tmp_path / "mapping.zip",
        "assays_zip": tmp_path / "assays.zip",
    }
    checksums = {
        "main_zip": write_zip(paths["main_zip"], adapter.MAIN_MEMBER, main_lines()),
        "mapping_zip": write_zip(
            paths["mapping_zip"],
            adapter.MAPPING_MEMBER,
            ["\t".join(adapter.MAPPING_HEADER)]
            + ["\t".join(pair) for pair in MAPPING_LINES],
        ),
        "assays_zip": write_zip(
            paths["assays_zip"],
            adapter.ASSAYS_MEMBER,
            ["\t".join(adapter.ASSAYS_HEADER)] + ["\t".join(row) for row in ASSAY_ROWS],
        ),
    }
    return paths, checksums


@pytest.fixture(name="fixture_zips")
def fixture_zips_fixture(tmp_path):
    return fixture_zips(tmp_path)


def make_config(checksums):
    return {
        "release": "fixture",
        "files": {
            key: {
                "url": f"https://fixtures.invalid/{key}.zip",
                "zip_name": f"{key}.zip",
                "member": adapter.DEFAULT_MEMBERS[key],
                "sha256": checksums[key],
            }
            for key in adapter.INPUT_KEYS
        },
    }


def run_fixture_pipeline(fixture_zips, tmp_path, **overrides):
    paths, checksums = fixture_zips
    kwargs = {
        "main_zip": paths["main_zip"],
        "mapping_zip": paths["mapping_zip"],
        "assays_zip": paths["assays_zip"],
        "config": make_config(checksums),
        "output_dir": tmp_path / "out",
    }
    kwargs.update(overrides)
    return adapter.run_pipeline(**kwargs)


def read_jsonl(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text().splitlines()]


# ---------------------------------------------------------------------------
# Checksum verification
# ---------------------------------------------------------------------------


def test_missing_release_aborts_before_parsing(fixture_zips, tmp_path):
    paths, checksums = fixture_zips
    config = make_config(checksums)
    config["release"] = ""
    with pytest.raises(ValueError, match="non-empty release"):
        adapter.run_pipeline(
            main_zip=paths["main_zip"],
            mapping_zip=paths["mapping_zip"],
            assays_zip=paths["assays_zip"],
            config=config,
            output_dir=tmp_path / "out",
        )


def test_checksum_mismatch_aborts_before_parsing(fixture_zips, tmp_path):
    paths, _ = fixture_zips
    bad_config = make_config({key: "0" * 64 for key in adapter.INPUT_KEYS})
    with pytest.raises(ValueError, match="SHA-256 mismatch"):
        adapter.run_pipeline(
            main_zip=paths["main_zip"],
            mapping_zip=paths["mapping_zip"],
            assays_zip=paths["assays_zip"],
            config=bad_config,
            output_dir=tmp_path / "out",
        )
    assert not (tmp_path / "out" / "source_records.jsonl").exists()


def test_verify_sha256_round_trip(tmp_path):
    payload = tmp_path / "x.bin"
    payload.write_bytes(b"synthetic")
    digest = hashlib.sha256(b"synthetic").hexdigest()
    assert adapter.verify_sha256(payload, digest) == digest
    with pytest.raises(ValueError, match="SHA-256 mismatch"):
        adapter.verify_sha256(payload, "f" * 64)


# ---------------------------------------------------------------------------
# Cell parsing: relations, censor reversal, malformed values
# ---------------------------------------------------------------------------


def parse_cell(raw):
    return adapter.parse_measurement_cell(
        raw,
        MeasurementType.KD,
        source_record_id="r|a|Kd",
        protein_sequence="ACD",
        smiles="CCO",
    )


def test_cell_relation_prefixes_parse_exactly():
    assert parse_cell(" 8.7").relation is MeasurementRelation.EXACT
    assert parse_cell("=8.7").relation is MeasurementRelation.EXACT
    assert parse_cell("<8.7").relation is MeasurementRelation.LESS_THAN
    assert parse_cell("<=8.7").relation is MeasurementRelation.LESS_THAN_OR_EQUAL
    assert parse_cell(">8.7").relation is MeasurementRelation.GREATER_THAN
    assert parse_cell(">=8.7").relation is MeasurementRelation.GREATER_THAN_OR_EQUAL
    assert parse_cell("~8.7").relation is MeasurementRelation.APPROXIMATE
    scientific = parse_cell(">1.00e+5")
    assert scientific.relation is MeasurementRelation.GREATER_THAN
    assert scientific.measurement.value == pytest.approx(100000.0)
    assert parse_cell("").measurement is None
    assert parse_cell("   ").measurement is None
    assert parse_cell(None).measurement is None


def test_cell_malformed_range_and_nonpositive_rejected():
    assert parse_cell("10-20").rejection_reason == "range_value:Kd"
    assert parse_cell("10 to 20").rejection_reason == "range_value:Kd"
    assert parse_cell("abc").rejection_reason == "non_numeric:Kd"
    assert parse_cell("8.7 nM").rejection_reason == "malformed_value:Kd"
    assert parse_cell("Kd (nM)").rejection_reason == "repeated_column_header:Kd"
    assert parse_cell("0").rejection_reason == "nonpositive_value:Kd"
    assert parse_cell("-5").rejection_reason == "nonpositive_value:Kd"
    assert parse_cell("1e-9999").rejection_reason == "nonpositive_value:Kd"
    assert parse_cell("NaN").rejection_reason == "nonfinite_value:Kd"
    assert parse_cell("inf").rejection_reason == "nonfinite_value:Kd"
    assert parse_cell("1e9999").rejection_reason == "nonfinite_value:Kd"


def test_censor_reversal_on_log_transform():
    censored = adapter.parse_measurement_cell(
        ">8.7",
        MeasurementType.KD,
        source_record_id="r|a|Kd",
        protein_sequence="ACD",
        smiles="CCO",
    ).measurement
    pbound, relation = censored.pbound()
    assert pbound == pytest.approx(8.060480747381382)
    assert relation is MeasurementRelation.LESS_THAN
    assert censored.relation is MeasurementRelation.GREATER_THAN


# ---------------------------------------------------------------------------
# Row processing and provenance
# ---------------------------------------------------------------------------


def fixture_context(tmp_path):
    paths, checksums = fixture_zips(tmp_path)
    assay_map = adapter.load_assay_map(paths["mapping_zip"])
    assay_info = adapter.load_assay_info(paths["assays_zip"])
    return assay_map, assay_info, make_config(checksums)["files"]


def process_single_row(tmp_path, row, row_number=1):
    assay_map, assay_info, files = fixture_context(tmp_path)
    return adapter.process_row(
        row,
        row_number=row_number,
        assay_map=assay_map,
        assay_info=assay_info,
        source_release="fixture",
        config_files=files,
        swissprot_column="UniProt (SwissProt) Primary ID of Target Chain 1",
        trembl_column="UniProt (TrEMBL) Primary ID of Target Chain 1",
    )


def test_multiple_measurement_types_from_one_row_no_relabel(tmp_path):
    records, row_rejections, cell_rejections, _ = process_single_row(
        tmp_path, _base_row("2", cells=ROW_MULTI_TYPE)
    )
    assert row_rejections == []
    assert cell_rejections == {}
    # 2 mapped assays x 4 populated cells, fixed type order per assay.
    assert len(records) == 8
    first_assay = [
        r for r in records if r["source_record"]["entryid_assayid"] == "100_1"
    ]
    assert [r["source_record"]["measurement"]["type"] for r in first_assay] == [
        "Ki",
        "IC50",
        "Kd",
        "EC50",
    ]
    # Each type keeps its own semantics; nothing is relabeled as Kd.
    ki = next(
        r for r in first_assay if r["source_record"]["measurement"]["type"] == "Ki"
    )
    assert ki["source_record"]["measurement"]["type"] == "Ki"
    assert ki["source_record"]["measurement"]["relation"] == ">"
    assert ki["source_record"]["measurement"]["value"] == pytest.approx(1000.0)
    # Per-(row, assay, type) record IDs are unique.
    ids = [r["source_record"]["source_record_id"] for r in records]
    assert len(ids) == len(set(ids)) == 8
    assert "2|row:1|100_1|Kd" in ids and "2|row:1|100_2|Ki" in ids


def test_repeated_reactant_set_id_uses_source_row_ordinal(tmp_path):
    first, _, _, _ = process_single_row(
        tmp_path, _base_row("2", cells=ROW_GOLD), row_number=10
    )
    second, _, _, _ = process_single_row(
        tmp_path, _base_row("2", cells=ROW_GOLD), row_number=11
    )
    first_ids = {record["source_record"]["source_record_id"] for record in first}
    second_ids = {record["source_record"]["source_record_id"] for record in second}
    assert first_ids.isdisjoint(second_ids)
    assert all("|row:10|" in record_id for record_id in first_ids)
    assert all("|row:11|" in record_id for record_id in second_ids)
    audit = adapter.AuditBuilder()
    audit.observe_source_row(_base_row("2", cells=ROW_GOLD))
    audit.observe_source_row(_base_row("2", cells=ROW_GOLD))
    stats = audit.duplicate_pair_stats()
    assert stats["duplicate_reactant_set_ids"] == 1
    assert stats["extra_reactant_set_rows"] == 1


def test_provenance_fields_survive_into_source_envelope(tmp_path):
    records, _, _, _ = process_single_row(tmp_path, MAIN_ROWS[2])
    assert records[0]["source_record"]["reactant_set_id"] == "3"
    assert len(records) == 2  # Kd + Ki on one mapped assay
    record = next(
        r for r in records if r["source_record"]["measurement"]["type"] == "Kd"
    )
    assert record["schema_version"] == adapter.SCHEMA_VERSION
    assert record["source_database"] == "BindingDB"
    assert record["source_release"] == "fixture"
    assert set(record["source_release_files"]) == set(adapter.INPUT_KEYS)
    for entry in record["source_release_files"].values():
        assert set(entry) == {"zip_name", "url", "member", "sha256"}
        assert len(entry["sha256"]) == 64
        assert not entry["url"].startswith("/")
    source = record["source_record"]
    assert source["reactant_set_id"] == "3"
    assert source["main_row_number"] == 1
    measurement = source["measurement"]
    assert measurement["raw_value"] == "8.7"
    assert measurement["unit"] == "nM"
    assert measurement["relation"] == "="
    assert measurement["is_exact"] is True
    provenance = source["provenance"]
    assert provenance["target_name"] == "Mutated Src kinase"
    assert provenance["target_organism"] == "Homo sapiens"
    assert provenance["chain_count"] == 1
    assert provenance["chain1_sequence"] == normalize_sequence(SEQ_MUT)
    assert provenance["swissprot_primary_id"] is None
    assert provenance["trembl_primary_id"] == "A0A023GPI8"
    assert provenance["article_doi"] is None
    assert provenance["pmid"] == "12345678"
    assert provenance["publication_date"] == "5/5/2010"
    assert provenance["bindingdb_date"] == "12/12/2017"
    assert (
        provenance["curation_datasource"] == "Curated from the literature by BindingDB"
    )
    assert provenance["ph"] is None
    assert provenance["temp_c"] is None
    assert provenance["assay"]["entryid_assayid"] == "200_1"
    assert provenance["assay"]["entry_id"] == "200"
    assert provenance["assay"]["assay_id"] == "1"
    assert provenance["assay"]["assay_name"] == "Displacement assay"
    assert "Radioligand" in provenance["assay"]["assay_description"]
    assert provenance["assay"]["joined"] is True
    assert provenance["ligand"]["smiles"] == SMILES_C
    assert provenance["ligand"]["inchi_key"] == "SYNTHETICINCHIKEY-AAAA"
    assert provenance["ligand"]["bindingdb_monomer_id"] == "4521"
    assert provenance["ligand"]["pubchem_cid"] == "5328914"
    assert provenance["ligand"]["chembl_id"] == "CHEMBL941"
    canonical = source["canonical"]
    assert canonical["smiles"] == canonicalize_smiles(SMILES_C)
    assert canonical["protein_id"] == stable_id("protein", normalize_sequence(SEQ_MUT))
    assert canonical["pair_id"].startswith("pair_")
    # The whole envelope is JSON-serializable as written.
    json.dumps(record, sort_keys=True)


def test_row_level_rejections(tmp_path):
    records, rejections, _, details = process_single_row(
        tmp_path, _base_row("5", chains="2", cells=ROW_KI_EXACT)
    )
    assert records == [] and details.multichain is True
    assert [r.reasons for r in rejections] == [("multichain_target",)]

    records, rejections, _, details = process_single_row(
        tmp_path, _base_row("6", sequence="  ", cells=ROW_KI_EXACT)
    )
    assert records == [] and details.missing_sequence is True
    assert [r.reasons for r in rejections] == [("missing_chain1_sequence",)]

    records, rejections, _, details = process_single_row(
        tmp_path, _base_row("7", smiles="not_a_smiles", cells=ROW_KI_EXACT)
    )
    assert records == [] and details.invalid_smiles is True
    assert [r.reasons for r in rejections] == [("invalid_smiles",)]

    records, rejections, cell_rejections, _ = process_single_row(
        tmp_path, _base_row("4", cells=ROW_MALFORMED)
    )
    assert records == []
    assert [r.reasons for r in rejections] == [("no_valid_measurement",)]
    assert sum(cell_rejections.values()) == 4

    records, rejections, cell_rejections, _ = process_single_row(
        tmp_path, _base_row("1", cells={})
    )
    assert records == [] and cell_rejections == {}
    assert [r.reasons for r in rejections] == [("no_supported_measurement",)]

    records, rejections, _, _ = process_single_row(
        tmp_path, _base_row("9999", cells=ROW_KI_EXACT)
    )
    assert rejections == []
    assert len(records) == 1
    source = records[0]["source_record"]
    assert source["source_record_id"] == "9999|row:1|UNMAPPED|Ki"
    assert source["entryid_assayid"] is None
    assert source["provenance"]["assay"]["joined"] is False

    records, rejections, _, _ = process_single_row(
        tmp_path, _base_row("8", chains="", cells=ROW_KI_EXACT)
    )
    assert records == []
    assert [r.reasons for r in rejections] == [("missing_chain_count",)]

    records, rejections, _, _ = process_single_row(
        tmp_path, _base_row("8", chains="1.0", cells=ROW_KI_EXACT)
    )
    assert records == []
    assert [r.reasons for r in rejections] == [("invalid_chain_count",)]


def test_missing_assay_join_kept_in_source_but_not_gold(tmp_path):
    records, rejections, cell_rejections, _ = process_single_row(
        tmp_path, _base_row("10", cells=ROW_KD_CENSORED)
    )
    assert rejections == [] and cell_rejections == {}
    assert len(records) == 1
    assay = records[0]["source_record"]["provenance"]["assay"]
    assert assay["joined"] is False
    assert assay["assay_name"] is None
    assert assay["assay_description"] is None
    audit = adapter.AuditBuilder()
    audit.observe_record(records[0])
    assert audit.records_missing_assay_join == 1


# ---------------------------------------------------------------------------
# End-to-end pipeline on synthetic zips
# ---------------------------------------------------------------------------


def test_end_to_end_counts_gold_and_audit(fixture_zips, tmp_path, capsys):
    audit = run_fixture_pipeline(fixture_zips, tmp_path)
    capsys.readouterr()  # discard the printed audit

    assert audit["source_rows"] == 9
    assert audit["source_release"] == "fixture"
    assert audit["rows_with_emitted_measurements"] == 5
    emitted = audit["emitted_records"]
    assert emitted["total"] == 13
    assert emitted["by_type"] == {"EC50": 2, "IC50": 2, "Kd": 4, "Ki": 5}
    assert emitted["by_relation"] == {"=": 8, ">": 3, "~": 2}
    assert emitted["by_type_relation"]["Ki|="] == 3
    assert emitted["by_type_relation"]["Ki|>"] == 2
    assert emitted["by_type_relation"]["Kd|="] == 3
    assert emitted["by_type_relation"]["Kd|>"] == 1
    assert emitted["by_type_relation"]["EC50|~"] == 2

    # Gold: only the exact '=' Kd rows (rsid 2 x2 assays, rsid 3 x1 assay).
    assert audit["exact_gold_kd_records"] == 3
    assert audit["gold_exact_kd"] == {
        "records": 3,
        "unique_pairs": 2,
        "unique_proteins": 2,
        "unique_assays": 3,
        "pkd_min": pytest.approx(8.060480747381382),
        "pkd_max": pytest.approx(8.060480747381382),
        "pkd_below_3_review_flag": 0,
        "pkd_above_12_review_flag": 0,
        "review_flags_are_not_filters": True,
    }
    assert audit["outputs"]["source_records.jsonl"]["records"] == 13
    assert audit["outputs"]["gold_exact_kd.jsonl"]["records"] == 3
    assert len(audit["outputs"]["source_records.jsonl"]["sha256"]) == 64
    assert len(audit["outputs"]["gold_exact_kd.jsonl"]["sha256"]) == 64
    assert audit["rejections"]["gold_excluded"] == {
        "missing_assay_join": 1,
        "non_exact_relation": 1,
        "non_kd_type": 9,
    }
    assert audit["rejections"]["row_level"] == {
        "invalid_smiles": 1,
        "missing_chain1_sequence": 1,
        "multichain_target": 1,
        "no_valid_measurement": 1,
    }
    assert audit["rejections"]["cell_level"] == {
        "range_value:IC50": 1,
        "non_numeric:Ki": 1,
        "nonpositive_value:EC50": 1,
        "nonpositive_value:Kd": 1,
    }
    assert audit["rejections"]["rows_flagged_multichain"] == 1
    assert audit["rejections"]["rows_flagged_invalid_smiles"] == 1
    assert audit["rejections"]["rows_flagged_missing_sequence"] == 1

    assert audit["coverage"]["records_with_assay_join"] == 11
    assert audit["coverage"]["records_missing_assay_join"] == 2
    assert audit["coverage"]["records_with_citation_doi_or_pmid"] == 13
    assert audit["coverage"]["records_with_publication_date"] == 13

    assert audit["uniques"] == {"proteins": 3, "ligands": 3, "pairs": 3, "assays": 4}
    assert audit["duplicates"] == {
        "duplicate_pairs_across_assays": 2,
        "extra_assay_instances": 2,
        "pairs_with_multiple_measurement_types": 3,
        "duplicate_reactant_set_ids": 0,
        "extra_reactant_set_rows": 0,
    }

    out = tmp_path / "out"
    source_records = read_jsonl(out / "source_records.jsonl")
    gold = read_jsonl(out / "gold_exact_kd.jsonl")
    assert len(source_records) == 13
    assert len(gold) == 3

    # No cross-type leakage: gold rows are Kd-only and carry pKd values.
    assert {row["measurement_type"] for row in gold} == {"Kd"}
    assert all(row["relation"] == "=" for row in gold)
    assert all(row["kd_nm"] > 0 for row in gold)
    gold_row = next(
        row for row in gold if row["source_record_id"] == "3|row:3|200_1|Kd"
    )
    assert gold_row["kd_nm"] == pytest.approx(8.7)
    assert gold_row["pkd"] == pytest.approx(8.060480747381382)
    assert gold_row["sequence"] == normalize_sequence(SEQ_MUT)
    assert gold_row["smiles"] == canonicalize_smiles(SMILES_C)
    assert gold_row["assay"]["assay_name"] == "Displacement assay"
    assert gold_row["citation"]["pmid"] == "12345678"
    assert gold_row["raw_value"] == "8.7"

    # Stable source order: nondecreasing row numbers, mapping order inside.
    row_numbers = [r["source_record"]["main_row_number"] for r in source_records]
    assert row_numbers == sorted(row_numbers)
    rsid2 = [r for r in source_records if r["source_record"]["reactant_set_id"] == "2"]
    assert [r["source_record"]["entryid_assayid"] for r in rsid2] == [
        "100_1",
        "100_1",
        "100_1",
        "100_1",
        "100_2",
        "100_2",
        "100_2",
        "100_2",
    ]
    assert [r["source_record"]["measurement"]["type"] for r in rsid2[:4]] == [
        "Ki",
        "IC50",
        "Kd",
        "EC50",
    ]
    # Unicode assay description survived the ZIP round trip.
    src_assay = next(
        r for r in source_records if r["source_record"]["entryid_assayid"] == "100_1"
    )
    assert (
        "Hot天"
        in src_assay["source_record"]["provenance"]["assay"]["assay_description"]
    )

    on_disk_audit = json.loads((out / "bindingdb_audit.json").read_text())
    assert on_disk_audit == audit


def test_deterministic_output_across_runs(fixture_zips, tmp_path, capsys):
    first_dir = tmp_path / "first"
    second_dir = tmp_path / "second"
    paths, checksums = fixture_zips
    config = make_config(checksums)
    adapter.run_pipeline(
        main_zip=paths["main_zip"],
        mapping_zip=paths["mapping_zip"],
        assays_zip=paths["assays_zip"],
        config=config,
        output_dir=first_dir,
    )
    adapter.run_pipeline(
        main_zip=paths["main_zip"],
        mapping_zip=paths["mapping_zip"],
        assays_zip=paths["assays_zip"],
        config=config,
        output_dir=second_dir,
    )
    capsys.readouterr()
    for name in ("source_records.jsonl", "gold_exact_kd.jsonl", "bindingdb_audit.json"):
        assert (first_dir / name).read_bytes() == (second_dir / name).read_bytes()


def test_max_source_rows_limits_processing(fixture_zips, tmp_path, capsys):
    audit = run_fixture_pipeline(fixture_zips, tmp_path, max_source_rows=1)
    capsys.readouterr()
    assert audit["source_rows"] == 1
    assert audit["emitted_records"]["total"] == 1
    assert audit["emitted_records"]["by_type"] == {"Ki": 1}
    assert audit["exact_gold_kd_records"] == 0


def test_cli_end_to_end_on_fixture_zips(fixture_zips, tmp_path, monkeypatch, capsys):
    paths, checksums = fixture_zips
    config_path = tmp_path / "config.json"
    config_path.write_text(json.dumps(make_config(checksums)))
    output_dir = tmp_path / "cli-out"
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "prepare_bindingdb_gold_kd.py",
            "--main-zip",
            str(paths["main_zip"]),
            "--mapping-zip",
            str(paths["mapping_zip"]),
            "--assays-zip",
            str(paths["assays_zip"]),
            "--config",
            str(config_path),
            "--output-dir",
            str(output_dir),
        ],
    )
    prepare_script.main()
    capsys.readouterr()
    audit = json.loads((output_dir / "bindingdb_audit.json").read_text())
    assert audit["source_rows"] == 9
    assert audit["exact_gold_kd_records"] == 3
    assert len(read_jsonl(output_dir / "source_records.jsonl")) == 13
    assert len(read_jsonl(output_dir / "gold_exact_kd.jsonl")) == 3


def test_cli_rejects_invalid_max_source_rows(fixture_zips, tmp_path, monkeypatch):
    paths, checksums = fixture_zips
    config_path = tmp_path / "config.json"
    config_path.write_text(json.dumps(make_config(checksums)))
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "prepare_bindingdb_gold_kd.py",
            "--main-zip",
            str(paths["main_zip"]),
            "--mapping-zip",
            str(paths["mapping_zip"]),
            "--assays-zip",
            str(paths["assays_zip"]),
            "--config",
            str(config_path),
            "--output-dir",
            str(tmp_path / "cli-out"),
            "--max-source-rows",
            "0",
        ],
    )
    with pytest.raises(SystemExit):
        prepare_script.main()


def test_load_config_validates_required_fields(tmp_path):
    bad = tmp_path / "bad.json"
    bad.write_text(json.dumps({"files": {}}))
    with pytest.raises(ValueError, match="config files entry"):
        prepare_script.load_config(bad)


# ---------------------------------------------------------------------------
# Structural guards
# ---------------------------------------------------------------------------


def test_missing_zip_member_raises(tmp_path):
    archive_path = tmp_path / "wrong.zip"
    write_zip(archive_path, "something_else.tsv", ["a\tb", "1\t2"])
    with pytest.raises(ValueError, match="does not contain member"):
        adapter.load_assay_map(archive_path, adapter.MAPPING_MEMBER)


def test_main_header_guard_rejects_unexpected_layout(tmp_path):
    archive_path = tmp_path / "bad_main.zip"
    write_zip(archive_path, adapter.MAIN_MEMBER, ["a\tb\tc", "1\t2\t3"])
    with pytest.raises(ValueError, match="unexpected header prefix"):
        list(adapter.stream_main_rows(archive_path))


def test_main_column_count_guard(tmp_path):
    header = list(adapter.MAIN_HEADER_PREFIX) + ["only one pad"]
    archive_path = tmp_path / "short_main.zip"
    write_zip(
        archive_path,
        adapter.MAIN_MEMBER,
        ["\t".join(header), "\t".join(["x"] * len(header))],
    )
    with pytest.raises(ValueError, match="expected 640 columns"):
        list(adapter.stream_main_rows(archive_path))


def test_ragged_data_row_raises(tmp_path):
    archive_path = tmp_path / "ragged.zip"
    lines = main_lines()
    lines[1] = "\t".join(lines[1].split("\t")[:10])  # truncate first data row
    write_zip(archive_path, adapter.MAIN_MEMBER, lines)
    with pytest.raises(ValueError, match="data row 1 has 10 fields"):
        list(adapter.stream_main_rows(archive_path))


def test_assay_map_collapses_duplicate_lines(fixture_zips):
    paths, _ = fixture_zips
    assay_map = adapter.load_assay_map(paths["mapping_zip"])
    assert assay_map["2"] == ["100_1", "100_2"]
    assert assay_map["1"] == ["100_1"]


def test_resolve_column_matches_suffix_variants():
    header = [
        "UniProt (SwissProt) Primary ID of Target Chain 1 Target Chain 1",
        "UniProt (TrEMBL) Primary ID of Target Chain 1",
    ]
    assert (
        adapter.resolve_column(header, adapter.SWISSPROT_PRIMARY_RE)
        == "UniProt (SwissProt) Primary ID of Target Chain 1 Target Chain 1"
    )
    assert (
        adapter.resolve_column(header, adapter.TREMBL_PRIMARY_RE)
        == "UniProt (TrEMBL) Primary ID of Target Chain 1"
    )


def test_blank_stream_is_explicitly_rejected(tmp_path):
    records, rejections, cell_rejections, _ = process_single_row(
        tmp_path,
        _base_row(
            "11",
            cells={"Ki (nM)": "", "IC50 (nM)": "", "Kd (nM)": "", "EC50 (nM)": ""},
        ),
    )
    assert records == [] and cell_rejections == {}
    assert [r.reasons for r in rejections] == [("no_supported_measurement",)]