Spaces:
Running
Running
File size: 3,829 Bytes
e477625 b018397 89bbd94 e477625 | 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 | """Tests for duplicate FASTA header handling in the web server.
Reviewer 2, major comment 3: "every sequence in a FASTA file came back with the
same assignment and confidence when the sequences were all named the same".
Cause: `parse_sequence_id` takes the first word of the header, and the merge in
`run_classification_with_prefilter` keys dicts on that id, so duplicates
collapsed to whichever row was written last.
"""
import sys
from pathlib import Path
import pytest
_HERE = Path(__file__).resolve().parent
# app.py imports the installed `taxonova` package (the version pinned in
# requirements.txt); the Space root is on the path for `webapp`.
# Order matters: space/app.py is a thin Hugging Face Spaces wrapper that also
# answers to `import app`. webapp/ must win, so it is inserted last.
sys.path.insert(0, str(_HERE.parent))
sys.path.insert(0, str(_HERE))
from app import ( # noqa: E402
restore_display_ids,
uniquify_sequence_ids,
)
TAX = {"domain": "unknown"}
def _seqs(*names):
return [(n, "ACGT", dict(TAX)) for n in names]
def test_distinct_names_are_untouched():
seqs = _seqs("a", "b", "c")
unique, display = uniquify_sequence_ids(seqs)
assert [u[0] for u in unique] == ["a", "b", "c"]
assert display == ["a", "b", "c"]
def test_duplicates_get_distinct_internal_ids():
"""The bug: three sequences named 'seq' shared one id."""
unique, display = uniquify_sequence_ids(_seqs("seq", "seq", "seq"))
ids = [u[0] for u in unique]
assert len(set(ids)) == 3, ids
assert ids[0] == "seq"
assert display == ["seq", "seq", "seq"]
def test_sequences_are_preserved_alongside_ids():
seqs = [("x", "AAAA", dict(TAX)), ("x", "CCCC", dict(TAX))]
unique, _ = uniquify_sequence_ids(seqs)
assert [u[1] for u in unique] == ["AAAA", "CCCC"]
def test_collision_with_a_generated_name_still_resolves():
"""A user who already has 'seq__dup2' must not collide with our suffix."""
unique, _ = uniquify_sequence_ids(_seqs("seq", "seq__dup2", "seq"))
ids = [u[0] for u in unique]
assert len(set(ids)) == 3, ids
def test_restore_puts_the_users_names_back():
display = ["seq", "seq", "seq"]
tsv = "sequence_id\tdomain\nseq\tBacteria\nseq__dup2\tArchaea\nseq__dup3\tBacteria\n"
results = [{"sequence_id": "seq"}, {"sequence_id": "seq__dup2"},
{"sequence_id": "seq__dup3"}]
out_tsv, out_results = restore_display_ids(tsv, results, display)
assert [r["sequence_id"] for r in out_results] == display
body = out_tsv.splitlines()[1:]
assert [ln.split("\t")[0] for ln in body] == display
# The calls themselves must survive intact and stay distinct.
assert [ln.split("\t")[1] for ln in body] == ["Bacteria", "Archaea", "Bacteria"]
def test_restore_refuses_a_length_mismatch():
"""Silently emitting rows under the wrong name is worse than failing."""
with pytest.raises(RuntimeError):
restore_display_ids("sequence_id\nA\nB\n", [{"sequence_id": "A"}], ["A", "B"])
def test_duplicates_can_receive_different_calls_end_to_end():
"""The regression itself: identical names, different sequences, different
calls, and each is reported under the user's name."""
seqs = [("mySeq", "AAAA", dict(TAX)), ("mySeq", "CCCC", dict(TAX))]
unique, display = uniquify_sequence_ids(seqs)
# Stand in for the classifier: distinct ids in, distinct rows out.
tsv = "sequence_id\tdomain\n" + "".join(
f"{uid}\tcall{i}\n" for i, (uid, _, _) in enumerate(unique))
results = [{"sequence_id": uid, "domain": f"call{i}"}
for i, (uid, _, _) in enumerate(unique)]
out_tsv, out_results = restore_display_ids(tsv, results, display)
assert [r["domain"] for r in out_results] == ["call0", "call1"]
assert {r["sequence_id"] for r in out_results} == {"mySeq"}
|