tests/test_assay.py

"""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
    <option>s, and with the list still empty it would silently do nothing and
    leave the last-used-anywhere default selected — which is exactly the
    mistake (a second batch on a different assay) it exists to prevent."""
    src = _app_js()
    assert "_assayLoad().then(() => _assayAdoptForLibrary(" in src, (
        "adoption must be chained off the list load, not fired beside it")


def test_adoption_declines_when_the_library_used_more_than_one_assay():
    """Picking one of them would look like a decision. dee/core/assay's
    pooling summary is where that mess gets reported."""
    assert "if (ids.length !== 1) return;" in _app_js()


def test_adoption_never_selects_an_assay_that_is_no_longer_in_the_list():
    """A free-plan assay can expire; selecting a dead id would log outcomes
    against a foreign key the server will refuse."""
    src = _app_js()
    assert "if (!opt) return;" in src and "CSS.escape(ids[0])" in src