mariano/soleprint
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updates 38.1 90

Browse Source
This commit is contained in:
buenosairesam
2026-09-14 06:03:48 -03:00
parent 4fe3718795
commit 5ff513fcc5
22 changed files with 1776 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
ctrl/spr.py
View File

@@ -73,7 +73,15 @@ def get_version(comp_type, source):
for line in (source / "pyproject.toml").read_text().splitlines(): for line in (source / "pyproject.toml").read_text().splitlines():
if line.strip().startswith("version"): if line.strip().startswith("version"):
return line.split('"')[1] return line.split('"')[1]
except FileNotFoundError: elif comp_type == "python":
# A plain package: no package.json, no pyproject.toml, just
# __version__ in __init__.py. station/tools/common is one -- it is
# imported and vendored, never installed, so giving it packaging
# metadata to satisfy this function would be the tail wagging the dog.
for line in (source / "__init__.py").read_text().splitlines():
if line.strip().startswith("__version__"):
return line.split('"')[1] if '"' in line else line.split("'")[1]
except (FileNotFoundError, IndexError):
pass pass
return "?" return "?"

10
registry.json
View File

@@ -1,10 +1,14 @@
{ {
"soleprint-ui": { "soleprint-cli": {
"type": "npm", "type": "python",
"path": "soleprint/common/ui" "path": "soleprint/station/tools/common"
}, },
"soleprint-modelgen": { "soleprint-modelgen": {
"type": "pip", "type": "pip",
"path": "soleprint/station/tools/modelgen" "path": "soleprint/station/tools/modelgen"
},
"soleprint-ui": {
"type": "npm",
"path": "soleprint/common/ui"
} }
} }

2
soleprint/station/tools/common/__init__.py
View File

@@ -12,3 +12,5 @@ it is circular.
""" """
__all__ = ["cli"] __all__ = ["cli"]
__version__ = "0.1.0"

21
soleprint/station/tools/common/cli.py
View File

@@ -65,7 +65,14 @@ Conventions it encodes, so they stop being re-decided:
Progress goes to stdout as plain print(). Errors go to stderr prefixed Progress goes to stdout as plain print(). Errors go to stderr prefixed
'Error: ' and exit 1 — never a traceback, which tells a user nothing they 'Error: ' and exit 1 — never a traceback, which tells a user nothing they
can act on. `fail()` is the only exit path. can act on. `fail()` is the only exit path for an *error*; a command that
ends badly but not wrongly (tests failed, drift found) returns an int and
`Tool.run` exits with it.
An alias for a flag that is `required=True` must be a SECOND OPTION STRING on
the same add_argument, never a second argument with the same dest. argparse
checks `required` per action, so a separate alias action leaves the real one
unsatisfied and can never be used. See shuntgen's --source/-s/--spec.
Heavy imports live inside the cmd_* function, never at module top, so Heavy imports live inside the cmd_* function, never at module top, so
`--help` stays instant and an optional dependency only costs the one `--help` stays instant and an optional dependency only costs the one
@@ -280,9 +287,19 @@ class Tool:
self._commands[verb].add_argument(*flags, **spec) self._commands[verb].add_argument(*flags, **spec)
def run(self, argv=None): def run(self, argv=None):
"""Dispatch, and turn the ways a command ends into exit codes.
A `cmd_*` returning a non-zero int exits with it. That is for an outcome
that is not an error — tester's "3 tests failed" is a true report, not a
malfunction, and routing it through fail() would print "Error:" in front
of a result the user asked for. Returning None, which almost every
command does, exits 0.
"""
args = self.parser.parse_args(argv) args = self.parser.parse_args(argv)
try: try:
args.func(args) result = args.func(args)
if isinstance(result, int) and result != 0:
sys.exit(result)
except KeyboardInterrupt: except KeyboardInterrupt:
# Ctrl-C is a decision, not a crash. Say so and leave quietly. # Ctrl-C is a decision, not a crash. Say so and leave quietly.
print("\nInterrupted.", file=sys.stderr) print("\nInterrupted.", file=sys.stderr)

84
soleprint/station/tools/common/selftest.py
View File

@@ -14,6 +14,7 @@ that an unknown registry name is refused -- not just that a known one loads.
python3 -m station.tools.common.selftest # from soleprint/ python3 -m station.tools.common.selftest # from soleprint/
python3 selftest.py # or from this directory python3 selftest.py # or from this directory
python3 selftest.py --revendor # push cli.py to the copies
Stdlib only. Nothing is written outside a temp directory. Stdlib only. Nothing is written outside a temp directory.
""" """
@@ -30,7 +31,14 @@ TOOLS = HERE.parent
# Tools that vendor a copy of cli.py rather than importing it. Adding one here # Tools that vendor a copy of cli.py rather than importing it. Adding one here
# is how its copy starts being checked; see cli.py's docstring for why the two # is how its copy starts being checked; see cli.py's docstring for why the two
# arrangements coexist. # arrangements coexist.
VENDORS = ("histgen",) VENDORS = ("histgen", "keymatrix")
# Tools that have adopted the scaffold. Every one gets the same assertions, so
# "adopted" is checkable rather than claimed. docgen and graphgen are NOT here
# and are not oversights: docgen is a library fronted by a Makefile of
# `python3 -c` one-liners, graphgen is a mounted FastAPI router. Neither has a
# CLI, and bolting one on to finish a sweep is the failure this list prevents.
ADOPTED = ("histgen", "shuntgen", "tester", "keymatrix")
def run_py(body, *, extra_path=None): def run_py(body, *, extra_path=None):
@@ -51,6 +59,33 @@ def run_py(body, *, extra_path=None):
) )
def split_docstring(path):
"""(docstring source, everything after it) for a module."""
src = Path(path).read_text(encoding="utf-8")
tree = ast.parse(src)
node = tree.body[0]
if not (isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant)):
raise SystemExit(f"{path} has no module docstring")
lines = src.split("\n")
return "\n".join(lines[:node.end_lineno]), "\n".join(lines[node.end_lineno:])
def revendor():
"""Give every vendored copy common/cli.py's code, keeping its own docstring.
The alternative is doing it by hand every time cli.py changes, which is how
the copies drift in the first place. The check below is what makes the
omission loud; this is what makes fixing it a command.
"""
_, source_body = split_docstring(HERE / "cli.py")
for tool in VENDORS:
copy = TOOLS / tool / "cli.py"
own_doc, _ = split_docstring(copy)
copy.write_text(own_doc + source_body, encoding="utf-8")
print(f" re-vendored {copy.relative_to(TOOLS)}")
return 0
def code_only(path): def code_only(path):
"""A file's AST with the module docstring dropped. """A file's AST with the module docstring dropped.
@@ -122,6 +157,20 @@ def main():
"t = Tool('x', 'd'); t.command('go', boom, 'h'); t.run(['go'])") "t = Tool('x', 'd'); t.command('go', boom, 'h'); t.run(['go'])")
check("Tool.run: BrokenPipeError -> exit 0", r.returncode == 0) check("Tool.run: BrokenPipeError -> exit 0", r.returncode == 0)
# A command that ends badly but not wrongly returns an int; fail() is for
# errors. tester's "3 tests failed" is a result, not a malfunction.
r = run_py("from common.cli import Tool\n"
"t = Tool('x', 'd'); t.command('go', lambda a: 3, 'h'); t.run(['go'])")
check("Tool.run: a non-zero int return becomes the exit code",
r.returncode == 3, f"exit {r.returncode}")
check("Tool.run: an int return prints no 'Error:'", "Error:" not in r.stderr)
r = run_py("from common.cli import Tool\n"
"t = Tool('x', 'd'); t.command('go', lambda a: 0, 'h'); t.run(['go'])")
check("Tool.run: returning 0 exits 0", r.returncode == 0)
r = run_py("from common.cli import Tool\n"
"t = Tool('x', 'd'); t.command('go', lambda a: None, 'h'); t.run(['go'])")
check("Tool.run: returning None exits 0", r.returncode == 0)
# ── a bare invocation must not report success ──────────────────── # ── a bare invocation must not report success ────────────────────
r = run_py("from common.cli import Tool\n" r = run_py("from common.cli import Tool\n"
"t = Tool('x', 'd'); t.command('go', lambda a: None, 'h'); t.run([])") "t = Tool('x', 'd'); t.command('go', lambda a: None, 'h'); t.run([])")
@@ -205,6 +254,37 @@ print("LEAKED", len(leaked))
and want in r.stderr and "Traceback" not in r.stderr, and want in r.stderr and "Traceback" not in r.stderr,
r.stderr.strip()) r.stderr.strip())
# ── every adopted tool behaves the same way ─────────────────────
soleprint = TOOLS.parent.parent
for tool in ADOPTED:
def as_tool(*argv, cwd=None):
return subprocess.run(
[sys.executable, "-m", f"station.tools.{tool}", *argv],
cwd=str(cwd or soleprint), capture_output=True, text=True,
env={"PATH": "/usr/bin:/bin", "HOME": str(Path.home()),
"PYTHONPATH": str(soleprint)})
r = as_tool()
check(f"{tool}: bare invocation exits non-zero",
r.returncode != 0, f"exit {r.returncode} -- a no-op reporting success")
check(f"{tool}: bare invocation puts usage on stderr, not stdout",
"usage:" in r.stderr and not r.stdout.strip(), r.stdout[:80])
r = as_tool("--help")
check(f"{tool}: --help exits 0", r.returncode == 0, r.stderr[:200])
check(f"{tool}: --help usage line names the module, not __main__.py",
f"python -m station.tools.{tool}" in r.stdout,
r.stdout.splitlines()[:1])
# --help must not read the filesystem. A poisoned .env in the working
# directory is the cheap version of that: tester used to import config.py
# at module scope, which reads .env, so --help died on a bad one.
with tempfile.TemporaryDirectory(prefix=f"{tool}-help-") as poison:
(Path(poison) / ".env").write_text("NOT = VALID = =\n\x00binary\n")
r = as_tool("--help", cwd=poison)
check(f"{tool}: --help does not read .env from the working directory",
r.returncode == 0, r.stderr[:200])
# ── the name `common` is taken one level up ───────────────────── # ── the name `common` is taken one level up ─────────────────────
# Not hypothetical: soleprint/common/ holds auth, theme and ui. Tools must # Not hypothetical: soleprint/common/ holds auth, theme and ui. Tools must
# reach the scaffold relatively (`from ..common.cli import Tool`) or by its # reach the scaffold relatively (`from ..common.cli import Tool`) or by its
@@ -255,4 +335,4 @@ print("LEAKED", len(leaked))
if __name__ == "__main__": if __name__ == "__main__":
sys.exit(main()) sys.exit(revendor() if "--revendor" in sys.argv else main())

12
soleprint/station/tools/histgen/cli.py
View File

@@ -240,9 +240,19 @@ class Tool:
self._commands[verb].add_argument(*flags, **spec) self._commands[verb].add_argument(*flags, **spec)
def run(self, argv=None): def run(self, argv=None):
"""Dispatch, and turn the ways a command ends into exit codes.
A `cmd_*` returning a non-zero int exits with it. That is for an outcome
that is not an error — tester's "3 tests failed" is a true report, not a
malfunction, and routing it through fail() would print "Error:" in front
of a result the user asked for. Returning None, which almost every
command does, exits 0.
"""
args = self.parser.parse_args(argv) args = self.parser.parse_args(argv)
try: try:
args.func(args) result = args.func(args)
if isinstance(result, int) and result != 0:
sys.exit(result)
except KeyboardInterrupt: except KeyboardInterrupt:
# Ctrl-C is a decision, not a crash. Say so and leave quietly. # Ctrl-C is a decision, not a crash. Say so and leave quietly.
print("\nInterrupted.", file=sys.stderr) print("\nInterrupted.", file=sys.stderr)

50
soleprint/station/tools/keymatrix/Makefile Normal file
View File

@@ -0,0 +1,50 @@
# keymatrix — one target per verb.
#
# The folder is meant to be copied out of soleprint and used on its own — it has
# to be, since it must run on ethbox's WSL, which has no spr on its path. So
# everything here is derived from where this file sits rather than written down:
# copy the directory anywhere, `cd` into it, and `make` works.
#
# make probe OUT=~/wdir/semester/peripheral-matrix read this machine
# make check SRC=~/wdir/semester/peripheral-matrix findings
# make shadow SRC=... the posture table
# make shadow SRC=... ARGS="--posture A" one posture
#
# There is no `make apply`, and there is no apply verb to wrap. nrft is managed
# by hand: report recipes, do not apply them.
#
# The logic lives in the Python, never here. Each target is one invocation.
HERE := $(patsubst %/,%,$(dir $(abspath $(lastword $(MAKEFILE_LIST)))))
PKG := $(notdir $(HERE))
PARENT := $(patsubst %/,%,$(dir $(HERE)))
PY ?= python3
KM := PYTHONPATH=$(PARENT) $(PY) -m $(PKG)
OUT ?= .
SRC ?= .
ARGS ?=
.PHONY: probe check shadow doctor help
help:
@sed -n '2,17p' $(lastword $(MAKEFILE_LIST)) | sed 's/^# \{0,1\}//'
probe:
$(KM) probe --output $(OUT) $(ARGS)
check:
$(KM) check --source $(SRC) $(ARGS)
shadow:
$(KM) shadow --source $(SRC) $(ARGS)
# Asks about the machine, never fails — rig's `make doctor` level.
doctor:
@echo "python : $$($(PY) --version)"
@echo "gsettings: $$(command -v gsettings || echo MISSING)"
@echo "dconf : $$(command -v dconf || echo MISSING)"
@echo "localectl: $$(command -v localectl || echo MISSING)"
@echo "session : $${XDG_SESSION_TYPE:-unknown} / $${XDG_CURRENT_DESKTOP:-unknown}"
@echo "endpoint : $$(PYTHONPATH=$(PARENT) $(PY) -c 'from $(PKG).endpoint import resolve_endpoint, platform_kind; print(resolve_endpoint(), "/", platform_kind())')"

28
soleprint/station/tools/keymatrix/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
"""
keymatrix - which endpoint actually receives a chord, given how you are connected.
The keyboard slice of semester/peripheral-matrix, sharing its vocabulary. Not a
`*gen` tool: it probes and checks, and only renders on request.
Chords are declared in four incompatible places across this fleet -- GNOME dconf
(configured by hand through a GUI, documented only as prose), AHK inline in .ahk
files, AutoKey per-item JSON, espanso YAML -- with no single source of truth and
drift already present. The harder half is that a chord's destination depends on
POSTURE: with the peripherals on nrft you VNC out to Windows and GNOME eats
first; with them on ethbox you RDP back into nrft and the arbitration inverts.
probe read one endpoint's live state into observed/<endpoint>.json
check drift, dead commands, collisions, doc drift
shadow the posture table: per action, which endpoint receives it
render recipes and CSV rows, never applied
**There is no apply verb.** Not a disabled one, not one behind a flag. The
owner's standing rule is that nrft is managed by hand: report recipes, do not
apply them. A tool that cannot write cannot write the wrong thing.
Stdlib only, no network. The directory can be copied out of soleprint and run on
its own -- it has to be, since it must run on ethbox's WSL, which has no spr on
its path.
"""
__version__ = "0.1.0"

136
soleprint/station/tools/keymatrix/__main__.py Normal file
View File

@@ -0,0 +1,136 @@
"""
keymatrix CLI.
python -m station.tools.keymatrix probe --output ~/wdir/semester/peripheral-matrix
python -m station.tools.keymatrix shadow --posture A
python -m station.tools.keymatrix check
Run from the soleprint/ directory so `station.tools...` resolves, or copy the
folder out and use its Makefile.
"""
import json
from pathlib import Path
from .cli import Tool, fail
def _load_observed(args):
"""Every observed/<endpoint>.json under --source, keyed by endpoint."""
root = Path(args.source).expanduser() if args.source else Path("observed")
if root.name != "observed":
root = root / "observed"
if not root.is_dir():
fail(f"no observed state at {root}",
"Run `keymatrix probe --output <dir>` first.")
out = {}
for path in sorted(root.glob("*.json")):
doc = json.loads(path.read_text())
index = {}
for b in doc.get("bindings", []):
index.setdefault(b["chord"], b) # first declaration wins
out[doc["endpoint"]] = index
if not out:
fail(f"{root} holds no observed state")
return out
def _load_docs(args):
"""The full observed documents, which the checks need (not the chord index)."""
root = Path(args.source).expanduser() if args.source else Path("observed")
if root.name != "observed":
root = root / "observed"
if not root.is_dir():
fail(f"no observed state at {root}",
"Run `keymatrix probe --output <dir>` first.")
docs = {}
for path in sorted(root.glob("*.json")):
doc = json.loads(path.read_text())
docs[doc["endpoint"]] = doc
if not docs:
fail(f"{root} holds no observed state")
return docs
def cmd_probe(args):
from . import probe
from .endpoint import resolve_endpoint
name = args.name or resolve_endpoint()
doc, warnings = probe.run(name)
for severity, message in warnings:
print(f" {severity}: {message}")
if args.dry_run:
print(json.dumps(doc, indent=2))
return
root = Path(args.output).expanduser() if args.output else Path(".")
out = root / "observed"
out.mkdir(parents=True, exist_ok=True)
path = out / f"{name}.json"
path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n")
print(f"Wrote {path} ({len(doc['bindings'])} bindings, "
f"{len(doc['unreadable'])} unreadable)")
def cmd_shadow(args):
from .chord import Chord
from . import shadow as sh
observed = _load_observed(args)
postures = [args.posture] if args.posture else sorted(sh.POSTURES)
for name in postures:
p = sh.POSTURES[name]
print(f"\nPosture {name} — {p['summary']}")
print(f" peripherals on {p['peripherals']}; "
f"chain: {' -> '.join(h['near'] + '/' + h['transport'] for h in p['hops'])}")
print()
print(f" {'chord':<22}{'verdict':<20}{'eaten by'}")
print(" " + "-" * 72)
for raw in args.chord or sorted({c for e in observed.values() for c in e}):
rows = sh.walk(Chord.parse(raw), name, observed)
hop, v = rows[0]
print(f" {raw:<22}{v.outcome:<20}{v.eater or '—'}")
def cmd_check(args):
from .check import run_checks
findings = run_checks(_load_docs(args))
for f in findings:
print(f" {f['severity']:<6} {f['kind']:<22} {f['message']}")
errors = sum(f["severity"] == "error" for f in findings)
print(f"\n{len(findings)} finding(s), {errors} at error severity")
return 1 if errors else 0
def main(argv=None):
tool = Tool("keymatrix", __doc__, package=__package__)
tool.command("probe", cmd_probe, "Read this endpoint's live keyboard state")
tool.common("probe", "output", "name", "dry_run",
output=dict(default=None,
help="Directory holding observed/ (default: here)"),
name=dict(default=None,
help="Endpoint name (default: resolved from the host)"))
tool.command("shadow", cmd_shadow, "Which endpoint receives a chord, per posture")
tool.common("shadow", "source",
source=dict(default=None, help="Directory holding observed/"))
tool.argument("shadow", "--posture", choices=("A", "B"), default=None,
help="Only this posture (default: both)")
tool.argument("shadow", "--chord", action="append", default=None,
help="Only this chord. Repeatable.")
tool.command("check", cmd_check, "Drift, dead commands, collisions")
tool.common("check", "source",
source=dict(default=None, help="Directory holding observed/"))
tool.run(argv)
if __name__ == "__main__":
main()

191
soleprint/station/tools/keymatrix/check.py Normal file
View File

@@ -0,0 +1,191 @@
"""
Findings, from observed state alone.
None of these needs a declared keymap. They are the checks that can run on the
first probe of a machine, which is the point -- the peripheral matrix's own next
step was "confirm the blank cells", and these confirm them mechanically instead.
Severities: error is something that is wrong now; warn is something that will be
wrong later; note is a fact worth stating that is nobody's fault.
"""
import os
import subprocess
from collections import defaultdict
from pathlib import Path
from .chord import Chord
# Directories that hold a second copy of a tree GNOME binds into. Declared
# rather than discovered, because "is this the checkout you are editing?" is a
# question about intent that no filesystem inspection answers.
CHECKOUTS = [
{"name": "dm",
"canonical": "/home/mariano/wdir/semester/dm",
"duplicates": ["/home/mariano/wdir/dm"]},
]
def _git(root, *args):
r = subprocess.run(["git", "-C", str(root), *args],
capture_output=True, text=True)
return r.stdout.strip() if r.returncode == 0 else None
def _finding(severity, kind, message, **extra):
return {"severity": severity, "kind": kind, "message": message, **extra}
def check_collisions(endpoint, bindings):
"""Two things on one chord, on one endpoint."""
by_chord = defaultdict(list)
for b in bindings:
by_chord[b["chord"]].append(b)
out = []
for chord, hits in sorted(by_chord.items()):
if len(hits) < 2:
continue
# GNOME legitimately lists one action under several chords; the reverse
# -- several actions under one chord -- is the collision.
sources = sorted({h["source"] for h in hits})
if len(sources) < 2:
continue
out.append(_finding("warn", "collision",
f"{endpoint}: {chord} is claimed by {len(sources)} "
f"bindings: {', '.join(sources)}"))
return out
def check_commands(endpoint, bindings):
"""A binding that runs something, pointed at something that is not there."""
out = []
for b in bindings:
command = b.get("command")
if not command or not command.startswith("/"):
continue
target = Path(command.split()[0])
if not target.exists():
out.append(_finding("error", "dead-command",
f"{endpoint}: {b['chord']} ({b.get('label')}) runs "
f"{target}, which does not exist"))
continue
if not os.access(target, os.X_OK):
out.append(_finding("error", "not-executable",
f"{endpoint}: {b['chord']} runs {target}, "
f"which is not executable"))
for entry in CHECKOUTS:
for dup in entry["duplicates"]:
if not str(target).startswith(dup.rstrip("/") + "/"):
continue
rel = str(target)[len(dup.rstrip("/")) + 1:]
other = Path(entry["canonical"]) / rel
# The check computes its own evidence rather than asserting the
# copies differ: today they do not, and a finding that cried wolf
# would be switched off before the day they do.
same_commit = (_git(dup, "rev-parse", "HEAD")
== _git(entry["canonical"], "rev-parse", "HEAD"))
same_bytes = (other.exists()
and other.read_bytes() == target.read_bytes())
if same_commit and same_bytes:
out.append(_finding(
"warn", "wrong-checkout",
f"{endpoint}: {b['chord']} runs {target}, in the '{dup}' "
f"copy rather than {entry['canonical']}. Same commit and "
f"byte-identical today, so nothing is broken -- but every "
f"edit to the canonical tree silently does not take effect."))
else:
out.append(_finding(
"error", "wrong-checkout",
f"{endpoint}: {b['chord']} runs {target} from the stale "
f"copy '{dup}', and it NO LONGER matches "
f"{entry['canonical']} (same_commit={same_commit}, "
f"same_bytes={same_bytes})."))
repo = target.parent
status = _git(repo, "status", "--porcelain", "--", str(target))
if status is not None and status.startswith("??"):
out.append(_finding("warn", "untracked-source",
f"{endpoint}: {b['chord']} runs {target.name}, "
f"which is in no commit. The chord works; the "
f"thing it runs exists only on this disk."))
return out
def check_keyd(endpoint, doc):
"""keyd emitting into the void, and keyd fighting the compositor."""
out = []
layer = doc.get("layers", {}).get("keyd")
if not layer or not layer.get("rules"):
return out
declared = {b["chord"] for b in doc.get("bindings", [])}
for rule in layer["rules"]:
if rule["to_chord"] not in declared:
out.append(_finding(
"warn", "keyd-dead-output",
f"{endpoint}: keyd maps {rule['from_chord']} -> "
f"{rule['to_chord']}, and nothing consumes {rule['to_chord']}. "
f"Either bind it in GNOME or drop the rule."))
if rule["from_chord"] in declared:
out.append(_finding(
"error", "keyd-below-compositor",
f"{endpoint}: keyd remaps {rule['from_chord']}, which GNOME also "
f"binds. keyd acts below the compositor, so this is an "
f"unfixable double-binding."))
if layer.get("same_as_authoritative") == "unknown":
out.append(_finding("note", "keyd-unverified",
f"{endpoint}: keyd rules were read from "
f"{layer.get('config_source')}, not from the "
f"authoritative root-only config."))
return out
def check_layout(endpoint, doc):
"""Two live authorities on layout, and chords that move between them."""
out = []
inp = doc.get("input") or {}
sources, opts = inp.get("sources") or [], inp.get("xkb_options") or []
lc = inp.get("localectl") or {}
if len(sources) > 1 and lc.get("x11_layout") and "," not in lc["x11_layout"]:
out.append(_finding(
"note", "layout-authority-split",
f"{endpoint}: GNOME lists {len(sources)} input sources "
f"({', '.join(s[1] for s in sources)}) while localectl reports one "
f"X11 layout ({lc['x11_layout']}). Both are live and they disagree."))
if len(sources) > 1 and not any(o.startswith("grp:") for o in opts):
out.append(_finding(
"note", "no-layout-switch-chord",
f"{endpoint}: {len(sources)} layouts with no grp: option, so "
f"Super+space is the only way to switch -- and Super+space is itself "
f"shadowed on the VNC link, so the far machine's layout cannot be "
f"changed from inside a fullscreen viewer."))
for b in doc.get("bindings", []):
try:
chord = Chord.parse(b["chord"])
except ValueError:
continue
if chord.layout_sensitive and len(sources) > 1:
out.append(_finding(
"note", "layout-sensitive-chord",
f"{endpoint}: {b['chord']} ({b['source']}) is on a punctuation "
f"key, which moves between the {len(sources)} layouts in rotation."))
return out
def run_checks(observed_docs):
"""`observed_docs` maps endpoint -> the full observed document."""
findings = []
for endpoint, doc in sorted(observed_docs.items()):
bindings = doc.get("bindings", [])
findings += check_collisions(endpoint, bindings)
findings += check_commands(endpoint, bindings)
findings += check_keyd(endpoint, doc)
findings += check_layout(endpoint, doc)
order = {"error": 0, "warn": 1, "note": 2}
findings.sort(key=lambda f: (order[f["severity"]], f["kind"], f["message"]))
return findings

224
soleprint/station/tools/keymatrix/chord.py Normal file
View File

@@ -0,0 +1,224 @@
"""
One normal form for a chord, and a parser/emitter per dialect.
Four systems in this fleet declare chords, and no two spell them alike:
GNOME dconf <Control><Super>c <Primary><Alt>1 <Super>Page_Up
AHK ^#c ^!1 #3
AutoKey ["<ctrl>","<super>"] + "w"
keyd leftcontrol+leftshift+f11 -> M-f11
A normal form is not tidiness here; it is the only way to ask whether two
declarations are the same chord. GNOME's own output is not internally
consistent -- `<Control><Super>c` and `<Super><Control>1` are both live in the
same session -- so even comparing GNOME to GNOME needs one.
Chord.parse_gnome("<Primary><Alt>1") -> Chord(ctrl, alt, key="1")
str(that) -> "Ctrl+Alt+1"
that.to_ahk() -> "^!1"
Modifier order in the normal form is fixed (Ctrl, Alt, Shift, Super) and is
arbitrary -- what matters is that it is the same every time.
"""
import re
# The four modifiers, in normal-form order.
MODS = ("Ctrl", "Alt", "Shift", "Super")
# Every spelling of a modifier that appears anywhere in the fleet.
# <Primary> is GNOME's own alias for <Control> and both are live in dconf.
_MOD_ALIASES = {
"ctrl": "Ctrl", "control": "Ctrl", "primary": "Ctrl", "leftcontrol": "Ctrl",
"rightcontrol": "Ctrl", "c": "Ctrl",
"alt": "Alt", "leftalt": "Alt", "rightalt": "Alt", "mod1": "Alt", "a": "Alt",
"shift": "Shift", "leftshift": "Shift", "rightshift": "Shift", "s": "Shift",
"super": "Super", "win": "Super", "meta": "Super", "mod4": "Super",
"leftmeta": "Super", "rightmeta": "Super", "m": "Super", "logo": "Super",
}
# AHK's prefix characters.
_AHK_MODS = {"^": "Ctrl", "!": "Alt", "+": "Shift", "#": "Super"}
_AHK_EMIT = {v: k for k, v in _AHK_MODS.items()}
# Key-name spellings that mean the same physical key. X keysym names are the
# canonical side, because dconf is the largest declaration site and the one
# that cannot be rewritten (it is hand-managed).
_KEY_ALIASES = {
"pgup": "Page_Up", "pageup": "Page_Up", "prior": "Page_Up",
"kp_prior": "Page_Up",
"pgdn": "Page_Down", "pagedown": "Page_Down", "next": "Page_Down",
"kp_next": "Page_Down",
"esc": "Escape", "return": "Return", "enter": "Return",
# (punctuation aliases stay verbatim; see _PUNCT_KEYSYMS below)
"del": "Delete", "ins": "Insert", "spc": "space", "spacebar": "space",
}
# X keysym names for punctuation. These are the keys that move between the us,
# latam and ru layouts nrft rotates, and they do not look like punctuation --
# `equal` and `bracketleft` are words. A length check alone misses every one.
_PUNCT_KEYSYMS = {
"equal", "minus", "plus", "underscore", "grave", "asciitilde",
"bracketleft", "bracketright", "braceleft", "braceright",
"semicolon", "colon", "apostrophe", "quotedbl",
"comma", "period", "less", "greater", "slash", "question",
"backslash", "bar", "exclam", "at", "numbersign", "dollar", "percent",
"asciicircum", "ampersand", "asterisk", "parenleft", "parenright",
}
class Chord:
"""A set of modifiers and one key, comparable across dialects."""
__slots__ = ("mods", "key")
def __init__(self, mods, key):
self.mods = tuple(m for m in MODS if m in set(mods))
self.key = key
# -- equality and hashing are the whole point --------------------------
def __eq__(self, other):
return (isinstance(other, Chord)
and self.mods == other.mods and self.key == other.key)
def __hash__(self):
return hash((self.mods, self.key))
def __repr__(self):
return f"Chord({str(self)!r})"
def __str__(self):
if not self.key: # `<Super>` alone: GNOME's overlay-key
return "+".join(self.mods)
return "+".join((*self.mods, self.key))
@property
def is_bare_modifier(self):
"""`<Super>` alone -- GNOME's overlay-key. A real binding, no key."""
return not self.key
@property
def layout_sensitive(self):
"""True if the key moves between us / latam / ru.
Letters, digits, F-keys and named keys are in the same place in all
three; punctuation is not. nrft rotates four layouts with an empty
xkb-options, so this is not hypothetical -- `<Alt><Super>equal` is live.
"""
k = self.key
if not k:
return False
if k.lower() in _PUNCT_KEYSYMS:
return True # `equal`, `bracketleft` -- words, but punctuation
if len(k) > 1:
return False # named keys: Tab, Page_Up, F11, XF86...
return not k.isalnum()
# -- parsing -----------------------------------------------------------
@classmethod
def _norm_key(cls, key):
"""Canonical spelling of one key name.
F-keys are upper-cased because the two sides of the *same keyd line*
disagree: keyd writes `f11`, GNOME writes `F11`. Without this, asking
"does anything consume what keyd emits?" compares Super+f11 against
Super+F11 and always answers no -- which happens to be the right answer
today, for the wrong reason, and would stay right after someone fixed it.
"""
if not key:
return ""
if re.fullmatch(r"[fF]\d{1,2}", key):
return "F" + key[1:]
return _KEY_ALIASES.get(key.lower(), key)
@classmethod
def parse_gnome(cls, text):
"""`<Control><Super>c`, `<Primary><Alt>1`, `<Super>`, `XF86Keyboard`."""
mods, rest = [], text
for raw in re.findall(r"<([^>]+)>", text):
name = _MOD_ALIASES.get(raw.lower())
if name is None:
raise ValueError(f"unknown GNOME modifier <{raw}> in {text!r}")
mods.append(name)
rest = re.sub(r"<[^>]+>", "", text)
return cls(mods, cls._norm_key(rest))
@classmethod
def parse_ahk(cls, text):
"""`^!1`, `^#c`, `^+#c`, `#3`.
AHK also allows the prefixes `*` `~` `$` and the side markers `<` `>`.
They change *how* a hotkey fires, not which chord it is, so they are
stripped -- but `parse_ahk` is not the place that decides a hotkey was
unparseable. The scanner reports what it could not read.
"""
text = text.strip()
for noise in ("*", "~", "$", "<", ">"):
text = text.replace(noise, "")
mods, i = [], 0
while i < len(text) and text[i] in _AHK_MODS:
mods.append(_AHK_MODS[text[i]])
i += 1
return cls(mods, cls._norm_key(text[i:]))
@classmethod
def parse_autokey(cls, modifiers, key):
"""AutoKey's JSON: `["<ctrl>", "<super>"]` plus `"w"`."""
mods = []
for raw in modifiers or ():
name = _MOD_ALIASES.get(raw.strip("<>").lower())
if name is None:
raise ValueError(f"unknown AutoKey modifier {raw!r}")
mods.append(name)
return cls(mods, cls._norm_key(key or ""))
@classmethod
def parse_keyd(cls, text):
"""keyd's two spellings: `leftcontrol+leftshift+f11` and `M-f11`.
keyd writes a *source* as plus-separated names and a *destination* in
emacs-style `C-`/`M-`/`A-`/`S-` prefixes. Both appear in the one config
on this machine, on opposite sides of the same line.
"""
text = text.strip()
if "-" in text and "+" not in text:
*prefixes, key = text.split("-")
mods = []
for p in prefixes:
name = _MOD_ALIASES.get(p.lower())
if name is None:
raise ValueError(f"unknown keyd modifier {p!r} in {text!r}")
mods.append(name)
return cls(mods, cls._norm_key(key))
parts = text.split("+")
mods = [_MOD_ALIASES[p.lower()] for p in parts[:-1]
if p.lower() in _MOD_ALIASES]
if len(mods) != len(parts) - 1:
raise ValueError(f"unknown keyd modifier in {text!r}")
return cls(mods, cls._norm_key(parts[-1]))
@classmethod
def parse(cls, text):
"""The normal form itself: `Ctrl+Super+c`."""
parts = [p for p in text.split("+") if p != ""]
if text.endswith("+"): # a literal plus as the key
parts.append("+")
mods, key = [], ""
for i, p in enumerate(parts):
name = _MOD_ALIASES.get(p.lower())
if name is not None and i < len(parts) - 1:
mods.append(name)
else:
key = p
return cls(mods, cls._norm_key(key))
# -- emitting ----------------------------------------------------------
def to_gnome(self):
return "".join(f"<{'Control' if m == 'Ctrl' else m}>" for m in self.mods) + self.key
def to_ahk(self):
return "".join(_AHK_EMIT[m] for m in self.mods) + self.key

236
soleprint/station/tools/keymatrix/cli.py Normal file
View File

@@ -0,0 +1,236 @@
"""
The shape a station tool's command line has - vendored copy.
keymatrix has to run on ethbox's WSL and, in principle, on an endpoint that has
never heard of soleprint, so it cannot import this from `..common`. The source
is `station/tools/common/cli.py`; `python3 ../common/selftest.py --revendor`
pushes changes here, and common/selftest.py fails if the two drift.
"""
import argparse
import importlib
import sys
# The flags worth having exactly one spelling of. A tool adds its own on top;
# it does not redefine these.
COMMON = {
"source": dict(flags=("--source", "-s"), help="What to read."),
"output": dict(flags=("--output", "-o"), help="Where to write."),
"name": dict(flags=("--name", "-n"), help="What to call it."),
"force": dict(flags=("--force", "-f"), action="store_true",
help="Write even if the destination is occupied."),
"dry_run": dict(flags=("--dry-run",), action="store_true",
help="Print what would happen; change nothing."),
}
def fail(message, hint=None):
"""The only way out on error: a line a user can act on, never a traceback."""
print(f"Error: {message}", file=sys.stderr)
if hint:
print(f" {hint}", file=sys.stderr)
sys.exit(1)
def refuse_to_clobber(path, force, marker, what):
"""
Regenerating is fine; overwriting something we did not write is not.
Lifted from shuntgen, which refuses a non-empty output directory unless it
carries the file its own generator leaves behind. The check is cheap and the
failure it prevents — silently eating a directory someone hand-wrote — is
not recoverable from.
"""
if force or not path.exists():
return
if not any(path.iterdir()):
return
if (path / marker).exists():
return
fail(f"{path} already exists and was not written by {what}.",
"Pick another path, or pass --force to write into it anyway.")
class Registry:
"""
A name -> implementation map that imports nothing in order to list itself.
The problem it solves, measured rather than assumed:
import modelgen 119 modules, 49 ms
import graphgen 1 module, 1 ms
import histgen 1 module, 3 ms
modelgen's `GENERATORS` is a dict of twelve names to nine *classes*, all
imported at the top of `generator/__init__.py`. So `modelgen --help` pays
for csv, zipfile, lzma, bz2, xml and ast -- the ODS extractor's machinery --
before it prints a line of text. That breaks the deferred-import rule this
module's docstring states, and it breaks it at the registry rather than
inside a cmd_*, which is why moving one import does not fix it.
The shape that works is a name -> **import path** map resolved on use,
instead of a name -> **class** dict resolved on import. `Profile.available()`
in atlas2/docgen reached the same answer from the other direction, by
globbing files and importing nothing to say what exists.
GENERATORS = Registry({
"pydantic": ".pydantic:PydanticGenerator",
"django": ".django:DjangoGenerator",
"typescript": ".typescript:TypeScriptGenerator",
}, package=__name__, what="generator",
aliases={"ts": "typescript"})
GENERATORS.names() # no import -- safe in help text and choices=
GENERATORS.load("ts") # imports exactly one module
Aliases are first-class because the real registry has them: `ts`, `proto`
and `jsonschema` are second names for generators that already exist. Keeping
them separate from `entries` is what lets `names()` list each generator once
while `choices()` still accepts every spelling a user might type.
"""
def __init__(self, entries, *, package=None, what="target", aliases=None):
self._entries = dict(entries)
self._aliases = dict(aliases or {})
self._package = package
self._what = what
self._loaded = {}
unknown = {a: t for a, t in self._aliases.items() if t not in self._entries}
if unknown:
# A typo here would surface as a user-facing "unknown generator" for
# a name the registry plainly lists. Cheaper to refuse at import.
raise ValueError(
f"alias target(s) not in the registry: {sorted(unknown.items())}")
# -- listing: none of this imports anything --------------------------
def names(self):
"""Canonical names, one per implementation."""
return sorted(self._entries)
def choices(self):
"""Every accepted spelling, for argparse `choices=`."""
return sorted(set(self._entries) | set(self._aliases))
def aliases_of(self, name):
return sorted(a for a, t in self._aliases.items() if t == name)
def help(self, lead=None):
"""A help string listing what exists, without importing any of it."""
parts = []
for name in self.names():
alts = self.aliases_of(name)
parts.append(f"{name} ({', '.join(alts)})" if alts else name)
listed = ", ".join(parts)
return f"{lead} {listed}." if lead else listed
def __contains__(self, name):
return name in self._entries or name in self._aliases
def __iter__(self):
return iter(self.names())
def __len__(self):
return len(self._entries)
# -- resolution ------------------------------------------------------
def resolve(self, name):
"""Follow an alias to its canonical name. Imports nothing."""
if name in self._aliases:
return self._aliases[name]
if name in self._entries:
return name
fail(f"unknown {self._what}: {name!r}",
f"Known: {', '.join(self.choices())}")
def load(self, name):
"""The implementation itself. Imports exactly one module, once."""
canonical = self.resolve(name)
if canonical not in self._loaded:
target = self._entries[canonical]
module_path, _, attr = target.partition(":")
if not attr:
fail(f"malformed registry entry for {canonical!r}: {target!r}",
'Expected "module.path:AttributeName".')
try:
module = importlib.import_module(module_path, self._package)
except ImportError as exc:
# An optional dependency is a different failure from a typo, and
# a traceback here would bury which of the two it is.
fail(f"{self._what} {canonical!r} could not be loaded: {exc}",
"Its dependency is probably not installed.")
try:
self._loaded[canonical] = getattr(module, attr)
except AttributeError:
fail(f"{module_path} has no {attr!r}, which {canonical!r} names.")
return self._loaded[canonical]
def __getitem__(self, name):
return self.load(name)
def _prog(package, name):
"""
How this tool was actually invoked, for the usage line.
The folder is meant to be copied out and run on its own, so a usage line
hardcoding `python -m station.tools.histgen` is wrong the moment it is —
it names a path that does not exist on the machine reading it.
"""
return f"python -m {package or name}"
class Tool:
"""A tool's whole command line: subcommands, shared flags, one exit path."""
def __init__(self, name, description, package=None):
self.name = name
self.parser = argparse.ArgumentParser(
prog=_prog(package, name),
description=(description or "").strip().split("\n\n")[0],
formatter_class=argparse.RawDescriptionHelpFormatter,
)
self._subparsers = self.parser.add_subparsers(dest="command", required=True)
self._commands = {}
def command(self, verb, func, help_text):
"""Register a subcommand. `func` takes parsed args and returns None."""
sub = self._subparsers.add_parser(verb, help=help_text, description=help_text)
sub.set_defaults(func=func)
self._commands[verb] = sub
return sub
def argument(self, verb, *args, **kwargs):
"""Add a positional or flag to one subcommand."""
self._commands[verb].add_argument(*args, **kwargs)
def common(self, verb, *names, **overrides):
"""Add shared flags by name, so their spelling is decided in one place."""
for key in names:
spec = dict(COMMON[key])
flags = spec.pop("flags")
spec.update(overrides.get(key, {}))
self._commands[verb].add_argument(*flags, **spec)
def run(self, argv=None):
"""Dispatch, and turn the ways a command ends into exit codes.
A `cmd_*` returning a non-zero int exits with it. That is for an outcome
that is not an error — tester's "3 tests failed" is a true report, not a
malfunction, and routing it through fail() would print "Error:" in front
of a result the user asked for. Returning None, which almost every
command does, exits 0.
"""
args = self.parser.parse_args(argv)
try:
result = args.func(args)
if isinstance(result, int) and result != 0:
sys.exit(result)
except KeyboardInterrupt:
# Ctrl-C is a decision, not a crash. Say so and leave quietly.
print("\nInterrupted.", file=sys.stderr)
sys.exit(130)
except BrokenPipeError:
# `... | head` closes the pipe early; that is the caller's business.
sys.exit(0)

55
soleprint/station/tools/keymatrix/endpoint.py Normal file
View File

@@ -0,0 +1,55 @@
"""
Which endpoint am I running on.
Same shape as `dm/dmapp/dmcore/config.py:resolve_device()` -- env var, then a
config file, then the hostname -- with one extension the fleet forces:
**WSL Debian inside ethbox is also called `ethbox`.** `peripheral-matrix/
endpoints.csv` already lists `ethbox-windows` and `ethbox-wsl` as separate rows
because they have entirely different probe surfaces, and a hostname cannot tell
them apart. `/proc/sys/kernel/osrelease` can.
"""
import os
import platform
import socket
from pathlib import Path
CONFIG_NAME = "keymatrix.json"
def is_wsl():
try:
return "microsoft" in Path("/proc/sys/kernel/osrelease").read_text().lower()
except OSError:
return False
def resolve_endpoint(config_path=None):
name = os.environ.get("KEYMATRIX_ENDPOINT")
if not name:
path = Path(config_path) if config_path else Path(__file__).parent / CONFIG_NAME
if path.exists():
import json
try:
name = json.loads(path.read_text()).get("endpoint") or ""
except ValueError:
name = ""
if not name:
name = socket.gethostname().split(".")[0]
if is_wsl():
name += "-wsl"
return name
def platform_kind():
"""What probe backend suits this machine, if none was asked for."""
if os.environ.get("KEYMATRIX_PROBE"):
return os.environ["KEYMATRIX_PROBE"]
if platform.system() == "Windows":
return "ahk"
if is_wsl():
return "wsl"
if os.environ.get("XDG_CURRENT_DESKTOP", "").upper().find("GNOME") >= 0:
return "gnome"
return "unknown"

55
soleprint/station/tools/keymatrix/probe/__init__.py Normal file
View File

@@ -0,0 +1,55 @@
"""
Read an endpoint's live keyboard state. Read-only, never sudo.
Backend selection follows `dm/dmapp/dmcore/backends/__init__.py` -- one env var,
`KEYMATRIX_PROBE` -- with one deliberate deviation: the default is auto-detected
from the platform rather than fixed. dmcore has one common case; this fleet has
four, and a wrong fixed default here produces an *empty* observed document,
which is worse than an error because it looks like an answer.
"""
import datetime
from ..endpoint import platform_kind
def run(endpoint, kind=None):
"""(observed document, warnings) for this machine."""
kind = kind or platform_kind()
warnings = []
def report(severity, message):
warnings.append((severity, message))
doc = {
"schema": 1,
"endpoint": endpoint,
"probed_at": datetime.datetime.now(datetime.timezone.utc)
.replace(microsecond=0).isoformat(),
"probe_kind": kind,
"platform": {},
"bindings": [],
"layers": {},
"input": {},
"transports": {},
# An absent fact, recorded as absent. The peripheral matrix leaves cells
# blank on purpose for the same reason: a blank must never be read as
# "no issue".
"unreadable": [],
}
if kind == "gnome":
from . import gnome, keyd, xkb, deskflow
doc["platform"] = gnome.session()
doc["bindings"], notes = gnome.collect(report)
doc["platform"]["gnome_notes"] = notes
doc["layers"]["keyd"] = keyd.collect(report, doc["unreadable"])
doc["input"] = xkb.collect(report)
doc["transports"]["deskflow"] = deskflow.collect(report)
else:
report("warn", f"no probe backend for {kind!r}; nothing was collected")
doc["unreadable"].append(
{"what": f"the whole endpoint ({kind})",
"reason": "no backend implemented for this platform yet"})
return doc, warnings

49
soleprint/station/tools/keymatrix/probe/deskflow.py Normal file
View File

@@ -0,0 +1,49 @@
"""
Deskflow's topology, from its own server config.
Deskflow's built-in hotkey facility is unused here (`hotkeys\\size=0`), which is
worth recording: adding a fifth place that declares chords would be the opposite
of this tool's point.
"""
import re
from pathlib import Path
CONF = Path.home() / ".config" / "Deskflow" / "deskflow-server.conf"
APP = Path.home() / ".config" / "Deskflow" / "Deskflow.conf"
def collect(report):
out = {"present": CONF.exists(), "links": {}, "options": {}, "hotkeys": None}
if not out["present"]:
return out
text = CONF.read_text()
block = re.search(r"section:\s*links(.*?)end", text, re.S)
if block:
current = None
for line in block.group(1).split("\n"):
line = line.strip()
if not line:
continue
if line.endswith(":"):
current = line[:-1].strip()
out["links"][current] = {}
elif "=" in line and current:
side, peer = (p.strip() for p in line.split("=", 1))
out["links"][current][side] = peer
opts = re.search(r"section:\s*options(.*?)end", text, re.S)
if opts:
for line in opts.group(1).split("\n"):
if "=" in line:
k, v = (p.strip() for p in line.split("=", 1))
out["options"][k] = v
if APP.exists():
m = re.search(r"hotkeys\\size=(\d+)", APP.read_text())
out["hotkeys"] = int(m.group(1)) if m else None
if out["hotkeys"] == 0:
report("note", "deskflow's own hotkey facility is unused "
"(hotkeys\\size=0), which is the right state")
return out

136
soleprint/station/tools/keymatrix/probe/gnome.py Normal file
View File

@@ -0,0 +1,136 @@
"""
Read this GNOME session's declared chords, exactly as they are.
Everything here shells out to `gsettings` / `dconf` and parses. Nothing is
written, nothing is changed, and nothing runs with sudo -- the owner's standing
rule is that nrft is managed by hand, so this reports and does not touch.
The five schemas are not an arbitrary list. Between them they are the entire
set of chords the compositor claims before any remote-desktop client sees a
keystroke, which is exactly the "eating set" the shadow rule needs.
"""
import re
import subprocess
from ..chord import Chord
# schema -> the short name a finding prints. Three of these end in
# ".keybindings", so the last dotted component is not a distinguishing label:
# wm, shell and mutter would all read "keybindings" and a finding could not say
# which one claimed a chord.
SCHEMAS = {
"org.gnome.desktop.wm.keybindings": "wm",
"org.gnome.shell.keybindings": "shell",
"org.gnome.mutter.keybindings": "mutter",
"org.gnome.mutter.wayland.keybindings": "mutter.wayland",
"org.gnome.settings-daemon.plugins.media-keys": "media-keys",
}
MEDIA_KEYS = "org.gnome.settings-daemon.plugins.media-keys"
CUSTOM_SCHEMA = MEDIA_KEYS + ".custom-keybinding"
def _sh(*argv):
r = subprocess.run(argv, capture_output=True, text=True)
return r.stdout.strip() if r.returncode == 0 else None
def _unquote(value):
"""GVariant scalars: `'<Super>h'`, `uint32 1`, `true`, `@as []`."""
value = value.strip()
value = re.sub(r"^(uint32|int32|@as|@a\(ss\))\s*", "", value).strip()
if value.startswith("'") and value.endswith("'"):
return value[1:-1]
return value
def _string_list(value):
"""`['<Super>Page_Up', '<Super>KP_Prior']` -> the two strings."""
return re.findall(r"'((?:[^'\\]|\\.)*)'", value or "")
def collect(report):
"""Every declared chord in this session, plus the input-source state.
`report` is called with (severity, message) for anything that could not be
read. An unreadable source is recorded as unreadable rather than skipped --
the peripheral matrix leaves cells blank on purpose for the same reason, and
a blank must never be mistaken for "no issue".
"""
bindings, notes = [], []
for schema, short in SCHEMAS.items():
dump = _sh("gsettings", "list-recursively", schema)
if dump is None:
report("warn", f"gsettings could not read {schema}")
continue
for line in dump.split("\n"):
parts = line.split(" ", 2)
if len(parts) < 3:
continue
_, key, value = parts
if schema == MEDIA_KEYS and key == "custom-keybindings":
# A list of dconf PATHS, not chords. Parsed as chords it yields
# six bindings named after directories. The customs behind those
# paths are collected below, where the labels and commands are.
continue
for raw in _string_list(value):
if not raw:
continue
if "/" in raw:
report("warn", f"{schema} {key} holds a path, not a chord: {raw!r}")
continue
try:
chord = Chord.parse_gnome(raw)
except ValueError as exc:
# An unparsed binding is an invisible binding, which is the
# failure this whole tool exists to end. Say so.
report("warn", f"unparsed chord {raw!r} in {schema} {key}: {exc}")
continue
bindings.append({
"chord": str(chord),
"raw": raw,
"source": f"gnome:{short}:{key}",
"label": None,
"command": None,
})
# Custom shortcuts live behind a relocatable schema, and the LIST is the
# only authority for which exist. Enumerating custom0..N instead returns
# empty defaults for the ones that do not, which get recorded as real
# bindings that nothing can explain.
paths = _string_list(_sh("gsettings", "get", MEDIA_KEYS, "custom-keybindings"))
for path in paths:
got = {}
for key in ("name", "command", "binding"):
got[key] = _unquote(_sh("gsettings", "get",
f"{CUSTOM_SCHEMA}:{path}", key) or "")
if not got["binding"]:
report("warn", f"custom keybinding {path} has no binding")
continue
try:
chord = Chord.parse_gnome(got["binding"])
except ValueError as exc:
report("warn", f"unparsed custom chord {got['binding']!r}: {exc}")
continue
bindings.append({
"chord": str(chord),
"raw": got["binding"],
"source": f"gnome:media-keys:{path.rstrip('/').rsplit('/', 1)[-1]}",
"label": got["name"] or None,
"command": got["command"] or None,
})
notes.append({"custom_keybinding_paths": paths})
return bindings, notes
def session():
"""What is running, so a verdict can say which session it describes."""
import os
return {
"session_type": os.environ.get("XDG_SESSION_TYPE"),
"desktop": os.environ.get("XDG_CURRENT_DESKTOP"),
"shell": _sh("gnome-shell", "--version"),
}

89
soleprint/station/tools/keymatrix/probe/keyd.py Normal file
View File

@@ -0,0 +1,89 @@
"""
keyd's state, including the part that cannot be read.
keyd remaps at evdev, BELOW the compositor, so it can change which chord you
press and never who wins the arbitration afterwards. That is why keymatrix
probes it and reports on it but does not use it to fix shadowing.
`/etc/keyd/default.conf` is 0600 root:root. **The tool never sudos**, so the
authoritative copy is simply unreadable, and that is recorded rather than
guessed around. A source copy may be declared; if it is, it is read and clearly
labelled as possibly-stale.
"""
import subprocess
from pathlib import Path
from ..chord import Chord
CONF = Path("/etc/keyd/default.conf")
SOURCE_COPY = Path.home() / "default.conf"
def _sh(*argv):
r = subprocess.run(argv, capture_output=True, text=True)
return r.stdout.strip() if r.returncode == 0 else None
def _parse(text, report):
rules, section = [], None
for line in text.split("\n"):
line = line.split("#", 1)[0].strip()
if not line:
continue
if line.startswith("[") and line.endswith("]"):
section = line[1:-1]
continue
if "=" not in line or section != "main":
continue
src, dst = (p.strip() for p in line.split("=", 1))
try:
rules.append({"from": src, "to": dst,
"from_chord": str(Chord.parse_keyd(src)),
"to_chord": str(Chord.parse_keyd(dst))})
except ValueError as exc:
report("warn", f"unparsed keyd rule {line!r}: {exc}")
return rules
def collect(report, unreadable):
state = _sh("systemctl", "show", "keyd", "-p", "ActiveState",
"-p", "UnitFileState", "-p", "FragmentPath", "-p", "ExecMainPID")
info = dict(l.split("=", 1) for l in (state or "").split("\n") if "=" in l)
out = {
"active": info.get("ActiveState") == "active",
"enabled": info.get("UnitFileState") == "enabled",
"unit": info.get("FragmentPath") or None,
"config_readable": False,
"config_source": None,
"same_as_authoritative": "unknown",
"rules": [],
}
# The binary is /usr/bin/keyd.rvaiya on Debian; there is no `keyd` on PATH.
# Anything shelling out to `keyd reload` is already broken here.
if out["unit"]:
for line in (Path(out["unit"]).read_text().split("\n")
if Path(out["unit"]).exists() else []):
if line.startswith("ExecStart"):
out["binary"] = line.split("=", 1)[1].split()[0]
break
try:
out["rules"] = _parse(CONF.read_text(), report)
out["config_readable"] = True
out["config_source"] = str(CONF)
out["same_as_authoritative"] = "yes"
except PermissionError:
unreadable.append({
"what": str(CONF),
"reason": "mode 0600 root:root; keymatrix never sudos"})
if SOURCE_COPY.exists():
out["rules"] = _parse(SOURCE_COPY.read_text(), report)
out["config_source"] = str(SOURCE_COPY)
report("warn", f"read keyd rules from {SOURCE_COPY}, which may be "
f"stale; {CONF} is root-only")
except FileNotFoundError:
out["config_source"] = None
return out

57
soleprint/station/tools/keymatrix/probe/xkb.py Normal file
View File

@@ -0,0 +1,57 @@
"""
Layout state, from the two authorities that disagree about it.
`localectl` describes the system X11 layout; GNOME's `input-sources` describes
the session's rotation. On nrft they say different things and both are live.
That matters for keyboards rather than for typography: `run/unstuckinput-wayland.sh`
records that layout translation happens on the *client* side of the deskflow
protocol, so a mismatch desyncs modifier state mid-session.
"""
import re
import subprocess
def _sh(*argv):
r = subprocess.run(argv, capture_output=True, text=True)
return r.stdout.strip() if r.returncode == 0 else None
def collect(report):
out = {"sources": [], "current": None, "xkb_options": [],
"xkb_model": None, "localectl": {}}
dump = _sh("gsettings", "list-recursively", "org.gnome.desktop.input-sources")
for line in (dump or "").split("\n"):
parts = line.split(" ", 2)
if len(parts) < 3:
continue
_, key, value = parts
if key == "sources":
out["sources"] = [list(t) for t in re.findall(r"\('([^']+)',\s*'([^']+)'\)", value)]
elif key == "current":
# `uint32 1`. Grabbing the first run of digits finds the 32 in the
# TYPE NAME, not the value, and reports source 32 of 4.
m = re.fullmatch(r"(?:uint32|int32)?\s*(\d+)", value.strip())
out["current"] = int(m.group(1)) if m else None
elif key == "xkb-options":
out["xkb_options"] = re.findall(r"'([^']+)'", value)
elif key == "xkb-model":
out["xkb_model"] = value.strip().strip("'")
status = _sh("localectl", "status") or ""
for line in status.split("\n"):
if ":" in line:
k, v = line.split(":", 1)
k = k.strip().lower().replace(" ", "_")
if k in ("x11_layout", "x11_model", "x11_options", "x11_variant",
"vc_keymap"):
v = v.strip()
# localectl prints "(unset)" where it means nothing is set.
out["localectl"][k] = None if v in ("", "(unset)", "n/a") else v
if not out["xkb_options"] and len(out["sources"]) > 1:
report("note", f"{len(out['sources'])} input sources with an empty "
f"xkb-options: no grp: switcher, so Super+space is the "
f"only way to change layout")
return out

190
soleprint/station/tools/keymatrix/shadow.py Normal file
View File

@@ -0,0 +1,190 @@
"""
Which endpoint actually receives a chord, given how you are connected.
The fleet has two POSTURES, and the direction of remote control flips between
them. That is the whole reason this module exists: a chord's destination is not
a property of the chord.
A peripherals on nrft nrft (GNOME/Wayland) --RealVNC--> ethbox (Windows)
and on to aws-workspace, three hops
B peripherals on ethbox ethbox (Windows) --RDP--> nrft (headless GNOME)
aws-workspace direct from ethbox, two hops
In A, GNOME arbitrates first and AHK's bindings are the unreachable ones. In B
that inverts exactly. `vnc-keyboard-routing.md`'s table is posture-A-only, and
reading it as general is how a fix gets designed for half the fleet.
## The rule
A chord reaches the far side unless the near side takes it first. Who "the near
side" is, is the posture. Three things decide the outcome, in order:
1. the transport's own grab -- can the client bind the keyboard at all;
2. chords the client intercepts for itself, whatever the compositor thinks;
3. otherwise, whatever the near endpoint has declared.
Step 3 consults **observed** state, never the keymap. The near side shadows with
chords nobody wrote down -- `<Super>l`, `<Alt><Super>8`, `<Super><Control>1..9`.
A rule that only knew about declared actions would miss all of them, and those
are the ones that surprise you.
## Nothing about a specific chord is hardcoded here
`Ctrl+Super+Left/Right` passes because it is *absent from the observed set*,
verified across all five keybinding schemas -- not because a constant says so.
Every transport-specific fact is data with an `evidence` string, so a claim can
be checked against the note it came from.
"""
from .chord import Chord
# How much of the keyboard a transport can take before the near compositor sees
# it. Every value here is evidence-backed; see TRANSPORTS below.
GRABS = ("none", "requested-ineffective", "effective", "exclusive", "unknown")
class Verdict:
__slots__ = ("outcome", "why", "eater")
def __init__(self, outcome, why="", eater=None):
self.outcome = outcome # passes | shadowed | eaten-by-transport | UNKNOWN
self.why = why
self.eater = eater
@property
def reaches_far_side(self):
return self.outcome == "passes"
def __repr__(self):
return f"<{self.outcome}{' by ' + self.eater if self.eater else ''}>"
def reaches(chord, transport, near_observed):
"""Does `chord`, pressed on the near side, arrive at the far side?
`near_observed` maps a normal-form chord string to the binding that claims
it -- the probe's output for whichever endpoint is near in this posture.
"""
grab = transport.get("grab", "unknown")
if grab not in GRABS:
raise ValueError(f"unknown grab kind {grab!r}")
if grab == "unknown":
# Never tested. Saying "passes" here would be a guess wearing a verdict's
# clothes, and the peripheral matrix's whole discipline is that a blank
# is not a "no issue".
return Verdict("UNKNOWN", "this transport's grab has never been tested")
key = str(chord)
for raw in transport.get("intercepts", ()):
if Chord.parse(raw) == chord:
return Verdict("eaten-by-transport",
transport.get("evidence", ""), transport["name"] + ":client")
if grab == "exclusive":
return Verdict("passes", "the near side receives no events at all")
if grab == "effective":
return Verdict("passes", "the client's grab binds; the compositor releases")
hit = near_observed.get(key)
if hit:
return Verdict("shadowed", "the near compositor claims it first",
hit.get("label") or hit["source"])
return Verdict("passes", "the near side binds nothing on this chord")
# ── the transports, as data ──────────────────────────────────────────────
TRANSPORTS = {
"realvnc-x11": {
"name": "realvnc-x11",
"grab": "requested-ineffective",
"intercepts": ["F8"],
"evidence": (
"GrabKeyboard already defaults to 2 ('always grab') and GNOME overrules "
"it -- nrft's own Ctrl+Super+c fires over the focused viewer, which is "
"the original complaint and the proof. RealVNC is X11/XWayland and "
"cannot use zwp_keyboard_shortcuts_inhibit_v1. F8 is its MenuKey, "
"consumed before forwarding. -- ahk/def/vnc-keyboard-routing.md"),
},
"rdp-grd": {
"name": "rdp-grd",
"grab": "unknown",
"intercepts": [],
"evidence": (
"Posture B: ethbox's RDP client into nrft's gnome-remote-desktop "
"(system unit active on *:3389, so a HEADLESS session -- a separate "
"GNOME instance, same dconf). mstsc has a documented 'Apply Windows "
"key combinations' setting with no RealVNC equivalent, so this "
"direction may be configurable where posture A is not. UNTESTED: the "
"setting's value here is unknown, so the grab is unknown."),
},
"physical-switch": {
"name": "physical-switch",
"grab": "exclusive",
"intercepts": [],
"evidence": (
"A USB/AV switch. The side it is not pointing at receives no events "
"at all, so nothing can shadow anything."),
},
"deskflow": {
"name": "deskflow",
"grab": "effective",
"intercepts": ["Super+Shift+Escape"],
"evidence": (
"Server captures at the screen edge via the libei/InputCapture portal, "
"so the near compositor sees nothing while switched away. REASONED, "
"not observed: grounded only in mutter binding cancel-input-capture at "
"<Super><Shift>Escape, which implies portal capture exists. The "
"nrft<->mcrn link is still 'candidate' in switching-matrix.csv."),
},
"workspaces-client": {
"name": "workspaces-client",
"grab": "unknown",
"intercepts": [],
"evidence": "Amazon WorkSpaces client. Never tested; nothing may be installed.",
},
}
# ── the postures ─────────────────────────────────────────────────────────
POSTURES = {
"A": {
"name": "A",
"summary": "peripherals on nrft; VNC out to ethbox",
"peripherals": "nrft",
"hops": [
{"near": "nrft", "far": "ethbox", "transport": "realvnc-x11"},
{"near": "ethbox", "far": "aws-workspace", "transport": "workspaces-client"},
],
},
"B": {
"name": "B",
"summary": "peripherals on ethbox, both monitors in Windows; RDP into nrft",
"peripherals": "ethbox",
"hops": [
{"near": "ethbox", "far": "nrft", "transport": "rdp-grd"},
{"near": "ethbox", "far": "aws-workspace", "transport": "workspaces-client"},
],
},
}
def walk(chord, posture, observed_by_endpoint):
"""Follow a chord along a posture's chain until something eats it.
Returns one row per hop, so a three-hop posture shows where the chord died
rather than only whether it reached the end.
"""
rows = []
for hop in POSTURES[posture]["hops"]:
near = observed_by_endpoint.get(hop["near"])
if near is None:
rows.append((hop, Verdict("UNKNOWN", f"{hop['near']} has never been probed")))
break
v = reaches(chord, TRANSPORTS[hop["transport"]], near)
rows.append((hop, v))
if not v.reaches_far_side:
break
return rows

156
soleprint/station/tools/shuntgen/__main__.py
View File

@@ -6,12 +6,16 @@ Shuntgen CLI.
python -m station.tools.shuntgen list python -m station.tools.shuntgen list
Run from the soleprint/ directory so `station.tools...` resolves. Run from the soleprint/ directory so `station.tools...` resolves.
Built on `station/tools/common/cli.py`, so the flag vocabulary, the single exit
path and the clobber guard are the scaffold's rather than this tool's. `-s` used
to mean --spec here and --source there, in one tool; it is --source on both now,
with --spec kept as a hidden alias.
""" """
import argparse
import sys
from pathlib import Path from pathlib import Path
from ..common.cli import COMMON, Tool, fail, refuse_to_clobber
from .emit import SPR_ROOT, ShuntEmitter from .emit import SPR_ROOT, ShuntEmitter
@@ -25,33 +29,16 @@ def _name_for(args, source: Path) -> str:
return source.stem return source.stem
def _refuse_to_clobber(output: Path, force: bool) -> None:
"""Regenerating is fine; overwriting something that is not a shunt is not."""
if not output.exists() or force:
return
if not any(output.iterdir()):
return
if (output / "depot" / "spec.json").exists():
return
print(
f"Error: {output} already exists and was not generated by shuntgen.\n"
" Pick another path, or pass --force to write into it anyway.",
file=sys.stderr,
)
sys.exit(1)
def cmd_from_openapi(args): def cmd_from_openapi(args):
"""Build a shunt from an OpenAPI 3.x / Swagger 2.0 document.""" """Build a shunt from an OpenAPI 3.x / Swagger 2.0 document."""
from ..modelgen.loader.extract.openapi import OpenAPIExtractor from ..modelgen.loader.extract.openapi import OpenAPIExtractor
spec_path = Path(args.spec) spec_path = Path(args.source)
if not spec_path.exists(): if not spec_path.exists():
print(f"Error: Spec not found: {spec_path}", file=sys.stderr) fail(f"spec not found: {spec_path}")
sys.exit(1)
output = Path(args.output) output = Path(args.output)
_refuse_to_clobber(output, args.force) refuse_to_clobber(output, args.force, "depot/spec.json", "shuntgen")
extractor = OpenAPIExtractor(spec_path) extractor = OpenAPIExtractor(spec_path)
print(f"Reading spec: {spec_path}") print(f"Reading spec: {spec_path}")
@@ -59,15 +46,10 @@ def cmd_from_openapi(args):
models, enums = extractor.extract() models, enums = extractor.extract()
endpoints = extractor.endpoints() endpoints = extractor.endpoints()
except (RuntimeError, ValueError) as e: except (RuntimeError, ValueError) as e:
print(f"Error: {e}", file=sys.stderr) fail(str(e))
sys.exit(1)
if not endpoints: if not endpoints:
print( fail("the spec declares no operations, so there is nothing to serve.")
"Error: the spec declares no operations, so there is nothing to serve.",
file=sys.stderr,
)
sys.exit(1)
name = _name_for(args, spec_path) name = _name_for(args, spec_path)
print(f"Found {len(models)} models, {len(endpoints)} endpoints") print(f"Found {len(models)} models, {len(endpoints)} endpoints")
@@ -94,19 +76,17 @@ def cmd_from_tabular(args):
source_path = Path(args.source) source_path = Path(args.source)
if not source_path.exists(): if not source_path.exists():
print(f"Error: Source not found: {source_path}", file=sys.stderr) fail(f"source not found: {source_path}")
sys.exit(1)
output = Path(args.output) output = Path(args.output)
_refuse_to_clobber(output, args.force) refuse_to_clobber(output, args.force, "depot/spec.json", "shuntgen")
extractor = TabularExtractor(source_path) extractor = TabularExtractor(source_path)
print(f"Reading sheets: {source_path}") print(f"Reading sheets: {source_path}")
try: try:
models, enums = extractor.extract() models, enums = extractor.extract()
except ValueError as e: except ValueError as e:
print(f"Error: {e}", file=sys.stderr) fail(str(e))
sys.exit(1)
datasets = extractor.datasets() datasets = extractor.datasets()
rows = sum(len(d.rows) for d in datasets) rows = sum(len(d.rows) for d in datasets)
@@ -174,65 +154,59 @@ def _report(output: Path, name: str, port: int) -> None:
print(f" open http://localhost:{port}/") print(f" open http://localhost:{port}/")
def _add_common(parser, port_default: int) -> None: def _add_common(tool, verb, port_default: int) -> None:
parser.add_argument( """The flags every generating verb shares.
"--output", "-o", type=str, required=True,
help="Where to write the shunt (e.g. artery/shunts/petstore)",
)
parser.add_argument(
"--name", "-n", type=str, default=None,
help="Shunt name (default: the output folder's name)",
)
parser.add_argument(
"--title", type=str, default=None,
help="Display title for the config UI (default: derived from the name)",
)
parser.add_argument(
"--port", "-p", type=int, default=port_default,
help=f"Default port for run.py (default: {port_default})",
)
parser.add_argument(
"--cabinet", type=str, default=None,
help="Comma-separated dependency containers to declare (e.g. postgres,redis)",
)
parser.add_argument(
"--force", action="store_true",
help="Write into a non-empty directory that shuntgen did not generate",
)
def main():
parser = argparse.ArgumentParser(
description="Shuntgen - generate runnable shunts",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
subparsers = parser.add_subparsers(dest="command", required=True)
openapi_parser = subparsers.add_parser( Spellings come from COMMON (cli.py:67) rather than being restated here.
"from-openapi", help="Generate a shunt from an OpenAPI / Swagger document" --output is required in this tool, which is the kind of per-tool decision
) the overrides exist for.
openapi_parser.add_argument( """
"--spec", "-s", type=str, required=True, tool.common(
help="Path to the spec (.json, .yaml or .yml)", verb, "output", "name", "force",
) output=dict(required=True,
_add_common(openapi_parser, 8099) help="Where to write the shunt (e.g. artery/shunts/petstore)"),
openapi_parser.set_defaults(func=cmd_from_openapi) name=dict(default=None, help="Shunt name (default: the output folder's name)"),
force=dict(help="Write into a non-empty directory shuntgen did not generate"),
tabular_parser = subparsers.add_parser(
"from-tabular", help="Generate a shunt from a directory of CSV/TSV/ODS sheets"
)
tabular_parser.add_argument(
"--source", "-s", type=str, required=True,
help="Directory of sheets (or a single .csv/.tsv/.ods file)",
) )
_add_common(tabular_parser, 8098) tool.argument(verb, "--title", default=None,
tabular_parser.set_defaults(func=cmd_from_tabular) help="Display title for the config UI (default: from the name)")
tool.argument(verb, "--port", "-p", type=int, default=port_default,
list_parser = subparsers.add_parser("list", help="List shunts under artery/shunts/") help=f"Default port for run.py (default: {port_default})")
list_parser.set_defaults(func=cmd_list) tool.argument(verb, "--cabinet", default=None,
help="Comma-separated dependency containers (e.g. postgres,redis)")
args = parser.parse_args()
args.func(args)
def main(argv=None):
tool = Tool("shuntgen", __doc__, package=__package__)
tool.command("from-openapi", cmd_from_openapi,
"Generate a shunt from an OpenAPI / Swagger document")
# --spec was this verb's spelling for the same idea, and it is kept.
#
# It is one action carrying three option strings rather than a second,
# hidden action, because argparse checks `required` per ACTION and not per
# dest: a separate `--spec` with dest="source" leaves the `--source` action
# unsatisfied, so `--spec x -o y` fails with "the following arguments are
# required: --source/-s" -- an alias that can never be used. Found by
# running it. The cost is that --spec shows in --help, which is the better
# trade anyway: someone with the old spelling in their fingers sees that it
# still works instead of guessing.
tool.argument("from-openapi", "--source", "-s", "--spec", required=True,
metavar="PATH",
help="Path to the spec (.json, .yaml or .yml). "
"--spec is the old name for this and still works.")
_add_common(tool, "from-openapi", 8099)
tool.command("from-tabular", cmd_from_tabular,
"Generate a shunt from a directory of CSV/TSV/ODS sheets")
tool.common("from-tabular", "source",
source=dict(required=True,
help="Directory of sheets (or a single .csv/.tsv/.ods file)"))
_add_common(tool, "from-tabular", 8098)
tool.command("list", cmd_list, "List shunts under artery/shunts/")
tool.run(argv)
if __name__ == "__main__": if __name__ == "__main__":

36
soleprint/station/tools/tester/__init__.py
View File

@@ -3,9 +3,39 @@ Tester - HTTP contract test runner with web UI.
Discovers and runs contract tests from tests/ directory. Discovers and runs contract tests from tests/ directory.
Tests can be symlinked from production repos or copied during deployment. Tests can be symlinked from production repos or copied during deployment.
"""
from .api import router ## Why the re-exports are lazy
from .core import discover_tests, start_test_run, get_run_status
`from .api import router` at module scope pulled FastAPI and pydantic in every
time anything touched this package -- including `python -m station.tools.tester
--help`, which then died with ModuleNotFoundError on a machine without pydantic
installed. The CLI does not need the web stack to print its own usage.
PEP 562 module `__getattr__` keeps the surface identical: `tester.router` and
`tester.discover_tests` resolve exactly as before, on first use. `soleprint/
run.py:582` imports `station.tools.tester.api` directly and is unaffected.
"""
__all__ = ["router", "discover_tests", "start_test_run", "get_run_status"] __all__ = ["router", "discover_tests", "start_test_run", "get_run_status"]
_SOURCES = {
"router": ".api",
"discover_tests": ".core",
"start_test_run": ".core",
"get_run_status": ".core",
}
def __getattr__(name):
if name not in _SOURCES:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
import importlib
module = importlib.import_module(_SOURCES[name], __name__)
value = getattr(module, name)
globals()[name] = value # resolve once, not on every access
return value
def __dir__():
return sorted(set(globals()) | set(_SOURCES))

74
soleprint/station/tools/tester/cli.py
View File

@@ -1,18 +1,32 @@
""" """
CLI for contracts_http tool. CLI for the contract HTTP test runner.
Built on `station/tools/common/cli.py`: one exit path, errors on stderr, a bare
invocation that does not report success.
Nothing that touches the filesystem is imported here. `config.py` runs
`load_config()` and `load_environments()` at module scope, so importing it at the
top of this file made `tester --help` read `.env` off disk -- the deferred-import
rule broken in the one place it is most visible. Both live inside the commands
that need them.
""" """
import argparse
import sys import sys
import time import time
from .config import config from ..common.cli import Tool, fail
from .core import discover_tests, start_test_run, get_run_status
def cmd_discover(args): def cmd_discover(args):
"""List discovered tests.""" """List discovered tests."""
from .core import discover_tests
tests = discover_tests() tests = discover_tests()
if not tests:
# It used to return None here whatever happened, and main() called it
# bare, so `discover` reported success on an empty room.
fail("no tests were discovered.",
"Check the test directory and that the files match the naming rule.")
if args.json: if args.json:
import json import json
@@ -44,7 +58,10 @@ def cmd_discover(args):
def cmd_run(args): def cmd_run(args):
"""Run tests.""" """Run tests. Returns the count-shaped exit code, not an error."""
from .config import config
from .core import discover_tests, get_run_status, start_test_run
print(f"Target: {config['CONTRACT_TEST_URL']}") print(f"Target: {config['CONTRACT_TEST_URL']}")
print() print()
@@ -57,8 +74,7 @@ def cmd_run(args):
if args.pattern.lower() in t.id.lower() if args.pattern.lower() in t.id.lower()
] ]
if not test_ids: if not test_ids:
print(f"No tests matching pattern: {args.pattern}") fail(f"no tests match the pattern: {args.pattern}")
return 1
print(f"Running {len(test_ids)} tests matching '{args.pattern}'") print(f"Running {len(test_ids)} tests matching '{args.pattern}'")
else: else:
print("Running all tests") print("Running all tests")
@@ -68,23 +84,26 @@ def cmd_run(args):
# Start run # Start run
run_id = start_test_run(test_ids) run_id = start_test_run(test_ids)
# Poll until complete # Poll until complete. KeyboardInterrupt is left to Tool.run, which exits
# 130; the only thing owed here is not leaving a half-drawn progress line
# behind on the terminal.
try:
while True: while True:
status = get_run_status(run_id) status = get_run_status(run_id)
if not status: if not status:
print("Error: Run not found") fail(f"run not found: {run_id}")
return 1
# Print progress
if status.current_test: if status.current_test:
sys.stdout.write(f"\r Running: {status.current_test[:60]}...") sys.stdout.write(f"\r Running: {status.current_test[:60]}...")
sys.stdout.flush() sys.stdout.flush()
if status.status in ("completed", "failed"): if status.status in ("completed", "failed"):
sys.stdout.write("\r" + " " * 80 + "\r") # Clear line
break break
time.sleep(0.5) time.sleep(0.5)
finally:
sys.stdout.write("\r" + " " * 80 + "\r")
sys.stdout.flush()
# Print results # Print results
print(f"Results: {status.passed} passed, {status.failed} failed, {status.skipped} skipped") print(f"Results: {status.passed} passed, {status.failed} failed, {status.skipped} skipped")
@@ -101,28 +120,21 @@ def cmd_run(args):
return 1 if failures else 0 return 1 if failures else 0
def main(args=None): def main(argv=None):
parser = argparse.ArgumentParser( # `Tool` puts required=True on the subparsers. Without it, a bare
description="Contract HTTP Tests - Pure HTTP test runner" # `python -m tester` printed help and exited 0: a no-op reporting success.
) tool = Tool("tester", __doc__, package=__package__)
subparsers = parser.add_subparsers(dest="command", help="Available commands")
# discover command tool.command("discover", cmd_discover, "List discovered tests")
discover_parser = subparsers.add_parser("discover", help="List discovered tests") tool.argument("discover", "--json", action="store_true", help="Output as JSON")
discover_parser.add_argument("--json", action="store_true", help="Output as JSON")
# run command tool.command("run", cmd_run, "Run tests")
run_parser = subparsers.add_parser("run", help="Run tests") tool.argument("run", "pattern", nargs="?",
run_parser.add_argument("pattern", nargs="?", help="Filter tests by pattern (e.g. 'customers', 'invoices')") help="Filter tests by pattern (e.g. 'customers', 'invoices')")
args = parser.parse_args(args) # cmd_run returns 1 when tests failed. Tool.run exits with it -- a real
# result, so it does not go through fail() and is not prefixed "Error:".
if args.command == "discover": tool.run(argv)
cmd_discover(args)
elif args.command == "run":
sys.exit(cmd_run(args))
else:
parser.print_help()
if __name__ == "__main__": if __name__ == "__main__":