"""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]}))