Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions tests/test_benchmark_evaluate.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,3 +128,10 @@ def test_counts_are_correct(self):
result = benchmark_summary(ITEMS, POSITIVES, ks=[2])
assert result["n_candidates"] == 5
assert result["n_positives"] == 2


class TestEnrichmentFactorEdgeCases:
def test_enrichment_factor_returns_zero_when_k_is_zero(self):
# k=0 → random_recall_at_k(n, n_pos, 0) == 0.0 → guard branch returns 0.0
ef = enrichment_factor(ITEMS, POSITIVES, k=0)
assert ef == 0.0
24 changes: 24 additions & 0 deletions tests/test_diversity.py
Original file line number Diff line number Diff line change
Expand Up @@ -288,3 +288,27 @@ def test_warning_contains_family_name(self):
warnings = family_structural_warnings(members, min_family_size=3)
for w in warnings:
assert "SEED-005" in w["message"]


class TestClusterPanelAlreadyAssignedSkip:
"""Cover diversity.py:73 — inner-loop skip when j already has a cluster."""

def _make_candidates(self, id_seq_pairs):
return [{"candidate_id": cid, "sequence": seq} for cid, seq in id_seq_pairs]

def test_already_assigned_cluster_member_is_skipped_in_inner_loop(self):
# Arrange: 4 sequences where seq_0 is similar to seq_1 AND seq_3 (not seq_2).
# When outer i=0 runs, inner assigns seq_1 and seq_3 to cluster_0.
# When outer i=2 runs (seq_2 gets a new cluster), inner j=3 already has
# cluster_0 → continue (line 73). Verified: 3 clusters [0,0,1,0].
candidates = self._make_candidates([
("A", "KWKLFKKIG"), # similar to B and D
("B", "KWKLFKKIA"), # similar to A (1 diff/9) — assigned to cluster 0 by i=0
("C", "RWRLFRKKG"), # different from A (4 diffs/9)
("D", "KWKLFKKIF"), # similar to A (1 diff/9) — assigned to cluster 0 by i=0
])
clustered = cluster_panel(candidates, similarity_threshold=0.85)
cluster_ids = [c["cluster_id"] for c in clustered]
assert cluster_ids[0] == cluster_ids[1], "A and B should be in the same cluster"
assert cluster_ids[0] == cluster_ids[3], "A and D should be in the same cluster"
assert cluster_ids[2] != cluster_ids[0], "C should be in a different cluster"
35 changes: 35 additions & 0 deletions tests/test_pilot_panel.py
Original file line number Diff line number Diff line change
Expand Up @@ -504,3 +504,38 @@ def test_selection_method_describes_full_formula(self, tmp_path):
assert "selectivity_proxy" in content
assert "novelty" in content
assert "serum_stability" in content


class TestPilotPhase3FillFromRemainder:
"""Cover pilot.py:173 — Phase 3 append when max_per_seed cap creates overflow.

max_per_seed=1 with 2 seeds means Phase 1 fills 2 slots. The third
slot can only be filled from the per-seed overflow remainder (Phase 3).
"""

def _c(self, cid, seed, seq, priority):
return {
"candidate_id": cid,
"sequence": seq,
"source": f"template_mutation_from_{seed}",
"scores": {
"ensemble": priority, "activity": priority, "boman_activity": 0.65,
"disagreement": 0.05, "safety": 0.9, "synthesis": 0.8, "novelty": 0.9,
},
"features": {"length": len(seq)},
}

def test_overflow_candidates_fill_remaining_slots(self):
# 2 seeds (S1, S2), max_per_seed=1, n=3:
# Phase 1 fills A1 (best S1) and B1 (best S2) → 2 selected.
# C1 (S1 second-best) is in cap-overflow remainder.
# Phase 3 appends C1 → line 173 reached, len(result) == 3.
candidates = [
self._c("A1", "S1", "KWKLFKKIG", 0.9),
self._c("B1", "S2", "RWRLFRRKG", 0.8),
self._c("C1", "S1", "KWKLKKILG", 0.7),
]
result = select_pilot_panel(candidates, n=3, max_per_seed=1, similarity_threshold=None)
assert len(result) == 3
ids = {r["candidate_id"] for r in result}
assert "C1" in ids, "Phase 3 overflow candidate must be selected"
13 changes: 12 additions & 1 deletion tests/test_pipeline_demo.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from pathlib import Path

from openamp_foundry.pipeline import run_ranking_pipeline
from openamp_foundry.pipeline import build_run_manifest, run_ranking_pipeline


def test_pipeline_demo(tmp_path):
Expand All @@ -18,3 +18,14 @@ def test_pipeline_demo(tmp_path):
assert out.exists()
assert report.exists()
assert any(Path(certs).glob("*.json"))


def test_manifest_missing_input_gets_na_hash(tmp_path):
# pipeline.py:105 — non-existent path → "N/A" in input_hashes (no sha256 key)
real = tmp_path / "a.csv"
real.write_text("id,sequence\nC1,KWKLFK\n")
missing = tmp_path / "missing.csv"
m = build_run_manifest("r1", {}, [real, missing], ["out.csv"], "2026-01-01T00:00:00Z")
hashes = m["input_hashes"]
assert hashes[str(real)] != "N/A", "Existing file must have a real sha256"
assert hashes[str(missing)] == "N/A", "Missing file must have N/A hash (line 105)"
17 changes: 17 additions & 0 deletions tests/test_retrospective.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from openamp_foundry.benchmark.retrospective import (
_auc_wilcoxon,
_bootstrap_auroc_ci,
_recall_at_k,
run_retrospective_benchmark,
)
Expand Down Expand Up @@ -55,6 +56,22 @@ def test_no_positives_returns_zero(self):
assert _recall_at_k([0, 0, 0], k=2) == pytest.approx(0.0)


class TestBootstrapAurocCi:
def test_empty_pos_returns_chance(self):
result = _bootstrap_auroc_ci([], [0.9, 0.5, 0.1])
assert result["mean"] == pytest.approx(0.5)
assert result["ci_lo"] == pytest.approx(0.5)
assert result["ci_hi"] == pytest.approx(0.5)
assert result["n_bootstrap"] == 0

def test_empty_neg_returns_chance(self):
result = _bootstrap_auroc_ci([0.9, 0.5, 0.1], [])
assert result["mean"] == pytest.approx(0.5)
assert result["ci_lo"] == pytest.approx(0.5)
assert result["ci_hi"] == pytest.approx(0.5)
assert result["n_bootstrap"] == 0


class TestRunRetrospectiveBenchmark:
@pytest.fixture
def mini_amp_csv(self, tmp_path):
Expand Down
16 changes: 16 additions & 0 deletions tests/test_template_mutator.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

from openamp_foundry.generators.template_mutator import (
_conservative_substitutes,
generate_aggregation_safe_double_variants,
generate_all_variants,
generate_candidate_pool,
generate_charge_enhanced_variants,
Expand Down Expand Up @@ -57,6 +58,11 @@ def test_never_cross_group_substitution(self):
cross = [s for s in subs if s in charged]
assert not cross, f"Hydrophobic AA {aa!r} has charged subs: {cross}"

def test_unknown_aa_returns_empty_list(self):
# AA not in any conservative group → guard at line 51 returns []
assert _conservative_substitutes("X") == []
assert _conservative_substitutes("B") == []


class TestSingleSubstitutionVariants:
def test_all_variants_same_length_as_seed(self):
Expand Down Expand Up @@ -126,6 +132,16 @@ def test_different_seeds_produce_different_variants(self):
v2 = generate_double_substitution_variants(SEED, n_samples=10, seed=2)
assert v1 != v2, "Different rng seeds should produce different variants"

def test_single_aa_sequence_returns_empty(self):
# < 2 substitutable positions → early-return guard in generate_double_substitution_variants
assert generate_double_substitution_variants("K") == []


class TestAggregationSafeDoubleVariants:
def test_single_aa_sequence_returns_empty(self):
# < 2 substitutable positions → early-return guard in generate_aggregation_safe_double_variants
assert generate_aggregation_safe_double_variants("K") == []


class TestChargeEnhancedVariants:
def test_replaces_polar_with_cationic(self):
Expand Down
Loading