"""The therapeutic compiler. The refusals are the product, so they are what this file spends its effort on. The substitution space is 12 ordered base pairs, so it is tested EXHAUSTIVELY rather than by example: every wild-type/patient combination is routed and the routing is checked against the two chemistries from first principles. A sampled test would have let a strand error through, and a strand error designs a guide against the wrong strand — silent in silico, expensive at the bench. """ import itertools import pytest from dee.core import compiler as C BASES = "ACGT" COMP = {"A": "T", "T": "A", "C": "G", "G": "C"} # ── the chemistry, restated independently of the implementation ───────── def _expected_family(wt, patient): """Derived here from scratch so this is a real second opinion, not a restatement of the module under test.""" if (patient, wt) == ("A", "G"): return "ABE", "sense" if (patient, wt) == ("C", "T"): return "CBE", "sense" if (COMP[patient], COMP[wt]) == ("A", "G"): return "ABE", "antisense" if (COMP[patient], COMP[wt]) == ("C", "T"): return "CBE", "antisense" return None, None @pytest.mark.parametrize("wt,patient", [(a, b) for a, b in itertools.product(BASES, BASES) if a != b]) def test_every_substitution_routes_correctly(wt, patient): call = C.classify_lesion(wt, patient) fam, strand = _expected_family(wt, patient) if fam is None: assert call.correction is None, f"{patient}>{wt} is a transversion" assert call.route == "prime_editing" # It compiles — to a pegRNA, never to a base editor. `correction is # None` above is the load-bearing assertion: no base-editing route # exists for a transversion, and prime editing is a different pass. assert call.compiles else: assert call.correction is not None, f"{patient}>{wt} should be {fam}" assert call.correction.editor_family == fam assert call.correction.strand == strand assert call.route == "base_editing" assert call.compiles def test_the_four_base_editable_corrections_are_exactly_the_transitions(): """Transitions are base-editable, transversions are not. If this ever reports more than four, a transversion has been mis-routed.""" editable = [(wt, pt) for wt, pt in itertools.product(BASES, BASES) if wt != pt and C.classify_lesion(wt, pt).correction is not None] assert len(editable) == 4 assert all(C.classify_lesion(wt, pt).is_transition for wt, pt in editable) def test_the_strand_derivation_a_reviewer_would_check_by_hand(): """Spelled out, because getting this backwards is the expensive error. A patient carrying C where wild-type is T needs T restored. On the sense strand that is C>T — a CBE edit. A patient carrying G where wild-type is A needs A restored: sense G>A, which on the ANTISENSE strand reads C>T, also CBE but engaging the other strand. """ sense = C.classify_lesion("T", "C").correction assert (sense.editor_family, sense.strand, sense.editor_change) == ("CBE", "sense", "C>T") anti = C.classify_lesion("A", "G").correction assert (anti.editor_family, anti.strand, anti.editor_change) == ("CBE", "antisense", "C>T") sense_abe = C.classify_lesion("G", "A").correction assert (sense_abe.editor_family, sense_abe.strand) == ("ABE", "sense") anti_abe = C.classify_lesion("C", "T").correction assert (anti_abe.editor_family, anti_abe.strand) == ("ABE", "antisense") # ── refusals ──────────────────────────────────────────────────────────── def test_identical_alleles_are_refused_not_silently_compiled(): call = C.classify_lesion("A", "A") assert not call.compiles assert call.errors()[0].code == "no_lesion" assert "swapped" in call.errors()[0].remedy def test_a_transversion_says_why_base_editing_is_out_and_routes_to_pe(): call = C.classify_lesion("A", "C") # correcting C>A codes = [d.code for d in call.diagnostics] assert "transversion_no_base_editor" in codes # The refusal that used to live here — prime_editing_unavailable — is gone # because prime editing is now designed. What must NOT come back is a base # editor for a transversion. assert "prime_editing_unavailable" not in codes assert call.correction is None msg = " ".join(d.message for d in call.diagnostics) assert "ABE writes A>G" in msg and "CBE writes C>T" in msg def test_a_large_deletion_is_refused_and_points_somewhere_real(): call = C.classify_lesion("A" * 400, "") assert not call.compiles err = [d for d in call.errors() if d.code == "lesion_too_large"][0] assert "400-base deletion" in err.message assert "integrase" in err.remedy or "recombinase" in err.remedy def test_a_small_indel_routes_to_prime_editing_not_to_a_base_editor(): call = C.classify_lesion("", "ATG") assert call.kind == "insertion" and call.size == 3 assert call.route == "prime_editing" assert call.correction is None, "base editors do not add or remove bases" assert any(d.code == "indel_not_base_editable" for d in call.diagnostics) def test_a_lesion_near_the_bound_is_flagged_marginal(): size = C.PRIME_EDIT_INSERT_BOUND // 2 + 2 call = C.classify_lesion("", "A" * size) assert any(d.code == "lesion_near_bound" for d in call.diagnostics) def test_non_dna_alleles_are_refused(): for bad in ("N", "R", "Q", "5"): call = C.classify_lesion("A", bad) assert not call.compiles assert call.errors()[0].code == "non_dna_allele" def test_ambiguity_codes_are_not_quietly_treated_as_bases(): """N is a real thing to receive from a VCF and must not route.""" assert C.classify_lesion("N", "A").errors()[0].code == "non_dna_allele" # ── verification against real sequence ────────────────────────────────── WINDOW = "GATTACAGATTACAGGCCTTAA" def test_it_verifies_the_reference_actually_has_the_wildtype_base(): off = WINDOW.index("G") # position 0, a G call = C.compile_correction("G", "A", window=WINDOW, offset=off) assert call.compiles def test_an_off_by_one_coordinate_is_caught_by_the_reference_check(): """The check that earns its keep: alleles alone still type-check when the coordinate has drifted.""" off = 1 # WINDOW[1] is 'A', not 'G' call = C.compile_correction("G", "A", window=WINDOW, offset=off) assert not call.compiles err = [d for d in call.errors() if d.code == "reference_mismatch"][0] assert "'A'" in err.message assert "opposite strand" in err.remedy def test_an_offset_outside_the_window_is_refused(): call = C.compile_correction("G", "A", window=WINDOW, offset=999) assert not call.compiles assert call.errors()[0].code == "offset_outside_window" def test_restores_wildtype_is_a_whole_sequence_comparison(): corr = C.classify_lesion("G", "A").correction assert C.restores_wildtype(WINDOW, 0, corr) assert not C.restores_wildtype(WINDOW, 1, corr) assert not C.restores_wildtype("", 0, corr) # ── the scope boundary ────────────────────────────────────────────────── def test_germline_raises_rather_than_returning_a_diagnostic(): """A refusal a caller can read past and keep going is not a refusal.""" with pytest.raises(C.GermlineRefused): C.compile_correction("G", "A", germline=True) def test_germline_is_refused_before_any_routing_happens(): """Even a perfectly compilable lesion must not be routed.""" with pytest.raises(C.GermlineRefused): C.compile_correction("G", "A", window=WINDOW, offset=0, germline=True) # ── the direction of correction ───────────────────────────────────────── def test_arguments_are_wildtype_first_patient_second(): """Swapping these designs an editor that INSTALLS the disease. The two orderings must not produce the same plan.""" a = C.classify_lesion("G", "A") # patient A -> restore G b = C.classify_lesion("A", "G") # patient G -> restore A assert a.correction.editor_family == "ABE" assert b.correction.editor_family == "CBE" assert a.correction.strand != b.correction.strand # ═══════════════════════════════════════════════════════════════════════ # The pass pipeline. The point of these is that a pass which did NOT run is # never reported as one that ran and passed. # ═══════════════════════════════════════════════════════════════════════ def _by_name(report): return {p.name: p for p in report.passes} def test_every_declared_pass_is_reported(): r = C.compile_report("G", "A", window=WINDOW, offset=0) assert [p.name for p in r.passes] == [n for n, _ in C.PASS_ORDER] def test_unrunnable_passes_are_unavailable_not_ok(): """The whole honesty contract. Defaults are False so a caller that forgets to check capability gets an incomplete record, not a falsely clean one.""" r = C.compile_report("G", "A", window=WINDOW, offset=0) p = _by_name(r) assert p["consequence"].status == "unavailable" assert p["specificity"].status == "unavailable" assert "Assess edit consequence" in r.incomplete_because assert "Assess specificity" in r.incomplete_because def test_specificity_still_carries_its_caveat_when_it_runs(): """A coding-sequence-only index reporting a clean 'ok' would read as a clean bill of health on a therapeutic guide. It must never say ok.""" r = C.compile_report("G", "A", window=WINDOW, offset=0, can_check_specificity=True) spec = _by_name(r)["specificity"] assert spec.status == "warn", "must not be reportable as unqualified ok" assert "CODING SEQUENCE ONLY" in spec.detail assert "GUIDE-seq" in spec.detail def test_consequence_reports_the_number_when_scoring_actually_ran(): r = C.compile_report("G", "A", window=WINDOW, offset=0, consequence={"ok": True, "label": "G1A", "delta_ll": -4.2}) p = _by_name(r)["consequence"] assert p.status == "ok" assert "-4.2000" in p.detail assert "less likely than wild-type" in p.detail, "direction stated in words" assert "zero-shot" in p.detail assert "no validated relationship to clinical outcome" in p.detail def test_a_positive_delta_is_described_in_the_other_direction(): r = C.compile_report("G", "A", window=WINDOW, offset=0, consequence={"ok": True, "label": "G1A", "delta_ll": 1.5}) assert "more likely than wild-type" in _by_name(r)["consequence"].detail def test_reachability_alone_never_makes_the_consequence_pass_succeed(): """The bug this replaced: the pass took a capability flag and reported 'passed' whenever the model was merely reachable, so the UI said the edit had been assessed when nothing had been scored.""" r = C.compile_report("G", "A", window=WINDOW, offset=0) assert _by_name(r)["consequence"].status == "unavailable" def test_a_failed_scoring_attempt_is_failed_not_passed_with_caveat(): """'passed with caveat' on a model call that errored is the soft version of reporting an unrun pass as ok — in both cases nothing was assessed.""" r = C.compile_report("G", "A", window=WINDOW, offset=0, consequence={"ok": False, "label": "G1A", "error": "backend refused"}) p = _by_name(r)["consequence"] assert p.status == "failed" assert p.status != "warn", "must not read as a pass" assert "backend refused" in p.detail assert "assumed-benign" in p.detail def test_a_failed_pass_leaves_the_record_incomplete(): """Never run and attempted-but-failed leave the same hole.""" r = C.compile_report("G", "A", window=WINDOW, offset=0, consequence={"ok": False, "error": "boom"}) assert "Assess edit consequence" in r.incomplete_because def test_a_failed_consequence_does_not_block_the_compile(): """The lesion still routes; only the record is incomplete.""" r = C.compile_report("G", "A", window=WINDOW, offset=0, consequence={"ok": False, "error": "boom"}) assert r.compiled def test_a_refused_lesion_skips_the_rest_rather_than_reporting_ok(): # A lesion beyond the prime-editing size bound. Transversions used to sit # here; they now route to prime editing and are designed, so the skip # behaviour is asserted against a lesion that really has no route. r = C.compile_report("A" * 400, "") p = _by_name(r) assert p["classify"].status == "error" assert all(p[n].status == "skipped" for n in ("verify", "enumerate", "consequence", "specificity", "emit")) assert not r.compiled def test_a_transversion_without_sequence_is_unavailable_not_ok_and_not_refused(): """The three-way distinction, on the route that now exists. A transversion has a prime-editing route, but a pegRNA cannot be designed from alleles alone — so enumerate must report `unavailable`, which is neither a promise nor a refusal.""" r = C.compile_report("A", "C") p = _by_name(r) assert r.lesion.route == "prime_editing" assert p["classify"].status == "warn" assert p["verify"].status == "unavailable" assert p["enumerate"].status == "unavailable" assert "reference window" in p["enumerate"].detail assert "Enumerate strategies" in r.incomplete_because def test_a_reference_mismatch_stops_the_build(): r = C.compile_report("G", "A", window=WINDOW, offset=1) p = _by_name(r) assert p["verify"].status == "error" assert p["enumerate"].status == "skipped" assert not r.compiled def test_no_window_makes_verify_unavailable_and_says_why(): r = C.compile_report("G", "A") v = _by_name(r)["verify"] assert v.status == "unavailable" assert "off-by-one" in v.detail def test_compiled_is_false_when_anything_was_skipped(): assert not C.compile_report("A", "A").compiled def test_scope_travels_with_every_report(): r = C.compile_report("G", "A", window=WINDOW, offset=0) assert "Somatic" in r.scope["application"] assert "Not IND-ready" in r.scope["status"] assert "Immunogenicity" in r.scope["silent_on"] def test_compile_report_refuses_germline_before_running_any_pass(): with pytest.raises(C.GermlineRefused): C.compile_report("G", "A", window=WINDOW, offset=0, germline=True) # ═══════════════════════════════════════════════════════════════════════ # Enumerate: real guides. The pass used to promise guides and produce none. # ═══════════════════════════════════════════════════════════════════════ HBB = ("CCTGAGGAGAAGGCTGCCGTCACCGCCCTGTGGGGCAAGGTGAACGTGGATGAAGTTGGTGGTGAGG" "CCCTGGGCAGGCTGCTGGTGGTCTACCCTTGGACCCAGAGGTTCTTTGAGTCC") COMPL = {"A": "T", "T": "A", "C": "G", "G": "C"} def test_spacer_index_maps_to_the_forward_strand_on_both_strands(): """The single most dangerous line in the planner. A '-' guide's spacer runs antiparallel, so its 5' base is the LAST base of the forward footprint.""" from dee.core import crispr guides = crispr.find_guides(HBB, mode="base_edit", base_editor="be4max", max_results=8) checked = 0 for g in guides: for pos in range(1, len(g.spacer) + 1): off = C.spacer_pos_to_offset(g.position, g.strand, len(g.spacer), pos) got = HBB[off] if g.strand == "+" else COMPL[HBB[off]] assert got == g.spacer[pos - 1], ( f"{g.strand} guide at {g.position}, spacer pos {pos}") checked += 1 assert checked > 100, "should have exercised both strands thoroughly" def test_guides_are_designed_against_the_patient_sequence_not_the_reference(): """The bug that made enumerate return nothing for every lesion: an ABE has to find an A to convert, and the wild-type reference has the correct G there. Searching the reference finds nothing, forever.""" corr = C.classify_lesion("G", "A").correction # ABE, sense found = [off for off in range(len(HBB)) if HBB[off] == "G" and C.plan_base_edit_strategies(HBB, off, corr)[0]] assert found, "at least one G must be reachable by an ABE guide" def test_a_strategy_carries_a_real_spacer_pam_and_editor(): corr = C.classify_lesion("G", "A").correction off = next(o for o in range(len(HBB)) if HBB[o] == "G" and C.plan_base_edit_strategies(HBB, o, corr)[0]) strategies, _ = C.plan_base_edit_strategies(HBB, off, corr) s = strategies[0] assert len(s.spacer) == 20 and set(s.spacer) <= set("ACGT") assert s.pam and s.editor_family == "ABE" assert s.strand == "+", "sense correction must engage the sense strand" assert 1 <= s.target_spacer_pos <= 20 def test_bystanders_carry_forward_strand_offsets_so_they_can_be_scored(): """A bystander without a genomic coordinate cannot be handed to Evo 2, which is the entire point of collecting them.""" corr = C.classify_lesion("G", "A").correction for off in range(len(HBB)): if HBB[off] != "G": continue strategies, _ = C.plan_base_edit_strategies(HBB, off, corr) for s in strategies: for b in s.bystanders: assert 0 <= b.offset < len(HBB) assert b.offset != off, "the target is not a bystander" assert b.from_base == "A" and b.to_base == "G" def test_an_editor_of_the_wrong_family_is_refused(): corr = C.classify_lesion("G", "A").correction # needs ABE strategies, diags = C.plan_base_edit_strategies(HBB, 12, corr, editor_id="be4max") assert not strategies assert diags[0].code == "editor_family_mismatch" assert "C>T" in diags[0].remedy def test_a_window_that_is_not_wildtype_is_refused(): corr = C.classify_lesion("G", "A").correction off = HBB.index("A") strategies, diags = C.plan_base_edit_strategies(HBB, off, corr) assert not strategies assert diags[0].code == "window_is_not_wildtype" def test_unreachable_target_says_it_is_a_pam_limit_not_a_score_threshold(): corr = C.classify_lesion("G", "A").correction for off in range(len(HBB)): if HBB[off] != "G": continue strategies, diags = C.plan_base_edit_strategies(HBB, off, corr) if not strategies and diags: assert diags[0].code == "no_guide_places_target_in_window" assert "PAM-availability" in diags[0].remedy return pytest.skip("every G in this fixture happens to be reachable") def test_the_enumerate_pass_reports_guides_not_a_promise_of_guides(): corr = C.classify_lesion("G", "A").correction off = next(o for o in range(len(HBB)) if HBB[o] == "G" and C.plan_base_edit_strategies(HBB, o, corr)[0]) st, dg = C.plan_base_edit_strategies(HBB, off, corr) r = C.compile_report("G", "A", window=HBB, offset=off, strategies=st, enumerate_diags=dg) p = _by_name(r)["enumerate"] assert p.status in ("ok", "warn") assert "guide(s) reach this base" in p.detail assert r.strategies, "the report carries the actual designs" def test_no_reachable_guide_stops_the_build_rather_than_emitting_a_record(): # A real G, so `verify` passes and `enumerate` is the pass that refuses. off = HBB.index("G") r = C.compile_report("G", "A", window=HBB, offset=off, strategies=[]) p = _by_name(r) assert p["enumerate"].status == "error" assert p["emit"].status == "skipped" assert not r.compiled # ═══════════════════════════════════════════════════════════════════════ # Bystander scoring — forward-strand labels for a genome model # ═══════════════════════════════════════════════════════════════════════ def test_a_sense_strand_bystander_label_needs_no_complementing(): win = "ACGTACGTAC" assert C.bystander_forward_label(win, 4, "A", "G", "+") == "A5G" def test_an_antisense_bystander_label_complements_the_alt(): """The editor sees A>G on the minus strand; the genome reads T at that position and becomes C. Getting this backwards asks the model about a change that never happens.""" win = "ACGTACGTAC" # offset 3 is 'T' assert C.bystander_forward_label(win, 3, "A", "G", "-") == "T4C" def test_a_label_is_refused_when_the_genome_disagrees_with_the_editor(): """If the spacer-to-genome mapping is broken, a label built on it would produce a confident wrong number. Refuse instead.""" win = "ACGTACGTAC" # offset 0 is 'A', not 'C' with pytest.raises(ValueError) as e: C.bystander_forward_label(win, 0, "C", "T", "+") assert "mapping is wrong" in str(e.value) def test_labels_are_collected_once_across_all_strategies(): """One flat set so they can be scored in ONE model call — a 7B round trip per base would make the feature unusable.""" corr = C.classify_lesion("G", "A").correction for off in range(len(HBB)): if HBB[off] != "G": continue st, _ = C.plan_base_edit_strategies(HBB, off, corr) if any(s.bystanders for s in st): labels = C.label_bystanders(HBB, st) assert labels and len(labels) == len(set(labels)), "deduplicated" for s in st: for b in s.bystanders: assert b.label, "every bystander is labelled" assert b.label in labels # the label's reference base must equal the genome assert b.label[0] == HBB[b.offset] return pytest.skip("fixture produced no bystanders") def test_unscored_bystanders_stay_none_and_are_never_coerced_to_zero(): """'not scored' and 'predicted neutral' are different claims, and a therapeutic reader acts on them differently.""" corr = C.classify_lesion("G", "A").correction for off in range(len(HBB)): if HBB[off] != "G": continue st, _ = C.plan_base_edit_strategies(HBB, off, corr) if not any(s.bystanders for s in st): continue C.label_bystanders(HBB, st) C.attach_bystander_scores(st, {}) # nothing came back for s in st: for b in s.bystanders: assert b.delta_ll is None, "must not become 0.0" return pytest.skip("fixture produced no bystanders") def test_scores_attach_by_label(): corr = C.classify_lesion("G", "A").correction for off in range(len(HBB)): if HBB[off] != "G": continue st, _ = C.plan_base_edit_strategies(HBB, off, corr) if not any(s.bystanders for s in st): continue labels = C.label_bystanders(HBB, st) C.attach_bystander_scores(st, {labels[0]: -3.75}) hit = [b for s in st for b in s.bystanders if b.label == labels[0]] assert hit and all(b.delta_ll == -3.75 for b in hit) return pytest.skip("fixture produced no bystanders") # ═══════════════════════════════════════════════════════════════════════ # The design record — the actual deliverable # ═══════════════════════════════════════════════════════════════════════ def test_the_record_is_deterministic(): """A record that changes between runs cannot be diffed, and one that cannot be diffed cannot be audited.""" r = C.compile_report("G", "A", window=WINDOW, offset=0) a = C.design_record(r, variant="NM_1:c.1G>A") b = C.design_record(r, variant="NM_1:c.1G>A") assert a == b and len(a) > 400 def test_the_record_carries_the_scope_limits(): r = C.compile_report("G", "A", window=WINDOW, offset=0) rec = C.design_record(r) assert "Somatic" in rec and "Not IND-ready" in rec assert "not a clinical decision" in rec assert "Predicted specificity is not measured specificity" in rec def test_the_record_names_what_it_did_not_establish(): """The section that makes it a record rather than a certificate.""" r = C.compile_report("G", "A", window=WINDOW, offset=0) rec = C.design_record(r) assert "NOT ESTABLISHED by this record" in rec assert "Assess specificity" in rec def test_a_refusal_produces_a_record_too(): """A refused design still deserves documentation — often more so.""" r = C.compile_report("A" * 400, "") # beyond every editing modality rec = C.design_record(r, variant="NM_1:c.1_400del") assert "compiled NO" in rec assert "lesion_too_large" in rec assert "remedy:" in rec def test_a_transversions_record_states_the_gap_even_though_it_compiles(): """`compiled yes` must never be read as `nothing is missing`.""" rec = C.design_record(C.compile_report("A", "C")) assert "compiled yes" in rec assert "NOT ESTABLISHED by this record:" in rec assert "Enumerate strategies" in rec assert "transversion_no_base_editor" in rec def test_bystander_scores_appear_in_the_record_and_unscored_says_so(): corr = C.classify_lesion("G", "A").correction for off in range(len(HBB)): if HBB[off] != "G": continue st, dg = C.plan_base_edit_strategies(HBB, off, corr) if not any(s.bystanders for s in st): continue C.label_bystanders(HBB, st) r = C.compile_report("G", "A", window=HBB, offset=off, strategies=st, enumerate_diags=dg) rec = C.design_record(r) assert "STRATEGIES" in rec and "bystanders" in rec assert "not scored" in rec, "unscored must be stated, not omitted" return pytest.skip("fixture produced no bystanders") def test_the_specificity_pass_hands_off_instead_of_shrugging(): r = C.compile_report("G", "A", window=WINDOW, offset=0) sp = _by_name(r)["specificity"] assert "Cas-OFFinder" in sp.detail or "CRISPRme" in sp.detail assert "GUIDE-seq" in sp.detail assert "Cas-independent deamination" in sp.detail, ( "a base editor is not cleared by a DSB-capture assay, and the handoff " "must say so")