"""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",)]