Spaces:
Running
Running
| """Region selector: every model the registry lists (and whose file is | |
| present) is offered in one Region list, and the page names the reference of | |
| the chosen model; deprecated and preview models are not offered. | |
| Model files are stand-ins (small files under the registry file names) and | |
| loading and settings reading are stubbed, so this runs without checkpoints. | |
| Expectations are derived from the registry, so a release that adds models | |
| needs no change here. | |
| """ | |
| import html | |
| import re | |
| import sys | |
| from pathlib import Path | |
| import pytest | |
| pytest.importorskip("httpx") | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| from fastapi.testclient import TestClient # noqa: E402 | |
| from taxonova.models import list_models # noqa: E402 | |
| import webapp.app as webapp_app # noqa: E402 | |
| from webapp import model_catalog as mc # noqa: E402 | |
| LISTED = [s for s in list_models() if s.status not in ("deprecated", "preview")] | |
| HIDDEN = [s for s in list_models(include_hidden=True) if s.status in ("deprecated", "preview")] | |
| SETTINGS = {"operating_point": None, "species_gate": False, "species_groups": "kept", "forward_primer": None} | |
| def _options(page: str, select_id: str): | |
| block = re.search(rf'<select id="{select_id}"[^>]*>(.*?)</select>', page, re.S).group(1) | |
| return [(v, html.unescape(t).strip(), bool(sel)) | |
| for v, sel, t in re.findall(r'<option value="([^"]*)"( selected)?>([^<]*)</option>', block)] | |
| def server(tmp_path, monkeypatch): | |
| """Start the app on a models dir that has every registry file; returns the page.""" | |
| for spec in list_models(include_hidden=True): | |
| (tmp_path / spec.filename).write_bytes(b"stand-in") | |
| monkeypatch.setenv("TAXONOVA_WEBAPP_MODELS_DIR", str(tmp_path)) | |
| loads = [] | |
| monkeypatch.setattr(webapp_app, "load_model", lambda path: loads.append(path) or {"path": path}) | |
| monkeypatch.setattr(webapp_app, "read_settings", lambda path: dict(SETTINGS)) | |
| def start(): | |
| webapp_app.startup() | |
| return TestClient(webapp_app.app).get("/").text | |
| start.loads = loads | |
| return start | |
| def _ref(spec): | |
| return mc.reference_label(spec.reference, spec.reference_version) | |
| def test_every_listed_model_is_offered_and_hidden_ones_are_not(server): | |
| page = server() | |
| assert list(webapp_app.app.state.models) == [s.id for s in LISTED] | |
| # one Region list, no reference selector | |
| assert 'id="reference"' not in page | |
| regions = _options(page, "model") | |
| operon_only = {_ref(s) for s in LISTED if all(mc.region_key(t.id) == "operon" for t in LISTED if _ref(t) == _ref(s))} | |
| amplicons = [k for k in mc.REGION_LABELS if k != "operon"] | |
| rank = dict({k: i for i, k in enumerate(amplicons)}, operon=len(amplicons) + 1) | |
| expected = [s.id for s in sorted(LISTED, key=lambda s: (_ref(s) in operon_only, | |
| rank.get(mc.region_key(s.id), len(amplicons))))] | |
| assert [v for v, _, _ in regions] == expected | |
| assert regions[0] == ("v4", "V4", True) and sum(sel for _, _, sel in regions) == 1 | |
| if {"v4v5", "v3v4", "16s", "operon"} <= {s.id for s in LISTED}: | |
| ids = [v for v, _, _ in regions] | |
| assert ids.index("v4") < ids.index("v4v5") < ids.index("v3v4") < ids.index("16s") < ids.index("operon") | |
| labels = [t for _, t, _ in regions] | |
| assert len(labels) == len(set(labels)), labels | |
| assert "Reference: " + _ref(mc.registry.get_model("v4")) in page | |
| for spec in HIDDEN: | |
| assert f'"{spec.id}"' not in page, spec.id | |
| visible = re.sub(r'<script id="cfg".*?</script>', "", page, flags=re.S) | |
| assert "preview" not in visible.lower() | |
| def test_start_up_loads_the_default_model_only(server): | |
| server() | |
| assert len(server.loads) == 1 and server.loads[0].endswith(mc.registry.get_model("v4").filename) | |
| assert webapp_app.app.state.cache.loaded_ids() == ["v4"] | |
| def test_every_region_has_a_plain_label(): | |
| for m in mc.discover_models(Path("/nonexistent")): | |
| assert m.region_label != m.id or m.id in mc.REGION_LABELS, m.id | |
| assert "16S V" not in m.region_label, m.region_label | |
| labels = {m.id: m.region_label for m in mc.discover_models(Path("/nonexistent"))} | |
| if "v4v5" in labels: | |
| assert labels["v4v5"] == "V4-V5" | |
| # a region the web app has never seen is named from the registry entry | |
| assert mc.region_label("v1v3", "16S V1-V3") == "V1-V3" | |
| assert mc.region_label("v1v9", "16S V1-V9 (whole gene)") == "V1-V9" | |
| def test_a_model_needs_its_file(server, tmp_path): | |
| victim = LISTED[-1] | |
| (tmp_path / victim.filename).unlink() | |
| server() | |
| assert victim.id not in webapp_app.app.state.models | |
| def test_deprecated_ids_get_a_message_naming_the_command_line(server): | |
| server() | |
| deprecated = [s for s in HIDDEN if s.status == "deprecated"] | |
| if not deprecated: | |
| pytest.skip("this taxonova release lists no deprecated models") | |
| for spec in deprecated: | |
| with pytest.raises(webapp_app.InputError) as e: | |
| webapp_app.resolve_model(spec.id) | |
| msg = str(e.value) | |
| assert "deprecated" in msg and f"taxonova classify reads.fasta --model {spec.id}" in msg, msg | |
| assert f"taxonova=={mc.package_version()}" in msg | |
| # its earlier file names name the same deprecated model | |
| for alias in spec.legacy_filenames: | |
| with pytest.raises(webapp_app.InputError, match="deprecated"): | |
| webapp_app.resolve_model(alias) | |
| #: ids and file names of the 26.11.0 V4-V5 models of the references withdrawn | |
| #: in 26.11.1: the web app must not know them (no deprecated ids replace them) | |
| WITHDRAWN = ("v4v5-silva", "v4v5-rdp", "v4v5-silva-26.11", "silva_v4v5", "rdp_v4v5", | |
| "taxonova-V4V5-silva-26.11.pt", "taxonova-V4V5-rdp-26.11.pt", "taxonova-V4V5-silva-26.11.1.pt") | |
| def test_unknown_and_withdrawn_ids_are_refused(server): | |
| server() | |
| for bad in WITHDRAWN + ("v4-nonexistent", "../../etc/passwd"): | |
| assert mc.canonical_id(bad) is None, bad | |
| assert mc.unavailable_message(bad) is None, bad | |
| with pytest.raises(webapp_app.InputError, match="Unknown model"): | |
| webapp_app.resolve_model(bad) | |
| def test_reference_labels(): | |
| assert mc.reference_label("Greengenes2", "2024.09") == "Greengenes2 2024.09" | |
| assert mc.reference_label("GROND", "GTDB R207") == "GTDB R207 (GROND)" | |