diff --git a/tests/test_novelty.py b/tests/test_novelty.py index 6a632c59..44f4661a 100644 --- a/tests/test_novelty.py +++ b/tests/test_novelty.py @@ -1,9 +1,117 @@ -from openamp_foundry.scoring.novelty import normalized_similarity +"""Tests for scoring/novelty.py — levenshtein, normalized_similarity, novelty_score.""" +from __future__ import annotations +from openamp_foundry.scoring.novelty import levenshtein, normalized_similarity, novelty_score +from openamp_foundry.types import PeptideCandidate -def test_similarity_exact(): - assert normalized_similarity("ABC", "ABC") == 1.0 +def _ref(cid: str, seq: str) -> PeptideCandidate: + return PeptideCandidate(candidate_id=cid, sequence=seq, source="test") -def test_similarity_lower_for_different_sequences(): - assert normalized_similarity("AAAA", "DDDD") < 0.5 + +class TestLevenshtein: + def test_identical_strings(self): + assert levenshtein("ABCDE", "ABCDE") == 0 + + def test_single_substitution(self): + assert levenshtein("ABCDE", "ABCDF") == 1 + + def test_single_insertion(self): + assert levenshtein("ABCD", "ABCDE") == 1 + + def test_single_deletion(self): + assert levenshtein("ABCDE", "ABCD") == 1 + + def test_empty_vs_string(self): + assert levenshtein("", "ABC") == 3 + + def test_string_vs_empty(self): + assert levenshtein("ABC", "") == 3 + + def test_both_empty(self): + assert levenshtein("", "") == 0 + + def test_completely_different_same_length(self): + assert levenshtein("AAA", "ZZZ") == 3 + + def test_symmetric(self): + a, b = "KWKLF", "DEDED" + assert levenshtein(a, b) == levenshtein(b, a) + + +class TestNormalizedSimilarity: + def test_identical_returns_one(self): + assert normalized_similarity("ABC", "ABC") == 1.0 + + def test_completely_different_short(self): + # 3 edits to go from AAA to ZZZ, max_len=3 → 0.0 + assert normalized_similarity("AAA", "ZZZ") == 0.0 + + def test_both_empty_returns_one(self): + assert normalized_similarity("", "") == 1.0 + + def test_one_empty_returns_zero(self): + assert normalized_similarity("ABC", "") == 0.0 + + def test_symmetric(self): + a, b = "KWKLF", "KWKLI" + assert normalized_similarity(a, b) == normalized_similarity(b, a) + + def test_single_substitution_in_short_peptide(self): + # ABCDE vs ABCDF: 1 edit, max_len=5 → 1 - 1/5 = 0.8 + result = normalized_similarity("ABCDE", "ABCDF") + assert abs(result - 0.8) < 1e-9 + + def test_range_between_zero_and_one(self): + result = normalized_similarity("KWKLFKKIGAVLKVL", "DEDEDEDEDEDEDEDE") + assert 0.0 <= result <= 1.0 + + def test_lower_for_different_than_similar(self): + similar = normalized_similarity("KWKLFKKIGAVLKVL", "KWKLFKKIGAVLKVK") + different = normalized_similarity("KWKLFKKIGAVLKVL", "DEDEDEDEDEDEDEDE") + assert similar > different + + +class TestNoveltyScore: + def test_no_references_returns_one(self): + score, nearest = novelty_score("KWKLF", []) + assert score == 1.0 + assert nearest is None + + def test_identical_to_reference_returns_zero(self): + refs = [_ref("R1", "KWKLF")] + score, nearest = novelty_score("KWKLF", refs) + assert score == 0.0 + assert nearest is not None + assert nearest["candidate_id"] == "R1" + assert nearest["similarity"] == 1.0 + + def test_score_between_zero_and_one(self): + refs = [_ref("R1", "DEDEDEDE")] + score, _ = novelty_score("KWKLF", refs) + assert 0.0 <= score <= 1.0 + + def test_nearest_has_required_keys(self): + refs = [_ref("R1", "DEDEDEDE")] + _, nearest = novelty_score("KWKLF", refs) + assert nearest is not None + for key in ("candidate_id", "source", "similarity", "sequence"): + assert key in nearest + + def test_picks_most_similar_reference(self): + refs = [ + _ref("far", "DEDEDEDEDEDEDEDE"), + _ref("close", "KWKLF"), + ] + _, nearest = novelty_score("KWKLF", refs) + assert nearest["candidate_id"] == "close" + + def test_high_novelty_for_unique_sequence(self): + refs = [_ref("R1", "DEDEDEDEDEDEDEDE")] + score, _ = novelty_score("KWKLFKKIGAVLKVL", refs) + assert score > 0.5 + + def test_low_novelty_for_near_duplicate(self): + refs = [_ref("R1", "KWKLFKKIGAVLKVL")] + score, _ = novelty_score("KWKLFKKIGAVLKVK", refs) # 1 substitution in 15 + assert score < 0.2 diff --git a/tests/test_selection_diversity.py b/tests/test_selection_diversity.py new file mode 100644 index 00000000..aa5557e9 --- /dev/null +++ b/tests/test_selection_diversity.py @@ -0,0 +1,87 @@ +"""Tests for selection/diversity.py — greedy_diverse_select.""" +from __future__ import annotations + +from openamp_foundry.features.physchem import compute_features +from openamp_foundry.selection.diversity import greedy_diverse_select +from openamp_foundry.types import PeptideCandidate, ScoredCandidate + + +def _make(cid: str, seq: str, ensemble: float = 0.80) -> ScoredCandidate: + return ScoredCandidate( + candidate=PeptideCandidate(candidate_id=cid, sequence=seq, source="test"), + features=compute_features(seq), + scores={"ensemble": ensemble}, + ) + + +# Clearly distinct sequences +_DISTINCT = [ + _make("AMP1", "KWKLFKKIGAVLKVL", 0.90), + _make("AMP2", "DEDEDEDEDEDEDEDE", 0.80), + _make("AMP3", "GGGGGGGGGGGGGGGG", 0.70), +] + +# Three nearly identical sequences, one clearly distinct +_NEAR_DUPLICATES = [ + _make("A", "KWKLFKKIGAVLKVL", 0.90), + _make("B", "KWKLFKKIGAVLKVK", 0.85), # 1 substitution → very similar to A + _make("C", "KWKLFKKIGAVLKVV", 0.80), # 1 substitution → very similar to A + _make("D", "DEDEDEDEDEDEDEDE", 0.70), # completely different +] + + +class TestGreedyDiverseSelect: + def test_returns_up_to_top_n(self): + # _DISTINCT has pairwise similarities well below 0.85, so top_n=2 returns exactly 2 + result = greedy_diverse_select(_DISTINCT, top_n=2, max_pairwise_similarity=0.85) + assert len(result) == 2 + + def test_distinct_sequences_all_selected(self): + result = greedy_diverse_select(_DISTINCT, top_n=3, max_pairwise_similarity=0.85) + assert len(result) == 3 + + def test_near_duplicates_excluded(self): + result = greedy_diverse_select(_NEAR_DUPLICATES, top_n=4, max_pairwise_similarity=0.85) + ids = {c.candidate.candidate_id for c in result} + # B and C are near-duplicates of A (>0.85 similar) — at most one of A/B/C should survive + amp_like = ids & {"A", "B", "C"} + assert len(amp_like) <= 1 + + def test_empty_input_returns_empty(self): + assert greedy_diverse_select([], top_n=5) == [] + + def test_top_n_zero_returns_empty(self): + assert greedy_diverse_select(_DISTINCT, top_n=0) == [] + + def test_first_in_list_wins_when_duplicates(self): + result = greedy_diverse_select(_NEAR_DUPLICATES, top_n=4, max_pairwise_similarity=0.85) + ids = {c.candidate.candidate_id for c in result} + # A is first and highest ensemble — it should always be selected + assert "A" in ids + + def test_distinct_survivor_included_despite_duplicates(self): + result = greedy_diverse_select(_NEAR_DUPLICATES, top_n=4, max_pairwise_similarity=0.85) + ids = {c.candidate.candidate_id for c in result} + # D is completely different — should always be selected + assert "D" in ids + + def test_threshold_one_accepts_all(self): + # max_pairwise_similarity=1.0 → condition is sim <= 1.0, always True (sim is in [0,1]) + # so ALL candidates are accepted, including exact duplicates + result = greedy_diverse_select(_NEAR_DUPLICATES, top_n=4, max_pairwise_similarity=1.0) + assert len(result) == 4 + + def test_low_threshold_rejects_most(self): + # With a very low threshold, sequences with any non-zero similarity are excluded. + # AMP1 (len 15, has G) and AMP3 (all-G, len 16) share similarity > 0 due to the + # one G in AMP1, so AMP3 is excluded after AMP1 is added. + result = greedy_diverse_select(_DISTINCT, top_n=3, max_pairwise_similarity=0.0) + assert len(result) < len(_DISTINCT) + + def test_input_order_determines_priority(self): + low_first = [ + _make("LOW", "KWKLFKKIGAVLKVL", 0.50), + _make("HIGH", "KWKLFKKIGAVLKVK", 0.99), # near-dup of LOW + ] + result = greedy_diverse_select(low_first, top_n=1, max_pairwise_similarity=0.85) + assert result[0].candidate.candidate_id == "LOW" diff --git a/tests/test_selection_pareto.py b/tests/test_selection_pareto.py new file mode 100644 index 00000000..42600b92 --- /dev/null +++ b/tests/test_selection_pareto.py @@ -0,0 +1,100 @@ +"""Tests for selection/pareto.py — rank_candidates and select_top.""" +from __future__ import annotations + +from openamp_foundry.features.physchem import compute_features +from openamp_foundry.selection.pareto import rank_candidates, select_top +from openamp_foundry.types import PeptideCandidate, ScoredCandidate + + +def _make(cid: str, ensemble: float, seq: str = "KWKLFKKIGAVLKVL") -> ScoredCandidate: + features = compute_features(seq) + return ScoredCandidate( + candidate=PeptideCandidate(candidate_id=cid, sequence=seq, source="test"), + features=features, + scores={"ensemble": ensemble, "activity": 0.7, "safety": 0.8}, + ) + + +_POOL = [ + _make("A", 0.70), + _make("B", 0.90), + _make("C", 0.50), + _make("D", 0.85), + _make("E", 0.60), +] + + +class TestRankCandidates: + def test_returns_descending_order(self): + ranked = rank_candidates(list(_POOL)) + scores = [c.scores["ensemble"] for c in ranked] + assert scores == sorted(scores, reverse=True) + + def test_highest_ensemble_is_first(self): + ranked = rank_candidates(list(_POOL)) + assert ranked[0].candidate.candidate_id == "B" + + def test_lowest_ensemble_is_last(self): + ranked = rank_candidates(list(_POOL)) + assert ranked[-1].candidate.candidate_id == "C" + + def test_does_not_mutate_input(self): + original_order = [c.candidate.candidate_id for c in _POOL] + rank_candidates(list(_POOL)) + assert [c.candidate.candidate_id for c in _POOL] == original_order + + def test_empty_list_returns_empty(self): + assert rank_candidates([]) == [] + + def test_single_candidate_returned(self): + single = [_make("X", 0.75)] + ranked = rank_candidates(single) + assert len(ranked) == 1 + assert ranked[0].candidate.candidate_id == "X" + + def test_equal_scores_preserves_all(self): + equal = [_make("P", 0.80), _make("Q", 0.80), _make("R", 0.80)] + ranked = rank_candidates(equal) + assert len(ranked) == 3 + assert all(c.scores["ensemble"] == 0.80 for c in ranked) + + def test_missing_ensemble_key_defaults_to_zero(self): + # rank_candidates uses .get("ensemble", 0.0) — a missing key silently ranks last + no_ensemble = ScoredCandidate( + candidate=PeptideCandidate(candidate_id="NO_ENS", sequence="AAAA", source="test"), + features=compute_features("AAAA"), + scores={"activity": 0.99}, # no "ensemble" key + ) + ranked = rank_candidates([no_ensemble, _make("HAS_ENS", 0.5)]) + assert ranked[-1].candidate.candidate_id == "NO_ENS" + + +class TestSelectTop: + def test_returns_correct_count(self): + selected = select_top(_POOL, top_n=3) + assert len(selected) == 3 + + def test_returns_top_ensemble(self): + selected = select_top(_POOL, top_n=2) + ids = {c.candidate.candidate_id for c in selected} + assert ids == {"B", "D"} + + def test_top_1_returns_best(self): + selected = select_top(_POOL, top_n=1) + assert selected[0].candidate.candidate_id == "B" + + def test_top_n_larger_than_pool_returns_all(self): + selected = select_top(_POOL, top_n=100) + assert len(selected) == len(_POOL) + + def test_top_0_returns_empty(self): + selected = select_top(_POOL, top_n=0) + assert selected == [] + + def test_empty_pool_returns_empty(self): + assert select_top([], top_n=5) == [] + + def test_result_is_sorted_descending(self): + selected = select_top(_POOL, top_n=4) + scores = [c.scores["ensemble"] for c in selected] + assert scores == sorted(scores, reverse=True) diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 00000000..d1b93ef1 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,156 @@ +"""Tests for utils/hashing.py and utils/io.py.""" +from __future__ import annotations + +import hashlib +import json +import tempfile +from pathlib import Path + +from openamp_foundry.utils.hashing import file_sha256, stable_json_hash +from openamp_foundry.utils.io import read_json, write_json, write_jsonl + + +class TestStableJsonHash: + def test_same_object_produces_same_hash(self): + obj = {"a": 1, "b": [2, 3]} + assert stable_json_hash(obj) == stable_json_hash(obj) + + def test_key_order_does_not_matter(self): + a = {"x": 1, "y": 2} + b = {"y": 2, "x": 1} + assert stable_json_hash(a) == stable_json_hash(b) + + def test_different_values_differ(self): + assert stable_json_hash({"v": 1}) != stable_json_hash({"v": 2}) + + def test_returns_64_char_hex(self): + h = stable_json_hash({"k": "v"}) + assert len(h) == 64 + assert all(c in "0123456789abcdef" for c in h) + + def test_empty_dict(self): + h = stable_json_hash({}) + assert len(h) == 64 + + def test_nested_structure(self): + h1 = stable_json_hash({"nested": {"a": [1, 2]}}) + h2 = stable_json_hash({"nested": {"a": [1, 3]}}) + assert h1 != h2 + + def test_list_input(self): + h = stable_json_hash([1, 2, 3]) + assert len(h) == 64 + + def test_string_input(self): + h = stable_json_hash("hello") + assert len(h) == 64 + + +class TestFileSha256: + def test_matches_content_hash(self): + content = b"hello world" + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "test.bin" + path.write_bytes(content) + expected = hashlib.sha256(content).hexdigest() + assert file_sha256(path) == expected + + def test_different_files_differ(self): + with tempfile.TemporaryDirectory() as d: + p1 = Path(d) / "a.bin" + p2 = Path(d) / "b.bin" + p1.write_bytes(b"abc") + p2.write_bytes(b"xyz") + assert file_sha256(p1) != file_sha256(p2) + + def test_returns_64_char_hex(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "data.bin" + path.write_bytes(b"data") + assert len(file_sha256(path)) == 64 + + def test_empty_file(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "empty.bin" + path.write_bytes(b"") + expected = hashlib.sha256(b"").hexdigest() + assert file_sha256(path) == expected + + +class TestWriteJsonl: + def test_writes_one_line_per_row(self): + rows = [{"a": 1}, {"b": 2}, {"c": 3}] + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "out.jsonl" + write_jsonl(path, rows) + lines = path.read_text().strip().splitlines() + assert len(lines) == 3 + + def test_each_line_is_valid_json(self): + rows = [{"x": i} for i in range(5)] + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "out.jsonl" + write_jsonl(path, rows) + lines = path.read_text().strip().splitlines() + parsed = [json.loads(ln) for ln in lines] + assert [p["x"] for p in parsed] == list(range(5)) + + def test_creates_parent_directories(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "sub" / "dir" / "out.jsonl" + write_jsonl(path, [{"k": "v"}]) + assert path.exists() + + def test_empty_iterable_creates_empty_file(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "out.jsonl" + write_jsonl(path, []) + assert path.read_text() == "" + + def test_unicode_values_preserved(self): + rows = [{"name": "café"}] + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "out.jsonl" + write_jsonl(path, rows) + lines = path.read_text().strip().splitlines() + assert json.loads(lines[0])["name"] == "café" + + +class TestReadWriteJson: + def test_round_trip(self): + payload = {"key": "value", "num": 42, "list": [1, 2, 3]} + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "data.json" + write_json(path, payload) + loaded = read_json(path) + assert loaded == payload + + def test_creates_parent_directories(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "nested" / "data.json" + write_json(path, {"x": 1}) + assert path.exists() + + def test_file_ends_with_newline(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "data.json" + write_json(path, {"a": 1}) + content = path.read_text() + assert content.endswith("\n") + + def test_keys_sorted_in_output(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "data.json" + write_json(path, {"z": 3, "a": 1, "m": 2}) + content = path.read_text() + first_key_pos = content.index('"a"') + m_pos = content.index('"m"') + z_pos = content.index('"z"') + assert first_key_pos < m_pos < z_pos + + def test_unicode_preserved(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "data.json" + write_json(path, {"name": "pré-synthèse"}) + loaded = read_json(path) + assert loaded["name"] == "pré-synthèse"