diff --git a/tests/test_benchmark_evaluate.py b/tests/test_benchmark_evaluate.py index 9610089c..ff28c428 100644 --- a/tests/test_benchmark_evaluate.py +++ b/tests/test_benchmark_evaluate.py @@ -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 diff --git a/tests/test_diversity.py b/tests/test_diversity.py index 76c31f14..b9b8aff9 100644 --- a/tests/test_diversity.py +++ b/tests/test_diversity.py @@ -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" diff --git a/tests/test_pilot_panel.py b/tests/test_pilot_panel.py index def862a3..c72fd021 100644 --- a/tests/test_pilot_panel.py +++ b/tests/test_pilot_panel.py @@ -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" diff --git a/tests/test_pipeline_demo.py b/tests/test_pipeline_demo.py index eb112402..a5366c95 100644 --- a/tests/test_pipeline_demo.py +++ b/tests/test_pipeline_demo.py @@ -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): @@ -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)" diff --git a/tests/test_retrospective.py b/tests/test_retrospective.py index e3539a9c..7a5bcf0f 100644 --- a/tests/test_retrospective.py +++ b/tests/test_retrospective.py @@ -7,6 +7,7 @@ from openamp_foundry.benchmark.retrospective import ( _auc_wilcoxon, + _bootstrap_auroc_ci, _recall_at_k, run_retrospective_benchmark, ) @@ -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): diff --git a/tests/test_template_mutator.py b/tests/test_template_mutator.py index d04aaba4..0efd9fec 100644 --- a/tests/test_template_mutator.py +++ b/tests/test_template_mutator.py @@ -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, @@ -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): @@ -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):