"""Assay normalisation — the thing that decides whether the commons is signal. A DE outcome used to be a bare float. "1.8" from one lab and "1.8" from another were stored identically and are very often not the same claim: one a Tm shift in °C, the other a fold-change in activity. Pooling them builds the cross-lab prior on a category error — and that prior is fed back into round 2, so it does not merely mislead a reader, it re-ranks the next experiment. The tests below pin the two decisions that make this defensible: anchoring on the wild-type measured in the same run, and REFUSING to pool anything without one. The refusal is the feature. A smaller honest prior beats a larger one nobody can trust. """ import math import pytest from dee.core.assay import (Assay, normalise, poolable, summarise_pooling) def A(**kw): base = dict(scale="ratio", direction="higher_is_better", wt_value=1.0) base.update(kw) return Assay(**base) # --------------------------------------------------------------------------- # # The refusals — these are the point # --------------------------------------------------------------------------- # def test_a_bare_number_with_no_assay_is_not_poolable(): r = normalise(1.8, None) assert r.ok is False and r.comparable is False assert "no unit" in r.why and "inventing its meaning" in r.why def test_no_wild_type_means_lab_local_not_worthless(): """The message matters: 'we ignored your data' is the wrong lesson. It is valid for the lab's own comparison and merely cannot be pooled.""" r = normalise(1.8, A(wt_value=None)) assert r.comparable is False assert "valid for your own comparison" in r.why def test_a_zero_wild_type_refuses_rather_than_dividing(): """A wild-type at the noise floor makes every ratio explode, and a prior built from those is dominated by division artefacts, not biology.""" r = normalise(5.0, A(wt_value=0.0)) assert r.comparable is False and "division artefact" in r.why def test_a_negative_ratio_refuses_rather_than_producing_a_nan(): r = normalise(-2.0, A(wt_value=1.0)) assert r.comparable is False and "sign convention" in r.why @pytest.mark.parametrize("bad", [None, "", "n/a", float("nan"), float("inf")]) def test_non_numeric_measurements_are_refused(bad): assert normalise(bad, A()).comparable is False def test_an_unknown_scale_or_direction_is_refused_not_defaulted(): """Defaulting would silently pick a normalisation, and the wrong one produces nonsense that still typechecks.""" assert normalise(2.0, A(scale="ordinal")).comparable is False assert normalise(2.0, A(direction="sideways")).comparable is False # --------------------------------------------------------------------------- # # Scale type is a real scientific distinction # --------------------------------------------------------------------------- # def test_interval_scales_subtract_because_dividing_celsius_is_meaningless(): """The ratio of 40 °C to 20 °C is not 'twice as hot' in any physical sense, so a Tm assay must normalise by subtraction.""" r = normalise(48.0, A(scale="interval", wt_value=45.0, unit="°C")) assert r.ok and r.basis == "delta" assert r.effect == 3.0 def test_ratio_scales_divide_and_report_log_fold(): """Log-fold so that 2x better and 2x worse are symmetric. A raw ratio compresses everything below 1 into (0,1) and lets one 10x outlier dominate any average taken over it.""" up = normalise(2.0, A(scale="ratio", wt_value=1.0)) down = normalise(0.5, A(scale="ratio", wt_value=1.0)) assert up.basis == "fold" assert up.effect == 1.0 and down.effect == -1.0 # symmetric in log2 def test_a_stdev_upgrades_an_interval_assay_to_an_effect_size(): """How large the change is relative to the NOISE is the only form that survives comparison across instruments.""" r = normalise(48.0, A(scale="interval", wt_value=45.0, wt_stdev=1.5)) assert r.basis == "z" and r.effect == 2.0 # --------------------------------------------------------------------------- # # Direction — without it, half the assays rank backwards # --------------------------------------------------------------------------- # def test_positive_always_means_better_whichever_way_the_assay_runs(): """A degradation assay and a stability assay look identical in the table; only `direction` keeps them from ranking opposite ways.""" better_up = normalise(48.0, A(scale="interval", wt_value=45.0, direction="higher_is_better")) better_down = normalise(42.0, A(scale="interval", wt_value=45.0, direction="lower_is_better")) assert better_up.effect > 0 and better_down.effect > 0 def test_a_worse_variant_is_negative_in_both_directions(): worse_up = normalise(42.0, A(scale="interval", wt_value=45.0, direction="higher_is_better")) worse_down = normalise(48.0, A(scale="interval", wt_value=45.0, direction="lower_is_better")) assert worse_up.effect < 0 and worse_down.effect < 0 def test_a_variant_identical_to_wild_type_has_no_effect(): assert normalise(45.0, A(scale="interval", wt_value=45.0)).effect == 0.0 assert normalise(1.0, A(scale="ratio", wt_value=1.0)).effect == 0.0 # --------------------------------------------------------------------------- # # Reporting what was excluded # --------------------------------------------------------------------------- # def test_exclusions_are_counted_by_reason_not_silently_dropped(): """'412 of 900 could not be pooled' with no explanation is the kind of number that quietly becomes policy. Named causes are fixable — usually by asking one lab to record its wild-type.""" rows = [normalise(2.0, A()), normalise(3.0, A()), normalise(2.0, A(wt_value=None)), normalise("x", A())] s = summarise_pooling(rows) assert s["total"] == 4 and s["pooled"] == 2 and s["excluded"] == 2 assert len(s["excluded_reasons"]) == 2 assert "2 of 4" in s["headline"] and "lab-local" in s["headline"] def test_summarise_handles_an_empty_set(): s = summarise_pooling([]) assert s["total"] == 0 and s["pooled"] == 0 def test_poolable_is_the_strict_question(): assert poolable(2.0, A()) is True assert poolable(2.0, A(wt_value=None)) is False # --------------------------------------------------------------------------- # # Round-tripping a stored row # --------------------------------------------------------------------------- # def test_an_assay_row_from_the_database_normalises(): a = Assay.from_row({"name": "Tm shift", "readout": "thermostability", "unit": "°C", "scale": "interval", "direction": "higher_is_better", "wt_value": 45.0, "wt_stdev": 1.5, "wt_n": 3}) r = normalise(48.0, a) assert r.ok and r.basis == "z" and r.effect == 2.0 def test_a_missing_row_is_no_assay_at_all(): assert Assay.from_row(None) is None assert normalise(1.0, Assay.from_row(None)).comparable is False def test_rows_predating_the_assays_table_stay_honestly_unattributed(): """Every de_outcomes row written before migration 0023 has no assay. Backfilling a guessed one would be inventing provenance.""" assert normalise(1.8, None).comparable is False # --------------------------------------------------------------------------- # # The endpoints # --------------------------------------------------------------------------- # from dee import auth as dee_auth from dee import server def _client(): app = server.create_app() app.config.update(TESTING=True) return app.test_client() class _Anon: user_id = None anonymous = True class _User: user_id = "u1" anonymous = False def test_listing_assays_signed_out_is_a_gate_not_an_error(monkeypatch): monkeypatch.setattr(server._auth, "get_auth", lambda: _Anon()) assert _client().get("/api/assays").get_json() == { "ok": True, "gated": True, "assays": []} def test_creating_an_assay_signed_out_is_refused(monkeypatch): monkeypatch.setattr(server._auth, "get_auth", lambda: _Anon()) res = _client().post("/api/assays", json={"name": "Tm"}) assert res.status_code == 401 and res.get_json()["kind"] == "signin_required" def test_an_assay_needs_a_name(monkeypatch): monkeypatch.setattr(server._auth, "get_auth", lambda: _User()) assert _client().post("/api/assays", json={}).status_code == 400 def test_creating_an_assay_reports_poolability_immediately(monkeypatch): """Telling a lab their results will be lab-local AFTER 40 measurements is too late. The missing field is almost always the wild-type, and it costs nothing to add before the plate is read.""" monkeypatch.setattr(server._auth, "get_auth", lambda: _User()) monkeypatch.setattr(server._auth, "save_assay", lambda uid, **kw: { "ok": True, "id": "a1", "assay": {"name": kw["name"], "scale": kw["scale"], "direction": kw["direction"], "wt_value": kw.get("wt_value")}}) with_wt = _client().post("/api/assays", json={ "name": "Tm shift", "scale": "interval", "wt_value": 45.0}).get_json() assert with_wt["poolable"] is True without = _client().post("/api/assays", json={ "name": "Tm shift", "scale": "interval"}).get_json() assert without["poolable"] is False assert "cannot be placed beside another lab" in without["poolable_why"] def test_a_bad_scale_is_refused_by_the_writer(monkeypatch): monkeypatch.setattr(dee_auth, "SUPABASE_URL", "https://x") monkeypatch.setattr(dee_auth, "SUPABASE_SERVICE_KEY", "k") res = dee_auth.save_assay("u1", name="x", scale="ordinal") assert res["ok"] is False and "scale" in res["error"] def test_nan_and_infinite_anchors_never_reach_the_row(monkeypatch): """They would poison every normalisation computed against them.""" captured = {} import urllib.request class _R: def __enter__(self): return self def __exit__(self, *a): return False def read(self): return b'[{"id":"a1"}]' monkeypatch.setattr(dee_auth, "SUPABASE_URL", "https://x") monkeypatch.setattr(dee_auth, "SUPABASE_SERVICE_KEY", "k") monkeypatch.setattr(dee_auth, "has_pro_plan", lambda uid: False) monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=0: (captured.update( __import__("json").loads(req.data.decode())), _R())[1]) dee_auth.save_assay("u1", name="x", wt_value=float("nan"), wt_stdev=float("inf")) assert captured["wt_value"] is None and captured["wt_stdev"] is None def test_round_two_reports_what_could_and_could_not_be_pooled(monkeypatch): """Shown where a scientist is most likely to act on it. Round 2 still uses all of their OWN measurements — those are comparable with each other by construction — but the split governs what can leave the account.""" monkeypatch.setattr(server._auth, "get_auth", lambda: _User()) monkeypatch.setattr(server._auth, "list_de_outcomes", lambda uid, lid: []) monkeypatch.setattr(server._auth, "cleanup_expired_de_outcomes_async", lambda uid: None) monkeypatch.setattr(server._auth, "get_assay", lambda uid, aid: { "name": "Tm", "scale": "interval", "direction": "higher_is_better", "wt_value": 45.0}) monkeypatch.setattr(server, "_de_round2_library", lambda wt, st, meas: ([], {"kind": "stub"})) body = _client().post("/api/de/round2", json={ "wt_protein": "MKVLAAGIVGL", "assay_id": "a1", "measurements": [{"mutations": "A1G", "measured_value": 48.0}, {"mutations": "T2C", "measured_value": 46.0}], }).get_json() assert body["ok"] is True assert body["pooling"]["pooled"] == 2 and body["pooling"]["excluded"] == 0 def test_round_two_without_an_assay_says_the_results_are_lab_local(monkeypatch): monkeypatch.setattr(server._auth, "get_auth", lambda: _User()) monkeypatch.setattr(server._auth, "list_de_outcomes", lambda uid, lid: []) monkeypatch.setattr(server._auth, "cleanup_expired_de_outcomes_async", lambda uid: None) monkeypatch.setattr(server, "_de_round2_library", lambda wt, st, meas: ([], {"kind": "stub"})) body = _client().post("/api/de/round2", json={ "wt_protein": "MKVLAAGIVGL", "measurements": [{"mutations": "A1G", "measured_value": 48.0}], }).get_json() p = body["pooling"] assert p["pooled"] == 0 and p["excluded"] == 1 assert any("no assay recorded" in why for why in p["excluded_reasons"]) # --------------------------------------------------------------------------- # # check_assay — asked at definition time, not after forty measurements # --------------------------------------------------------------------------- # from dee.core.assay import check_assay def test_check_assay_judges_the_assay_not_a_probe_measurement(): """Passing a null measurement to normalise() to find this out answers 'no numeric measurement' — true, and useless: it describes the probe rather than the assay.""" r = check_assay(A(wt_value=None)) assert r.comparable is False assert "no wild-type recorded for this assay" in r.why assert "numeric measurement" not in r.why def test_a_ready_assay_names_the_basis_it_will_use(): assert check_assay(A(scale="interval", wt_value=45.0)).basis == "delta" assert check_assay(A(scale="interval", wt_value=45.0, wt_stdev=1.5)).basis == "z" assert check_assay(A(scale="ratio", wt_value=1.0)).basis == "fold" def test_check_assay_catches_a_zero_anchor_before_any_data_is_entered(): r = check_assay(A(scale="ratio", wt_value=0.0)) assert r.comparable is False and "division artefact" in r.why def test_check_assay_on_nothing(): assert check_assay(None).comparable is False # --------------------------------------------------------------------------- # # Capture: an assay is attached to the outcomes it produced # --------------------------------------------------------------------------- # def test_listing_assays_carries_each_rows_own_verdict(monkeypatch): """The client must never re-derive poolability. A second implementation in JavaScript would be a second set of rules about what counts as comparable evidence, and the two would drift.""" monkeypatch.setattr(server._auth, "get_auth", lambda: _User()) monkeypatch.setattr(server._auth, "list_assays", lambda uid: [ {"id": "a1", "name": "Tm", "scale": "interval", "wt_value": 45.0, "direction": "higher_is_better"}, {"id": "a2", "name": "Activity", "scale": "ratio", "wt_value": None, "direction": "higher_is_better"}]) rows = _client().get("/api/assays").get_json()["assays"] assert rows[0]["poolable"] is True assert rows[1]["poolable"] is False assert "wild-type" in rows[1]["poolable_why"] def _stub_outcome_save(monkeypatch, *, assay=None): seen = {} monkeypatch.setattr(server._auth, "get_auth", lambda: _User()) monkeypatch.setattr(server._auth, "get_assay", lambda uid, aid: assay) monkeypatch.setattr(server._auth, "cleanup_expired_de_outcomes_async", lambda uid: None) monkeypatch.setattr(server._auth, "record_edge", lambda *a, **kw: seen.update(edge=True)) monkeypatch.setattr( server._auth, "save_de_outcomes", lambda uid, lid, rows, assay_id=None: ( seen.update(assay_id=assay_id, n=len(rows)), {"ok": True, "n": len(rows)})[1]) return seen def test_an_assay_id_that_is_not_yours_never_reaches_the_row(monkeypatch): """Forwarding an unverified id would attach another lab's readout, unit and wild-type to these numbers — worse than no assay at all, because the result would then look poolable.""" seen = _stub_outcome_save(monkeypatch, assay=None) # ownership lookup fails res = _client().post("/api/de/outcomes", json={ "library_id": "lib1", "assay_id": "somebody-elses", "outcomes": [{"mutations": "A1G", "measured_value": 2.0}]}) assert res.status_code == 404 assert "assay_id" not in seen # the write never happened at all def test_saving_outcomes_says_how_many_were_actually_banked(monkeypatch): """'Saved' alone is how a lab logs forty measurements and only learns months later that none of them could be pooled.""" seen = _stub_outcome_save(monkeypatch, assay={ "id": "a1", "name": "Activity", "scale": "ratio", "wt_value": 1.0, "direction": "higher_is_better"}) body = _client().post("/api/de/outcomes", json={ "library_id": "lib1", "assay_id": "a1", "outcomes": [{"mutations": "A1G", "measured_value": 2.0}, {"mutations": "T2C", "measured_value": 0.5}]}).get_json() assert body["ok"] is True and seen["assay_id"] == "a1" assert body["pooling"]["pooled"] == 2 and body["pooling"]["excluded"] == 0 def test_outcomes_without_an_assay_still_save_and_say_they_are_lab_local(monkeypatch): """The refusal to pool is not a refusal to store. A result with no assay is perfectly good evidence in the lab that produced it, and still trains that lab's own round 2.""" seen = _stub_outcome_save(monkeypatch, assay=None) body = _client().post("/api/de/outcomes", json={ "library_id": "lib1", "outcomes": [{"mutations": "A1G", "measured_value": 2.0}]}).get_json() assert body["ok"] is True and seen["assay_id"] is None p = body["pooling"] assert p["pooled"] == 0 and p["excluded"] == 1 assert any("no assay recorded" in why for why in p["excluded_reasons"]) def _capture_outcome_insert(monkeypatch): """Return the list of rows POSTed to de_outcomes.""" import urllib.request posted = [] class _R: def __enter__(self): return self def __exit__(self, *a): return False def read(self): return b"[]" def _urlopen(req, timeout=0): if req.get_method() == "POST" and req.data: posted.extend(__import__("json").loads(req.data.decode())) return _R() monkeypatch.setattr(dee_auth, "SUPABASE_URL", "https://x") monkeypatch.setattr(dee_auth, "SUPABASE_SERVICE_KEY", "k") monkeypatch.setattr(dee_auth, "has_pro_plan", lambda uid: False) monkeypatch.setattr(urllib.request, "urlopen", _urlopen) return posted def test_the_assay_id_is_stamped_on_every_outcome_row(monkeypatch): posted = _capture_outcome_insert(monkeypatch) dee_auth.save_de_outcomes( "00000000-0000-0000-0000-000000000001", "a" * 32, [{"mutations": "A1G", "measured_value": 2.0}, {"mutations": "T2C", "measured_value": 0.5}], assay_id="a1") assert len(posted) == 2 assert all(r["assay_id"] == "a1" for r in posted) def test_an_unattributed_outcome_omits_the_column_entirely(monkeypatch): """Sent only when there is one, so a database that has not run 0023 yet still accepts the unattributed case rather than rejecting the insert on an unknown column.""" posted = _capture_outcome_insert(monkeypatch) dee_auth.save_de_outcomes( "00000000-0000-0000-0000-000000000001", "a" * 32, [{"mutations": "A1G", "measured_value": 2.0}]) assert posted and "assay_id" not in posted[0] # --------------------------------------------------------------------------- # # The capture UI's ordering guarantee (source assertion — the browser check # lives in the preview run, this is what stops a refactor undoing it) # --------------------------------------------------------------------------- # import pathlib def _app_js() -> str: return pathlib.Path("dee/static/app.js").read_text(encoding="utf-8") def test_the_library_s_own_assay_is_chosen_after_the_list_has_loaded(): """Two parallel fetches would race: adoption looks the assay up among the