test(astra): validate decision anchors and universe pins

Author: cailmdaley

tests/helpers/astra_record.py

@@ -0,0 +1,284 @@
+"""Reusable parsing and resolution helpers for ShapePipe's ASTRA record."""
+
+import ast
+import configparser
+from dataclasses import dataclass
+from pathlib import Path
+import re
+
+import yaml
+
+
+@dataclass(frozen=True)
+class Anchor:
+    """An anchor sentence found in a YAML value."""
+
+    location: str
+    references: tuple[str, ...]
+    error: str | None = None
+
+
+def load_yaml(path):
+    """Load YAML with PyYAML's safe loader."""
+
+    return yaml.safe_load(Path(path).read_text(encoding="utf-8"))
+
+
+def _walk(value, location=""):
+    if isinstance(value, dict):
+        for key, child in value.items():
+            path = f"{location}.{key}" if location else str(key)
+            yield from _walk(child, path)
+    elif isinstance(value, list):
+        for index, child in enumerate(value):
+            yield from _walk(child, f"{location}[{index}]")
+    else:
+        yield location, value
+
+
+def _rationales(document):
+    for location, value in _walk(document):
+        if location.endswith(".rationale"):
+            yield location, value
+
+
+def extract_anchors(document):
+    """Parse all ``Anchor:`` sentences and check every rationale has one."""
+
+    anchors = []
+    for location, value in _walk(document):
+        if not isinstance(value, str) or "Anchor:" not in value:
+            continue
+        tail = value.split("Anchor:", 1)[1].strip()
+        error = None
+        refs = ()
+        if value.count("Anchor:") != 1:
+            count = value.count("Anchor:")
+            error = f"expected one Anchor: marker, found {count}"
+        elif not tail.endswith("."):
+            error = "anchor sentence must end with a period"
+        else:
+            refs = tuple(part.strip() for part in tail[:-1].split(";"))
+            if not refs or any(not ref for ref in refs):
+                error = "anchor sentence contains an empty ref"
+        anchors.append(Anchor(location, refs, error))
+
+    for location, value in _rationales(document):
+        if (
+            not isinstance(value, str)
+            or value.count("Anchor:") != 1
+            or not value.rstrip().endswith(".")
+        ):
+            anchors.append(
+                Anchor(
+                    location,
+                    (),
+                    "rationale must end with exactly one Anchor: sentence",
+                )
+            )
+    return anchors
+
+
+def _parse_reference(reference):
+    if "::" in reference:
+        path, symbol = reference.split("::", 1)
+        return "code", path, symbol
+    if "#" in reference:
+        path, key = reference.split("#", 1)
+        return "config", path, key
+    return "path", reference, ""
+
+
+def resolve_anchor(root, reference):
+    """Return ``None`` if a reference resolves, otherwise a diagnostic."""
+
+    kind, relative, selector = _parse_reference(reference)
+    path = Path(relative)
+    if path.is_absolute() or ".." in path.parts:
+        return "path must be relative to the repository root"
+    target = Path(root) / path
+    if not target.exists():
+        return "path does not exist"
+    if kind == "path":
+        return None
+    if not target.is_file():
+        return "code/config refs must name a file"
+
+    try:
+        text = target.read_text(encoding="utf-8")
+    except (OSError, UnicodeError) as error:
+        return f"cannot read file: {error}"
+
+    if kind == "code":
+        if target.suffix != ".py":
+            return "code-symbol refs must name a .py file"
+        try:
+            tree = ast.parse(text, filename=str(target))
+        except SyntaxError as error:
+            return f"cannot parse Python file: {error}"
+        if not _has_symbol(tree, selector):
+            return f"no def/class/assignment target named {selector!r}"
+        return None
+
+    suffix = target.suffix.lower()
+    if suffix == ".ini":
+        return _ini_key(text, selector)
+    if suffix == ".setools":
+        return _setools_key(text, selector)
+    if suffix in {".sex", ".psfex", ".ww", ".param", ".conf"}:
+        key = selector.rsplit(".", 1)[-1]
+        pattern = re.compile(rf"^\s*(?:#\s*)?{re.escape(key)}(?=$|\s|=|\()")
+        if any(pattern.search(line) for line in text.splitlines()):
+            return None
+        return f"no line starts with key {key!r} (commented keys are allowed)"
+    return f"unsupported config-key file type {suffix or '(no extension)'}"
+
+
+def _ini_key(text, selector):
+    if "." not in selector:
+        return "INI config ref needs SECTION.KEY"
+    section, key = selector.rsplit(".", 1)
+    parser = configparser.ConfigParser(
+        interpolation=None, strict=False, allow_no_value=True
+    )
+    try:
+        parser.read_string(text)
+    except configparser.Error as error:
+        return f"cannot parse INI file: {error}"
+    if not parser.has_section(section):
+        return f"INI section {section!r} is missing"
+    if not parser.has_option(section, key):
+        return f"INI key {key!r} is missing from section {section!r}"
+    return None
+
+
+def _setools_key(text, selector):
+    if "." not in selector:
+        return "SETools config ref needs SECTION.KEY"
+    section, key = selector.rsplit(".", 1)
+    pattern = re.compile(rf"^\s*(?:#\s*)?{re.escape(key)}(?=$|\s|=|<|>)")
+    active = False
+    for line in text.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("[") and stripped.endswith("]"):
+            active = stripped[1:-1].strip() == section
+        elif active and pattern.search(line):
+            return None
+    return f"SETools key {key!r} is missing from section {section!r}"
+
+
+def _target_names(target):
+    if isinstance(target, ast.Name):
+        return [target.id]
+    if isinstance(target, ast.Attribute):
+        return [target.attr]
+    if isinstance(target, (ast.Tuple, ast.List)):
+        return [name for item in target.elts for name in _target_names(item)]
+    if isinstance(target, ast.Starred):
+        return _target_names(target.value)
+    return []
+
+
+def _bindings(scope):
+    """Collect declarations and assignment targets in one lexical scope."""
+
+    result = {}
+
+    def visit(node):
+        if isinstance(
+            node,
+            (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef),
+        ):
+            result[node.name] = node
+            return
+        if isinstance(node, ast.Lambda):
+            return
+        if isinstance(node, ast.Assign):
+            targets = node.targets
+        elif isinstance(node, (ast.AnnAssign, ast.AugAssign, ast.NamedExpr)):
+            targets = [node.target]
+        elif isinstance(node, (ast.For, ast.AsyncFor)):
+            targets = [node.target]
+        elif isinstance(node, (ast.With, ast.AsyncWith)):
+            targets = [item.optional_vars for item in node.items]
+        else:
+            targets = []
+        for target in targets:
+            if target is not None:
+                result.update(dict.fromkeys(_target_names(target), node))
+        if isinstance(node, ast.ExceptHandler) and node.name:
+            result[node.name] = node
+        for child in ast.iter_child_nodes(node):
+            visit(child)
+
+    for statement in scope.body:
+        visit(statement)
+    return result
+
+
+def _has_symbol(tree, symbol):
+    scope = tree
+    parts = symbol.split(".")
+    for index, part in enumerate(parts):
+        declaration = _bindings(scope).get(part)
+        if declaration is None:
+            return False
+        if index == len(parts) - 1:
+            return True
+        if not isinstance(
+            declaration,
+            (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef),
+        ):
+            return False
+        scope = declaration
+    return False
+
+
+def universe_errors(record, universe):
+    """Check scoped decision IDs and options against the ASTRA record."""
+
+    record_decisions = _decisions(record)
+    pinned = _decisions(universe)
+    errors = []
+    for location in sorted(pinned.keys() - record_decisions.keys()):
+        errors.append(
+            f"{location}: universe decision is absent from astra.yaml"
+        )
+    for location in sorted(record_decisions.keys() - pinned.keys()):
+        errors.append(
+            f"{location}: astra.yaml decision is not pinned in the universe"
+        )
+    for location in sorted(record_decisions.keys() & pinned.keys()):
+        definition = record_decisions[location]
+        options = (
+            definition.get("options", {})
+            if isinstance(definition, dict)
+            else {}
+        )
+        if not isinstance(options, dict) or pinned[location] not in options:
+            errors.append(
+                f"{location}: pinned option {pinned[location]!r} is not in "
+                "ASTRA options"
+            )
+    return errors
+
+
+def _decisions(document, location=""):
+    scope = document if isinstance(document, dict) else {}
+    result = {}
+    for decision_id, definition in (scope.get("decisions") or {}).items():
+        key = (
+            f"{location}.decisions.{decision_id}"
+            if location
+            else f"decisions.{decision_id}"
+        )
+        result[key] = definition
+    for analysis_id, analysis in (scope.get("analyses") or {}).items():
+        child = (
+            f"{location}.analyses.{analysis_id}"
+            if location
+            else f"analyses.{analysis_id}"
+        )
+        if isinstance(analysis, dict):
+            result.update(_decisions(analysis, child))
+    return result

tests/unit/test_astra_anchors.py

@@ -0,0 +1,50 @@
+"""Keep ASTRA decision anchors and the committed universe resolvable."""
+
+from pathlib import Path
+
+from tests.helpers.astra_record import (
+    extract_anchors,
+    load_yaml,
+    resolve_anchor,
+    universe_errors,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_every_astra_anchor_resolves():
+    record = load_yaml(REPO_ROOT / "astra.yaml")
+    anchors = extract_anchors(record)
+    errors = []
+
+    assert anchors, "astra.yaml contains no Anchor: sentences"
+    for anchor in anchors:
+        if anchor.error:
+            errors.append(f"{anchor.location}: {anchor.error}")
+            continue
+        for reference in anchor.references:
+            problem = resolve_anchor(REPO_ROOT, reference)
+            if problem:
+                errors.append(
+                    f"{anchor.location}: {reference}: {problem}"
+                )
+
+    message = (
+        "Unresolved ASTRA anchors or rationales:\n - "
+        + "\n - ".join(errors)
+    )
+    assert not errors, message
+
+
+def test_committed_universe_matches_astra_decisions():
+    record = load_yaml(REPO_ROOT / "astra.yaml")
+    universe = load_yaml(REPO_ROOT / "universes" / "committed.yaml")
+
+    errors = universe_errors(record, universe)
+
+    message = (
+        "ASTRA / committed universe mismatch:\n - "
+        + "\n - ".join(errors)
+    )
+    assert not errors, message