"""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"}