syntheogenesis / tests /test_compiler.py
github-actions[bot]
Deploy 1dc745d
0bdb418
Raw
History Blame Contribute Delete
27.7 kB
"""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")