diff --git a/Makefile b/Makefile index 297af65..138af4c 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test lint classify classify-hprc classify-and-report download classify-bam classify-vcf classify-fastq classify-fasta classify-gfa classify-headers classify-bed coverage-report validation-report all-reports download-hprc validate-hprc clean help +.PHONY: test lint validate-metadata classify classify-hprc classify-and-report download classify-bam classify-vcf classify-fastq classify-fasta classify-gfa classify-headers classify-bed coverage-report validation-report all-reports download-hprc validate-hprc clean help help: @echo "meta-disco — AnVIL file metadata classification" @@ -8,6 +8,7 @@ help: @echo " make classify Run full classification pipeline (all file types, parallel)" @echo " make classify-and-report Run classify + regenerate all reports" @echo " make download Download fresh AnVIL metadata from API" + @echo " make validate-metadata Check downloaded metadata against the record schema" @echo "" @echo " make classify-bam Classify BAM/CRAM files (network required)" @echo " make classify-vcf Classify VCF files (network required)" @@ -42,6 +43,9 @@ classify-and-report: classify classify-hprc all-reports download: python scripts/download_anvil_metadata.py +validate-metadata: + python scripts/validate_metadata.py + classify-headers: classify-bam classify-vcf classify-fastq classify-fasta classify-gfa classify-bam: diff --git a/scripts/validate_metadata.py b/scripts/validate_metadata.py new file mode 100644 index 0000000..bf6a181 --- /dev/null +++ b/scripts/validate_metadata.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Validate downloaded AnVIL metadata against the record schema. + +Run this after `make download` and before a multi-hour classification run: a +renamed key or a newly-nullable column otherwise surfaces as a crash somewhere +deep in the corpus, or as a silently wrong classification. + + python scripts/validate_metadata.py + python scripts/validate_metadata.py --input data/anvil/anvil_files_metadata.json + +Exit code 1 when the records do not match the schema. +""" + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.meta_disco.metadata_schema import ( # noqa: E402 + ANVIL_RECORD_SCHEMA, + validate_anvil_records, +) + +DEFAULT_INPUT = Path("data/anvil/anvil_files_metadata.json") + + +def load_records(path: Path) -> list[dict]: + """Read the records array from a metadata file (JSON or NDJSON).""" + if path.suffix == ".ndjson": + with open(path) as f: + return [json.loads(line) for line in f if line.strip()] + + with open(path) as f: + data = json.load(f) + if not isinstance(data, dict): + raise TypeError(f"Expected a JSON object, got {type(data).__name__}") + records = data.get("files") or data.get("results") + if records is None: + raise ValueError("JSON object must contain a 'files' or 'results' key") + return records + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", type=Path, default=DEFAULT_INPUT, + help=f"Metadata file to check (default: {DEFAULT_INPUT})") + args = parser.parse_args() + + if not args.input.is_file(): + print(f"No such file: {args.input}", file=sys.stderr) + return 1 + + records = load_records(args.input) + report = validate_anvil_records(records) + + print(f"Schema: {len(ANVIL_RECORD_SCHEMA)} fields") + print(f"Input : {args.input}") + print(report.summary()) + + if report.is_valid: + return 0 + + print( + "\nThe AnVIL record shape has changed, or the download is incomplete.\n" + "Fix the schema in src/meta_disco/metadata_schema.py if the change is\n" + "intended — do NOT coerce the values, or the drift is hidden rather than\n" + "caught.", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/meta_disco/metadata_schema.py b/src/meta_disco/metadata_schema.py new file mode 100644 index 0000000..7e3babf --- /dev/null +++ b/src/meta_disco/metadata_schema.py @@ -0,0 +1,173 @@ +"""Shape validation for AnVIL file-metadata records. + +Nothing between the AnVIL API and the classifiers checks the shape of a record. +A renamed key, a newly-nullable column, or a stringified size flows silently into +758k records and surfaces as a crash — or a wrong classification — somewhere deep +in a multi-hour run. + +Two contracts, deliberately separate: + +* :data:`ANVIL_RECORD_SCHEMA` — the full 12-field record the download script + produces. Checked by ``scripts/validate_metadata.py`` and at download time. +* :func:`validate_pipeline_records` — the far smaller promise ``ClassifyPipeline`` + relies on. Derived inputs (a filtered subset, a hand-built test fixture) carry + only a few fields, so requiring the full AnVIL record here would be wrong. + +Nothing here mutates a record. A ``None`` is never coerced to ``""``: that would +hide the drift this exists to catch, and it is what makes ``rec.get(k, "")`` +dangerous — the default fires only when the key is *absent*, so a present-but-null +value returns ``None`` and crashes the first ``.lower()`` downstream. +""" + +import re +from collections import Counter +from dataclasses import dataclass, field + +MD5_PATTERN = re.compile(r"^[0-9a-f]{32}$") + +# Bounded so a systematically broken download reports a summary, not 758k lines. +MAX_SAMPLES_PER_PROBLEM = 3 + + +@dataclass(frozen=True) +class FieldSpec: + """The contract for one record field.""" + + types: tuple[type, ...] + nullable: bool = False + allow_empty: bool = False + pattern: re.Pattern | None = None + min_value: int | None = None + + +# Measured against the 758,658-record corpus. Two constraints are easy to get +# wrong and are called out rather than inferred: +# * file_size may be 0 — three records are. `min_value=0`, not 1. +# * data_modality and reference_assembly are AnVIL's own declarations and are +# null for ~99% of files. They are nullable by design, not by accident. +ANVIL_RECORD_SCHEMA: dict[str, FieldSpec] = { + "entry_id": FieldSpec((str,)), + "file_id": FieldSpec((str,)), + "file_name": FieldSpec((str,)), + "file_format": FieldSpec((str,)), + "file_size": FieldSpec((int,), min_value=0), + "file_md5sum": FieldSpec((str,), pattern=MD5_PATTERN), + "drs_uri": FieldSpec((str,)), + "dataset_id": FieldSpec((str,)), + "dataset_title": FieldSpec((str,)), + "is_supplementary": FieldSpec((bool,)), + "data_modality": FieldSpec((str,), nullable=True), + "reference_assembly": FieldSpec((str,), nullable=True), +} + +# Fields ClassifyPipeline reads. Each may be absent (a derived input need not +# carry it) but must never be present-and-null, which is the failure this guards. +PIPELINE_STRING_FIELDS = ("file_md5sum", "file_name", "file_format", "dataset_title") + + +@dataclass +class ValidationReport: + """Problems found, counted by kind, with a bounded sample of each.""" + + total: int = 0 + counts: Counter = field(default_factory=Counter) + samples: dict[str, list[str]] = field(default_factory=dict) + + @property + def is_valid(self) -> bool: + return not self.counts + + @property + def problem_records(self) -> int: + """Number of distinct problems recorded, not of distinct records. + + One record may contribute several problems, so this is an upper bound on + the number of bad records — it is a signal of scale, not a count. + """ + return sum(self.counts.values()) + + def add(self, kind: str, detail: str) -> None: + self.counts[kind] += 1 + bucket = self.samples.setdefault(kind, []) + if len(bucket) < MAX_SAMPLES_PER_PROBLEM: + bucket.append(detail) + + def summary(self) -> str: + if self.is_valid: + return f"{self.total:,} records, no problems." + lines = [f"{self.total:,} records, {len(self.counts)} kinds of problem:"] + for kind, n in self.counts.most_common(): + lines.append(f" {kind}: {n:,}") + for detail in self.samples.get(kind, []): + lines.append(f" e.g. {detail}") + return "\n".join(lines) + + +def _check_value(name: str, value, spec: FieldSpec) -> str | None: + """The problem this value has against its spec, or None.""" + if value is None: + return None if spec.nullable else "null" + + # bool is a subclass of int, so an `is_supplementary` in a `file_size` slot + # would silently pass an isinstance(int) check. + if isinstance(value, bool) and bool not in spec.types: + return "type" + if not isinstance(value, spec.types): + return "type" + + if isinstance(value, str): + if not value and not spec.allow_empty: + return "empty" + if spec.pattern and not spec.pattern.match(value): + return "pattern" + if spec.min_value is not None and value < spec.min_value: + return "range" + return None + + +def validate_anvil_records( + records: list[dict], schema: dict[str, FieldSpec] = ANVIL_RECORD_SCHEMA +) -> ValidationReport: + """Validate records against the full AnVIL contract. + + Unknown fields are *not* a problem: the API may add a column, and that should + not fail a download. A renamed column shows up as `missing:`. + """ + report = ValidationReport(total=len(records)) + for i, rec in enumerate(records): + if not isinstance(rec, dict): + report.add("not_an_object", f"record {i}: {type(rec).__name__}") + continue + for name, spec in schema.items(): + if name not in rec: + report.add(f"missing:{name}", f"record {i}") + continue + problem = _check_value(name, rec[name], spec) + if problem: + report.add(f"{problem}:{name}", f"record {i}: {rec[name]!r}") + return report + + +def validate_pipeline_records(records: list[dict]) -> ValidationReport: + """Validate the small promise ClassifyPipeline relies on. + + Each of PIPELINE_STRING_FIELDS may be absent, but must not be present-and-null: + `rec.get("file_name", "")` returns None for a null, and the first `.lower()` or + slice downstream then raises. `file_size`, when present, must be a non-negative + int (or null — some derived inputs omit the size rather than the key). + """ + report = ValidationReport(total=len(records)) + for i, rec in enumerate(records): + if not isinstance(rec, dict): + report.add("not_an_object", f"record {i}: {type(rec).__name__}") + continue + for name in PIPELINE_STRING_FIELDS: + if name in rec and not isinstance(rec[name], str): + kind = "null" if rec[name] is None else "type" + report.add(f"{kind}:{name}", f"record {i}: {rec[name]!r}") + size = rec.get("file_size") + if size is not None and (isinstance(size, bool) or not isinstance(size, int)): + report.add("type:file_size", f"record {i}: {size!r}") + elif isinstance(size, int) and not isinstance(size, bool) and size < 0: + report.add("range:file_size", f"record {i}: {size!r}") + return report diff --git a/src/meta_disco/pipeline.py b/src/meta_disco/pipeline.py index 86c71bb..bb5179b 100644 --- a/src/meta_disco/pipeline.py +++ b/src/meta_disco/pipeline.py @@ -14,6 +14,7 @@ from .fetchers import FetchError from .header_classifier import classify_without_content +from .metadata_schema import validate_pipeline_records @dataclass(frozen=True) @@ -204,21 +205,35 @@ def classify_single( # --- Internal --- def _load_input(self) -> list[dict]: - """Load NDJSON or JSON input, extracting records array.""" + """Load NDJSON or JSON input, extract the records array, and validate it. + + Validates only the fields this pipeline reads (``validate_pipeline_records``), + not the full AnVIL record: a derived input may legitimately carry a subset. + Raises rather than dropping the offenders — a dropped record is + indistinguishable from a file that was never seen (see #155). + """ with open(self.input_path) as f: if self.input_path.suffix == ".ndjson": - return [json.loads(line) for line in f if line.strip()] - data = json.load(f) - if not isinstance(data, dict): - raise TypeError( - f"Expected JSON object with 'results' key, got {type(data).__name__}" - ) - records = data.get("results") or data.get("files") - if records is None: - raise ValueError( - "JSON object must contain a 'results' or 'files' key" - ) - return records + records = [json.loads(line) for line in f if line.strip()] + else: + data = json.load(f) + if not isinstance(data, dict): + raise TypeError( + f"Expected JSON object with 'results' key, got {type(data).__name__}" + ) + records = data.get("results") or data.get("files") + if records is None: + raise ValueError( + "JSON object must contain a 'results' or 'files' key" + ) + + report = validate_pipeline_records(records) + if not report.is_valid: + raise ValueError( + f"{self.input_path} holds records this pipeline cannot read.\n" + f"{report.summary()}" + ) + return records def _filter_records(self, records: list[dict]) -> list[dict]: """Filter to records matching file_type.extensions with valid MD5.""" diff --git a/tests/test_metadata_schema.py b/tests/test_metadata_schema.py new file mode 100644 index 0000000..9540014 --- /dev/null +++ b/tests/test_metadata_schema.py @@ -0,0 +1,156 @@ +"""Tests for AnVIL metadata record validation. + +The constraints here were measured against the 758,658-record corpus, not +assumed. Two of them are counterintuitive and are pinned explicitly: `file_size` +may be 0, and `data_modality`/`reference_assembly` are nullable by design. +""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.meta_disco.metadata_schema import ( + ANVIL_RECORD_SCHEMA, + validate_anvil_records, + validate_pipeline_records, +) + + +def _record(**overrides): + """A record that satisfies the full AnVIL schema.""" + rec = { + "entry_id": "e1", + "file_id": "f1", + "file_name": "sample.bam", + "file_format": ".bam", + "file_size": 1234, + "file_md5sum": "0" * 32, + "drs_uri": "drs://drs.anv0:v2_abc", + "dataset_id": "d1", + "dataset_title": "ANVIL_HPRC", + "is_supplementary": False, + "data_modality": None, + "reference_assembly": None, + } + rec.update(overrides) + return rec + + +class TestAnvilSchema: + def test_a_good_record_has_no_problems(self): + assert validate_anvil_records([_record()]).is_valid + + def test_zero_file_size_is_valid(self): + """Three corpus records have size 0. `min_value=0`, not 1.""" + assert validate_anvil_records([_record(file_size=0)]).is_valid + + def test_negative_file_size_is_a_range_problem(self): + report = validate_anvil_records([_record(file_size=-1)]) + assert "range:file_size" in report.counts + + def test_declared_modality_and_assembly_are_nullable(self): + """AnVIL's own declarations, null for ~99% of the corpus.""" + report = validate_anvil_records( + [_record(data_modality=None, reference_assembly=None)] + ) + assert report.is_valid + + def test_a_null_required_field_is_a_null_problem(self): + report = validate_anvil_records([_record(file_name=None)]) + assert report.counts["null:file_name"] == 1 + + def test_a_renamed_key_shows_up_as_missing(self): + rec = _record() + rec["fileName"] = rec.pop("file_name") + report = validate_anvil_records([rec]) + assert report.counts["missing:file_name"] == 1 + + def test_an_added_key_is_not_a_problem(self): + """The API may add a column; that must not fail a download.""" + assert validate_anvil_records([_record(new_column="x")]).is_valid + + def test_a_stringified_size_is_a_type_problem(self): + report = validate_anvil_records([_record(file_size="1234")]) + assert report.counts["type:file_size"] == 1 + + def test_a_bool_is_not_an_int(self): + """bool subclasses int, so isinstance(True, int) is True.""" + report = validate_anvil_records([_record(file_size=True)]) + assert report.counts["type:file_size"] == 1 + + def test_an_empty_string_is_an_empty_problem(self): + report = validate_anvil_records([_record(dataset_title="")]) + assert report.counts["empty:dataset_title"] == 1 + + def test_a_malformed_md5_is_a_pattern_problem(self): + report = validate_anvil_records([_record(file_md5sum="NOTHEX")]) + assert report.counts["pattern:file_md5sum"] == 1 + + def test_uppercase_md5_is_rejected(self): + report = validate_anvil_records([_record(file_md5sum="A" * 32)]) + assert report.counts["pattern:file_md5sum"] == 1 + + def test_a_non_object_record_is_reported_not_crashed_on(self): + report = validate_anvil_records([_record(), "not a dict", 42]) + assert report.counts["not_an_object"] == 2 + + def test_samples_are_bounded(self): + report = validate_anvil_records([_record(file_name=None) for _ in range(50)]) + assert report.counts["null:file_name"] == 50 + assert len(report.samples["null:file_name"]) == 3 + + def test_summary_names_every_problem_kind(self): + report = validate_anvil_records([_record(file_name=None, file_size=-1)]) + summary = report.summary() + assert "null:file_name" in summary + assert "range:file_size" in summary + + def test_schema_covers_the_twelve_fields_the_download_produces(self): + assert len(ANVIL_RECORD_SCHEMA) == 12 + + +class TestPipelineContract: + """The pipeline reads four string fields and a size; a derived input may omit + any of them, but must never carry a present-and-null value.""" + + def test_a_minimal_record_is_valid(self): + assert validate_pipeline_records([{"file_md5sum": "x"}]).is_valid + + def test_absent_fields_are_fine(self): + assert validate_pipeline_records([{}]).is_valid + + def test_a_null_file_name_is_rejected(self): + """`rec.get("file_name", "")` returns None for a null, and the first + `.lower()` or slice downstream then raises.""" + report = validate_pipeline_records([{"file_md5sum": "x", "file_name": None}]) + assert report.counts["null:file_name"] == 1 + + @pytest.mark.parametrize( + "fld", ["file_md5sum", "file_name", "file_format", "dataset_title"] + ) + def test_every_pipeline_string_field_rejects_null(self, fld): + report = validate_pipeline_records([{fld: None}]) + assert report.counts[f"null:{fld}"] == 1 + + def test_a_null_file_size_is_allowed(self): + """Some derived inputs omit the size rather than the key.""" + assert validate_pipeline_records([{"file_size": None}]).is_valid + + def test_a_stringified_size_is_rejected(self): + report = validate_pipeline_records([{"file_size": "100"}]) + assert report.counts["type:file_size"] == 1 + + def test_a_bool_size_is_rejected(self): + report = validate_pipeline_records([{"file_size": True}]) + assert report.counts["type:file_size"] == 1 + + def test_zero_size_is_allowed_negative_is_not(self): + assert validate_pipeline_records([{"file_size": 0}]).is_valid + assert not validate_pipeline_records([{"file_size": -1}]).is_valid + + def test_the_pipeline_contract_does_not_require_the_anvil_fields(self): + """drs_uri, entry_id, etc. are absent from a derived input and that is fine.""" + assert validate_pipeline_records([{"file_md5sum": "x", "file_name": "a.bam"}]).is_valid diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 2aae740..b7abf05 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -157,24 +157,46 @@ def test_fetcher_returns_none(self, input_file, tmp_path): assert data["metadata"]["failed"] == 2 @pytest.mark.parametrize("workers", [1, 2]) - def test_null_file_name_does_not_crash_or_lose_the_record(self, tmp_path, workers): - """`record.get("file_name", "")` returns None for a present-but-null key, and - _filter_records admits such a record when its file_format matches. + def test_null_file_name_is_rejected_at_load(self, tmp_path, workers): + """The outer gate: `_load_input` validates the fields the pipeline reads, so a + present-but-null `file_name` never reaches the run loop (#161).""" + config = _make_config() + input_path = tmp_path / "in.json" + input_path.write_text(json.dumps({"results": [ + {"file_md5sum": "n1", "file_name": None, "file_format": ".test"}, + ]})) + with pytest.raises(ValueError, match="null:file_name"): + ClassifyPipeline( + config, input_path, tmp_path / "out.json", + evidence_base=tmp_path / "evidence", workers=workers, + ).run() + + @pytest.mark.parametrize("workers", [1, 2]) + def test_null_file_name_in_the_run_loop_neither_crashes_nor_loses_the_record( + self, tmp_path, workers + ): + """Defense in depth, below the load-time gate. + `record.get("file_name", "")` returns None for a present-but-null key. Sequentially that aborted the run. In the parallel path it was worse: the raise landed in the executor's `except Exception`, so a record that had classified successfully was counted as errored and never written. + + Drives `_run_parallel` directly, since `_load_input` now rejects such a + record before it can reach here. """ config = _make_config() input_path = tmp_path / "in.json" - input_path.write_text(json.dumps({"results": [ - {"file_md5sum": "n1", "file_name": None, "file_format": ".test"}, - ]})) + input_path.write_text(json.dumps({"results": []})) output = tmp_path / "out.json" - results = ClassifyPipeline( + pipeline = ClassifyPipeline( config, input_path, output, evidence_base=tmp_path / "evidence", workers=workers, - ).run() + ) + pipeline.evidence_dir.mkdir(parents=True, exist_ok=True) + results = pipeline._run_parallel( + [{"file_md5sum": "n1", "file_name": None, "file_format": ".test"}] + ) assert len(results) == 1, "the record must be classified, not dropped" meta = json.loads(output.read_text())["metadata"]