File size: 6,340 Bytes
2a0d367
 
 
a69199c
9e4e9f9
 
 
2a0d367
a69199c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
9e4e9f9
 
 
a69199c
 
 
 
 
 
 
 
 
 
9e4e9f9
 
a69199c
 
9e4e9f9
 
 
a69199c
9e4e9f9
a69199c
 
 
9e4e9f9
a69199c
 
 
2a0d367
 
9e4e9f9
 
 
 
 
2a0d367
 
a69199c
2a0d367
7f63629
 
 
 
2a0d367
 
7f63629
 
 
2a0d367
 
 
9e4e9f9
 
a69199c
 
9e4e9f9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2a0d367
 
 
 
 
 
 
 
 
 
 
 
 
a69199c
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
"""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)]


@pytest.fixture
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)"