scripts.derive_test_categories

Reference › scripts › derive_test_categories

Derives a category for each test module from its syntax tree: the RUL-039 / RUL-036#13 pass. It writes the result as a committed artifact. tests/scripts/test_derive_test_categories.py imports it.

Invocation: main with no options derives the categories and writes the artifact. --stdout prints a human table instead of writing. --check recomputes and compares with the committed artifact, and returns 2 if the artifact is missing or unreadable, if its derivation version differs from the script's, or if the stored fingerprint of the tests tree no longer matches, with a message telling the user to re-derive.

derive goes through the test modules (_test_modules) and calls _derive_markers on each. That function parses the module and looks for evidence of its category: explicit pytest marker names (_pytest_marker_names), subprocess calls and the command words in their arguments (_is_subprocess_call, _argv_strings, _tokenize), and other string constants. _body_hash and _fingerprint hash the modules and the whole set, and the fingerprint is what --check compares, so any change to the tests tree makes the artifact stale.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/derive_test_categories.py

#!/usr/bin/env python3
"""Derive test-module categories from AST — the RUL-039 / RUL-036#13 pass.

Computes, per test module under ``tests/``, a category row. This pass derives ONE
class: ``rehearsal``. The category is COMPUTED from what a test actually does —
never hand-set (RUL-036's own words) — so a label always carries its file:line
evidence; a label without evidence is invalid.

The rehearsal class (the founding incident — dead run 906142, ``docs/lessons/
L23-exoneration.md``) is the set of tests exercising real deploy surfaces:
- subprocess/CLI argv string literals whose tokens include ``incus``, ``lxc``,
  ``docker``, ``terraform``, ``tofu`` (the RUL-038/ADR-0111 marker set) in
  EXECUTION position — not prose, not docstrings, not asserted-against data;
- env-gated live lanes: guards on ``INCUS_HARNESS_IT``-style live-run env keys,
  and ``@pytest.mark.<...>_live`` / ``@pytest.mark.live`` markers;
- real deploy-surface invocations: subprocess calls into ``deploy/``, ``ship``,
  ``promote`` or the installer scripts (the mechanism, not a directory name).

Deliberately NOT markers: importing a deploy module to unit-test its pure
functions; ``tmp_path`` fixture deploy configs; the word "deploy" in test names
or docstrings.

Schema (designed to EXTEND, not be replaced):

.. code-block:: json

    {"derivation_version": 1,
     "generated_at": "...",
     "tests_tree_fingerprint": "<sha of sorted (module, body_hash)>",
     "rows": [{"module": "tests/harness/test_incus_isolation.py",
               "category": "rehearsal",
               "markers": [{"kind": "argv-literal", "value": "incus", "at": "f:l"}, ...]}]}

``--check`` recomputes and compares against the committed artifact, exiting
non-zero on drift (the staleness gate consumes this). ``--stdout`` prints a
human table. Default: write ``reports/rehearsal-class-modules.json``.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import sys
from datetime import UTC, datetime
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
ARTIFACT = REPO_ROOT / "reports" / "rehearsal-class-modules.json"
DERIVATION_VERSION = 1

#: subprocess/CLI argv tokens whose presence in EXECUTION position marks the
#: RUL-038/ADR-0111 rehearsal marker set (container + IaC surfaces).
_ARGV_TOKENS = ("incus", "lxc", "docker", "terraform", "tofu")

#: Live-run env keys — a guard on one of these is an env-gated live lane.
_LIVE_ENV_KEYS = ("INCUS_HARNESS_IT",)

#: Deploy-surface argv references — invoking one of the repo's deploy scripts is
#: a real deploy surface. Script names (with ``.sh``) are unambiguous, unlike a
#: bare "ship"/"deploy" word.
_DEPLOY_SURFACE = (
    "deploy.sh",
    "install-claude-config.sh",
    "install-gitlab-runner-executor.sh",
    "install-orchestrata.sh",
    "install-tools.sh",
    "provision_environments.sh",
    "restart-tool-instances.sh",
    "retire-plain-units.sh",
    "ship-merge-master.sh",
    "ship.sh",
    "toolchain.sh",
)

#: subprocess/CLI entry points whose argv we inspect.
_SUBPROCESS_ATTRS = {
    "subprocess.run",
    "subprocess.Popen",
    "subprocess.call",
    "subprocess.check_call",
    "subprocess.check_output",
    "os.system",
    "os.popen",
}


def _lineno(node: ast.AST) -> int:
    return getattr(node, "lineno", 0)


def _is_subprocess_call(node: ast.AST) -> bool:
    """True for ``subprocess.run(...)`` / ``os.system(...)`` etc."""
    if not isinstance(node, ast.Call):
        return False
    func = node.func
    if isinstance(func, ast.Attribute):
        if isinstance(func.value, ast.Name):
            return f"{func.value.id}.{func.attr}" in _SUBPROCESS_ATTRS
    return False


def _argv_strings(node: ast.AST) -> list[str]:
    """The string literals in a subprocess call's argv (first positional arg
    as a list, or the whole string for ``os.system``). Execution position only."""
    if not isinstance(node, ast.Call):
        return []
    args = node.args
    if not args:
        return []
    first = args[0]
    out: list[str] = []
    if isinstance(first, ast.List):
        for elt in first.elts:
            if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
                out.append(elt.value)
    elif isinstance(first, ast.Constant) and isinstance(first.value, str):
        out.append(first.value)
    return out


def _string_constants(node: ast.AST) -> list[str]:
    """Every string constant in a subtree (for env-key guard detection)."""
    out: list[str] = []
    for child in ast.walk(node):
        if isinstance(child, ast.Constant) and isinstance(child.value, str):
            out.append(child.value)
    return out


def _tokenize(s: str) -> set[str]:
    """Execution-position tokens in an argv string: the executable basename and
    any bare alphanumeric word. A flag (``--incus-json``) or a filename
    (``incus.json``) contributes no bare token — only an executable named
    ``incus`` or a standalone ``incus`` word does."""
    tokens: set[str] = set()
    words = s.split()
    if not words:
        return tokens
    tokens.add(words[0].rsplit("/", 1)[-1])
    for w in words:
        if w.isalnum():
            tokens.add(w)
    return tokens


def _pytest_marker_names(tree: ast.AST) -> list[str]:
    """Names of ``@pytest.mark.<name>`` decorators on any node."""
    names: list[str] = []
    for node in ast.walk(tree):
        for dec in getattr(node, "decorator_list", []):
            if (
                isinstance(dec, ast.Attribute)
                and isinstance(dec.value, ast.Attribute)
                and isinstance(dec.value.value, ast.Name)
                and dec.value.value.id == "pytest"
                and dec.value.attr == "mark"
            ):
                names.append(dec.attr)
    return names


def _derive_markers(module: Path, body: str) -> list[dict]:
    """The rehearsal markers present in one test module, with file:line evidence.

    Empty list == not rehearsal (the only category this pass derives)."""
    try:
        tree = ast.parse(body, filename=str(module))
    except SyntaxError:
        # An unparsable test module cannot be categorized — but the suite
        # itself refuses to collect it, so it never reaches the gate. Skip.
        return []

    markers: list[dict] = []

    # 1. argv-literal tokens (incus/lxc/docker/terraform/tofu) in subprocess argv.
    # Match whole-word TOKENS, not substrings of flags/filenames: "incus" in
    # "--incus-json" or "incus.json" is a CLI/fixture reference, not a container
    # surface.
    argv_tokens: set[str] = set()
    for node in ast.walk(tree):
        if _is_subprocess_call(node):
            for s in _argv_strings(node):
                for word in _tokenize(s):
                    if word in _ARGV_TOKENS:
                        argv_tokens.add(word)
    for tok in sorted(argv_tokens):
        at = ""
        for node in ast.walk(tree):
            if _is_subprocess_call(node):
                if any(tok in _tokenize(s) for s in _argv_strings(node)):
                    at = f"{module}:{_lineno(node)}"
                    break
        markers.append({"kind": "argv-literal", "value": tok, "at": at})

    # 2. env-gated live lanes (INCUS_HARNESS_IT-style guards).
    for key in _LIVE_ENV_KEYS:
        for node in ast.walk(tree):
            found = False
            if isinstance(node, ast.Call):
                # os.environ.get("KEY")
                func = node.func
                if (
                    isinstance(func, ast.Attribute)
                    and isinstance(func.value, ast.Attribute)
                    and isinstance(func.value.value, ast.Name)
                    and func.value.value.id == "os"
                    and func.value.attr == "environ"
                ):
                    found = any(key == c for c in _string_constants(node))
            elif isinstance(node, ast.Subscript):
                # os.environ["KEY"]
                value = node.value
                if (
                    isinstance(value, ast.Attribute)
                    and isinstance(value.value, ast.Name)
                    and value.value.id == "os"
                    and value.attr == "environ"
                ):
                    found = any(key == c for c in _string_constants(node))
            if found:
                markers.append(
                    {
                        "kind": "env-gated",
                        "value": key,
                        "at": f"{module}:{_lineno(node)}",
                    }
                )
                break

    # 3. pytest live markers (@pytest.mark.incus_live / *_live / live).
    for name in sorted(set(_pytest_marker_names(tree))):
        if name == "live" or name.endswith("_live") or "incus" in name:
            markers.append(
                {
                    "kind": "pytest-marker",
                    "value": name,
                    "at": f"{module}:{_lineno(tree)}",
                }
            )

    # 4. deploy-surface invocations — a subprocess call that reaches a deploy
    # script, either directly (argv literal names the script) or via a variable
    # assigned to a deploy path (``SHIP = REPO_ROOT / "deploy" / "ship.sh"``
    # then ``subprocess.run(["bash", str(SHIP), ...])``). Referencing a deploy
    # script as *data* (a fixture input to some other tool) is NOT a marker —
    # the subprocess must actually reach the script.
    deploy_refs: set[str] = set()

    # 4a. direct: the deploy script is the executable (argv[0]) or the script
    # argument to a shell interpreter (argv[1] when argv[0] is bash/sh). A deploy
    # script passed as *data* to another tool (argv[1] of `python other_tool.py`)
    # is NOT a marker.
    _SHELLS = ("bash", "sh", "/bin/bash", "/bin/sh", "bash.exe")
    for node in ast.walk(tree):
        if not _is_subprocess_call(node):
            continue
        argv = _argv_strings(node)
        if not argv:
            continue
        positions = [argv[0]]
        if argv[0].rsplit("/", 1)[-1] in _SHELLS and len(argv) > 1:
            positions.append(argv[1])
        for s in positions:
            for surf in _DEPLOY_SURFACE:
                if surf in s:
                    deploy_refs.add(surf)

    # 4b. indirect: a Name assigned to a deploy-path expression is referenced
    # by a subprocess argv (str(SHIP) etc.).
    for node in ast.walk(tree):
        targets: list[ast.Name] = []
        value = None
        if isinstance(node, ast.Assign):
            targets = [t for t in node.targets if isinstance(t, ast.Name)]
            value = node.value
        elif isinstance(node, ast.AnnAssign):
            targets = [node.target] if isinstance(node.target, ast.Name) else []
            value = node.value
        if value is None or not targets:
            continue
        matched = {
            surf
            for c in _string_constants(value)
            for surf in _DEPLOY_SURFACE
            if surf in c
        }
        if not matched:
            continue
        names = {t.id for t in targets}
        for sub in ast.walk(tree):
            if not _is_subprocess_call(sub):
                continue
            first = sub.args[0] if sub.args else None
            if first is not None and any(
                isinstance(n, ast.Name) and n.id in names for n in ast.walk(first)
            ):
                deploy_refs |= matched

    for surf in sorted(deploy_refs):
        at = ""
        for node in ast.walk(tree):
            if _is_subprocess_call(node):
                at = f"{module}:{_lineno(node)}"
                break
        markers.append({"kind": "deploy-surface", "value": surf, "at": at})

    return markers


def _test_modules() -> list[Path]:
    return sorted(REPO_ROOT.glob("tests/**/test_*.py"))


def _body_hash(body: str) -> str:
    return hashlib.sha256(body.encode("utf-8")).hexdigest()


def _fingerprint(rows: list[dict]) -> str:
    """The tests-tree fingerprint: sha of the sorted (module, body_hash) list."""
    parts: list[str] = []
    for m in _test_modules():
        try:
            body = m.read_text(encoding="utf-8")
        except OSError:
            continue
        parts.append(f"{m.relative_to(REPO_ROOT)}\0{_body_hash(body)}")
    digest = hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()
    return digest


def derive() -> dict:
    rows: list[dict] = []
    for m in _test_modules():
        try:
            body = m.read_text(encoding="utf-8")
        except OSError:
            continue
        markers = _derive_markers(m, body)
        if not markers:
            continue
        rows.append(
            {
                "module": str(m.relative_to(REPO_ROOT)),
                "category": "rehearsal",
                "markers": markers,
            }
        )
    return {
        "derivation_version": DERIVATION_VERSION,
        "generated_at": datetime.now(UTC).isoformat(),
        "tests_tree_fingerprint": _fingerprint(rows),
        "rows": rows,
    }


def _load_artifact() -> dict | None:
    if not ARTIFACT.is_file():
        return None
    try:
        return json.loads(ARTIFACT.read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        return None


def _stdout_table(data: dict) -> None:
    print(f"# rehearsal-class modules ({len(data['rows'])}):")
    for row in data["rows"]:
        kinds = ", ".join(f"{m['kind']}:{m['value']}" for m in row["markers"])
        print(f"  {row['module']}  —  {kinds}")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--check",
        action="store_true",
        help="recompute and compare against the committed artifact; exit non-zero on drift",
    )
    parser.add_argument(
        "--stdout",
        action="store_true",
        help="print a human table instead of writing the artifact",
    )
    args = parser.parse_args(argv)

    data = derive()

    if args.stdout:
        _stdout_table(data)
        return 0

    if args.check:
        artifact = _load_artifact()
        if artifact is None:
            print(
                "rehearsal-class: artifact missing or unparsable — regenerate "
                "with `python scripts/derive_test_categories.py`",
                file=sys.stderr,
            )
            return 2
        if artifact.get("derivation_version") != DERIVATION_VERSION:
            print(
                f"rehearsal-class: derivation_version mismatch "
                f"({artifact.get('derivation_version')} != {DERIVATION_VERSION})",
                file=sys.stderr,
            )
            return 2
        if artifact.get("tests_tree_fingerprint") != data["tests_tree_fingerprint"]:
            print(
                "rehearsal-class: tests_tree_fingerprint mismatch — the tree "
                "changed since the artifact was derived. Re-derive with "
                "`python scripts/derive_test_categories.py`",
                file=sys.stderr,
            )
            return 2
        if artifact.get("rows") != data["rows"]:
            print(
                "rehearsal-class: rows drift — regenerate with "
                "`python scripts/derive_test_categories.py`",
                file=sys.stderr,
            )
            return 2
        print("rehearsal-class: artifact == tree ✅")
        return 0

    ARTIFACT.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    print(f"wrote {ARTIFACT} ({len(data['rows'])} rehearsal-class modules)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())