forgefirm/forgetest/forgetest/catalog.py

# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier:    MIT

"""The acceptance catalog: test definitions and the registry.

A test is a function decorated with @test(...). The decorator records
what the release gate needs to know without running anything: the id,
the subsystem, the kind (auto / operator / live), how it takes the
hardware (api / takeover), the controller mode it needs (if any), what
source it covers, what it requires, and whether it belongs to the
always-required core. Two more fields describe the operator's part
without affecting the gate: `actions`, the machine actions the test asks
for by name (the page lists them before a start; a bench actuator can
perform them), and `precheck`, a condition the machine must meet for the
test to start at all (a reason string refuses the start, the way an
unmet prerequisite does, and records no result). The function body runs
under the runner with a Context (log, prompts, evidence, hardware
helpers) and reports by returning normally (PASS) or raising
runner.Failed (FAIL).
"""
import ast
import hashlib
import inspect
import os
import re
import threading

from . import manifest as _manifest

KINDS = ("auto", "operator", "live")
HARDWARE = ("api", "takeover")
MODES = ("grbl", "cloud")
# The machine actions a test may ask of the operator (Context.act):
# the lid, the remote-interlock loop, and the big button. Everything a
# test needs done to the machine is one of these, so a bench actuator
# that covers a channel can stand in for the hands on it.
ACTIONS = ("lid", "interlock", "button")
_ID_RX = re.compile(r"^[a-z][a-z0-9-]*\.[a-z][a-z0-9-]*$")

REGISTRY = {}


class Test:
    def __init__(self, id, title, subsystem, kind, hardware, covers, requires,
                 always, est_min, steps, description, fn, mode=None, actions=(),
                 precheck=None, hands=()):
        self.id = id
        self.title = title
        self.subsystem = subsystem
        self.kind = kind
        self.hardware = hardware
        self.mode = mode
        self.covers = tuple((str(c), str(g)) for c, g in covers)
        self.requires = tuple(requires)
        self.always = bool(always)
        self.est_min = est_min
        self.steps = tuple(steps)
        self.actions = tuple(actions)
        self.hands = tuple(hands)
        self.precheck = precheck
        self.description = description or (fn.__doc__ or "").strip()
        self.fn = fn
        self._source_sha = None
        self._fp = (None, None)   # (manifest content sha, fingerprint)

    @property
    def source_sha(self):
        """The hash of the test's own implementation: its function (the
        decorator included) together with the code its module shares
        among its tests (everything outside the @test functions), the
        shared code of every sibling module that either of the two
        imports, and the shared judges (SHARED_JUDGES). Part of the
        fingerprint: a change inside one test's body invalidates that
        test's earlier passes and no other; a change to a helper
        invalidates the tests of that module and the tests that import
        it; a change to a shared judge invalidates every test. A test the
        module does not define in the @test form hashes its whole file."""
        if self._source_sha is None:
            # One filler at a time: a second poll that arrives while the
            # first computes waits for it rather than repeating the work.
            with _IMPL_LOCK:
                if self._source_sha is None:
                    path = inspect.getsourcefile(self.fn) or inspect.getfile(self.fn)
                    self._source_sha = implementation_sha(path, self.id)
        return self._source_sha

    def fingerprint(self, manifest):
        """The domain fingerprint on this manifest, memoized by the
        manifest's content hash: the page recomputes every test's
        fingerprint on every poll, and a manifest never changes under a
        running tool."""
        key = manifest.content_sha
        if key and self._fp[0] == key:
            return self._fp[1]
        fp = _manifest.fingerprint(manifest, self.covers, extra=[self.source_sha])
        if key:
            self._fp = (key, fp)
        return fp

    def definition(self):
        """The gate-visible definition (no implementation, no prose)."""
        return {
            "id": self.id,
            "subsystem": self.subsystem,
            "kind": self.kind,
            "hardware": self.hardware,
            "covers": [list(c) for c in self.covers],
            "requires": list(self.requires),
            "always": self.always,
        }

    def describe(self):
        d = self.definition()
        d.update({"title": self.title, "est_min": self.est_min, "mode": self.mode,
                  "steps": list(self.steps), "actions": list(self.actions),
                  "hands": list(self.hands), "precheck": bool(self.precheck),
                  "description": self.description})
        return d

    def fixture_runnable(self, channels):
        """Whether a bench actuator covering `channels` lets this test
        run with nobody in the room: an operator test whose every action
        is a covered channel and which asks a person for nothing else. A
        live test never qualifies (the arm press and the fire watch are
        a person's); an auto test needs no fixture."""
        if self.kind != "operator" or self.hands or not self.actions:
            return False
        return all(a in channels for a in self.actions)

    def cannot_start(self):
        """The reason the test cannot start on the machine as it is, or
        None. Evaluated right before a start; never a result."""
        if self.precheck is None:
            return None
        try:
            reason = self.precheck()
        except Exception as e:  # noqa: BLE001 - a broken precheck refuses, it never crashes the runner
            return "precheck errored: %s: %s" % (type(e).__name__, e)
        return reason or None


def source_file_sha(path):
    with open(path, "rb") as f:
        data = f.read().replace(b"\r\n", b"\n")
    return hashlib.sha256(data).hexdigest()


# The package modules that judge the machine. Every test runs through
# the first three without importing them: the Context a test is handed
# and the takeover (runner.py), the sysfs, LED, switch and HTTP readers
# behind it (hw.py), and the fresh-boot idle state judged around every
# run (baseline.py). puls.py is the dark job the cloud and extension
# tests print: suite/cloud.py imports it from the package (`from .. import
# puls`), an import the sibling walk does not follow. Their text joins
# every implementation hash: a change to one of them moves every test.
# The rest of the package (the server, the page, the campaign rules, the
# artifact, the manifest, this registry, the bench actuator) judges
# nothing about the machine and stays out.
SHARED_JUDGES = ("baseline.py", "hw.py", "puls.py", "runner.py")


def shared_judge_paths():
    """The SHARED_JUDGES files, next to this module. Every one has to be
    there: a judge missing from the package is an error, never a hash
    computed without it."""
    here = os.path.dirname(os.path.abspath(__file__))
    return [os.path.join(here, name) for name in SHARED_JUDGES]


# The files never change under a run, so each is read and parsed once:
# path -> (text, tree or None), path -> (shared sha, {test id: own sha}),
# path -> the sibling modules it imports itself.
_PARSED = {}
_PARTS = {}
_SIBLINGS = {}
_PARTS_LOCK = threading.RLock()
_IMPL_LOCK = threading.Lock()


def forget(path):
    """Drop what is known about a module: its file changed (the unit tests
    edit their modules between readings)."""
    with _PARTS_LOCK:
        for cache in (_PARSED, _PARTS, _SIBLINGS):
            cache.pop(path, None)


def _parsed(path):
    """(text, tree) of a suite module, line endings normalized; the tree
    is None when the file does not parse."""
    with _PARTS_LOCK:
        hit = _PARSED.get(path)
        if hit is None:
            with open(path, "rb") as f:
                text = f.read().replace(b"\r\n", b"\n").decode("utf-8")
            try:
                tree = ast.parse(text)
            except SyntaxError:
                tree = None
            hit = _PARSED[path] = (text, tree)
        return hit


def _test_defs(tree):
    """{test id: its function} for the @test("...") functions at the top
    level of a module's tree."""
    defs = {}
    for node in (tree.body if tree is not None else []):
        if not isinstance(node, ast.FunctionDef):
            continue
        for d in node.decorator_list:
            if (isinstance(d, ast.Call) and isinstance(d.func, ast.Name) and d.func.id == "test"
                    and d.args and isinstance(d.args[0], ast.Constant) and isinstance(d.args[0].value, str)):
                defs[d.args[0].value] = node
    return defs


def module_parts(path):
    """(shared sha, {test id: own sha}) for a suite module: the test ids
    are read from the @test("...") decorators on the module's top-level
    functions; a test's own text runs from its first decorator line to
    the end of its body; the shared text is every other line of the
    file. Line endings normalized."""
    with _PARTS_LOCK:
        hit = _PARTS.get(path)
    if hit is not None:
        return hit
    text, tree = _parsed(path)
    lines = text.split("\n")
    spans = {}
    for tid, node in _test_defs(tree).items():
        start = min([dd.lineno for dd in node.decorator_list] + [node.lineno])
        spans[tid] = (start, node.end_lineno)
    owned = set()
    for a, b in spans.values():
        owned.update(range(a, b + 1))
    shared = "\n".join(ln for i, ln in enumerate(lines, 1) if i not in owned)
    own = {tid: _manifest.sha256_text("\n".join(lines[a - 1:b])) for tid, (a, b) in spans.items()}
    parts = (_manifest.sha256_text(shared), own)
    with _PARTS_LOCK:
        _PARTS[path] = parts
    return parts


def sibling_imports(path, seen=None):
    """The suite modules a module's shared text imports with `from .x
    import ...` or `from . import x`, as paths, transitively, in a stable
    order. A helper a test calls from a sibling module is part of the
    judge, so its text belongs in the test's fingerprint."""
    if seen is None:
        seen = []
    for sib in _direct_siblings(path):
        if sib not in seen and sib != path:
            seen.append(sib)
            sibling_imports(sib, seen)
    return seen


def _imported(path, nodes):
    """The suite modules the statements `nodes` import with `from .x
    import` or `from . import x` (each name a module of the suite), at any
    depth (inside a function or a block too), in the order they are
    written."""
    here = os.path.dirname(path)
    found = sorted((n for top in nodes for n in ast.walk(top)
                    if isinstance(n, ast.ImportFrom) and n.level == 1),
                   key=lambda n: (n.lineno, n.col_offset))
    out = []
    for node in found:
        for name in ([node.module] if node.module else [a.name for a in node.names]):
            sib = os.path.join(here, name.replace(".", os.sep) + ".py")
            if os.path.isfile(sib) and sib not in out:
                out.append(sib)
    return out


def _direct_siblings(path):
    """The suite modules one module's shared text imports itself: at its
    top level, and inside its helpers and blocks. What a @test function
    imports is that test's own (_own_siblings)."""
    with _PARTS_LOCK:
        hit = _SIBLINGS.get(path)
        if hit is None:
            _text, tree = _parsed(path)
            tests = list(_test_defs(tree).values())
            shared = [n for n in (tree.body if tree is not None else []) if not any(n is t for t in tests)]
            hit = _SIBLINGS[path] = _imported(path, shared)
        return hit


def _own_siblings(path, test_id):
    """The suite modules one test's own function imports: part of that
    test's judge, and of no other test's."""
    _text, tree = _parsed(path)
    node = _test_defs(tree).get(test_id)
    return _imported(path, [node]) if node is not None else []


def implementation_sha(path, test_id, judges=None):
    """The implementation hash of one test (see Test.source_sha): the
    test's own text, the shared text of its module, the shared text of
    every sibling suite module the module's shared text or the test's
    own function imports (transitively), and the text of the shared
    judges. `judges` names files to take as the judges in place of
    SHARED_JUDGES (the unit tests edit their own)."""
    shared, own = module_parts(path)
    if test_id not in own:
        return source_file_sha(path)
    sibs = sibling_imports(path)
    for sib in _own_siblings(path, test_id):
        if sib not in sibs and sib != path:
            sibs.append(sib)
            sibling_imports(sib, sibs)
    helpers = [module_parts(p)[0] for p in sorted(sibs) if p != path]
    paths = shared_judge_paths() if judges is None else list(judges)
    judged = [module_parts(p)[0] for p in paths]
    return _manifest.sha256_text("%s:%s" % (":".join([shared] + helpers + judged), own[test_id]))


def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers=(),
         requires=(), always=False, est_min=1, steps=(), description="", actions=(),
         precheck=None, hands=()):
    """`mode` names the controller mode the test needs live when it starts
    ("grbl" or "cloud"); the runner switches the machine there before the
    test and leaves it there, so a queue crosses modes only where a test
    asks it to. None means the test runs in whatever mode it finds (or
    manages the mode itself, as the cloud tests do through enter_cloud,
    which also waits for the service session).

    `actions` names the machine actions (ACTIONS) the test performs
    through Context.act; an `auto` test declares none. `hands` names what
    the test asks of a person beyond those actions ("app" for a job in
    the Glowforge app, "scrap", "mark" for an eye on the result...): an
    operator test with none, whose actions a bench actuator covers, runs
    unattended when one is up. `precheck` is a callable returning a
    reason string when the machine cannot run the test as it is (None
    when it can)."""
    if not _ID_RX.match(id):
        raise ValueError("test id %r must look like subsystem.name" % id)
    if kind not in KINDS:
        raise ValueError("test %s: kind %r" % (id, kind))
    if hardware not in HARDWARE:
        raise ValueError("test %s: hardware %r" % (id, hardware))
    if mode is not None and mode not in MODES:
        raise ValueError("test %s: mode %r" % (id, mode))
    for c, g in covers:
        if c in _manifest.DEV_ONLY_COMPONENTS:
            raise ValueError("test %s: may not cover dev-only component %r" % (id, c))
    for a in actions:
        if a not in ACTIONS:
            raise ValueError("test %s: action %r" % (id, a))
    if actions and kind == "auto":
        raise ValueError("test %s: an auto test asks for no machine actions" % id)
    if hands and kind == "auto":
        raise ValueError("test %s: an auto test asks nothing of a person" % id)
    if precheck is not None and not callable(precheck):
        raise ValueError("test %s: precheck must be callable" % id)

    def deco(fn):
        if id in REGISTRY:
            raise ValueError("duplicate test id %r" % id)
        REGISTRY[id] = Test(id, title, subsystem, kind, hardware, covers, requires,
                            always, est_min, steps, description, fn, mode=mode,
                            actions=actions, precheck=precheck, hands=hands)
        return fn
    return deco


def all_tests(registry=None):
    """Tests in registration order (the suite modules import in
    subsystem order, so this is the display order)."""
    return list((registry if registry is not None else REGISTRY).values())


def get(id, registry=None):
    return (registry if registry is not None else REGISTRY).get(id)


def order_by_requires(tests, selected):
    """`selected` ids in an order that runs a prerequisite before the test
    that names it.

    Registration order otherwise, so a run reads down the page. A
    prerequisite outside the selection places nothing: either it is
    already satisfied, or the run that needs it will be refused and
    recorded as skipped. validate() has ruled out cycles; the stack check
    only keeps a malformed registry from recursing forever.
    """
    want = set(selected)
    by_id = {t.id: t for t in tests}
    out, placed = [], set()

    def visit(tid, stack):
        if tid in placed or tid not in want or tid not in by_id or tid in stack:
            return
        stack.add(tid)
        for r in by_id[tid].requires:
            visit(r, stack)
        stack.discard(tid)
        placed.add(tid)
        out.append(tid)

    for t in tests:
        visit(t.id, set())
    return out


def validate(registry=None):
    """Every `requires` names a known test and there are no cycles."""
    reg = registry if registry is not None else REGISTRY
    for t in reg.values():
        for r in t.requires:
            if r not in reg:
                raise ValueError("test %s requires unknown test %s" % (t.id, r))
    seen = {}

    def visit(tid, stack):
        if tid in stack:
            raise ValueError("requires cycle: %s" % " -> ".join(stack + [tid]))
        if seen.get(tid):
            return
        for r in reg[tid].requires:
            visit(r, stack + [tid])
        seen[tid] = True
    for tid in reg:
        visit(tid, [])


def catalog_hash(registry=None):
    """Identity of the catalog's definitions (ids, kinds, coverage,
    requirements, core membership) - not of the implementations, which
    the per-test fingerprints carry."""
    defs = sorted((t.definition() for t in all_tests(registry)), key=lambda d: d["id"])
    return _manifest.sha256_text(_manifest.canonical(defs))


def load_suite():
    """Import the suite modules (each registers its tests) and validate."""
    from . import suite  # noqa: F401  (registers on import)
    validate()
    return REGISTRY


def suite_dir():
    return os.path.join(os.path.dirname(os.path.abspath(__file__)), "suite")