Blame for webapp/test_api.py

"""HTTP API with a stand-in classifier: jobs, downloads, errors, headers.

Needs httpx (for Starlette's TestClient); skipped without it.
"""
import os
import sys
import tempfile
import time
from pathlib import Path

import pytest

pytest.importorskip("httpx")
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
os.environ["TAXONOVA_WEBAPP_MODELS_DIR"] = tempfile.mkdtemp(prefix="no_models_")

from fastapi.testclient import TestClient  # noqa: E402

import webapp.app as webapp_app  # noqa: E402
from webapp.classifier import parse_tsv, tsv_header  # noqa: E402
from webapp.model_catalog import WebModel  # noqa: E402

LEVELS = webapp_app.LEVEL_ORDER


def fake_info(confidence=None, overrides=None):
    user = confidence is not None or bool(overrides)
    gates = {lvl: 0.5 for lvl in LEVELS}
    gates.update(genus=0.56, species=0.5)
    if user:
        c = 0.8 if confidence is None else confidence
        gates = {lvl: min(1.0, 0.5 * c / 0.8) for lvl in LEVELS}
        gates.update(genus=min(1.0, 0.7 * c / 0.8), species=c)
        gates.update(overrides or {})
    return {"thresholds": gates, "source": "user" if user else "checkpoint", "species_gate": True,
            "species_groups": "kept", "gated": 1, "collapsed": 0, "low_information": 0,
            "higher_rank": 2, "rounding": 0, "other_rule": 1,
            "withheld_by_rank": {lvl: {"low_information": 0, "species_gate": int(lvl == "species"),
                                       "species_group": 0, "rounding": 0,
                                       "higher_rank": 2 * int(lvl == "species"),
                                       "other_rule": int(lvl == "species")} for lvl in LEVELS},
            "low_information_limits": [80, 0.05], "short_read": 0, "strand": None}


def fake_classify(cached, seqs, confidence, overrides=None, progress=None, chunk_size=None):
    if progress:
        progress("Classifying", 0.5)
    lines = [tsv_header()]
    for sid, seq, _ in seqs:
        if set(seq.upper()) <= set("ACGT") and len(seq) >= 20:
            lines.append("\t".join([sid] + ["Bacteria", "1.0000", "Bacillota", "0.9900", "Bacilli", "0.9000",
                                            "Lactobacillales", "0.9000", "Lactobacillaceae", "0.9000",
                                            "Lactobacillus", "0.8000", "Unclassified", "0.4000"]))
        else:
            lines.append("\t".join([sid] + ["Unclassified", "0.0000"] * 7))
    tsv = "\n".join(lines) + "\n"
    rows = parse_tsv(tsv)
    for r in rows:
        r["prefilter_pass"] = r["domain"] != "Unclassified"
    return tsv, rows, fake_info(confidence, overrides)


@pytest.fixture(scope="module")
def client():
    with TestClient(webapp_app.app) as c:
        st = webapp_app.app.state
        st.models["v4"] = WebModel(id="v4", label="V4 \u00b7 Greengenes2", region_key="v4",
                                   reference="Greengenes2", model_version="26.11", status="stable",
                                   filename="x.pt")
        st.cache.put("v4", {})
        st.prefilter_available = True
        webapp_app.classify_with_prefilter = fake_classify
        yield c


def wait(client, job_id):
    for _ in range(100):
        j = client.get(f"/api/jobs/{job_id}").json()
        if j["status"] in ("done", "error"):
            return j
        time.sleep(0.05)
    raise AssertionError("job did not finish")


SEQ = "TACGTAGGGGGCAAGCGTTATCCGGATTTACTGGGTGTAAAGGG"


def test_page_renders_with_models_and_security_headers(client):
    r = client.get("/")
    assert r.status_code == 200
    assert "V4 \\u00b7 Greengenes2" in r.text  # JSON-escaped in the page config
    assert "frame-ancestors" in r.headers["content-security-policy"]
    assert "script-src 'self'" in r.headers["content-security-policy"]
    assert "cdn" not in r.text.lower()


def test_page_references_every_static_asset_with_its_content_hash(client):
    import hashlib
    import re
    static = Path(webapp_app.__file__).resolve().parent / "static"
    html = client.get("/").text
    refs = re.findall(r'(?:src|href)="/static/([^"?]+)(?:\?v=([^"]*))?"', html)
    assert {p for p, _ in refs} >= {"app.css", "app.js", "theme.js", "favicon.svg", "vendor/echarts-5.5.1.min.js"}
    for path, token in refs:
        assert token == hashlib.sha256((static / path).read_bytes()).hexdigest()[:10], path
        assert client.get(f"/static/{path}?v={token}").status_code == 200
    assert "?v=" + str(getattr(webapp_app.app.state, "version", "unknown")) not in html


def test_job_with_counts_and_all_downloads(client):
    fasta = f">ASV1\n{SEQ}\n>ASV2\n{SEQ}\n>junk\nNNNN\n"
    counts = "id\tS1\tS2\nASV1\t10\t0\nASV2\t5\t5\njunk\t1\t1\n"
    r = client.post("/api/jobs", data={"sequences": fasta, "model": "v4", "confidence": "0.8"},
                    files={"counts": ("c.tsv", counts, "text/plain")})
    assert r.status_code == 200, r.text
    jid = r.json()["job_id"]
    j = wait(client, jid)
    assert j["status"] == "done"
    res = j["result"]
    assert res["summary"]["submitted"] == 3 and res["summary"]["passed_filter"] == 2
    assert res["hasCounts"] and res["byRank"]["samples"] == ["S1", "S2"]
    assert res["cli"] == "taxonova classify reads.fasta --model v4 --confidence 0.8"
    assert res["thresholds"]["source"] == "user" and res["gates"]["genus"] == pytest.approx(0.7)
    assert res["rows"][0][0] == "ASV1" and res["rows"][0][-1] == 10
    for kind in ("results", "qiime", "phyloseq", "merged", "long"):
        d = client.get(f"/api/jobs/{jid}/download/{kind}")
        assert d.status_code == 200, kind
        assert "attachment" in d.headers["content-disposition"]
    assert client.get(f"/api/jobs/{jid}/download/results").text.startswith("sequence_id\tdomain")


def test_duplicate_ids_keep_the_users_names(client):
    fasta = f">same\n{SEQ}\n>same\nNNNN\n"
    j = wait(client, client.post("/api/jobs", data={"sequences": fasta}).json()["job_id"])
    rows = j["result"]["rows"]
    assert [r[0] for r in rows] == ["same", "same"]
    assert rows[0][2] == "Bacteria" and rows[1][2] == "Unclassified"


def test_per_rank_override_changes_the_cli_hint(client):
    r = client.post("/api/jobs", data={"sequences": f">a\n{SEQ}\n", "thresholds": '{"genus": 0.9}'})
    res = wait(client, r.json()["job_id"])["result"]
    assert res["gates"]["genus"] == 0.9
    assert res["cli"] == "taxonova classify reads.fasta --model v4 --genus-threshold 0.9"


def test_no_confidence_means_the_models_operating_point(client):
    r = client.post("/api/jobs", data={"sequences": f">a\n{SEQ}\n", "model": "v4"})
    res = wait(client, r.json()["job_id"])["result"]
    assert res["cli"] == "taxonova classify reads.fasta --model v4"
    assert res["thresholds"] == {"source": "checkpoint", "species_gate": True, "species_groups": "kept"}
    assert res["withheld"]["gated"] == 1
    assert any(w.startswith("The species gate withheld the species call for 1 sequence") for w in res["warnings"])
    assert res["pip"] == f"pip install taxonova=={webapp_app.package_version()}"
    assert len(res["labelled"]["species"]) == len(res["confidences"]["species"])
    assert res["labelled"]["species"] == [False] and res["labelled"]["genus"] == [True]


def test_every_reason_for_a_withheld_species_reaches_the_page(client):
    """Species Unclassified at or above the threshold: each reason has its
    count and a plain statement, and the per-rank counts reach the chart."""
    r = client.post("/api/jobs", data={"sequences": f">a\n{SEQ}\n", "model": "v4"})
    res = wait(client, r.json()["job_id"])["result"]
    w = res["withheld"]
    assert (w["gated"], w["higher_rank"], w["rounding"], w["other_rule"]) == (1, 2, 0, 1)
    assert w["withheld_by_rank"]["species"]["higher_rank"] == 2
    assert w["withheld_by_rank"]["genus"] == {k: 0 for k in w["withheld_by_rank"]["genus"]}
    notes = res["warnings"]
    assert any(n.startswith("2 sequence(s) have a species confidence at or above the species threshold but no "
                            "species call, because a higher rank did not reach its own threshold") for n in notes)
    assert any(n.startswith("1 sequence(s) have a species confidence at or above the threshold but no species "
                            "call, withheld by another rule") for n in notes)
    assert not any("before rounding" in n for n in notes)


def test_page_sends_no_confidence_unless_asked(client):
    js = (Path(webapp_app.__file__).parent / "static" / "app.js").read_text()
    assert "defaultConfidence" not in js
    page = client.get("/").text
    assert "0.8, gives 0.7 at genus" not in page


@pytest.mark.parametrize("data,message", [
    ({"sequences": ""}, "No sequences"),
    ({"sequences": "@r\nACGT\n+\nIIII\n"}, "FASTQ"),
    ({"sequences": ">ok\nACGT\n>p1\nMKVLAAGIV\n"}, "first is 'p1': 'L' at position 4"),
    ({"sequences": ">a\nACGT\n", "model": "nonsense"}, "Unknown model"),
    ({"sequences": ">a\nACGT\n", "confidence": "1.5"}, "between 0 and 1"),
    ({"sequences": ">a\nACGT\n", "thresholds": '{"genus": 2}'}, "between 0 and 1"),
    ({"sequences": ">a\nACGT\n", "thresholds": '{"strain": 0.5}'}, "Unknown rank"),
])
def test_input_errors_are_400_with_a_message(client, data, message):
    r = client.post("/api/jobs", data=data)
    assert r.status_code == 400 and message in r.json()["error"]


def test_count_table_that_does_not_match(client):
    r = client.post("/api/jobs", data={"sequences": f">ASV1\n{SEQ}\n"},
                    files={"counts": ("c.csv", "id,S1\nOTU_1,3\n", "text/csv")})
    assert r.status_code == 400 and "None of the feature IDs" in r.json()["error"]


def test_old_region_names_still_resolve(client):
    assert webapp_app.resolve_model("V4") == "v4"
    assert webapp_app.resolve_model("gg2_v4") == "v4"


def test_unknown_job_is_404(client):
    assert client.get("/api/jobs/doesnotexist").status_code == 404
    assert client.get("/api/jobs/doesnotexist/download/results").status_code == 404


def test_oversized_request_refused_before_reading(client):
    r = client.post("/api/jobs", content=b"x", headers={"content-length": str(10 ** 9),
                                                         "content-type": "application/octet-stream"})
    assert r.status_code == 413


def test_legacy_classify_endpoint(client, monkeypatch):
    seen = []

    def spy(cached, seqs, confidence, overrides=None, progress=None, chunk_size=None):
        seen.append(confidence)
        return fake_classify(cached, seqs, confidence, overrides, progress, chunk_size)

    monkeypatch.setattr(webapp_app, "classify_with_prefilter", spy)
    r = client.post("/classify", data={"sequences": f">a\n{SEQ}\n", "v_region": "V4"}).json()
    # 0.8 with the scaled profile, the 26.10 default of this endpoint
    assert seen == [0.8]
    assert r["results"][0]["genus"] == "Lactobacillus"
    assert r["results"][0]["v_region"] == "V4"
    assert r["tsv"].startswith("sequence_id")


def test_legacy_classify_defaults_to_v3v4_as_in_26_10_0(client):
    st = webapp_app.app.state
    st.models["v3v4"] = WebModel(id="v3v4", label="16S V3-V4", region_key="v3v4", reference="Greengenes2",
                                 model_version="26.10", status="stable", filename="y.pt")
    st.cache.put("v3v4", {})
    try:
        r = client.post("/classify", data={"sequences": f">a\n{SEQ}\n"}).json()
        assert r["results"][0]["v_region"] == "V3-V4"
        # The page and the new API keep V4 as their default.
        assert webapp_app.resolve_model(None) == "v4"
    finally:
        del st.models["v3v4"]


def test_pasted_fasta_over_one_megabyte_is_accepted(client):
    fasta = "".join(f">ASV{i}\n{SEQ * 6}\n" for i in range(4700))
    assert len(fasta) > 1_100_000
    r = client.post("/api/jobs", data={"sequences": fasta})
    assert r.status_code == 200, r.text
    j = wait(client, r.json()["job_id"])
    assert j["result"]["summary"]["submitted"] == 4700


def test_chunked_upload_over_the_limit_is_413(client, monkeypatch):
    monkeypatch.setattr(webapp_app, "MAX_REQUEST_BYTES", 10_000)
    mw = _body_limit()
    old = mw.limit
    mw.limit = 10_000
    try:
        def body():
            for _ in range(50):
                yield b"x" * 1000
        r = client.post("/api/jobs", content=body(),
                        headers={"content-type": "multipart/form-data; boundary=zzz"})
        assert r.status_code == 413
        assert "larger than" in r.json()["error"]
    finally:
        mw.limit = old


def _body_limit():
    node = webapp_app.app.middleware_stack
    while node is not None and not isinstance(node, webapp_app.BodyLimit):
        node = getattr(node, "app", None)
    assert node is not None
    return node


def test_rna_input_is_converted_with_a_note(client):
    rna = SEQ.replace("T", "U")
    j = wait(client, client.post("/api/jobs", data={"sequences": f">r\n{rna}\n"}).json()["job_id"])
    res = j["result"]
    assert res["rows"][0][2] == "Bacteria"
    assert any("contained U" in w for w in res["warnings"])


def test_bad_confidence_is_a_400_with_a_message(client):
    r = client.post("/api/jobs", data={"sequences": f">a\n{SEQ}\n", "confidence": "high"})
    assert r.status_code == 400 and "Confidence" in r.json()["error"]


def test_non_finite_count_is_400_not_500(client):
    r = client.post("/api/jobs", data={"sequences": f">ASV1\n{SEQ}\n"},
                    files={"counts": ("c.tsv", "id\tS1\nASV1\tinf\n", "text/plain")})
    assert r.status_code == 400 and "not a count" in r.json()["error"]


def test_gzip_upload_with_several_members(client):
    import gzip
    text = "".join(f">s{i}\n{SEQ}\n" for i in range(40)).encode()
    data = gzip.compress(text[:500]) + gzip.compress(text[500:])
    r = client.post("/api/jobs", files={"file": ("x.fasta.gz", data, "application/gzip")})
    assert r.status_code == 200, r.text
    assert wait(client, r.json()["job_id"])["result"]["summary"]["submitted"] == 40


def test_greengenes2_names_shown_in_full_and_exact_in_downloads(client, monkeypatch):
    def gg2_classify(cached, seqs, confidence, overrides=None, progress=None, chunk_size=None):
        tsv, rows, info = fake_classify(cached, seqs, confidence, overrides, progress, chunk_size)
        tsv = tsv.replace("\tLactobacillus\t", "\tBacteroides_H_857956\t")
        for r in rows:
            if r["genus"] == "Lactobacillus":
                r["genus"] = "Bacteroides_H_857956"
        return tsv, rows, info

    monkeypatch.setattr(webapp_app, "classify_with_prefilter", gg2_classify)
    counts = "id\tS1\nASV1\t10\n"
    r = client.post("/api/jobs", data={"sequences": f">ASV1\n{SEQ}\n", "model": "v4"},
                    files={"counts": ("c.tsv", counts, "text/plain")})
    jid = r.json()["job_id"]
    res = wait(client, jid)["result"]
    # No display-only names or switch labels in the payload.
    assert "shortNames" not in res and "rankNames" not in res
    assert "full_names_label" not in res["model"] and "profile" not in res["model"]
    genus = [t for t in res["byRank"]["ranks"]["genus"] if t["label"] == "Bacteroides_H_857956"][0]
    assert genus["name"] == "Bacteroides_H_857956" and "short" not in genus
    assert res["rows"][0][2 + 2 * LEVELS.index("genus")] == "Bacteroides_H_857956"
    for kind in ("results", "qiime", "phyloseq", "merged", "long"):
        body = client.get(f"/api/jobs/{jid}/download/{kind}").text
        assert "Bacteroides_H_857956" in body, kind


def test_page_has_no_full_names_switch(client):
    page = client.get("/").text
    for text in ("full-names", "names-switch", "Full names", "Full Greengenes2"):
        assert text not in page, text
    js = (Path(webapp_app.__file__).parent / "static" / "app.js").read_text()
    for text in ("fullNames", "full-names", "shortNames", "shortened", ".short"):
        assert text not in js, text


def test_legacy_classify_refuses_non_nucleotide_input(client):
    r = client.post("/classify", data={"sequences": ">p1\nMKVLAAGIV\n"})
    # the 26.10 endpoint reports input errors in the body, with status 200, as it did
    assert r.status_code == 200 and "'p1'" in r.json()["error"]


def test_the_page_words_every_withheld_reason():
    """The confidence caption has text for every reason the server counts,
    and takes the counts from the server rather than listing rules that may
    not apply."""
    import re
    from webapp.package_adapter import WITHHELD_REASONS
    js = (Path(webapp_app.__file__).parent / "static" / "app.js").read_text()
    block = js[js.index("var WITHHELD_TEXT = {"):js.index("var WITHHELD_ORDER")]
    keys = re.findall(r"^\s+(\w+): function", block, re.M)
    assert sorted(keys) == sorted(WITHHELD_REASONS)
    order = re.search(r"var WITHHELD_ORDER = \[([^\]]*)\]", js).group(1)
    assert sorted(re.findall(r'"(\w+)"', order)) == sorted(WITHHELD_REASONS)
    assert "withheld_by_rank" in js


def test_unknown_is_shown_as_no_name_in_the_reference(client, monkeypatch):
    """A Greengenes2 'Unknown' at genus and species: counted apart from the
    named taxa and from Unclassified, with one note; the results download is
    the classifier's TSV unchanged."""
    made = {}

    def unknown_classify(cached, seqs, confidence, overrides=None, progress=None, chunk_size=None):
        tsv, rows, info = fake_classify(cached, seqs, confidence, overrides, progress, chunk_size)
        lines = tsv.splitlines(True)
        f = lines[2].split("\t")
        f[11], f[13] = "Unknown", "Unknown"
        lines[2] = "\t".join(f)
        made["tsv"] = "".join(lines)
        rows[1]["genus"] = rows[1]["species"] = "Unknown"
        return made["tsv"], rows, info

    monkeypatch.setattr(webapp_app, "classify_with_prefilter", unknown_classify)
    r = client.post("/api/jobs", data={"sequences": f">a\n{SEQ}\n>u\n{SEQ}\n", "model": "v4"})
    jid = r.json()["job_id"]
    res = wait(client, jid)["result"]
    s = res["summary"]
    assert s["classified"]["genus"] == 1 and s["classified"]["species"] == 0
    assert s["no_name"]["genus"] == 1 and s["no_name"]["species"] == 1
    assert res["noName"]["species"] == [False, True] and res["labelled"]["species"] == [False, True]
    assert [t["name"] for t in res["byRank"]["ranks"]["genus"] if t["no_name"]] == ["Unknown (Lactobacillaceae)"]
    notes = [n for n in res["warnings"] if "no name" in n]
    assert notes == ["1 sequence(s) at genus and 1 at species are labelled Unknown: the reference itself had no "
                     "name at that rank for the predicted lineage. They are counted as no name in the reference, "
                     "apart from the classified and the Unclassified sequences."]
    assert client.get(f"/api/jobs/{jid}/download/results").text == made["tsv"]


def test_the_page_shows_no_name_apart_from_unclassified():
    js = (Path(webapp_app.__file__).parent / "static" / "app.js").read_text()
    css = (Path(webapp_app.__file__).parent / "static" / "app.css").read_text()
    for text in ("No name in the reference", "no_name", "noName", "--s-noname"):
        assert text in js, text
    # light, dark by preference and dark by choice
    assert css.count("--s-noname:") == 3


def test_short_reads_get_a_note_and_full_length_reads_none():
    from webapp.app import run_notes
    notes = run_notes({"short_read": 3, "low_information_limits": [80, 0.05]})
    assert any(n.startswith("3 sequence(s) are short (80 to 119 valid bases). They are classified") for n in notes)
    assert not any("are short" in n for n in run_notes({"short_read": 0, "low_information_limits": [80, 0.05]}))