test: external_predict and config coverage — final pre-synthesis gates

src/openamp_foundry/config.py

@@ -11,4 +11,7 @@
 
 def load_config(path: str | Path = DEFAULT_CONFIG_PATH) -> dict[str, Any]:
     with Path(path).open("r", encoding="utf-8") as f:
-        return yaml.safe_load(f)
+        result = yaml.safe_load(f)
+    if result is None:
+        raise ValueError(f"Config file is empty: {path}")
+    return result

tests/test_config.py

@@ -0,0 +1,103 @@
+"""Tests for config.py — load_config loading and error handling."""
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from openamp_foundry.config import load_config
+
+
+class TestLoadConfig:
+    def _write_config(self, content: str, path: Path) -> None:
+        path.write_text(content, encoding="utf-8")
+
+    def test_loads_valid_yaml(self):
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "config.yaml"
+            self._write_config("key: value\nnested:\n  a: 1\n", cfg_path)
+            result = load_config(cfg_path)
+        assert result["key"] == "value"
+        assert result["nested"]["a"] == 1
+
+    def test_loads_default_pipeline_config(self):
+        default = Path(__file__).parents[1] / "configs" / "pipeline.yaml"
+        result = load_config(default)
+        assert "filters" in result
+        assert "weights" in result
+        assert "selection" in result
+
+    def test_default_config_weights_are_positive(self):
+        default = Path(__file__).parents[1] / "configs" / "pipeline.yaml"
+        result = load_config(default)
+        for name, w in result["weights"].items():
+            assert w > 0, f"Weight '{name}' must be positive, got {w}"
+
+    def test_default_config_filters_present(self):
+        default = Path(__file__).parents[1] / "configs" / "pipeline.yaml"
+        result = load_config(default)
+        filters = result["filters"]
+        assert "min_length" in filters
+        assert "max_length" in filters
+        assert filters["min_length"] < filters["max_length"]
+
+    def test_default_config_selection_present(self):
+        default = Path(__file__).parents[1] / "configs" / "pipeline.yaml"
+        result = load_config(default)
+        sel = result["selection"]
+        assert "top_n" in sel
+        assert "min_novelty" in sel
+        assert "max_safety_risk" in sel
+        assert 0.0 < sel["max_safety_risk"] <= 1.0
+
+    def test_string_path_accepted(self):
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "config.yaml"
+            self._write_config("x: 42\n", cfg_path)
+            result = load_config(str(cfg_path))
+        assert result["x"] == 42
+
+    def test_missing_file_raises(self):
+        with pytest.raises((FileNotFoundError, OSError)):
+            load_config("/nonexistent/path/config.yaml")
+
+    def test_empty_yaml_raises(self):
+        # yaml.safe_load("") returns None; load_config rejects this to protect
+        # downstream callers that always expect a dict.
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "config.yaml"
+            self._write_config("", cfg_path)
+            with pytest.raises(ValueError, match="empty"):
+                load_config(cfg_path)
+
+    def test_list_yaml_preserved(self):
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "config.yaml"
+            self._write_config("items:\n  - a\n  - b\n", cfg_path)
+            result = load_config(cfg_path)
+        assert result["items"] == ["a", "b"]
+
+    def test_numeric_values_preserved(self):
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "config.yaml"
+            self._write_config("threshold: 0.75\ncount: 42\n", cfg_path)
+            result = load_config(cfg_path)
+        assert abs(result["threshold"] - 0.75) < 1e-9
+        assert result["count"] == 42
+
+    def test_bool_values_preserved(self):
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "config.yaml"
+            self._write_config("enabled: true\ndisabled: false\n", cfg_path)
+            result = load_config(cfg_path)
+        assert result["enabled"] is True
+        assert result["disabled"] is False
+
+    def test_malformed_yaml_raises(self):
+        import yaml as _yaml
+        with tempfile.TemporaryDirectory() as d:
+            cfg_path = Path(d) / "bad.yaml"
+            self._write_config("key: [unclosed\n", cfg_path)
+            with pytest.raises(_yaml.YAMLError):
+                load_config(cfg_path)

tests/test_external_predict.py

@@ -0,0 +1,275 @@
+"""Tests for reports/external_predict.py — FASTA generation, checklist, confident panel."""
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+from openamp_foundry.reports.external_predict import (
+    write_confident_panel,
+    write_external_predict_checklist,
+    write_pilot_fasta,
+)
+
+_PANEL = [
+    {"candidate_id": "SEED-001_VAR_001", "sequence": "KWKLFKKIGAVLKVL", "pilot_rank": 1, "seed": "SEED-001"},
+    {"candidate_id": "SEED-002_VAR_002", "sequence": "GIGKFLHSAKKFGKA", "pilot_rank": 2, "seed": "SEED-002"},
+    {"candidate_id": "SEED-003_VAR_003", "sequence": "FLPAIGRLLNGIL", "pilot_rank": 3, "seed": "SEED-003"},
+]
+
+
+class TestWritePilotFasta:
+    def test_creates_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            assert path.exists()
+
+    def test_creates_parent_dirs(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "sub" / "dir" / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            assert path.exists()
+
+    def test_one_header_per_candidate(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            headers = [ln for ln in path.read_text().splitlines() if ln.startswith(">")]
+        assert len(headers) == len(_PANEL)
+
+    def test_header_contains_candidate_id(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            content = path.read_text()
+        assert "SEED-001_VAR_001" in content
+
+    def test_header_contains_rank(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            lines = path.read_text().splitlines()
+        header_lines = [line for line in lines if line.startswith(">")]
+        assert "rank=1" in header_lines[0]
+
+    def test_header_contains_seed(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            lines = path.read_text().splitlines()
+        assert "seed=SEED-001" in lines[0]
+
+    def test_sequences_present_in_body(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            content = path.read_text()
+        for c in _PANEL:
+            assert c["sequence"] in content
+
+    def test_file_ends_with_newline(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(_PANEL, path)
+            content = path.read_text()
+        assert content.endswith("\n")
+
+    def test_empty_panel_produces_empty_like_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta([], path)
+            content = path.read_text()
+        assert content.strip() == ""
+
+    def test_missing_optional_fields_do_not_crash(self):
+        minimal = [{"sequence": "KWKLF"}]
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(minimal, path)
+            content = path.read_text()
+        assert "KWKLF" in content
+
+    def test_empty_sequence_in_candidate_writes_blank_body_line(self):
+        # write_pilot_fasta writes an empty sequence line for candidates with seq="".
+        # External predictor tools may reject or misparse this — callers should
+        # validate sequences before calling write_pilot_fasta.
+        panel_with_empty = [{"candidate_id": "EMPTY", "sequence": "", "pilot_rank": 1}]
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "pilot.fasta"
+            write_pilot_fasta(panel_with_empty, path)
+            lines = path.read_text().splitlines()
+        assert lines[0].startswith(">EMPTY")
+        assert lines[1] == ""  # blank body line — documents the current contract
+
+
+class TestWriteExternalPredictChecklist:
+    def test_creates_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            assert out.exists()
+
+    def test_creates_parent_dirs(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "sub" / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            assert out.exists()
+
+    def test_contains_title_header(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "External Predictor Checklist" in content
+
+    def test_contains_all_three_tools(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "CAMPR4" in content
+        assert "AMPScanner" in content
+        assert "dbAMP" in content
+
+    def test_contains_panel_size(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "3 candidates" in content
+
+    def test_contains_result_table_headers(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "CAMPR4" in content
+        assert "Tools Agree" in content
+
+    def test_each_candidate_in_result_table(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        for c in _PANEL:
+            assert c["candidate_id"] in content
+
+    def test_contains_synthesis_gate_decision_table(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "STOP" in content or "Proceed" in content
+
+    def test_contains_disclaimer_text(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "wet-lab" in content.lower() or "expert" in content.lower()
+
+    def test_generated_at_included_when_provided(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(
+                _PANEL, "pilot.fasta", out, generated_at="2026-01-01T00:00:00+00:00"
+            )
+            content = out.read_text()
+        assert "2026-01-01" in content
+
+    def test_generated_at_omitted_when_empty(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out, generated_at="")
+            content = out.read_text()
+        assert "Generated:" not in content
+
+    def test_file_ends_with_newline(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert content.endswith("\n")
+
+    def test_fasta_path_referenced_in_checklist(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "my_panel.fasta", out)
+            content = out.read_text()
+        assert "my_panel.fasta" in content
+
+    def test_two_thirds_agreement_rule_stated(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "checklist.md"
+            write_external_predict_checklist(_PANEL, "pilot.fasta", out)
+            content = out.read_text()
+        assert "≥2" in content or "2/3" in content
+
+
+class TestWriteConfidentPanel:
+    def test_filters_to_kept_ids(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            result = write_confident_panel(
+                _PANEL,
+                ["SEED-001_VAR_001", "SEED-003_VAR_003"],
+                out,
+            )
+        ids = {c["candidate_id"] for c in result}
+        assert ids == {"SEED-001_VAR_001", "SEED-003_VAR_003"}
+
+    def test_reranks_confident_panel(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            result = write_confident_panel(
+                _PANEL,
+                ["SEED-002_VAR_002", "SEED-003_VAR_003"],
+                out,
+            )
+        ranks = [c["pilot_rank"] for c in result]
+        assert ranks == [1, 2]
+
+    def test_empty_keep_returns_empty(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            result = write_confident_panel(_PANEL, [], out)
+        assert result == []
+
+    def test_unknown_keep_id_ignored(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            result = write_confident_panel(_PANEL, ["NONEXISTENT"], out)
+        assert result == []
+
+    def test_creates_csv_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            write_confident_panel(_PANEL, ["SEED-001_VAR_001"], out)
+            assert (Path(d) / "confident.csv").exists()
+
+    def test_creates_markdown_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            write_confident_panel(_PANEL, ["SEED-001_VAR_001"], out)
+            assert (Path(d) / "confident.md").exists()
+
+    def test_returns_list_of_dicts(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            result = write_confident_panel(_PANEL, ["SEED-001_VAR_001"], out)
+        assert isinstance(result, list)
+        assert isinstance(result[0], dict)
+
+    def test_output_order_follows_panel_not_keep_ids(self):
+        # keep_ids is converted to a set internally; output order is determined
+        # by the original panel order, NOT the order of keep_ids.
+        # This documents the current contract so any future reordering is explicit.
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "confident"
+            result = write_confident_panel(
+                _PANEL,
+                ["SEED-003_VAR_003", "SEED-001_VAR_001"],  # reversed from panel order
+                out,
+            )
+        assert result[0]["candidate_id"] == "SEED-001_VAR_001"  # panel order wins
+        assert result[1]["candidate_id"] == "SEED-003_VAR_003"