diff --git a/scripts/validate_negative_archive.py b/scripts/validate_negative_archive.py new file mode 100644 index 00000000..ae64c44b --- /dev/null +++ b/scripts/validate_negative_archive.py @@ -0,0 +1,140 @@ +"""Validate negative-result archive CSV entries. + +Checks each row against the schema defined in +docs/evidence/NEGATIVE_RESULT_ARCHIVE.md. + +Usage: + python scripts/validate_negative_archive.py \ + --archive outputs/negative_result_archive.csv +""" + +from __future__ import annotations + +import csv +import json +import sys +from datetime import datetime +from pathlib import Path + +REQUIRED = {"entry_id", "date", "candidate_id", "sequence", "reason_category", + "reason_detail", "pipeline_version", "source_batch"} + +VALID_CATEGORIES = {"pre_selection_reject", "selected_untested", "lab_inactive", + "lab_toxic", "control_failure", "synthesis_failure"} + +CONDITIONAL_CATEGORIES = {"lab_inactive", "lab_toxic"} + +DATE_FORMAT = "%Y-%m-%d" + + +def validate_archive(archive_path: str) -> dict: + p = Path(archive_path) + if not p.exists(): + return {"error": f"Archive not found: {archive_path}"} + + rows: list[dict] = [] + with p.open("r", newline="") as f: + reader = csv.DictReader(f) + for row in reader: + rows.append(row) + + if not rows: + return {"error": "Archive is empty"} + + errors: list[dict] = [] + warnings: list[dict] = [] + valid_count = 0 + + for i, row in enumerate(rows): + row_errors = [] + row_warnings = [] + + # Required fields + for field in REQUIRED: + val = row.get(field, "").strip() + if not val: + row_errors.append(f"Missing required field: {field}") + + # Reason category + cat = row.get("reason_category", "").strip() + if cat and cat not in VALID_CATEGORIES: + row_errors.append(f"Invalid reason_category: {cat}") + + # Date format + date_val = row.get("date", "").strip() + if date_val: + try: + datetime.strptime(date_val, DATE_FORMAT) + except ValueError: + row_errors.append(f"Invalid date format: {date_val} (expected YYYY-MM-DD)") + + # Conditional fields for lab-tested categories + if cat in CONDITIONAL_CATEGORIES: + for cf in ["assay_type", "assay_result", "assay_unit"]: + if not row.get(cf, "").strip(): + row_warnings.append(f"Conditional field '{cf}' recommended for {cat}") + + # Superseded_by should reference an existing entry_id + superseded = row.get("superseded_by", "").strip() + if superseded: + found = any(r.get("entry_id", "").strip() == superseded for r in rows) + if not found: + row_warnings.append(f"superseded_by={superseded} does not match any entry_id") + + # pipeline_version should not be empty if present + pv = row.get("pipeline_version", "").strip() + if not pv and date_val: + row_warnings.append("pipeline_version is recommended") + + if row_errors: + errors.append({ + "row": i + 1, + "entry_id": row.get("entry_id", "?"), + "errors": row_errors, + }) + else: + valid_count += 1 + + if row_warnings: + warnings.append({ + "row": i + 1, + "entry_id": row.get("entry_id", "?"), + "warnings": row_warnings, + }) + + return { + "archive": archive_path, + "total_rows": len(rows), + "valid": valid_count, + "errors": len(errors), + "warnings": len(warnings), + "per_row_errors": errors, + "per_row_warnings": warnings, + "all_valid": len(errors) == 0, + } + + +def main() -> int: + import argparse + parser = argparse.ArgumentParser(description="Validate negative-result archive") + parser.add_argument("--archive", required=True) + parser.add_argument("--out", default=None) + args = parser.parse_args() + + result = validate_archive(args.archive) + if "error" in result: + print(f"Error: {result['error']}", file=sys.stderr) + return 2 + + print(json.dumps(result, indent=2)) + print(f"\nValid: {result['valid']}/{result['total_rows']}, " + f"Errors: {result['errors']}, Warnings: {result['warnings']}") + + if args.out: + Path(args.out).write_text(json.dumps(result, indent=2) + "\n") + + return 0 if result["all_valid"] else 3 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_validate_negative_archive.py b/tests/test_validate_negative_archive.py new file mode 100644 index 00000000..bf82d2b4 --- /dev/null +++ b/tests/test_validate_negative_archive.py @@ -0,0 +1,151 @@ +"""Tests for the negative-result archive validator.""" + +import csv +import json +import subprocess +import sys +import tempfile +from pathlib import Path + +import pytest + +from scripts.validate_negative_archive import validate_archive + + +def _make_archive(tmp: Path, rows: list[dict]) -> str: + p = tmp / "archive.csv" + with p.open("w", newline="") as f: + if rows: + w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) + w.writeheader() + w.writerows(rows) + return str(p) + + +_VALID_ROW = { + "entry_id": "1", + "date": "2026-08-01", + "candidate_id": "WAVE0.5-001", + "sequence": "ACDEFGHIKLMNPQRSTVWY", + "reason_category": "lab_inactive", + "reason_detail": "MIC > 64 ug/mL", + "pipeline_version": "v0.5.49", + "source_batch": "wave0.5", + "assay_type": "MIC", + "assay_result": "128", + "assay_unit": "ug/mL", + "score_activity": "0.85", + "score_safety": "0.79", + "score_novelty": "0.88", + "score_ensemble": "0.83", + "recalibration_used": "no", + "superseded_by": "", + "reviewer_notes": "", +} + + +def test_valid_entry(): + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), [_VALID_ROW])) + assert result["all_valid"] is True + assert result["valid"] == 1 + + +def test_missing_required_field(): + row = dict(_VALID_ROW) + del row["candidate_id"] + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), [row])) + assert result["all_valid"] is False + assert result["errors"] == 1 + + +def test_invalid_category(): + row = dict(_VALID_ROW, reason_category="invalid_cat") + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), [row])) + assert result["all_valid"] is False + + +def test_invalid_date(): + row = dict(_VALID_ROW, date="not-a-date") + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), [row])) + assert result["all_valid"] is False + + +def test_conditional_fields_warning(): + row = dict(_VALID_ROW, assay_type="", assay_result="", assay_unit="") + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), [row])) + assert result["warnings"] >= 1 + + +def test_superseded_refers_to_existing(): + rows = [ + dict(_VALID_ROW, entry_id="1", superseded_by=""), + dict(_VALID_ROW, entry_id="2", superseded_by="1"), + ] + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), rows)) + assert result["all_valid"] is True + + +def test_superseded_refers_to_nonexistent(): + rows = [ + dict(_VALID_ROW, entry_id="1", superseded_by="999"), + ] + with tempfile.TemporaryDirectory() as tmp: + result = validate_archive(_make_archive(Path(tmp), rows)) + assert result["warnings"] >= 1 + + +def test_empty_archive(): + with tempfile.TemporaryDirectory() as tmp: + p = Path(tmp) / "empty.csv" + # CSV with only header + with p.open("w", newline="") as f: + w = csv.writer(f) + w.writerow(list(_VALID_ROW.keys())) + result = validate_archive(str(p)) + assert "error" in result or result["total_rows"] == 0 + + +def test_missing_file(): + result = validate_archive("/nonexistent.csv") + assert "error" in result + + +def test_cli_exit_valid(tmp_path): + p = _make_archive(tmp_path, [_VALID_ROW]) + r = subprocess.run( + [sys.executable, "scripts/validate_negative_archive.py", + "--archive", p], + capture_output=True, text=True, + ) + assert r.returncode == 0 + + +def test_cli_exit_invalid(tmp_path): + row = dict(_VALID_ROW, reason_category="") + p = _make_archive(tmp_path, [row]) + r = subprocess.run( + [sys.executable, "scripts/validate_negative_archive.py", + "--archive", p], + capture_output=True, text=True, + ) + assert r.returncode == 3 + + +def test_cli_writes_output(tmp_path): + p = _make_archive(tmp_path, [_VALID_ROW]) + out = tmp_path / "report.json" + subprocess.run( + [sys.executable, "scripts/validate_negative_archive.py", + "--archive", p, "--out", str(out)], + capture_output=True, + env={"PYTHONPATH": "src"}, + ) + assert out.exists() + data = json.loads(out.read_text()) + assert "valid" in data