From 3461327aa6585b5a4f2562437db2d7595a3dd240 Mon Sep 17 00:00:00 2001 From: j Date: Sun, 28 Jun 2026 08:12:02 +0700 Subject: [PATCH 1/2] =?UTF-8?q?test:=20add=20external=5Fpredict=20and=20co?= =?UTF-8?q?nfig=20coverage=20=E2=80=94=20final=20pre-synthesis=20gates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/test_external_predict.py (31 tests): write_pilot_fasta: file creation, FASTA header format (id/rank/seed), sequence body, empty panel, missing optional fields, newline termination write_external_predict_checklist: all 3 tools named, panel size, result table rows, 2/3-agreement gate, disclaimer, generated_at conditional, fasta path ref write_confident_panel: keeps only specified IDs, re-ranks 1..n, handles empty and nonexistent IDs, writes .csv and .md outputs - tests/test_config.py (11 tests): load_config: valid YAML, real pipeline config (filters/weights/selection), weights sum to 1.0, string path, missing file raises, empty YAML, list/numeric/bool values preserved 705 tests total (was 663). --- tests/test_config.py | 92 ++++++++++++ tests/test_external_predict.py | 249 +++++++++++++++++++++++++++++++++ 2 files changed, 341 insertions(+) create mode 100644 tests/test_config.py create mode 100644 tests/test_external_predict.py diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 00000000..982345cd --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,92 @@ +"""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_sum_to_one(self): + default = Path(__file__).parents[1] / "configs" / "pipeline.yaml" + result = load_config(default) + total = sum(result["weights"].values()) + assert abs(total - 1.0) < 0.01, f"Weights sum to {total}, expected 1.0" + + 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 + + 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_returns_none_or_dict(self): + with tempfile.TemporaryDirectory() as d: + cfg_path = Path(d) / "config.yaml" + self._write_config("", cfg_path) + result = load_config(cfg_path) + # yaml.safe_load("") returns None + assert result is None or isinstance(result, dict) + + 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 diff --git a/tests/test_external_predict.py b/tests/test_external_predict.py new file mode 100644 index 00000000..b4458d04 --- /dev/null +++ b/tests/test_external_predict.py @@ -0,0 +1,249 @@ +"""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 = [l for l in lines if l.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 + + +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 and ("tool" in content.lower() or "≥2" 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) From d0e8240042c465dd807c6061d3b369afef8f5274 Mon Sep 17 00:00:00 2001 From: j Date: Sun, 28 Jun 2026 08:17:52 +0700 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20address=20code=20review=20=E2=80=94?= =?UTF-8?q?=20E741,=20ordering=20doc,=20empty=20YAML=20guard,=20malformed?= =?UTF-8?q?=20YAML=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_external_predict.py: - fix E741: rename l → line in list comprehension - add test_empty_sequence_in_candidate_writes_blank_body_line: documents that callers must validate sequences before write_pilot_fasta (blank FASTA body line risks silent misparse by external prediction tools) - add test_output_order_follows_panel_not_keep_ids: documents that keep_ids order is not honoured — output order is always panel order - fix test_two_thirds_agreement_rule_stated: use '≥2' or '2/3' instead of '2' config.py: - load_config now raises ValueError('Config file is empty') when yaml.safe_load returns None; protects all downstream callers expecting a dict test_config.py: - test_empty_yaml_raises: replaces test_empty_yaml_returns_none_or_dict (was accepting None which would crash any caller doing result['key']) - test_malformed_yaml_raises: new test verifying yaml.YAMLError propagates - test_default_config_weights_are_positive: replaces fragile sum-to-1.0 check (ensemble scorer normalises by total weight so sum doesn't have to be 1.0) - test_default_config_selection_present: add max_safety_risk assertion --- src/openamp_foundry/config.py | 5 ++++- tests/test_config.py | 25 ++++++++++++++++++------- tests/test_external_predict.py | 30 ++++++++++++++++++++++++++++-- 3 files changed, 50 insertions(+), 10 deletions(-) diff --git a/src/openamp_foundry/config.py b/src/openamp_foundry/config.py index 62eb3f96..955d0271 100644 --- a/src/openamp_foundry/config.py +++ b/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 diff --git a/tests/test_config.py b/tests/test_config.py index 982345cd..628003e0 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -28,11 +28,11 @@ def test_loads_default_pipeline_config(self): assert "weights" in result assert "selection" in result - def test_default_config_weights_sum_to_one(self): + def test_default_config_weights_are_positive(self): default = Path(__file__).parents[1] / "configs" / "pipeline.yaml" result = load_config(default) - total = sum(result["weights"].values()) - assert abs(total - 1.0) < 0.01, f"Weights sum to {total}, expected 1.0" + 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" @@ -48,6 +48,8 @@ def test_default_config_selection_present(self): 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: @@ -60,13 +62,14 @@ def test_missing_file_raises(self): with pytest.raises((FileNotFoundError, OSError)): load_config("/nonexistent/path/config.yaml") - def test_empty_yaml_returns_none_or_dict(self): + 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) - result = load_config(cfg_path) - # yaml.safe_load("") returns None - assert result is None or isinstance(result, dict) + with pytest.raises(ValueError, match="empty"): + load_config(cfg_path) def test_list_yaml_preserved(self): with tempfile.TemporaryDirectory() as d: @@ -90,3 +93,11 @@ def test_bool_values_preserved(self): 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) diff --git a/tests/test_external_predict.py b/tests/test_external_predict.py index b4458d04..00edfa79 100644 --- a/tests/test_external_predict.py +++ b/tests/test_external_predict.py @@ -49,7 +49,7 @@ def test_header_contains_rank(self): path = Path(d) / "pilot.fasta" write_pilot_fasta(_PANEL, path) lines = path.read_text().splitlines() - header_lines = [l for l in lines if l.startswith(">")] + header_lines = [line for line in lines if line.startswith(">")] assert "rank=1" in header_lines[0] def test_header_contains_seed(self): @@ -89,6 +89,18 @@ def test_missing_optional_fields_do_not_crash(self): 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): @@ -191,7 +203,7 @@ def test_two_thirds_agreement_rule_stated(self): out = Path(d) / "checklist.md" write_external_predict_checklist(_PANEL, "pilot.fasta", out) content = out.read_text() - assert "2" in content and ("tool" in content.lower() or "≥2" in content) + assert "≥2" in content or "2/3" in content class TestWriteConfidentPanel: @@ -247,3 +259,17 @@ def test_returns_list_of_dicts(self): 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"