b0b51f1e71
- fail closed on MCP input and compile database policy - add shared scan catalogs, finding queries, and tiered gates
107 lines
4.4 KiB
Python
107 lines
4.4 KiB
Python
"""Reviewed changed-path to regression-test selection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def string_list(value: object, field: str) -> list[str]:
|
|
if not isinstance(value, list) or not all(
|
|
isinstance(item, str) and item for item in value
|
|
):
|
|
raise ValueError(f"{field} must be a non-empty string array")
|
|
return list(value)
|
|
|
|
|
|
def load_manifest(path: Path) -> dict[str, object]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
raise ValueError(f"failed to read impact manifest {path}: {error}") from error
|
|
if not isinstance(value, dict) or value.get("schema_version") != 1:
|
|
raise ValueError("impact manifest schema_version must be 1")
|
|
always = value.get("always")
|
|
rules = value.get("rules")
|
|
if not isinstance(always, list) or not isinstance(rules, list):
|
|
raise ValueError("impact manifest always and rules must be arrays")
|
|
seen_rule_ids: set[str] = set()
|
|
for index, item in enumerate(always):
|
|
if not isinstance(item, dict):
|
|
raise ValueError(f"always[{index}] must be an object")
|
|
if not isinstance(item.get("test"), str) or not item["test"]:
|
|
raise ValueError(f"always[{index}].test must be a non-empty string")
|
|
if not isinstance(item.get("rationale"), str) or not item["rationale"]:
|
|
raise ValueError(f"always[{index}].rationale must be a non-empty string")
|
|
for index, item in enumerate(rules):
|
|
if not isinstance(item, dict):
|
|
raise ValueError(f"rules[{index}] must be an object")
|
|
rule_id = item.get("id")
|
|
rationale = item.get("rationale")
|
|
if not isinstance(rule_id, str) or not rule_id:
|
|
raise ValueError(f"rules[{index}].id must be a non-empty string")
|
|
if rule_id in seen_rule_ids:
|
|
raise ValueError(f"duplicate impact rule id: {rule_id}")
|
|
seen_rule_ids.add(rule_id)
|
|
if not isinstance(rationale, str) or not rationale:
|
|
raise ValueError(f"rules[{index}].rationale must be a non-empty string")
|
|
string_list(item.get("globs"), f"rules[{index}].globs")
|
|
string_list(item.get("tests"), f"rules[{index}].tests")
|
|
string_list(value.get("fallback_globs"), "fallback_globs")
|
|
return value
|
|
|
|
|
|
def matches(path: str, patterns: list[str]) -> bool:
|
|
return any(fnmatch.fnmatchcase(path, pattern) for pattern in patterns)
|
|
|
|
|
|
def select_tests(changed_paths: list[str], manifest_path: Path) -> dict[str, object]:
|
|
manifest = load_manifest(manifest_path)
|
|
always = manifest.get("always")
|
|
rules = manifest.get("rules")
|
|
if not isinstance(always, list) or not isinstance(rules, list):
|
|
raise ValueError("impact manifest always and rules must be arrays")
|
|
reasons: dict[str, set[str]] = {}
|
|
for item in always:
|
|
if isinstance(item, dict):
|
|
reasons.setdefault(str(item["test"]), set()).add(str(item["rationale"]))
|
|
|
|
matched_rules: list[dict[str, object]] = []
|
|
matched_paths: set[str] = set()
|
|
for rule in rules:
|
|
if not isinstance(rule, dict):
|
|
continue
|
|
globs = string_list(rule["globs"], f"rule {rule['id']} globs")
|
|
paths = sorted(path for path in set(changed_paths) if matches(path, globs))
|
|
if not paths:
|
|
continue
|
|
matched_paths.update(paths)
|
|
reason = f"{rule['id']}: {rule['rationale']}"
|
|
for test in string_list(rule["tests"], f"rule {rule['id']} tests"):
|
|
reasons.setdefault(test, set()).add(reason)
|
|
matched_rules.append({"id": rule["id"], "paths": paths})
|
|
|
|
fallback_globs = string_list(manifest["fallback_globs"], "fallback_globs")
|
|
unmapped_paths = sorted(
|
|
path
|
|
for path in set(changed_paths)
|
|
if path not in matched_paths and matches(path, fallback_globs)
|
|
)
|
|
return {
|
|
"schema_version": 1,
|
|
"manifest": manifest_path.as_posix(),
|
|
"changed_paths": sorted(set(changed_paths)),
|
|
"full_suite": bool(unmapped_paths),
|
|
"fallback_reasons": (
|
|
["unmapped production/build/test paths require the full suite"]
|
|
if unmapped_paths
|
|
else []
|
|
),
|
|
"unmapped_paths": unmapped_paths,
|
|
"matched_rules": sorted(matched_rules, key=lambda item: str(item["id"])),
|
|
"selected_tests": [
|
|
{"name": name, "reasons": sorted(values)}
|
|
for name, values in sorted(reasons.items())
|
|
],
|
|
}
|