Spaces:
Running
Running
| """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"} | |