From 7c0412075c9e08bd0c1333bfc0d4d104cd63973f Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Mon, 14 Sep 2026 10:39:08 -0400 Subject: [PATCH] Normalize line endings to LF .gitattributes sets text=auto with eol=lf, so every text file is stored and checked out with LF, and a patch keeps its bytes. The files that carried CRLF from a Windows editor are renormalized. No content changes. --- .gitattributes | 4 + AGENTS.md | 796 +++++++++--------- forgetest/tests/test_fixture.py | 1354 +++++++++++++++---------------- 3 files changed, 1079 insertions(+), 1075 deletions(-) create mode 100644 .gitattributes diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..1b30e1b --- /dev/null +++ b/.gitattributes @@ -0,0 +1,4 @@ +# Every text file is stored and checked out with LF; a patch keeps its bytes. +* text=auto eol=lf +*.patch -text +*.diff -text diff --git a/AGENTS.md b/AGENTS.md index fc1fd0a..cc52857 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,398 +1,398 @@ -# AGENTS.md - -Instructions for AI coding agents that work in this repository. A human -contributor is welcome to read it too: it is the same set of rules. - -## ForgeFIRM repository rules - -### Safety on the machine - -- The laser never fires with the coolant pump deliberately off. The heater-only - flow-calibrate and flow-verify tools are the only pump-off state, and the - tube stays dark during them. -- The 40 V motor rail stays up while the machine is on. Never cycle it for a - controller handover. A rail glitch can leave the stepper drivers - unserviceable. - -### The iteration loop - -- Edit, cross-build in the Yocto build environment, hot-deploy to the - machine, prove. Then one commit, one push, one pin bump for the proven - change. -- A full image build happens once, at the end of a work item, never per fix. -- Kernel and BSP changes ride one image flash, batched. Validate a `.ko` or - overlay change on the image that ships it. Never hot-swap a `.ko` onto a - board that is about to be reflashed. - -### Acceptance coverage - -- Evaluate every change against the acceptance catalog in - `forgetest/forgetest/suite/`. Does an existing test exercise the changed - behavior? Does its `covers` map name the files you touched? If not, add or - widen one in the same change. The coverage lint - (`python3 -m forgetest.coverage --enforce`) is the floor. -- The `SRCREV` of a component and the `PV` that moves with it live in - `-pin.inc` next to the recipe. Nothing else goes in that file. The - image manifest keeps `*-pin.inc` out of the layer content hash. Thus a pin - bump invalidates only the acceptance tests that cover that component. A - pin in a recipe body counts as a platform change and needs a full - acceptance campaign. -- Test a harness change yourself, on the host, before you hand it to the - operator for a bench run. A harness that fails on its own defect costs the - operator a full re-run cycle. -- A test declares its `mode`, and the runner switches. Cloud tests stay in - cloud mode. Keep `requires` lists minimal, and never gate one test behind a - chain of others. -- Never exercise a gate or a limit through the `GFCOOL_*` environment - overrides in a test. Use the settings API, and restore it in the teardown. - -### CI couplings that set the push order - -- The CI of `grblHAL-glowforge` gets the laser harnesses from `scripts/bench/` - at the head of `master`, unpinned. Push a harness change here before the - driver change that needs it. -- The acceptance page shares its theme and its vendored Bootstrap with the - `forgectrl` panel, byte for byte. Land a UI change in `forgectrl` first. - Push it and pin it. Only then push this repository. -- The CI of `kernel-module-glowforge` cross-builds against the BSP from - `meta-openglow`. Push `meta-openglow` before a module change that needs a - BSP change. - -### Builds and images - -- Builds run only in the dedicated Yocto build environment that the Build page - of the site defines (kas and bitbake). Never use a native host build of a - component as proof, not even for a compile check. -- Every build makes both images: `forgefirm-image` (the release image) and - `forgefirm-image-dev` (the bench image, with `forgetest` on port 8090). The - dev image needs an explicit `--target`. Report both artifact names. -- After a kernel-module pin bump, rebuild the kernel and the module in the - same run: `bitbake -c cleansstate linux-fslc kernel-module-glowforge`, then - the images. The kernel version suffix is not reproducible across - re-checkouts, and a rootfs whose kernel and module disagree fails. -- Read the layer HEAD lines at the top of a build log before you trust a build - that moves a pin. - -### The record of the work - -This repository carries no status document and no dated log. Both were -retired: every fact about the machine and the firmware is on the -documentation site, and **the record of a piece of work is the commit message -that carries it**: what was done, how it was proven, and what it replaced, in -the same place as the change itself. - -A measurement session or a bench drill is written up in the commit that -lands its result. Nothing accumulates in a file. - -### On the bench - -These rules apply whenever an agent works with the operator at a machine. - -- **One armed run per turn.** Start the run, let it finish, stop, report what - happened against what was expected, and wait for the operator's - confirmation before any further armed run or machine command. Never chain a - jog, a fix, or a re-run onto a burn in the same step. Dry runs (latch - locked, no emission) can be batched. Say which is which before you start. -- **Zero inference during live fire.** Report only what the combined - observations confirm: the operator's eyes, the drill output, the logs, the - sysfs readbacks. Everything else is labeled "unconfirmed" and is not acted - on. -- **One Grbl connection.** Never open a second connection to TCP port 23 while - a drill, an acceptance test, or a sender holds one. The last connection - wins, the displaced client's cleanup never arrives, and its queued commands - wait behind the arm gate. Read state through `forgectrl` (HTTP) and sysfs - only. After a displaced or aborted drill: soft reset, confirm that the RX - buffer reads full, `M5`, confirm that `armed` is false, before anyone - touches the button. -- **The operator touches the machine.** The operator runs the install, the - service restart, and the reflash on the machine. Never modify the software - of a running machine unless the operator asks for that exact step. -- **Bench hygiene.** Stage test files in `/tmp`. A file that must survive a - reboot goes in `/data/bench-scratch/`, and that directory is deleted whole - at the end of the session. Nothing goes loose in `/data`. Remove everything - you put on the board, in the same session, and list `/data` before you end. - A bench tool worth a second use goes in `scripts/bench/`. The dev image - installs those tools under `/usr/share/forgetest/bench/`. Run them from - there. - -## Project-wide rules - -The sections above are specific to this repository. The rules below apply to -every repository of the OpenGlow ForgeFIRM project: the firmware components, -the build base, the documentation, and the hardware designs. Where a -repository rule and a rule below conflict, the safety, proof, and hygiene -rules below win. The documentation site, , is -the source of truth for every fact about the machine and the firmware. Read -its Developers section before you change code. - -### What ForgeFIRM is - -ForgeFIRM is open firmware for stock Glowforge lasers (Basic, Plus, Pro). It -runs on the factory NXP i.MX6 control board with no hardware modification. It -replaces the cloud-dependent factory userspace with an open Linux image that -controls the machine locally. In GRBL mode the machine is a standard grblHAL -controller for LightBurn and other senders on TCP port 23. Cloud mode runs the -factory experience through the Glowforge web service, and it is a deliberate, -maintained feature. The stack is hardware validated: GRBL mode cuts real jobs, -and cloud mode runs end to end. - -One controller runs at a time under `forgectrl`, the machine-services daemon, -which owns the pulse device, the cooling engine, the cameras, and the control -panel. The kernel module plays the pulse stream into the stepper drivers and -owns the laser latch and the safety readbacks. The Technical section of the -site describes the whole stack. - -### The repositories - -All repositories live under the OpenGlow organization on GitHub, -. For development, check every repository out -as a sibling in one base directory named `openglow-forgefirm`. The build -scripts expect that layout. Each repository is its own git repository. The -base directory is not. - -| Repository | Role | Default branch | License | -|---|---|---|---| -| `forgefirm` | The build and release base: the `meta-forgefirm` Yocto layer, the kas configuration, the image recipes, the install and release scripts, the acceptance tool `forgetest/`, the bench tools `scripts/bench/`, and the bench actuator firmware `fixture/`. It carries no project documentation. | `master` | MIT (layer metadata) | -| `meta-openglow` | The BSP layers `meta-glowforge-bsp` and `meta-openglow-core`: kernel, device tree, U-Boot, board recipes. | `scarthgap` | MIT (layer metadata) | -| `forgectrl` | The machine-services daemon (C) on HTTPS port 443, with the read-only routes on HTTP port 80: supervisor, pulse-device broker, motion-liveness gate, cooling engine, cameras, telemetry, settings, diagnostics, control panel, A/B updates. | `main` | MIT | -| `grblHAL-glowforge` | The grblHAL driver for the stock board, the GRBL-mode controller. The core is a submodule at `src/grbl`. Machine constants are in `src/boards/glowforge.h`. | `main` | GPL-3.0-or-later | -| `grblHAL-core` | The fork of the grblHAL core that the driver uses. Minimum change, no rationale comments. | `forgefirm` | The upstream grblHAL license | -| `kernel-module-glowforge` | `glowforge.ko`: the SDMA + EPIT pulse engine, the laser latch, the safety readbacks, the sensors. | `master` | GPL-2.0-or-later | -| `python3-gfhardware` | The `gfhardware` hardware library and the cloud-mode applications in `forgefirm-app/`. | `master` | MIT, with one LGPL-2.1-or-later component | -| `Glowforge-Utilities` | `gfutilities` on PyPI: the factory protocol and service layer that cloud mode uses. | `master` | MIT | -| `forgefirm-docs` | The documentation site. | `main` | CC BY-SA 4.0 | -| `openglow-serial-adapter` | The USB-C serial-console adapter for the control board. | `main` | CC BY-NC-SA 4.0 | - -The full table, the license details, and the component diagram are on the -site: . - -### Safety first, and in that order - -The machine is a laser, and the operator is the safety authority at it. These -rules bind every change, in every repository, that can affect what the machine -does. - -- The hardware chain is the safety boundary. Software only adds gates on top - of it. `LASER_ON` is never a bare GPIO. The kernel laser latch is locked by - default, and every close of the pulse device locks it again. -- Never defeat a switch, an interlock, or a readback, in hardware or in - software, not even for a test. -- Never change a fire gate, a thermal limit, or a safety default to make a - test pass. -- The order of work is absolute: emission and motion first, robustness and - hygiene after. Review a change that touches emission or motion for that - before anything else. -- A change that can put energy where it was not commanded gets a regression - test with the fix, in the same commit, never after. -- Position counters, homing anchors, and a homed flag are not proof of - physical motion. The head accelerometer is, and so are the operator's eyes. - -### Proof before done - -A feature or fix is not complete until it is proven. The order of preference: - -1. A host test that runs in CI. The Test page of the site lists the host tests - of each repository. -2. A bench drill on the bench reference, recorded in the commit message that - carries the change. -3. Documented reasoning. - -Also: - -- Run the host tests of this repository before every commit. Where the build - uses `-Werror`, a warning is a failure. -- A component that ships in the image is also evaluated against the release - acceptance catalog in the `forgefirm` repository (Test page, "Coverage - currency"). A behavior change with no catalog consequence gets a sentence - of justification in the commit message. -- For a component that ships in the image, the proof build is the Yocto - cross-build that the Build page of the site defines, never a native host - build. -- Report outcomes faithfully. Quote failing output as it is. Name a skipped - step as skipped. Never narrate an expected result as an observed one. - -### The workflow: prove locally, push when proven - -- Nothing goes to a public repository until it is proven: in the host tests - for host-only changes, on the bench for changes that alter what the machine - does. -- Iterate locally. Then make one commit and one push for the proven change. - No chains of fix-up pushes. Every push runs CI, and CI minutes are a budget. -- Commit only finished work, and only when asked. Never commit intermediate - findings, working notes, plans, audits, or status files. If a conclusion is - still moving, it is not ready for a repository. Unpushed mistakes come out - with `git reset --mixed `, which keeps the working tree. -- "Commit" and "commit and build" include the push of every touched - repository, in dependency order. Never hold a push back to protect earlier - unpushed commits on the same branch. The only exception is an explicit "do - not push" for that session. -- Never push, force-push, tag, cut a release, refresh a lockfile, bump a - submodule pointer, or bump a pin unless asked. - -### Pins - -A firmware component reaches the image only through its pin: an exact -`SRCREV` in `meta-forgefirm` (in the `forgefirm` repository) for ForgeFIRM -components, or in `meta-openglow` for BSP components. There is no `AUTOREV`. -A commit in a component repository changes nothing on a machine until its pin -moves. The order is always: push the source repository, bump its pin, then -run `bitbake -c fetch ` to make sure that the pin resolves. Some -pairs of repositories have a CI coupling that fixes the push order between -them. The Release flow page of the site lists them, and each affected -repository names its own in the section above. - -### Git conventions - -- A commit is attributed to the human who is responsible for it, in both the - author and the committer fields. Never list an AI assistant as an author or - a co-author. Never add an AI attribution trailer or a "Generated with" line - to a commit message or a pull request. In a fresh clone, set the local - `user.name` and `user.email` before the first commit. -- A commit message says what changed and why, in the present tense, in - Simplified Technical English. The "why" of a change to fork code lives here - and nowhere else. -- Keep line endings LF in every file that Linux tooling consumes: patches, - shell scripts, recipes, device trees. Check the bytes after an edit from a - Windows tool. A CRLF patch breaks `do_patch`. -- Never commit build output, `__pycache__`, virtual environments, editor - files, a local knowledge graph (`graphify-out/`), captured machine data that - carries an identity, tokens, keys, or passwords. -- Never rewrite published history unless the operator asks for it. -- A pull request to an upstream repository is never automatic. The developer - decides if and when to send one. You can remind them that upstream - candidates exist, but the decision is theirs. - -### Writing rules - -They apply to everything: documentation, README files, code comments, -docstrings, log and console text, commit messages, issue text, and replies. - -- **American English.** analyze, behavior, color, center, gray, catalog, - license, judgment, program, artifact, percent, toward, among, while, - learned. The style lint in `forgefirm-docs` rejects the British forms. -- **ASD-STE100 Simplified Technical English.** Short sentences: 20 words in - a procedure, 25 in a description. One topic per sentence, one instruction - per sentence. Active voice, imperative in procedures, simple present in - descriptions. "Must" for necessity, "can" for possibility. No "should", - "would", "could", "may". Prefer: do, make sure, use, start, show, occur, - before, after, because, for example, that is. No "etc.", no "and/or". - Noun clusters of three words at most. Warnings and cautions come before the - step they protect. Warm and direct is good. A joke that shades a fact is - not. -- **No em dashes.** Never, anywhere. Use a colon, a period, or a comma. -- **Present tense, present state.** Text describes the code and the machine - as they are. No history narrative, no "the old version did", no "evolved - from", no story of how the code got here. Upstream copyright and - derivation attribution to public projects stays: that is license credit, - not narrative. -- **No workstation paths.** Nothing outside a repository appears as a path in - it: no drive letters, no WSL mounts, no home directories, no sibling-repo - relative paths, no private reference directories. Name the artifact - instead: "the captured factory pulse files", "that bench session". -- **No bench-machine identity.** Never a hostname, a serial number, a fuse - value, a private IP address, or a login of the bench machine, in any file, - comment, commit message, sample, or test fixture. Use neutral placeholders - of the `ABC-123` form. Record the fact of a verification, never the values. -- **No factory firmware code.** Never cite the factory firmware binary or its - code: no offsets, no addresses, no quoted or decompiled code. State the - recovered fact (a formula, a constant, a behavior) and stop. Hardware - addresses (IOMUX pad values, sensor registers, boot offsets) are fine. -- **No rationale comments in fork code.** Code destined upstream (for - example, the grblHAL core fork) carries the minimum change and no "why we - did this" comments. The reasoning goes in the commit message. Patches - carried in the Yocto layers are exempt. The `ForgeFIRM:` comment convention - there is deliberate. -- **Interface docs describe the lever, not the policy.** A sysfs attribute - doc says what the attribute is, its range, its units, its conversion. How - the factory firmware drives it does not belong there. -- **No definitive legal or regulatory statements.** The project is not a - lawyer. The only permitted form: modifying the machine, including - replacing its firmware, may have legal and regulatory ramifications, and it - is the end user's responsibility to adhere to the laws, regulations, - certifications, and insurance terms that apply where they are. Warranty - wording is "may void your warranty", never "voids". Never cite a regulation, - a standard, or a certification regime as applying or not applying. The - affiliation disclaimer lives in the site footer only. It is also the user's - responsibility to make sure that they stay within the vendor's terms of - service when they use a feature that calls that vendor's cloud services. -- **Copyright.** Copyright in this project belongs to 514 LLC d/b/a - OpenGlow. Every new source file carries two lines in its header, in the - comment style of the file: `Copyright 514 LLC d/b/a OpenGlow` and - `Written by Scott Wiederhold`. Keep the year, or the year range, that - the file already has. The copyright notices of other holders stay - unchanged. -- **SPDX.** Every new source file carries an `SPDX-License-Identifier` line - under the license of its repository. - -### Where documentation goes - -- **One home.** A fact lives on the documentation site, or it does not exist. - A repository README says what the repository is, how to build and test it, - and where the documentation is. Nothing else. -- **The currency rule.** A change carries a documentation commit when it adds, - removes, or renames an interface (a sysfs attribute, an HTTP route, a - settings key, a G-code or `$` setting), or when it corrects a measured - fact. A measured fact says how it was obtained, and on what: unless the page - says otherwise, a measurement on the site was taken on the bench reference, - which the site defines. -- **A moved document is deleted.** No stub and no redirect stays at the old - path. -- **Every diagram is Mermaid.** No ASCII art and no image of a diagram. -- There is no roadmap page. Open items are tracked as GitHub issues once the - repositories accept them. -- Do not add plan files, audit files, working notes, or status files to any - repository. The project has **no status document and no dated log**: the - site describes the present, and the record of what was done, how it was - proven, and what it replaced goes in the commit message that carries the - change. - -### Analysis and reporting - -- Rank engineering options on technical merit only: performance, correctness, - robustness, verifiability. Process cost (an acceptance campaign, an image - flash, a new dependency) is a one-line footnote, never a ranking factor. - The operator decides the process cost. -- Never infer or assume a hardware fact. Label every unverified claim as - unverified, and say how to verify it. -- Report only what the observations confirm. Everything else is labeled - "unconfirmed" and is not acted on. - -### General practice for agents - -- Read the README of this repository, this file, and the Developers section - of the site before you change code. -- Orient in the code before you edit. If a local knowledge graph - (`graphify-out/`) exists in the working tree or its parent, query it first. - It is a local artifact and is never committed. -- Keep a change minimal and scoped to the request. No drive-by reformatting, - no unrelated cleanups, no renames for taste. Match the style of the - surrounding code. -- Do not add a dependency without a stated reason. The target is a - single-core machine with a small eMMC, and every package rides the image. -- Extend an existing test or tool instead of adding a parallel one. -- Never create a file in the repository that the request does not need. -- Ask before anything destructive or outward-facing: a push, a force-push, a - tag, a release, a history rewrite, or any change to a machine. -- When a fact is unknown, say so and say how to find out. Never fill a gap - with a plausible number. -- Use the exact names of the project: kas, bitbake, pin, SRCREV, campaign, - fingerprint, drill, sysfs, the latch, the armed window. Do not invent names. - -### Before you commit - -1. The host tests of this repository pass, with `-Werror` where the build - uses it. -2. The change is proven at the highest level available: CI test, bench drill, - or documented reasoning. -3. For a component that ships in the image: the acceptance catalog covers the - changed behavior, or the commit message says why it has no catalog - consequence. -4. New files carry the 514 LLC d/b/a OpenGlow copyright, the `Written by` - line, and an SPDX line. Line endings are LF. -5. No em dashes, no British spellings, no workstation paths, no bench - identity, no factory firmware code, no history narrative, no AI - attribution. -6. Documentation is current: interfaces and measured facts (the currency - rule). -7. The commit is one settled change, attributed to the human author, and it - is pushed only when the operator asks. +# AGENTS.md + +Instructions for AI coding agents that work in this repository. A human +contributor is welcome to read it too: it is the same set of rules. + +## ForgeFIRM repository rules + +### Safety on the machine + +- The laser never fires with the coolant pump deliberately off. The heater-only + flow-calibrate and flow-verify tools are the only pump-off state, and the + tube stays dark during them. +- The 40 V motor rail stays up while the machine is on. Never cycle it for a + controller handover. A rail glitch can leave the stepper drivers + unserviceable. + +### The iteration loop + +- Edit, cross-build in the Yocto build environment, hot-deploy to the + machine, prove. Then one commit, one push, one pin bump for the proven + change. +- A full image build happens once, at the end of a work item, never per fix. +- Kernel and BSP changes ride one image flash, batched. Validate a `.ko` or + overlay change on the image that ships it. Never hot-swap a `.ko` onto a + board that is about to be reflashed. + +### Acceptance coverage + +- Evaluate every change against the acceptance catalog in + `forgetest/forgetest/suite/`. Does an existing test exercise the changed + behavior? Does its `covers` map name the files you touched? If not, add or + widen one in the same change. The coverage lint + (`python3 -m forgetest.coverage --enforce`) is the floor. +- The `SRCREV` of a component and the `PV` that moves with it live in + `-pin.inc` next to the recipe. Nothing else goes in that file. The + image manifest keeps `*-pin.inc` out of the layer content hash. Thus a pin + bump invalidates only the acceptance tests that cover that component. A + pin in a recipe body counts as a platform change and needs a full + acceptance campaign. +- Test a harness change yourself, on the host, before you hand it to the + operator for a bench run. A harness that fails on its own defect costs the + operator a full re-run cycle. +- A test declares its `mode`, and the runner switches. Cloud tests stay in + cloud mode. Keep `requires` lists minimal, and never gate one test behind a + chain of others. +- Never exercise a gate or a limit through the `GFCOOL_*` environment + overrides in a test. Use the settings API, and restore it in the teardown. + +### CI couplings that set the push order + +- The CI of `grblHAL-glowforge` gets the laser harnesses from `scripts/bench/` + at the head of `master`, unpinned. Push a harness change here before the + driver change that needs it. +- The acceptance page shares its theme and its vendored Bootstrap with the + `forgectrl` panel, byte for byte. Land a UI change in `forgectrl` first. + Push it and pin it. Only then push this repository. +- The CI of `kernel-module-glowforge` cross-builds against the BSP from + `meta-openglow`. Push `meta-openglow` before a module change that needs a + BSP change. + +### Builds and images + +- Builds run only in the dedicated Yocto build environment that the Build page + of the site defines (kas and bitbake). Never use a native host build of a + component as proof, not even for a compile check. +- Every build makes both images: `forgefirm-image` (the release image) and + `forgefirm-image-dev` (the bench image, with `forgetest` on port 8090). The + dev image needs an explicit `--target`. Report both artifact names. +- After a kernel-module pin bump, rebuild the kernel and the module in the + same run: `bitbake -c cleansstate linux-fslc kernel-module-glowforge`, then + the images. The kernel version suffix is not reproducible across + re-checkouts, and a rootfs whose kernel and module disagree fails. +- Read the layer HEAD lines at the top of a build log before you trust a build + that moves a pin. + +### The record of the work + +This repository carries no status document and no dated log. Both were +retired: every fact about the machine and the firmware is on the +documentation site, and **the record of a piece of work is the commit message +that carries it**: what was done, how it was proven, and what it replaced, in +the same place as the change itself. + +A measurement session or a bench drill is written up in the commit that +lands its result. Nothing accumulates in a file. + +### On the bench + +These rules apply whenever an agent works with the operator at a machine. + +- **One armed run per turn.** Start the run, let it finish, stop, report what + happened against what was expected, and wait for the operator's + confirmation before any further armed run or machine command. Never chain a + jog, a fix, or a re-run onto a burn in the same step. Dry runs (latch + locked, no emission) can be batched. Say which is which before you start. +- **Zero inference during live fire.** Report only what the combined + observations confirm: the operator's eyes, the drill output, the logs, the + sysfs readbacks. Everything else is labeled "unconfirmed" and is not acted + on. +- **One Grbl connection.** Never open a second connection to TCP port 23 while + a drill, an acceptance test, or a sender holds one. The last connection + wins, the displaced client's cleanup never arrives, and its queued commands + wait behind the arm gate. Read state through `forgectrl` (HTTP) and sysfs + only. After a displaced or aborted drill: soft reset, confirm that the RX + buffer reads full, `M5`, confirm that `armed` is false, before anyone + touches the button. +- **The operator touches the machine.** The operator runs the install, the + service restart, and the reflash on the machine. Never modify the software + of a running machine unless the operator asks for that exact step. +- **Bench hygiene.** Stage test files in `/tmp`. A file that must survive a + reboot goes in `/data/bench-scratch/`, and that directory is deleted whole + at the end of the session. Nothing goes loose in `/data`. Remove everything + you put on the board, in the same session, and list `/data` before you end. + A bench tool worth a second use goes in `scripts/bench/`. The dev image + installs those tools under `/usr/share/forgetest/bench/`. Run them from + there. + +## Project-wide rules + +The sections above are specific to this repository. The rules below apply to +every repository of the OpenGlow ForgeFIRM project: the firmware components, +the build base, the documentation, and the hardware designs. Where a +repository rule and a rule below conflict, the safety, proof, and hygiene +rules below win. The documentation site, , is +the source of truth for every fact about the machine and the firmware. Read +its Developers section before you change code. + +### What ForgeFIRM is + +ForgeFIRM is open firmware for stock Glowforge lasers (Basic, Plus, Pro). It +runs on the factory NXP i.MX6 control board with no hardware modification. It +replaces the cloud-dependent factory userspace with an open Linux image that +controls the machine locally. In GRBL mode the machine is a standard grblHAL +controller for LightBurn and other senders on TCP port 23. Cloud mode runs the +factory experience through the Glowforge web service, and it is a deliberate, +maintained feature. The stack is hardware validated: GRBL mode cuts real jobs, +and cloud mode runs end to end. + +One controller runs at a time under `forgectrl`, the machine-services daemon, +which owns the pulse device, the cooling engine, the cameras, and the control +panel. The kernel module plays the pulse stream into the stepper drivers and +owns the laser latch and the safety readbacks. The Technical section of the +site describes the whole stack. + +### The repositories + +All repositories live under the OpenGlow organization on GitHub, +. For development, check every repository out +as a sibling in one base directory named `openglow-forgefirm`. The build +scripts expect that layout. Each repository is its own git repository. The +base directory is not. + +| Repository | Role | Default branch | License | +|---|---|---|---| +| `forgefirm` | The build and release base: the `meta-forgefirm` Yocto layer, the kas configuration, the image recipes, the install and release scripts, the acceptance tool `forgetest/`, the bench tools `scripts/bench/`, and the bench actuator firmware `fixture/`. It carries no project documentation. | `master` | MIT (layer metadata) | +| `meta-openglow` | The BSP layers `meta-glowforge-bsp` and `meta-openglow-core`: kernel, device tree, U-Boot, board recipes. | `scarthgap` | MIT (layer metadata) | +| `forgectrl` | The machine-services daemon (C) on HTTPS port 443, with the read-only routes on HTTP port 80: supervisor, pulse-device broker, motion-liveness gate, cooling engine, cameras, telemetry, settings, diagnostics, control panel, A/B updates. | `main` | MIT | +| `grblHAL-glowforge` | The grblHAL driver for the stock board, the GRBL-mode controller. The core is a submodule at `src/grbl`. Machine constants are in `src/boards/glowforge.h`. | `main` | GPL-3.0-or-later | +| `grblHAL-core` | The fork of the grblHAL core that the driver uses. Minimum change, no rationale comments. | `forgefirm` | The upstream grblHAL license | +| `kernel-module-glowforge` | `glowforge.ko`: the SDMA + EPIT pulse engine, the laser latch, the safety readbacks, the sensors. | `master` | GPL-2.0-or-later | +| `python3-gfhardware` | The `gfhardware` hardware library and the cloud-mode applications in `forgefirm-app/`. | `master` | MIT, with one LGPL-2.1-or-later component | +| `Glowforge-Utilities` | `gfutilities` on PyPI: the factory protocol and service layer that cloud mode uses. | `master` | MIT | +| `forgefirm-docs` | The documentation site. | `main` | CC BY-SA 4.0 | +| `openglow-serial-adapter` | The USB-C serial-console adapter for the control board. | `main` | CC BY-NC-SA 4.0 | + +The full table, the license details, and the component diagram are on the +site: . + +### Safety first, and in that order + +The machine is a laser, and the operator is the safety authority at it. These +rules bind every change, in every repository, that can affect what the machine +does. + +- The hardware chain is the safety boundary. Software only adds gates on top + of it. `LASER_ON` is never a bare GPIO. The kernel laser latch is locked by + default, and every close of the pulse device locks it again. +- Never defeat a switch, an interlock, or a readback, in hardware or in + software, not even for a test. +- Never change a fire gate, a thermal limit, or a safety default to make a + test pass. +- The order of work is absolute: emission and motion first, robustness and + hygiene after. Review a change that touches emission or motion for that + before anything else. +- A change that can put energy where it was not commanded gets a regression + test with the fix, in the same commit, never after. +- Position counters, homing anchors, and a homed flag are not proof of + physical motion. The head accelerometer is, and so are the operator's eyes. + +### Proof before done + +A feature or fix is not complete until it is proven. The order of preference: + +1. A host test that runs in CI. The Test page of the site lists the host tests + of each repository. +2. A bench drill on the bench reference, recorded in the commit message that + carries the change. +3. Documented reasoning. + +Also: + +- Run the host tests of this repository before every commit. Where the build + uses `-Werror`, a warning is a failure. +- A component that ships in the image is also evaluated against the release + acceptance catalog in the `forgefirm` repository (Test page, "Coverage + currency"). A behavior change with no catalog consequence gets a sentence + of justification in the commit message. +- For a component that ships in the image, the proof build is the Yocto + cross-build that the Build page of the site defines, never a native host + build. +- Report outcomes faithfully. Quote failing output as it is. Name a skipped + step as skipped. Never narrate an expected result as an observed one. + +### The workflow: prove locally, push when proven + +- Nothing goes to a public repository until it is proven: in the host tests + for host-only changes, on the bench for changes that alter what the machine + does. +- Iterate locally. Then make one commit and one push for the proven change. + No chains of fix-up pushes. Every push runs CI, and CI minutes are a budget. +- Commit only finished work, and only when asked. Never commit intermediate + findings, working notes, plans, audits, or status files. If a conclusion is + still moving, it is not ready for a repository. Unpushed mistakes come out + with `git reset --mixed `, which keeps the working tree. +- "Commit" and "commit and build" include the push of every touched + repository, in dependency order. Never hold a push back to protect earlier + unpushed commits on the same branch. The only exception is an explicit "do + not push" for that session. +- Never push, force-push, tag, cut a release, refresh a lockfile, bump a + submodule pointer, or bump a pin unless asked. + +### Pins + +A firmware component reaches the image only through its pin: an exact +`SRCREV` in `meta-forgefirm` (in the `forgefirm` repository) for ForgeFIRM +components, or in `meta-openglow` for BSP components. There is no `AUTOREV`. +A commit in a component repository changes nothing on a machine until its pin +moves. The order is always: push the source repository, bump its pin, then +run `bitbake -c fetch ` to make sure that the pin resolves. Some +pairs of repositories have a CI coupling that fixes the push order between +them. The Release flow page of the site lists them, and each affected +repository names its own in the section above. + +### Git conventions + +- A commit is attributed to the human who is responsible for it, in both the + author and the committer fields. Never list an AI assistant as an author or + a co-author. Never add an AI attribution trailer or a "Generated with" line + to a commit message or a pull request. In a fresh clone, set the local + `user.name` and `user.email` before the first commit. +- A commit message says what changed and why, in the present tense, in + Simplified Technical English. The "why" of a change to fork code lives here + and nowhere else. +- Keep line endings LF in every file that Linux tooling consumes: patches, + shell scripts, recipes, device trees. Check the bytes after an edit from a + Windows tool. A CRLF patch breaks `do_patch`. +- Never commit build output, `__pycache__`, virtual environments, editor + files, a local knowledge graph (`graphify-out/`), captured machine data that + carries an identity, tokens, keys, or passwords. +- Never rewrite published history unless the operator asks for it. +- A pull request to an upstream repository is never automatic. The developer + decides if and when to send one. You can remind them that upstream + candidates exist, but the decision is theirs. + +### Writing rules + +They apply to everything: documentation, README files, code comments, +docstrings, log and console text, commit messages, issue text, and replies. + +- **American English.** analyze, behavior, color, center, gray, catalog, + license, judgment, program, artifact, percent, toward, among, while, + learned. The style lint in `forgefirm-docs` rejects the British forms. +- **ASD-STE100 Simplified Technical English.** Short sentences: 20 words in + a procedure, 25 in a description. One topic per sentence, one instruction + per sentence. Active voice, imperative in procedures, simple present in + descriptions. "Must" for necessity, "can" for possibility. No "should", + "would", "could", "may". Prefer: do, make sure, use, start, show, occur, + before, after, because, for example, that is. No "etc.", no "and/or". + Noun clusters of three words at most. Warnings and cautions come before the + step they protect. Warm and direct is good. A joke that shades a fact is + not. +- **No em dashes.** Never, anywhere. Use a colon, a period, or a comma. +- **Present tense, present state.** Text describes the code and the machine + as they are. No history narrative, no "the old version did", no "evolved + from", no story of how the code got here. Upstream copyright and + derivation attribution to public projects stays: that is license credit, + not narrative. +- **No workstation paths.** Nothing outside a repository appears as a path in + it: no drive letters, no WSL mounts, no home directories, no sibling-repo + relative paths, no private reference directories. Name the artifact + instead: "the captured factory pulse files", "that bench session". +- **No bench-machine identity.** Never a hostname, a serial number, a fuse + value, a private IP address, or a login of the bench machine, in any file, + comment, commit message, sample, or test fixture. Use neutral placeholders + of the `ABC-123` form. Record the fact of a verification, never the values. +- **No factory firmware code.** Never cite the factory firmware binary or its + code: no offsets, no addresses, no quoted or decompiled code. State the + recovered fact (a formula, a constant, a behavior) and stop. Hardware + addresses (IOMUX pad values, sensor registers, boot offsets) are fine. +- **No rationale comments in fork code.** Code destined upstream (for + example, the grblHAL core fork) carries the minimum change and no "why we + did this" comments. The reasoning goes in the commit message. Patches + carried in the Yocto layers are exempt. The `ForgeFIRM:` comment convention + there is deliberate. +- **Interface docs describe the lever, not the policy.** A sysfs attribute + doc says what the attribute is, its range, its units, its conversion. How + the factory firmware drives it does not belong there. +- **No definitive legal or regulatory statements.** The project is not a + lawyer. The only permitted form: modifying the machine, including + replacing its firmware, may have legal and regulatory ramifications, and it + is the end user's responsibility to adhere to the laws, regulations, + certifications, and insurance terms that apply where they are. Warranty + wording is "may void your warranty", never "voids". Never cite a regulation, + a standard, or a certification regime as applying or not applying. The + affiliation disclaimer lives in the site footer only. It is also the user's + responsibility to make sure that they stay within the vendor's terms of + service when they use a feature that calls that vendor's cloud services. +- **Copyright.** Copyright in this project belongs to 514 LLC d/b/a + OpenGlow. Every new source file carries two lines in its header, in the + comment style of the file: `Copyright 514 LLC d/b/a OpenGlow` and + `Written by Scott Wiederhold`. Keep the year, or the year range, that + the file already has. The copyright notices of other holders stay + unchanged. +- **SPDX.** Every new source file carries an `SPDX-License-Identifier` line + under the license of its repository. + +### Where documentation goes + +- **One home.** A fact lives on the documentation site, or it does not exist. + A repository README says what the repository is, how to build and test it, + and where the documentation is. Nothing else. +- **The currency rule.** A change carries a documentation commit when it adds, + removes, or renames an interface (a sysfs attribute, an HTTP route, a + settings key, a G-code or `$` setting), or when it corrects a measured + fact. A measured fact says how it was obtained, and on what: unless the page + says otherwise, a measurement on the site was taken on the bench reference, + which the site defines. +- **A moved document is deleted.** No stub and no redirect stays at the old + path. +- **Every diagram is Mermaid.** No ASCII art and no image of a diagram. +- There is no roadmap page. Open items are tracked as GitHub issues once the + repositories accept them. +- Do not add plan files, audit files, working notes, or status files to any + repository. The project has **no status document and no dated log**: the + site describes the present, and the record of what was done, how it was + proven, and what it replaced goes in the commit message that carries the + change. + +### Analysis and reporting + +- Rank engineering options on technical merit only: performance, correctness, + robustness, verifiability. Process cost (an acceptance campaign, an image + flash, a new dependency) is a one-line footnote, never a ranking factor. + The operator decides the process cost. +- Never infer or assume a hardware fact. Label every unverified claim as + unverified, and say how to verify it. +- Report only what the observations confirm. Everything else is labeled + "unconfirmed" and is not acted on. + +### General practice for agents + +- Read the README of this repository, this file, and the Developers section + of the site before you change code. +- Orient in the code before you edit. If a local knowledge graph + (`graphify-out/`) exists in the working tree or its parent, query it first. + It is a local artifact and is never committed. +- Keep a change minimal and scoped to the request. No drive-by reformatting, + no unrelated cleanups, no renames for taste. Match the style of the + surrounding code. +- Do not add a dependency without a stated reason. The target is a + single-core machine with a small eMMC, and every package rides the image. +- Extend an existing test or tool instead of adding a parallel one. +- Never create a file in the repository that the request does not need. +- Ask before anything destructive or outward-facing: a push, a force-push, a + tag, a release, a history rewrite, or any change to a machine. +- When a fact is unknown, say so and say how to find out. Never fill a gap + with a plausible number. +- Use the exact names of the project: kas, bitbake, pin, SRCREV, campaign, + fingerprint, drill, sysfs, the latch, the armed window. Do not invent names. + +### Before you commit + +1. The host tests of this repository pass, with `-Werror` where the build + uses it. +2. The change is proven at the highest level available: CI test, bench drill, + or documented reasoning. +3. For a component that ships in the image: the acceptance catalog covers the + changed behavior, or the commit message says why it has no catalog + consequence. +4. New files carry the 514 LLC d/b/a OpenGlow copyright, the `Written by` + line, and an SPDX line. Line endings are LF. +5. No em dashes, no British spellings, no workstation paths, no bench + identity, no factory firmware code, no history narrative, no AI + attribution. +6. Documentation is current: interfaces and measured facts (the currency + rule). +7. The commit is one settled change, attributed to the human author, and it + is pushed only when the operator asks. diff --git a/forgetest/tests/test_fixture.py b/forgetest/tests/test_fixture.py index cfdfa2f..c19685d 100644 --- a/forgetest/tests/test_fixture.py +++ b/forgetest/tests/test_fixture.py @@ -1,591 +1,591 @@ -"""The bench actuator as the tool sees it. - -The fixture is a box on the network that opens the lid loop, pulls the -interlock loop and presses the button on request. What has to hold: the -tool finds it by name without a resolver on the image, speaks its API -under the key, falls back to the operator when it cannot act, moves an -operator test it can perform alone into the unattended queue (never a -live one, never one that needs a person for anything else), refuses a -prompt during such a run rather than hanging on it, releases what the -box still holds after every run, and presses the arm button only where -the bench opted in. -""" -import json -import os -import shutil -import socket -import struct -import tempfile -import threading -import time -import unittest -from http.server import BaseHTTPRequestHandler, HTTPServer - -import helpers -from forgetest import fixture as fx -from forgetest import runner as runner_mod -from forgetest.log import Log -from forgetest.runner import Context, Failed, Run, Runner - -KEY = "0123456789abcdef0123456789abcdef" - - -class FakeFixture: - """The device's API, as fixture/README.md describes it.""" - - def __init__(self, button_enabled=True): - self.state = {"lid": "closed", "interlock": "closed", "button": "idle"} - self.button_enabled = button_enabled - self.calls = [] - self.presses = 0 # presses the device performed - srv = self - - class H(BaseHTTPRequestHandler): - def log_message(self, *a): - pass - - def _json(self, code, obj): - body = json.dumps(obj).encode() - self.send_response(code) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def _state(self): - return {"device": "forgefixture", "hostname": "forgefixture", "version": "1.0.0", - "idf": "v5.5.5", "uptime_s": 12, "channels": dict(srv.state), - "button_enabled": srv.button_enabled, - "button_pulsing": srv.state["button"] == "pressed", - "wifi": {"connected": True, "ip": "127.0.0.1", "rssi": -50}} - - def _auth(self): - if self.headers.get("X-Fixture-Key") != KEY: - self._json(401, {"error": "X-Fixture-Key missing or wrong"}) - return False - return True - - def do_GET(self): - srv.calls.append(("GET", self.path)) - if not self._auth(): - return - self._json(200, self._state()) - - def do_POST(self): - n = int(self.headers.get("Content-Length") or 0) - body = json.loads(self.rfile.read(n).decode() or "{}") - srv.calls.append(("POST", self.path, body)) - if not self._auth(): - return - if self.path in ("/lid", "/interlock"): - st = body.get("state") - if st not in ("open", "close", "closed"): - return self._json(400, {"error": "state must be \"open\" or \"close\""}) - srv.state[self.path[1:]] = "open" if st == "open" else "closed" - return self._json(200, self._state()) - if self.path == "/button": - if not srv.button_enabled: - return self._json(409, {"error": "button disabled: the enable jumper is out"}) - if srv.state["button"] == "pressed": - return self._json(409, {"error": "a button pulse is in progress"}) - srv.state["button"] = "pressed" - srv.presses += 1 - threading.Timer(0.05, lambda: srv.state.__setitem__("button", "idle")).start() - s = self._state() - s["pulse_ms"] = 200 - return self._json(200, s) - if self.path == "/release": - srv.state = {"lid": "closed", "interlock": "closed", "button": "idle"} - return self._json(200, self._state()) - self._json(404, {"error": "no such path"}) - - self.httpd = HTTPServer(("127.0.0.1", 0), H) - self.port = self.httpd.server_address[1] - threading.Thread(target=self.httpd.serve_forever, daemon=True).start() - - def stop(self): - self.httpd.shutdown() - self.httpd.server_close() - - -class MdnsTests(unittest.TestCase): - def test_query_packet(self): - q = fx.mdns_query("forgefixture.local") - self.assertEqual(q[:12], struct.pack("!HHHHHH", 0, 0, 1, 0, 0, 0)) - self.assertEqual(q[12:], b"\x0cforgefixture\x05local\x00" + struct.pack("!HH", 1, 0x8001)) - - def _response(self, name, ip, extra_name=None): - labels = b"".join(struct.pack("B", len(p)) + p.encode() for p in name.split(".")) + b"\x00" - hdr = struct.pack("!HHHHHH", 0, 0x8400, 0, 2 if extra_name else 1, 0, 0) - rr = labels + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton(ip) - if extra_name: - # a second answer naming the first through a compression pointer - # to offset 12 (the first name), plus its own first label - other = struct.pack("B", len(extra_name)) + extra_name.encode() + struct.pack("!H", 0xC000 | 12) - rr += other + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton("10.0.0.9") - return hdr + rr - - def test_answers_are_the_named_a_records(self): - data = self._response("forgefixture.local", "192.0.2.50") - self.assertEqual(fx.mdns_answers(data, "forgefixture.local"), ["192.0.2.50"]) - self.assertEqual(fx.mdns_answers(data, "FORGEFIXTURE.local."), ["192.0.2.50"]) - self.assertEqual(fx.mdns_answers(data, "other.local"), []) - # a query (QR clear) is never an answer - self.assertEqual(fx.mdns_answers(fx.mdns_query("forgefixture.local"), "forgefixture.local"), []) - # compression pointers are followed; the other record is not ours - data2 = self._response("forgefixture.local", "192.0.2.50", extra_name="printer") - self.assertEqual(fx.mdns_answers(data2, "forgefixture.local"), ["192.0.2.50"]) - self.assertEqual(fx.mdns_answers(data2, "printer.forgefixture.local"), ["10.0.0.9"]) - # garbage does not raise - self.assertEqual(fx.mdns_answers(b"\x00\x01", "forgefixture.local"), []) - - -class ConfigTests(unittest.TestCase): - def setUp(self): - self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-") - - def tearDown(self): - shutil.rmtree(self.tmp, ignore_errors=True) - - def write(self, obj): - p = os.path.join(self.tmp, "fixture.json") - with open(p, "w") as f: - json.dump(obj, f) - return p - - def test_no_file_is_no_fixture(self): - self.assertIsNone(fx.load_config(os.path.join(self.tmp, "none.json"))) - - def test_defaults_and_checks(self): - cfg = fx.load_config(self.write({"key": KEY})) - self.assertEqual(cfg["hostname"], "forgefixture") - self.assertEqual(cfg["channels"], ["lid", "interlock", "button"]) - self.assertFalse(cfg["arm_press"]) - self.assertIsNone(cfg["ip"]) - with self.assertRaises(fx.FixtureError): - fx.load_config(self.write({"hostname": "x"})) # no key - with self.assertRaises(fx.FixtureError): - fx.load_config(self.write({"key": KEY, "channels": ["lid", "laser"]})) - p = self.write({"key": KEY}) - with open(p, "w") as f: - f.write("{not json") - with self.assertRaises(fx.FixtureError): - fx.load_config(p) - - -class ClientTests(unittest.TestCase): - def setUp(self): - self.dev = FakeFixture() - self.resolved = [] - - def tearDown(self): - self.dev.stop() - - def resolver(self, hostname): - self.resolved.append(hostname) - return "127.0.0.1" - - def client(self, **over): - cfg = {"hostname": "forgefixture", "key": KEY, "channels": ["lid", "interlock", "button"], - "port": self.dev.port} - cfg.update(over) - return fx.Fixture(cfg, resolver=self.resolver) - - def test_status_covers_act_release(self): - f = self.client() - st = f.status() - self.assertEqual(st["device"], "forgefixture") - self.assertEqual(self.resolved, ["forgefixture"]) # resolved once - self.assertTrue(f.covers("lid") and f.covers("interlock") and f.covers("button")) - f.act("lid", "open") - self.assertEqual(self.dev.state["lid"], "open") - f.act("interlock", "open") - f.act("button", "press") - self.assertEqual(self.dev.calls[-1], ("POST", "/button", {})) - time.sleep(0.15) # the fake's pulse ends - self.assertEqual(sorted(fx.Fixture.energized(f.status())), ["interlock", "lid"]) - f.release() - self.assertEqual(fx.Fixture.energized(f.status()), []) - self.assertEqual(self.resolved, ["forgefixture"]) # the address is cached - with self.assertRaises(fx.FixtureError): - f.act("button", "hold") - with self.assertRaises(fx.FixtureError): - f.act("lid", "ajar") - - def test_the_button_is_covered_only_with_the_jumper_in(self): - self.dev.button_enabled = False - f = self.client() - f.status() - self.assertTrue(f.covers("lid")) - self.assertFalse(f.covers("button")) - with self.assertRaises(fx.FixtureError) as cm: - f.act("button", "press") - self.assertIn("jumper", str(cm.exception)) - - def test_two_presses_are_spaced_so_the_controller_sees_the_release(self): - f = self.client() - f.status() - t0 = time.time() - f.act("button", "press") - f.act("button", "press") - took = time.time() - t0 - posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] - self.assertEqual(len(posts), 2) # no 409 round trip was needed - self.assertEqual(self.dev.presses, 2) - # the second waited for the first pulse (200 ms as reported) and the gap - self.assertGreaterEqual(took, 0.2 + fx.BUTTON_GAP_S - 0.05) - - def test_a_pulse_in_progress_is_waited_out_then_retried(self): - f = self.client() - f.status() - self.dev.state["button"] = "pressed" # a press this client did not time - threading.Timer(0.3, lambda: self.dev.state.__setitem__("button", "idle")).start() - f.act("button", "press") - posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] - self.assertEqual(len(posts), 2) # the 409, then the press - self.assertEqual(self.dev.presses, 1) - - def test_a_pulse_that_never_ends_is_the_fixtures_error(self): - f = self.client() - f.status() - self.dev.state["button"] = "pressed" - with self.assertRaises(fx.FixtureError) as cm: - f.act("button", "press") - self.assertIn("in progress", str(cm.exception)) - - def test_a_wrong_key_is_refused(self): - f = self.client(key="wrong") - with self.assertRaises(fx.FixtureError) as cm: - f.status() - self.assertIn("refused the key", str(cm.exception)) - - def test_an_ip_override_skips_the_lookup(self): - f = self.client(ip="127.0.0.1") - f.status() - self.assertEqual(self.resolved, []) - - def test_channels_not_wired_are_not_covered(self): - f = self.client(channels=["lid"]) - f.status() - self.assertTrue(f.covers("lid")) - self.assertFalse(f.covers("interlock")) - - def test_probe_without_a_config_is_none_and_a_silent_box_is_logged(self): - lines = [] - self.assertIsNone(fx.probe(lines.append, path=os.path.join(tempfile.gettempdir(), "no-such-fixture.json"))) - self.assertEqual(lines, []) - tmp = tempfile.mkdtemp(prefix="forgetest-fixture-") - try: - p = os.path.join(tmp, "fixture.json") - with open(p, "w") as f: - json.dump({"key": KEY, "ip": "127.0.0.1", "port": 1}, f) - # port 1 on localhost: nothing listens there - f2 = fx.probe(lines.append, path=p, resolver=lambda h: None) - self.assertIsNone(f2) - self.assertTrue(lines and "running without it" in lines[-1], lines) - finally: - shutil.rmtree(tmp, ignore_errors=True) - - -class CatalogTests(unittest.TestCase): - def test_fixture_runnable(self): - op = helpers.make_test("m.op", [], kind="operator", actions=("lid", "button")) - self.assertTrue(op.fixture_runnable(("lid", "interlock", "button"))) - self.assertFalse(op.fixture_runnable(("lid",))) # the button is not covered - self.assertFalse(helpers.make_test("m.live", [], kind="live", actions=("lid",)) - .fixture_runnable(("lid",))) # live never downgrades - self.assertFalse(helpers.make_test("m.app", [], kind="operator", hands=("app",)) - .fixture_runnable(("lid", "interlock", "button"))) - self.assertFalse(helpers.make_test("m.none", [], kind="operator") - .fixture_runnable(("lid",))) # no actions: a person does something - self.assertFalse(helpers.make_test("m.auto", []).fixture_runnable(("lid",))) - - def test_an_auto_test_asks_nothing_of_a_person(self): - from forgetest import catalog - with self.assertRaises(ValueError): - catalog.test("z.auto", title="t", subsystem="z", hands=("app",))(lambda ctx: None) - - -class StubFixture: - """What the runner needs of a fixture, scripted.""" - - def __init__(self, channels=("lid", "interlock", "button"), button_enabled=True, arm_press=False, - fail=False, fc=None): - self.fc = fc # the fake forgectrl whose switches follow the actions - self.hostname = "forgefixture" - self._ip = "127.0.0.1" - self.channels = tuple(channels) - self.button_enabled = button_enabled - self.arm_press = arm_press - self.fail = fail - self.acts = [] - self.held = [] - self.released = 0 - - def covers(self, channel): - return channel in self.channels and (channel != "button" or self.button_enabled) - - def act(self, channel, state): - if self.fail: - raise fx.FixtureError("the box is off") - self.acts.append((channel, state)) - if state == "open": - self.held.append(channel) - if self.fc is not None and channel in ("lid", "interlock"): - sw = self.fc.state["status"]["switches"] - sw["lid" if channel == "lid" else "interlock_ok"] = state != "open" - - def status(self): - return {"channels": {c: ("open" if c in self.held else "closed") for c in ("lid", "interlock")}} - - @staticmethod - def energized(state): - return fx.Fixture.energized(state) - - def release(self): - self.released += 1 - self.held = [] - - def summary(self): - return {"hostname": self.hostname, "ip": self._ip, "channels": list(self.channels), - "button_enabled": self.button_enabled, "arm_press": self.arm_press} - - -def t_lid(ctx): - ctx.ready("On Ready the lid opens") - ctx.act("lid", "open") - ctx.act("lid", "close") - - -def t_asks(ctx): - ctx.act("lid", "open") - ctx.confirm("Did it?") - - -class RoutingTests(unittest.TestCase): - """The queues with a fixture up: the runner's probe is scripted.""" - - def setUp(self): - self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-run-") - os.environ["FORGETEST_DATA"] = self.tmp - os.environ["FORGETEST_MARKER"] = os.path.join(self.tmp, "marker") - self.fc = helpers.FakeForgectrl().start() - self.man = helpers.make_manifest() - self.reg = helpers.registry( - helpers.make_test("r.auto", [("forgectrl", "src/ui.c")]), - helpers.make_test("r.lid", [("forgectrl", "src/auth.c")], kind="operator", actions=("lid",), fn=t_lid), - helpers.make_test("r.btn", [("forgectrl", "src/cool.c")], kind="operator", actions=("button",), fn=t_lid), - helpers.make_test("r.app", [("forgectrl", "src/main.c")], kind="operator", actions=("lid",), - hands=("app",), fn=t_lid), - helpers.make_test("r.live", [("grblhal-glowforge", "src/**")], kind="live", actions=("lid",)), - helpers.make_test("r.asks", [("linux-fslc", "**")], kind="operator", actions=("lid",), fn=t_asks), - ) - self.log = Log(os.path.join(self.tmp, "results.jsonl")) - self.runner = Runner(self.log, self.man, self.reg) - self.saved_probe = fx.probe - self.stub = StubFixture(fc=self.fc) - fx.probe = lambda log, path=None, resolver=None: self.stub - - def tearDown(self): - fx.probe = self.saved_probe - self.fc.stop() - for k in ("FORGETEST_DATA", "FORGETEST_MARKER"): - os.environ.pop(k, None) - shutil.rmtree(self.tmp, ignore_errors=True) - - def selection(self): - st, _ = self.runner.state() - return st["batch_available"], st["fixture"] - - def test_operator_tests_the_fixture_covers_move_to_the_unattended_queue(self): - av, summary = self.selection() - self.assertEqual(av["unattended"], ["r.auto", "r.lid", "r.btn", "r.asks"]) - self.assertEqual(av["attended"], ["r.app", "r.live"]) # needs the app; live never moves - self.assertEqual(summary["hostname"], "forgefixture") - - def test_without_the_jumper_the_button_tests_stay_attended(self): - self.stub.button_enabled = False - av, _ = self.selection() - self.assertIn("r.btn", av["attended"]) - self.assertIn("r.lid", av["unattended"]) - - def test_no_fixture_is_the_old_routing(self): - fx.probe = lambda log, path=None, resolver=None: None - self.runner._fixture_probed = 0.0 - av, summary = self.selection() - self.assertEqual(av["unattended"], ["r.auto"]) - self.assertIsNone(summary) - - def test_a_queue_started_during_a_probe_waits_for_the_fixture(self): - # The daemon just came up: nothing probed yet, a page poll starts - # the first probe (slow: an mDNS answer), and the queue start - # lands while it is in flight. The start must see the fixture. - stub = self.stub - - def slow_probe(log, path=None, resolver=None): - time.sleep(0.5) - return stub - fx.probe = slow_probe - self.runner._fixture_probed = 0.0 - self.runner.fixture = None - poll = threading.Thread(target=self.runner.probe_fixture) - poll.start() - time.sleep(0.1) - order = self.runner.batch_selection("unattended") - poll.join() - self.assertIn("r.lid", order) - self.assertIn("r.btn", order) - - def run_queue(self, group, ack_live=False): - ok, msg, order = self.runner.start_batch(group, ack_live=ack_live) - self.assertTrue(ok, msg) - deadline = time.time() + 30 - while not self.runner.batch["finished"] and time.time() < deadline: - time.sleep(0.05) - return self.runner.batch_snapshot(), order - - def last_result(self, tid): - for rec in reversed(self.log.read()): - if rec.get("t") == "result" and rec["test"] == tid: - return rec - return None - - def test_the_fixture_performs_the_actions_and_the_ready_gate_passes(self): - # r.asks confirms after its action: a person is asked -> a FAIL that - # says so, and the queue stops there, after r.lid and r.btn passed - b, order = self.run_queue("unattended") - self.assertEqual(order, ["r.auto", "r.lid", "r.btn", "r.asks"]) - results = {x["test"]: x["result"] for x in b["done"]} - self.assertEqual(results["r.lid"], "PASS") - self.assertEqual(results["r.btn"], "PASS") - self.assertEqual(results["r.asks"], "FAIL") - rec = self.last_result("r.lid") - acts = rec["evidence"]["actions"] - self.assertEqual([(a["channel"], a["state"], a["by"]) for a in acts], - [("lid", "open", "fixture"), ("lid", "close", "fixture")]) - self.assertTrue(any("READY (fixture performs the step)" in l for l in rec["log"]), rec["log"]) - self.assertEqual(rec["evidence"]["fixture"]["channels"], ["lid", "interlock", "button"]) - asks = self.last_result("r.asks") - self.assertIn("declare the step in hands=", asks["message"]) - - def test_a_failing_box_hands_the_action_to_the_operator(self): - self.stub.fail = True - run = Run("test", "r.lid", "r.lid") - ctx = Context(run, self.runner, self.reg["r.lid"]) - self.runner.probe_fixture(force=True) - seen = [] - ctx.wait_for = lambda cond, timeout, poll=0.25: 1.0 # the machine shows the state - run.set_notice = lambda text: seen.append(text) - ctx.act("lid", "open") - rec = run.evidence["actions"][0] - self.assertEqual(rec["by"], "operator") - self.assertIn("the box is off", rec["fixture_error"]) - self.assertTrue(seen and "lid" in seen[0].lower()) - - def test_a_failing_box_in_an_unattended_run_ends_the_test_as_an_error(self): - # nobody is in the room: the run must not wait ACT_TIMEOUT_S for a - # hand that is not there; it ends at once, as the harness's error - self.stub.fail = True - t0 = time.time() - b, order = self.run_queue("unattended") - results = {x["test"]: x["result"] for x in b["done"]} - self.assertEqual(results["r.lid"], "ERROR") - self.assertLess(time.time() - t0, 20) - rec = self.last_result("r.lid") - self.assertIn("the fixture could not perform a step", rec["message"]) - self.assertIn("the box is off", rec["message"]) - act = rec["evidence"]["actions"][0] - self.assertEqual((act["by"], act["fixture_error"]), ("fixture", "the box is off")) - self.assertFalse(any("asking the operator" in l for l in rec["log"]), rec["log"]) - - def test_what_the_box_still_holds_is_released_after_a_run(self): - def holds(ctx): - ctx.act("lid", "open") # and never closes it - self.reg["r.hold"] = helpers.make_test("r.hold", [("forgectrl", "src/ui.c")], kind="operator", - actions=("lid",), fn=holds) - self.runner.registry = self.reg - ok, msg, run = self.runner._start_test("r.hold") - self.assertTrue(ok, msg) - while run.finished is None: - time.sleep(0.05) - self.assertEqual(self.stub.released, 1) - rec = self.last_result("r.hold") - self.assertEqual(rec["evidence"]["fixture"]["released"], ["lid"]) - - def press_button(self, down): - self.fc.state["status"]["switches"]["button"] = bool(down) - - def test_ready_takes_a_press_on_the_machine_as_the_presence_check(self): - """With an actuator wired to the button, the operator proves they - are at the machine by pressing it, not by clicking the page. The - actuator then owns every press in the test, so no press in a live - cut is a person's and none can go uncounted.""" - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.probe_fixture(force=True) - asked = [] - run.ask = lambda q, o: asked.append(q) - self.press_button(False) - threading.Timer(0.3, self.press_button, args=(True,)).start() - threading.Timer(0.8, self.press_button, args=(False,)).start() - ctx.ready("LIVE FIRE. Scrap under the head.") - self.assertEqual(asked, []) # no page click asked for - self.assertTrue(run.fixture_takeover) - rec = [r for r in run.evidence["actions"] if r["state"] == "presence"] - self.assertEqual(len(rec), 1) - self.assertEqual(rec[0]["by"], "operator") - self.assertTrue(any("press the button on the machine" in ln.lower() for ln in run.lines)) - - def test_presence_is_proved_once_per_test_not_once_per_gate(self): - """A test with two armed halves reaches the ready gate twice. The - second one must not ask for another press: the operator proved - presence a moment earlier and the actuator has the presses. The - setup line still goes up, because the second half may want the - scrap moved.""" - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.probe_fixture(force=True) - asked = [] - run.ask = lambda q, o: asked.append(q) - self.press_button(False) - threading.Timer(0.3, self.press_button, args=(True,)).start() - threading.Timer(0.8, self.press_button, args=(False,)).start() - ctx.ready("First half.") - self.assertTrue(run.fixture_takeover) - presses = [r for r in run.evidence["actions"] if r["state"] == "presence"] - self.assertEqual(len(presses), 1) - - # the second gate: returns at once, no new press, notice still shown - seen = [] - run.set_notice = lambda text: seen.append(text) - t0 = time.time() - ctx.ready("Second half. Move the scrap.") - self.assertLess(time.time() - t0, 1.0) - self.assertEqual(asked, []) - self.assertEqual(len([r for r in run.evidence["actions"] if r["state"] == "presence"]), 1) - self.assertIn("Second half. Move the scrap.", seen) - self.assertTrue(any("presence already proved" in ln for ln in run.lines)) - - def test_the_presence_press_hands_the_arm_press_to_the_actuator(self): - """The bench's standing opt-in is not needed once the operator has - proved presence: that press is what the opt-in existed to - establish.""" - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.probe_fixture(force=True) - self.stub.arm_press = False - run.fixture_takeover = True - saved = runner_mod.hw.button_lit - runner_mod.hw.button_lit = lambda: True - try: - self.assertTrue(ctx.arm_press()) - deadline = time.time() + 5 - while ("button", "press") not in self.stub.acts and time.time() < deadline: - time.sleep(0.05) - self.assertIn(("button", "press"), self.stub.acts) - finally: - runner_mod.hw.button_lit = saved - +"""The bench actuator as the tool sees it. + +The fixture is a box on the network that opens the lid loop, pulls the +interlock loop and presses the button on request. What has to hold: the +tool finds it by name without a resolver on the image, speaks its API +under the key, falls back to the operator when it cannot act, moves an +operator test it can perform alone into the unattended queue (never a +live one, never one that needs a person for anything else), refuses a +prompt during such a run rather than hanging on it, releases what the +box still holds after every run, and presses the arm button only where +the bench opted in. +""" +import json +import os +import shutil +import socket +import struct +import tempfile +import threading +import time +import unittest +from http.server import BaseHTTPRequestHandler, HTTPServer + +import helpers +from forgetest import fixture as fx +from forgetest import runner as runner_mod +from forgetest.log import Log +from forgetest.runner import Context, Failed, Run, Runner + +KEY = "0123456789abcdef0123456789abcdef" + + +class FakeFixture: + """The device's API, as fixture/README.md describes it.""" + + def __init__(self, button_enabled=True): + self.state = {"lid": "closed", "interlock": "closed", "button": "idle"} + self.button_enabled = button_enabled + self.calls = [] + self.presses = 0 # presses the device performed + srv = self + + class H(BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def _json(self, code, obj): + body = json.dumps(obj).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _state(self): + return {"device": "forgefixture", "hostname": "forgefixture", "version": "1.0.0", + "idf": "v5.5.5", "uptime_s": 12, "channels": dict(srv.state), + "button_enabled": srv.button_enabled, + "button_pulsing": srv.state["button"] == "pressed", + "wifi": {"connected": True, "ip": "127.0.0.1", "rssi": -50}} + + def _auth(self): + if self.headers.get("X-Fixture-Key") != KEY: + self._json(401, {"error": "X-Fixture-Key missing or wrong"}) + return False + return True + + def do_GET(self): + srv.calls.append(("GET", self.path)) + if not self._auth(): + return + self._json(200, self._state()) + + def do_POST(self): + n = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(n).decode() or "{}") + srv.calls.append(("POST", self.path, body)) + if not self._auth(): + return + if self.path in ("/lid", "/interlock"): + st = body.get("state") + if st not in ("open", "close", "closed"): + return self._json(400, {"error": "state must be \"open\" or \"close\""}) + srv.state[self.path[1:]] = "open" if st == "open" else "closed" + return self._json(200, self._state()) + if self.path == "/button": + if not srv.button_enabled: + return self._json(409, {"error": "button disabled: the enable jumper is out"}) + if srv.state["button"] == "pressed": + return self._json(409, {"error": "a button pulse is in progress"}) + srv.state["button"] = "pressed" + srv.presses += 1 + threading.Timer(0.05, lambda: srv.state.__setitem__("button", "idle")).start() + s = self._state() + s["pulse_ms"] = 200 + return self._json(200, s) + if self.path == "/release": + srv.state = {"lid": "closed", "interlock": "closed", "button": "idle"} + return self._json(200, self._state()) + self._json(404, {"error": "no such path"}) + + self.httpd = HTTPServer(("127.0.0.1", 0), H) + self.port = self.httpd.server_address[1] + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def stop(self): + self.httpd.shutdown() + self.httpd.server_close() + + +class MdnsTests(unittest.TestCase): + def test_query_packet(self): + q = fx.mdns_query("forgefixture.local") + self.assertEqual(q[:12], struct.pack("!HHHHHH", 0, 0, 1, 0, 0, 0)) + self.assertEqual(q[12:], b"\x0cforgefixture\x05local\x00" + struct.pack("!HH", 1, 0x8001)) + + def _response(self, name, ip, extra_name=None): + labels = b"".join(struct.pack("B", len(p)) + p.encode() for p in name.split(".")) + b"\x00" + hdr = struct.pack("!HHHHHH", 0, 0x8400, 0, 2 if extra_name else 1, 0, 0) + rr = labels + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton(ip) + if extra_name: + # a second answer naming the first through a compression pointer + # to offset 12 (the first name), plus its own first label + other = struct.pack("B", len(extra_name)) + extra_name.encode() + struct.pack("!H", 0xC000 | 12) + rr += other + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton("10.0.0.9") + return hdr + rr + + def test_answers_are_the_named_a_records(self): + data = self._response("forgefixture.local", "192.0.2.50") + self.assertEqual(fx.mdns_answers(data, "forgefixture.local"), ["192.0.2.50"]) + self.assertEqual(fx.mdns_answers(data, "FORGEFIXTURE.local."), ["192.0.2.50"]) + self.assertEqual(fx.mdns_answers(data, "other.local"), []) + # a query (QR clear) is never an answer + self.assertEqual(fx.mdns_answers(fx.mdns_query("forgefixture.local"), "forgefixture.local"), []) + # compression pointers are followed; the other record is not ours + data2 = self._response("forgefixture.local", "192.0.2.50", extra_name="printer") + self.assertEqual(fx.mdns_answers(data2, "forgefixture.local"), ["192.0.2.50"]) + self.assertEqual(fx.mdns_answers(data2, "printer.forgefixture.local"), ["10.0.0.9"]) + # garbage does not raise + self.assertEqual(fx.mdns_answers(b"\x00\x01", "forgefixture.local"), []) + + +class ConfigTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def write(self, obj): + p = os.path.join(self.tmp, "fixture.json") + with open(p, "w") as f: + json.dump(obj, f) + return p + + def test_no_file_is_no_fixture(self): + self.assertIsNone(fx.load_config(os.path.join(self.tmp, "none.json"))) + + def test_defaults_and_checks(self): + cfg = fx.load_config(self.write({"key": KEY})) + self.assertEqual(cfg["hostname"], "forgefixture") + self.assertEqual(cfg["channels"], ["lid", "interlock", "button"]) + self.assertFalse(cfg["arm_press"]) + self.assertIsNone(cfg["ip"]) + with self.assertRaises(fx.FixtureError): + fx.load_config(self.write({"hostname": "x"})) # no key + with self.assertRaises(fx.FixtureError): + fx.load_config(self.write({"key": KEY, "channels": ["lid", "laser"]})) + p = self.write({"key": KEY}) + with open(p, "w") as f: + f.write("{not json") + with self.assertRaises(fx.FixtureError): + fx.load_config(p) + + +class ClientTests(unittest.TestCase): + def setUp(self): + self.dev = FakeFixture() + self.resolved = [] + + def tearDown(self): + self.dev.stop() + + def resolver(self, hostname): + self.resolved.append(hostname) + return "127.0.0.1" + + def client(self, **over): + cfg = {"hostname": "forgefixture", "key": KEY, "channels": ["lid", "interlock", "button"], + "port": self.dev.port} + cfg.update(over) + return fx.Fixture(cfg, resolver=self.resolver) + + def test_status_covers_act_release(self): + f = self.client() + st = f.status() + self.assertEqual(st["device"], "forgefixture") + self.assertEqual(self.resolved, ["forgefixture"]) # resolved once + self.assertTrue(f.covers("lid") and f.covers("interlock") and f.covers("button")) + f.act("lid", "open") + self.assertEqual(self.dev.state["lid"], "open") + f.act("interlock", "open") + f.act("button", "press") + self.assertEqual(self.dev.calls[-1], ("POST", "/button", {})) + time.sleep(0.15) # the fake's pulse ends + self.assertEqual(sorted(fx.Fixture.energized(f.status())), ["interlock", "lid"]) + f.release() + self.assertEqual(fx.Fixture.energized(f.status()), []) + self.assertEqual(self.resolved, ["forgefixture"]) # the address is cached + with self.assertRaises(fx.FixtureError): + f.act("button", "hold") + with self.assertRaises(fx.FixtureError): + f.act("lid", "ajar") + + def test_the_button_is_covered_only_with_the_jumper_in(self): + self.dev.button_enabled = False + f = self.client() + f.status() + self.assertTrue(f.covers("lid")) + self.assertFalse(f.covers("button")) + with self.assertRaises(fx.FixtureError) as cm: + f.act("button", "press") + self.assertIn("jumper", str(cm.exception)) + + def test_two_presses_are_spaced_so_the_controller_sees_the_release(self): + f = self.client() + f.status() + t0 = time.time() + f.act("button", "press") + f.act("button", "press") + took = time.time() - t0 + posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] + self.assertEqual(len(posts), 2) # no 409 round trip was needed + self.assertEqual(self.dev.presses, 2) + # the second waited for the first pulse (200 ms as reported) and the gap + self.assertGreaterEqual(took, 0.2 + fx.BUTTON_GAP_S - 0.05) + + def test_a_pulse_in_progress_is_waited_out_then_retried(self): + f = self.client() + f.status() + self.dev.state["button"] = "pressed" # a press this client did not time + threading.Timer(0.3, lambda: self.dev.state.__setitem__("button", "idle")).start() + f.act("button", "press") + posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] + self.assertEqual(len(posts), 2) # the 409, then the press + self.assertEqual(self.dev.presses, 1) + + def test_a_pulse_that_never_ends_is_the_fixtures_error(self): + f = self.client() + f.status() + self.dev.state["button"] = "pressed" + with self.assertRaises(fx.FixtureError) as cm: + f.act("button", "press") + self.assertIn("in progress", str(cm.exception)) + + def test_a_wrong_key_is_refused(self): + f = self.client(key="wrong") + with self.assertRaises(fx.FixtureError) as cm: + f.status() + self.assertIn("refused the key", str(cm.exception)) + + def test_an_ip_override_skips_the_lookup(self): + f = self.client(ip="127.0.0.1") + f.status() + self.assertEqual(self.resolved, []) + + def test_channels_not_wired_are_not_covered(self): + f = self.client(channels=["lid"]) + f.status() + self.assertTrue(f.covers("lid")) + self.assertFalse(f.covers("interlock")) + + def test_probe_without_a_config_is_none_and_a_silent_box_is_logged(self): + lines = [] + self.assertIsNone(fx.probe(lines.append, path=os.path.join(tempfile.gettempdir(), "no-such-fixture.json"))) + self.assertEqual(lines, []) + tmp = tempfile.mkdtemp(prefix="forgetest-fixture-") + try: + p = os.path.join(tmp, "fixture.json") + with open(p, "w") as f: + json.dump({"key": KEY, "ip": "127.0.0.1", "port": 1}, f) + # port 1 on localhost: nothing listens there + f2 = fx.probe(lines.append, path=p, resolver=lambda h: None) + self.assertIsNone(f2) + self.assertTrue(lines and "running without it" in lines[-1], lines) + finally: + shutil.rmtree(tmp, ignore_errors=True) + + +class CatalogTests(unittest.TestCase): + def test_fixture_runnable(self): + op = helpers.make_test("m.op", [], kind="operator", actions=("lid", "button")) + self.assertTrue(op.fixture_runnable(("lid", "interlock", "button"))) + self.assertFalse(op.fixture_runnable(("lid",))) # the button is not covered + self.assertFalse(helpers.make_test("m.live", [], kind="live", actions=("lid",)) + .fixture_runnable(("lid",))) # live never downgrades + self.assertFalse(helpers.make_test("m.app", [], kind="operator", hands=("app",)) + .fixture_runnable(("lid", "interlock", "button"))) + self.assertFalse(helpers.make_test("m.none", [], kind="operator") + .fixture_runnable(("lid",))) # no actions: a person does something + self.assertFalse(helpers.make_test("m.auto", []).fixture_runnable(("lid",))) + + def test_an_auto_test_asks_nothing_of_a_person(self): + from forgetest import catalog + with self.assertRaises(ValueError): + catalog.test("z.auto", title="t", subsystem="z", hands=("app",))(lambda ctx: None) + + +class StubFixture: + """What the runner needs of a fixture, scripted.""" + + def __init__(self, channels=("lid", "interlock", "button"), button_enabled=True, arm_press=False, + fail=False, fc=None): + self.fc = fc # the fake forgectrl whose switches follow the actions + self.hostname = "forgefixture" + self._ip = "127.0.0.1" + self.channels = tuple(channels) + self.button_enabled = button_enabled + self.arm_press = arm_press + self.fail = fail + self.acts = [] + self.held = [] + self.released = 0 + + def covers(self, channel): + return channel in self.channels and (channel != "button" or self.button_enabled) + + def act(self, channel, state): + if self.fail: + raise fx.FixtureError("the box is off") + self.acts.append((channel, state)) + if state == "open": + self.held.append(channel) + if self.fc is not None and channel in ("lid", "interlock"): + sw = self.fc.state["status"]["switches"] + sw["lid" if channel == "lid" else "interlock_ok"] = state != "open" + + def status(self): + return {"channels": {c: ("open" if c in self.held else "closed") for c in ("lid", "interlock")}} + + @staticmethod + def energized(state): + return fx.Fixture.energized(state) + + def release(self): + self.released += 1 + self.held = [] + + def summary(self): + return {"hostname": self.hostname, "ip": self._ip, "channels": list(self.channels), + "button_enabled": self.button_enabled, "arm_press": self.arm_press} + + +def t_lid(ctx): + ctx.ready("On Ready the lid opens") + ctx.act("lid", "open") + ctx.act("lid", "close") + + +def t_asks(ctx): + ctx.act("lid", "open") + ctx.confirm("Did it?") + + +class RoutingTests(unittest.TestCase): + """The queues with a fixture up: the runner's probe is scripted.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-run-") + os.environ["FORGETEST_DATA"] = self.tmp + os.environ["FORGETEST_MARKER"] = os.path.join(self.tmp, "marker") + self.fc = helpers.FakeForgectrl().start() + self.man = helpers.make_manifest() + self.reg = helpers.registry( + helpers.make_test("r.auto", [("forgectrl", "src/ui.c")]), + helpers.make_test("r.lid", [("forgectrl", "src/auth.c")], kind="operator", actions=("lid",), fn=t_lid), + helpers.make_test("r.btn", [("forgectrl", "src/cool.c")], kind="operator", actions=("button",), fn=t_lid), + helpers.make_test("r.app", [("forgectrl", "src/main.c")], kind="operator", actions=("lid",), + hands=("app",), fn=t_lid), + helpers.make_test("r.live", [("grblhal-glowforge", "src/**")], kind="live", actions=("lid",)), + helpers.make_test("r.asks", [("linux-fslc", "**")], kind="operator", actions=("lid",), fn=t_asks), + ) + self.log = Log(os.path.join(self.tmp, "results.jsonl")) + self.runner = Runner(self.log, self.man, self.reg) + self.saved_probe = fx.probe + self.stub = StubFixture(fc=self.fc) + fx.probe = lambda log, path=None, resolver=None: self.stub + + def tearDown(self): + fx.probe = self.saved_probe + self.fc.stop() + for k in ("FORGETEST_DATA", "FORGETEST_MARKER"): + os.environ.pop(k, None) + shutil.rmtree(self.tmp, ignore_errors=True) + + def selection(self): + st, _ = self.runner.state() + return st["batch_available"], st["fixture"] + + def test_operator_tests_the_fixture_covers_move_to_the_unattended_queue(self): + av, summary = self.selection() + self.assertEqual(av["unattended"], ["r.auto", "r.lid", "r.btn", "r.asks"]) + self.assertEqual(av["attended"], ["r.app", "r.live"]) # needs the app; live never moves + self.assertEqual(summary["hostname"], "forgefixture") + + def test_without_the_jumper_the_button_tests_stay_attended(self): + self.stub.button_enabled = False + av, _ = self.selection() + self.assertIn("r.btn", av["attended"]) + self.assertIn("r.lid", av["unattended"]) + + def test_no_fixture_is_the_old_routing(self): + fx.probe = lambda log, path=None, resolver=None: None + self.runner._fixture_probed = 0.0 + av, summary = self.selection() + self.assertEqual(av["unattended"], ["r.auto"]) + self.assertIsNone(summary) + + def test_a_queue_started_during_a_probe_waits_for_the_fixture(self): + # The daemon just came up: nothing probed yet, a page poll starts + # the first probe (slow: an mDNS answer), and the queue start + # lands while it is in flight. The start must see the fixture. + stub = self.stub + + def slow_probe(log, path=None, resolver=None): + time.sleep(0.5) + return stub + fx.probe = slow_probe + self.runner._fixture_probed = 0.0 + self.runner.fixture = None + poll = threading.Thread(target=self.runner.probe_fixture) + poll.start() + time.sleep(0.1) + order = self.runner.batch_selection("unattended") + poll.join() + self.assertIn("r.lid", order) + self.assertIn("r.btn", order) + + def run_queue(self, group, ack_live=False): + ok, msg, order = self.runner.start_batch(group, ack_live=ack_live) + self.assertTrue(ok, msg) + deadline = time.time() + 30 + while not self.runner.batch["finished"] and time.time() < deadline: + time.sleep(0.05) + return self.runner.batch_snapshot(), order + + def last_result(self, tid): + for rec in reversed(self.log.read()): + if rec.get("t") == "result" and rec["test"] == tid: + return rec + return None + + def test_the_fixture_performs_the_actions_and_the_ready_gate_passes(self): + # r.asks confirms after its action: a person is asked -> a FAIL that + # says so, and the queue stops there, after r.lid and r.btn passed + b, order = self.run_queue("unattended") + self.assertEqual(order, ["r.auto", "r.lid", "r.btn", "r.asks"]) + results = {x["test"]: x["result"] for x in b["done"]} + self.assertEqual(results["r.lid"], "PASS") + self.assertEqual(results["r.btn"], "PASS") + self.assertEqual(results["r.asks"], "FAIL") + rec = self.last_result("r.lid") + acts = rec["evidence"]["actions"] + self.assertEqual([(a["channel"], a["state"], a["by"]) for a in acts], + [("lid", "open", "fixture"), ("lid", "close", "fixture")]) + self.assertTrue(any("READY (fixture performs the step)" in l for l in rec["log"]), rec["log"]) + self.assertEqual(rec["evidence"]["fixture"]["channels"], ["lid", "interlock", "button"]) + asks = self.last_result("r.asks") + self.assertIn("declare the step in hands=", asks["message"]) + + def test_a_failing_box_hands_the_action_to_the_operator(self): + self.stub.fail = True + run = Run("test", "r.lid", "r.lid") + ctx = Context(run, self.runner, self.reg["r.lid"]) + self.runner.probe_fixture(force=True) + seen = [] + ctx.wait_for = lambda cond, timeout, poll=0.25: 1.0 # the machine shows the state + run.set_notice = lambda text: seen.append(text) + ctx.act("lid", "open") + rec = run.evidence["actions"][0] + self.assertEqual(rec["by"], "operator") + self.assertIn("the box is off", rec["fixture_error"]) + self.assertTrue(seen and "lid" in seen[0].lower()) + + def test_a_failing_box_in_an_unattended_run_ends_the_test_as_an_error(self): + # nobody is in the room: the run must not wait ACT_TIMEOUT_S for a + # hand that is not there; it ends at once, as the harness's error + self.stub.fail = True + t0 = time.time() + b, order = self.run_queue("unattended") + results = {x["test"]: x["result"] for x in b["done"]} + self.assertEqual(results["r.lid"], "ERROR") + self.assertLess(time.time() - t0, 20) + rec = self.last_result("r.lid") + self.assertIn("the fixture could not perform a step", rec["message"]) + self.assertIn("the box is off", rec["message"]) + act = rec["evidence"]["actions"][0] + self.assertEqual((act["by"], act["fixture_error"]), ("fixture", "the box is off")) + self.assertFalse(any("asking the operator" in l for l in rec["log"]), rec["log"]) + + def test_what_the_box_still_holds_is_released_after_a_run(self): + def holds(ctx): + ctx.act("lid", "open") # and never closes it + self.reg["r.hold"] = helpers.make_test("r.hold", [("forgectrl", "src/ui.c")], kind="operator", + actions=("lid",), fn=holds) + self.runner.registry = self.reg + ok, msg, run = self.runner._start_test("r.hold") + self.assertTrue(ok, msg) + while run.finished is None: + time.sleep(0.05) + self.assertEqual(self.stub.released, 1) + rec = self.last_result("r.hold") + self.assertEqual(rec["evidence"]["fixture"]["released"], ["lid"]) + + def press_button(self, down): + self.fc.state["status"]["switches"]["button"] = bool(down) + + def test_ready_takes_a_press_on_the_machine_as_the_presence_check(self): + """With an actuator wired to the button, the operator proves they + are at the machine by pressing it, not by clicking the page. The + actuator then owns every press in the test, so no press in a live + cut is a person's and none can go uncounted.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + asked = [] + run.ask = lambda q, o: asked.append(q) + self.press_button(False) + threading.Timer(0.3, self.press_button, args=(True,)).start() + threading.Timer(0.8, self.press_button, args=(False,)).start() + ctx.ready("LIVE FIRE. Scrap under the head.") + self.assertEqual(asked, []) # no page click asked for + self.assertTrue(run.fixture_takeover) + rec = [r for r in run.evidence["actions"] if r["state"] == "presence"] + self.assertEqual(len(rec), 1) + self.assertEqual(rec[0]["by"], "operator") + self.assertTrue(any("press the button on the machine" in ln.lower() for ln in run.lines)) + + def test_presence_is_proved_once_per_test_not_once_per_gate(self): + """A test with two armed halves reaches the ready gate twice. The + second one must not ask for another press: the operator proved + presence a moment earlier and the actuator has the presses. The + setup line still goes up, because the second half may want the + scrap moved.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + asked = [] + run.ask = lambda q, o: asked.append(q) + self.press_button(False) + threading.Timer(0.3, self.press_button, args=(True,)).start() + threading.Timer(0.8, self.press_button, args=(False,)).start() + ctx.ready("First half.") + self.assertTrue(run.fixture_takeover) + presses = [r for r in run.evidence["actions"] if r["state"] == "presence"] + self.assertEqual(len(presses), 1) + + # the second gate: returns at once, no new press, notice still shown + seen = [] + run.set_notice = lambda text: seen.append(text) + t0 = time.time() + ctx.ready("Second half. Move the scrap.") + self.assertLess(time.time() - t0, 1.0) + self.assertEqual(asked, []) + self.assertEqual(len([r for r in run.evidence["actions"] if r["state"] == "presence"]), 1) + self.assertIn("Second half. Move the scrap.", seen) + self.assertTrue(any("presence already proved" in ln for ln in run.lines)) + + def test_the_presence_press_hands_the_arm_press_to_the_actuator(self): + """The bench's standing opt-in is not needed once the operator has + proved presence: that press is what the opt-in existed to + establish.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + self.stub.arm_press = False + run.fixture_takeover = True + saved = runner_mod.hw.button_lit + runner_mod.hw.button_lit = lambda: True + try: + self.assertTrue(ctx.arm_press()) + deadline = time.time() + 5 + while ("button", "press") not in self.stub.acts and time.time() < deadline: + time.sleep(0.05) + self.assertIn(("button", "press"), self.stub.acts) + finally: + runner_mod.hw.button_lit = saved + def test_the_press_prompt_is_what_the_actuator_presses_on(self): """The machine's own press prompt, not a button LED read from here: the button is lit through parts of a live check that are @@ -621,92 +621,92 @@ class RoutingTests(unittest.TestCase): self.assertNotIn(("button", "press"), self.stub.acts) self.assertIn("Press it now.", seen) - def test_the_arm_press_waits_as_long_as_the_card_says_for_the_light(self): - """A card that settles the coolant before it lights the button - takes minutes; the caller names the wait, and the actuator presses - when the light comes instead of handing the press to a person at - the default 60 s (found on the bench by the flow-load card).""" - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.probe_fixture(force=True) - run.fixture_takeover = True - lit_at = time.time() + 0.6 - saved = runner_mod.hw.button_lit - runner_mod.hw.button_lit = lambda: time.time() >= lit_at - seen = [] - run.set_notice = lambda text: seen.append(text) - try: - self.assertTrue(ctx.arm_press(lit_timeout=5)) - deadline = time.time() + 5 - while ("button", "press") not in self.stub.acts and time.time() < deadline: - time.sleep(0.05) - self.assertIn(("button", "press"), self.stub.acts) - self.assertEqual([n for n in seen if n], []) # nobody was asked - rec = run.evidence["actions"][-1] - self.assertEqual(rec["by"], "fixture") - self.assertGreaterEqual(rec["took_s"], 0.5) - finally: - runner_mod.hw.button_lit = saved - - def test_an_actuator_lost_after_the_takeover_is_said_out_loud(self): - """Falling back to the operator without a word is how one dropped - actuator becomes a press nobody can account for afterwards.""" - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.probe_fixture(force=True) - run.fixture_takeover = True - self.runner.fixture = None # the box drops off mid-test - self.fc.state["status"]["switches"]["lid"] = True - threading.Timer(0.3, lambda: self.fc.state["status"]["switches"].__setitem__("lid", False)).start() - ctx.act("lid", "open", timeout=5) - rec = run.evidence["actions"][-1] - self.assertTrue(rec.get("fixture_lost")) - self.assertEqual(rec["by"], "operator") - self.assertTrue(any("WARNING" in ln and "now gone" in ln for ln in run.lines)) - - def test_ready_still_asks_the_page_with_no_actuator(self): - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.fixture = None - asked = [] - - def ask(q, o): - asked.append((q, list(o))) - return "Ready" - run.ask = ask - ctx.ready("LIVE FIRE. Scrap under the head.") - self.assertEqual(asked, [("LIVE FIRE. Scrap under the head.", ["Ready", "Cannot"])]) - self.assertFalse(run.fixture_takeover) - - def test_the_arm_press_is_the_operators_unless_the_bench_opted_in(self): - run = Run("test", "r.live", "r.live") - ctx = Context(run, self.runner, self.reg["r.live"]) - self.runner.probe_fixture(force=True) - seen = [] - run.set_notice = lambda text: seen.append(text) - self.assertFalse(ctx.arm_press()) - self.assertEqual(run.evidence["actions"][0]["by"], "operator") - self.assertTrue(seen) - # opted in: the press waits for the button to light - self.stub.arm_press = True - lit = {"v": False} - saved = runner_mod.hw.button_lit - runner_mod.hw.button_lit = lambda: lit["v"] - try: - seen[:] = [] - self.assertTrue(ctx.arm_press()) - time.sleep(0.3) - self.assertEqual(self.stub.acts, []) # not yet: dark button - lit["v"] = True - deadline = time.time() + 5 - while not self.stub.acts and time.time() < deadline: - time.sleep(0.05) - self.assertEqual(self.stub.acts, [("button", "press")]) - self.assertEqual(run.evidence["actions"][1]["by"], "fixture") - self.assertEqual(seen, []) # no notice went up - finally: - runner_mod.hw.button_lit = saved - - -if __name__ == "__main__": - unittest.main() + def test_the_arm_press_waits_as_long_as_the_card_says_for_the_light(self): + """A card that settles the coolant before it lights the button + takes minutes; the caller names the wait, and the actuator presses + when the light comes instead of handing the press to a person at + the default 60 s (found on the bench by the flow-load card).""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + run.fixture_takeover = True + lit_at = time.time() + 0.6 + saved = runner_mod.hw.button_lit + runner_mod.hw.button_lit = lambda: time.time() >= lit_at + seen = [] + run.set_notice = lambda text: seen.append(text) + try: + self.assertTrue(ctx.arm_press(lit_timeout=5)) + deadline = time.time() + 5 + while ("button", "press") not in self.stub.acts and time.time() < deadline: + time.sleep(0.05) + self.assertIn(("button", "press"), self.stub.acts) + self.assertEqual([n for n in seen if n], []) # nobody was asked + rec = run.evidence["actions"][-1] + self.assertEqual(rec["by"], "fixture") + self.assertGreaterEqual(rec["took_s"], 0.5) + finally: + runner_mod.hw.button_lit = saved + + def test_an_actuator_lost_after_the_takeover_is_said_out_loud(self): + """Falling back to the operator without a word is how one dropped + actuator becomes a press nobody can account for afterwards.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + run.fixture_takeover = True + self.runner.fixture = None # the box drops off mid-test + self.fc.state["status"]["switches"]["lid"] = True + threading.Timer(0.3, lambda: self.fc.state["status"]["switches"].__setitem__("lid", False)).start() + ctx.act("lid", "open", timeout=5) + rec = run.evidence["actions"][-1] + self.assertTrue(rec.get("fixture_lost")) + self.assertEqual(rec["by"], "operator") + self.assertTrue(any("WARNING" in ln and "now gone" in ln for ln in run.lines)) + + def test_ready_still_asks_the_page_with_no_actuator(self): + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.fixture = None + asked = [] + + def ask(q, o): + asked.append((q, list(o))) + return "Ready" + run.ask = ask + ctx.ready("LIVE FIRE. Scrap under the head.") + self.assertEqual(asked, [("LIVE FIRE. Scrap under the head.", ["Ready", "Cannot"])]) + self.assertFalse(run.fixture_takeover) + + def test_the_arm_press_is_the_operators_unless_the_bench_opted_in(self): + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + seen = [] + run.set_notice = lambda text: seen.append(text) + self.assertFalse(ctx.arm_press()) + self.assertEqual(run.evidence["actions"][0]["by"], "operator") + self.assertTrue(seen) + # opted in: the press waits for the button to light + self.stub.arm_press = True + lit = {"v": False} + saved = runner_mod.hw.button_lit + runner_mod.hw.button_lit = lambda: lit["v"] + try: + seen[:] = [] + self.assertTrue(ctx.arm_press()) + time.sleep(0.3) + self.assertEqual(self.stub.acts, []) # not yet: dark button + lit["v"] = True + deadline = time.time() + 5 + while not self.stub.acts and time.time() < deadline: + time.sleep(0.05) + self.assertEqual(self.stub.acts, [("button", "press")]) + self.assertEqual(run.evidence["actions"][1]["by"], "fixture") + self.assertEqual(seen, []) # no notice went up + finally: + runner_mod.hw.button_lit = saved + + +if __name__ == "__main__": + unittest.main()