Spaces:
Running
Running
MeluXina user
taxonova 26.11.3: short reads (80 to 119 valid bases) are classified, with a note in the results and a hint before the run; pin 26.11.3
4ffc1bd | """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) | |
| 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 | |
| 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]})) | |