data_validation: shared dataset-validator core; findings annotated on PR diffs

.github/workflows/datasets-validate-all.yml

@@ -39,7 +39,7 @@ jobs:
         run: python -m pip install --quiet -r data_validation/requirements.txt
 
       - name: Run every dataset validator and the shared checks
-        run: python data_validation/run_all_checks.py --verbose
+        run: python data_validation/run_all_checks.py --verbose --github
 
       - name: Cross-dataset consistency
         run: python data_validation/qc_tools/consistency_check.py

data_validation/README.md

@@ -22,7 +22,7 @@ The tooling has two jobs: keep the research data behind OWASP deliverables (the
 
 `run_all_checks.py` runs two layers.
 
-**1. Dataset validators.** Each dataset that has a `validate.py` (at its root, or in a sub-collection folder) is run in its own directory. These are the authoritative checks for their data: schema conformance against the dataset's own `schema.json`, plus rules a schema cannot express. Datasets without one are listed as `NO VALIDATOR` so the gap is visible.
+**1. Dataset validators.** Each dataset that has a `validate.py` (at its root, or in a sub-collection folder) is run in its own directory. Most are thin wrappers around a shared core, [`datasets/_shared/validate_lib.py`](../datasets/_shared/validate_lib.py) (schema with formats, DSGAI IDs, ID matches file name, duplicate IDs), plus the dataset's own extra rules. These are the authoritative checks for their data: schema conformance against the dataset's own `schema.json`, plus rules a schema cannot express. Datasets without one are listed as `NO VALIDATOR` so the gap is visible.
 
 | Dataset | Own validator |
 |---|---|
@@ -92,6 +92,7 @@ python data_validation/run_all_checks.py                       # everything
 python data_validation/run_all_checks.py --dataset datasets/exploit_dataset
 python data_validation/run_all_checks.py --verbose             # every validator's output and every warning
 python data_validation/run_all_checks.py --strict              # warnings fail too
+python data_validation/run_all_checks.py --github              # also emit annotations (CI uses this)
 
 # One check, on chosen files or datasets (each script's docstring has details)
 python data_validation/validators/schema_validator.py --file datasets/rag_dataset/entries/RAG-0001.json
@@ -107,7 +108,7 @@ GITHUB_TOKEN=... python data_validation/qc_tools/online_check.py --summary onlin
 python -m pytest data_validation/tests
 ```
 
-`run_all_checks.py` exits 0 when everything passed, 1 on any failure, and 2 when there was nothing to check.
+`run_all_checks.py` exits 0 when everything passed, 1 on any failure, and 2 when there was nothing to check. In CI it runs with `--github`, so every finding also appears as an annotation on the exact line of the file in the pull request diff.
 
 ---
 

data_validation/run_all_checks.py

@@ -28,6 +28,7 @@
 from __future__ import annotations
 
 import argparse
+import re
 import subprocess
 import sys
 from pathlib import Path
@@ -41,6 +42,7 @@
 from validators._common import (  # noqa: E402
     ERROR, SKIP_DIRS, Finding, ensure_utf8_stdout, iter_data_files, load_json, load_taxonomy_ids,
 )
+from validators.json_lines import line_for  # noqa: E402
 
 DATASETS_ROOT = Path(__file__).resolve().parent.parent / "datasets"
 DEFAULT_TIMEOUT = 300
@@ -88,6 +90,44 @@ def run_validator(dataset: Path, timeout: float = DEFAULT_TIMEOUT) -> tuple[str,
     return "FAIL", output
 
 
+VALIDATOR_LINE_RE = re.compile(r"^(?P<file>[^:\s][^:]*\.json): (?P<loc>[^:]+): (?P<msg>.+)$")
+
+
+def _escape(text: str, prop: bool = False) -> str:
+    text = text.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A")
+    return text.replace(":", "%3A").replace(",", "%2C") if prop else text
+
+
+def annotation(level: str, path: Path, pointer: str, title: str, message: str) -> str:
+    """A GitHub Actions workflow command that shows the finding on the PR diff."""
+    command = "error" if level == ERROR else "warning"
+    try:
+        shown = path.resolve().relative_to(DATASETS_ROOT.resolve().parent).as_posix()
+    except ValueError:
+        shown = path.as_posix()
+    props = f"file={_escape(shown, True)}"
+    line = line_for(path, pointer)
+    if line:
+        props += f",line={line}"
+    return f"::{command} {props},title={_escape(title, True)}::{_escape(message)}"
+
+
+def validator_annotations(script: Path, output: str) -> list[str]:
+    """Annotations for a dataset validator's "file.json: location: message" lines."""
+    out = []
+    for raw in output.splitlines():
+        m = VALIDATOR_LINE_RE.match(raw.strip())
+        if not m:
+            continue
+        name = m.group("file")
+        path = next((p for p in (script.parent / "entries" / name, script.parent / name) if p.is_file()),
+                    script.parent / name)
+        loc = m.group("loc").strip()
+        pointer = "" if loc == "<root>" else "/" + loc.lstrip("/")
+        out.append(annotation(ERROR, path, pointer, f"{script.parent.name}/validate.py", m.group("msg")))
+    return out
+
+
 def shared_checks(dataset: Path, datasets_root: Path) -> tuple[int, list[Finding]]:
     """(files checked, findings) for the shared checks on one dataset."""
     files = iter_data_files(dataset)
@@ -129,6 +169,8 @@ def main(argv: list[str] | None = None) -> int:
                         help="Treat shared-check warnings as failures")
     parser.add_argument("--no-shared", action="store_true",
                         help="Run only the dataset validators")
+    parser.add_argument("--github", action="store_true",
+                        help="Also print GitHub Actions annotations, so findings show on the PR diff")
     parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
                         help=f"Seconds each dataset validator may run (default {DEFAULT_TIMEOUT})")
     args = parser.parse_args(argv)
@@ -149,6 +191,7 @@ def main(argv: list[str] | None = None) -> int:
         return 2
 
     results = []
+    annotations: list[str] = []
     for dataset in datasets:
         validators = find_validators(dataset)
         if not validators:
@@ -165,6 +208,8 @@ def main(argv: list[str] | None = None) -> int:
             if output and (args.verbose or status in ("FAIL", "ERROR")):
                 for line in output.splitlines():
                     print(f"    {line}")
+            if status == "FAIL":
+                annotations += validator_annotations(script, output)
 
     checked_files = errors = warnings = 0
     if not args.no_shared:
@@ -181,6 +226,11 @@ def main(argv: list[str] | None = None) -> int:
             for f in findings:
                 if f.level == ERROR or args.verbose or args.strict:
                     print(f"    {f.format(datasets_root.parent)}")
+                annotations.append(annotation(f.level, f.path, f.location, f.check, f.message))
+
+    if args.github:
+        for line in annotations:
+            print(line)
 
     ran = [s for s in results if s != "NO VALIDATOR"]
     failed = [s for s in ran if s != "PASS"]

data_validation/tests/test_json_lines.py

@@ -0,0 +1,45 @@
+"""Tests for json_lines.py and the runner's GitHub annotations."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import run_all_checks  # noqa: E402
+from validators.json_lines import pointer_line  # noqa: E402
+
+TEXT = """{
+  "id": "X-1",
+  "tricky": "a \\" quote, a } and a ] inside",
+  "list": [
+    "first",
+    {"inner": ["q",
+               "r"]}
+  ],
+  "a/b": 1
+}"""
+
+
+def test_pointer_lines():
+    assert pointer_line(TEXT, "/id") == 2
+    assert pointer_line(TEXT, "/list/0") == 5
+    assert pointer_line(TEXT, "/list/1/inner/1") == 7
+    assert pointer_line(TEXT, "/a~1b") == 9
+    assert pointer_line(TEXT, "<root>") == 1
+    assert pointer_line(TEXT, "/missing") is None
+
+
+def test_annotation_format(tmp_path):
+    f = tmp_path / "e.json"
+    f.write_text(TEXT)
+    line = run_all_checks.annotation("ERROR", f, "/list/1/inner/1", "dsgai", "bad: value, 100%")
+    assert line.startswith("::error file=") and ",line=7,title=dsgai::bad: value, 100%25" in line
+
+
+def test_validator_output_is_annotated(tmp_path):
+    (tmp_path / "entries").mkdir()
+    (tmp_path / "entries" / "e.json").write_text(TEXT)
+    out = "e.json: list/1/inner/1: 'r' is not valid\nunrelated line\n\nFAIL: 1 issue(s) across 1 entries."
+    [line] = run_all_checks.validator_annotations(tmp_path / "validate.py", out)
+    assert "entries/e.json,line=7" in line.replace("\\", "/") and line.endswith("::'r' is not valid")

data_validation/tests/test_standards.py

@@ -66,9 +66,16 @@ def test_owasp_table_is_complete_and_crosswalked():
 
 @pytest.mark.parametrize("validate", sorted(DATASETS_ROOT.glob("*/validate.py")), ids=lambda p: p.parent.name)
 def test_dataset_validators_enforce_format(validate):
-    """Without a format checker, "format": "date" and "uri" are never checked."""
+    """Without a format checker, "format": "date" and "uri" are never checked.
+
+    A validator either builds its own (agentdataflow) or uses the shared core,
+    which does."""
     source = validate.read_text(encoding="utf-8")
-    assert re.search(r"format_checker\s*=", source), f"{validate} builds its validator without a format checker"
+    uses_shared = "from validate_lib import" in source
+    assert uses_shared or re.search(r"format_checker\s*=", source), f"{validate} has no format checker"
+    if uses_shared:
+        lib = (DATASETS_ROOT / "_shared" / "validate_lib.py").read_text(encoding="utf-8")
+        assert re.search(r"format_checker\s*=", lib)
 
 
 def test_nist_table_matches_the_publication():

data_validation/tests/test_validate_lib.py

@@ -0,0 +1,90 @@
+"""Tests for datasets/_shared/validate_lib.py, the core of the dataset validators."""
+from __future__ import annotations
+
+import json
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "datasets" / "_shared"))
+
+import validate_lib  # noqa: E402
+
+SCHEMA = {
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "type": "object",
+    "required": ["item_id", "date_added"],
+    "properties": {"item_id": {"type": "string"}, "date_added": {"type": "string", "format": "date"},
+                   "dsgai_mapping": {"type": "array"}},
+}
+
+
+def make(tmp_path: Path, entries: dict[str, dict]) -> Path:
+    (tmp_path / "schema.json").write_text(json.dumps(SCHEMA))
+    (tmp_path / "entries").mkdir()
+    for name, entry in entries.items():
+        (tmp_path / "entries" / f"{name}.json").write_text(json.dumps(entry))
+    return tmp_path
+
+
+def run(tmp_path, capsys, **kwargs) -> tuple[int, str]:
+    code = validate_lib.run(tmp_path, id_field="item_id", **kwargs)
+    return code, capsys.readouterr().out
+
+
+def test_valid_entries_pass(tmp_path, capsys):
+    make(tmp_path, {"A-1": {"item_id": "A-1", "date_added": "2026-01-01", "dsgai_mapping": ["DSGAI01"]}})
+    assert run(tmp_path, capsys) == (0, "OK: 1 entries validated against schema.\n")
+
+
+@pytest.mark.parametrize("entry,expected", [
+    ({"item_id": "A-1"}, "'date_added' is a required property"),
+    ({"item_id": "A-1", "date_added": "2026-13-01"}, "is not a 'date'"),
+    ({"item_id": "A-1", "date_added": "2026-01-01", "dsgai_mapping": ["DSGAI99"]}, "dsgai_mapping 'DSGAI99' not in taxonomy"),
+    ({"item_id": "A-1", "date_added": "2026-01-01", "risks": [{"dsgai_id": "DSGAI00"}]}, "risks/0/dsgai_id: dsgai_id 'DSGAI00'"),
+    ({"item_id": "B-2", "date_added": "2026-01-01"}, "does not match the file name"),
+])
+def test_each_rule_fails(tmp_path, capsys, entry, expected):
+    make(tmp_path, {"A-1": entry})
+    code, out = run(tmp_path, capsys)
+    assert code == 1 and expected in out
+
+
+def test_duplicate_ids(tmp_path, capsys):
+    make(tmp_path, {"A-1": {"item_id": "A-1", "date_added": "2026-01-01"},
+                    "A-2": {"item_id": "A-1", "date_added": "2026-01-01"}})
+    code, out = run(tmp_path, capsys)
+    assert code == 1 and "is also used by A-1.json" in out
+
+
+def test_extra_rule(tmp_path, capsys):
+    make(tmp_path, {"A-1": {"item_id": "A-1", "date_added": "2026-01-01"}})
+    code, out = run(tmp_path, capsys, extra=lambda entry, path: ["custom problem"])
+    assert code == 1 and "A-1.json: custom problem" in out
+
+
+def test_empty_entries(tmp_path, capsys):
+    (tmp_path / "schema.json").write_text(json.dumps(SCHEMA))
+    assert run(tmp_path, capsys, allow_empty=True)[0] == 0
+    assert run(tmp_path, capsys)[0] == 1
+
+
+@pytest.mark.parametrize("dataset,mutate,expected", [
+    ("exploit_dataset", lambda e: e.update(mitre_atlas_mapping=[]), "mitre_atlas_mapping does not include it"),
+    ("vulnerability_dataset", lambda e: e.update(cve_id="CVE-2000-0001"), "does not match cve_id"),
+])
+def test_dataset_specific_rules_still_fire(tmp_path, dataset, mutate, expected):
+    """Run the real validate.py on a copy with one entry broken."""
+    import subprocess
+    copy = tmp_path / "datasets"
+    shutil.copytree(ROOT / "datasets" / "_shared", copy / "_shared")
+    shutil.copytree(ROOT / "datasets" / dataset, copy / dataset)
+    entry_path = sorted((copy / dataset / "entries").glob("CVE-*.json" if dataset == "vulnerability_dataset" else "AML.*.json"))[0]
+    entry = json.loads(entry_path.read_text(encoding="utf-8"))
+    mutate(entry)
+    entry_path.write_text(json.dumps(entry), encoding="utf-8")
+    proc = subprocess.run([sys.executable, "validate.py"], cwd=copy / dataset, capture_output=True, text=True)
+    assert proc.returncode == 1 and expected in proc.stdout