diff --git a/ctrl/spr.py b/ctrl/spr.py index 37e3e46..f38f82d 100755 --- a/ctrl/spr.py +++ b/ctrl/spr.py @@ -73,7 +73,15 @@ def get_version(comp_type, source): for line in (source / "pyproject.toml").read_text().splitlines(): if line.strip().startswith("version"): 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 return "?" diff --git a/registry.json b/registry.json index 7282cee..9f45c92 100644 --- a/registry.json +++ b/registry.json @@ -1,10 +1,14 @@ { - "soleprint-ui": { - "type": "npm", - "path": "soleprint/common/ui" + "soleprint-cli": { + "type": "python", + "path": "soleprint/station/tools/common" }, "soleprint-modelgen": { "type": "pip", "path": "soleprint/station/tools/modelgen" + }, + "soleprint-ui": { + "type": "npm", + "path": "soleprint/common/ui" } } diff --git a/soleprint/station/tools/common/__init__.py b/soleprint/station/tools/common/__init__.py index bca429e..f61fad0 100644 --- a/soleprint/station/tools/common/__init__.py +++ b/soleprint/station/tools/common/__init__.py @@ -12,3 +12,5 @@ it is circular. """ __all__ = ["cli"] + +__version__ = "0.1.0" diff --git a/soleprint/station/tools/common/cli.py b/soleprint/station/tools/common/cli.py index 1185a27..ee24f1a 100644 --- a/soleprint/station/tools/common/cli.py +++ b/soleprint/station/tools/common/cli.py @@ -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 '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 `--help` stays instant and an optional dependency only costs the one @@ -280,9 +287,19 @@ class Tool: 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: - args.func(args) + 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) diff --git a/soleprint/station/tools/common/selftest.py b/soleprint/station/tools/common/selftest.py index 009b7a3..69ed41d 100644 --- a/soleprint/station/tools/common/selftest.py +++ b/soleprint/station/tools/common/selftest.py @@ -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 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. """ @@ -30,7 +31,14 @@ TOOLS = HERE.parent # 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 # 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): @@ -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): """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'])") 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 ──────────────────── r = run_py("from common.cli import Tool\n" "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, 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 ───────────────────── # Not hypothetical: soleprint/common/ holds auth, theme and ui. Tools must # reach the scaffold relatively (`from ..common.cli import Tool`) or by its @@ -255,4 +335,4 @@ print("LEAKED", len(leaked)) if __name__ == "__main__": - sys.exit(main()) + sys.exit(revendor() if "--revendor" in sys.argv else main()) diff --git a/soleprint/station/tools/histgen/cli.py b/soleprint/station/tools/histgen/cli.py index f8c89e4..2091e23 100644 --- a/soleprint/station/tools/histgen/cli.py +++ b/soleprint/station/tools/histgen/cli.py @@ -240,9 +240,19 @@ class Tool: 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: - args.func(args) + 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) diff --git a/soleprint/station/tools/keymatrix/Makefile b/soleprint/station/tools/keymatrix/Makefile new file mode 100644 index 0000000..ee33d07 --- /dev/null +++ b/soleprint/station/tools/keymatrix/Makefile @@ -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())')" diff --git a/soleprint/station/tools/keymatrix/__init__.py b/soleprint/station/tools/keymatrix/__init__.py new file mode 100644 index 0000000..62f9b17 --- /dev/null +++ b/soleprint/station/tools/keymatrix/__init__.py @@ -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/.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" diff --git a/soleprint/station/tools/keymatrix/__main__.py b/soleprint/station/tools/keymatrix/__main__.py new file mode 100644 index 0000000..5ca64a2 --- /dev/null +++ b/soleprint/station/tools/keymatrix/__main__.py @@ -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/.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 ` 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 ` 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() diff --git a/soleprint/station/tools/keymatrix/check.py b/soleprint/station/tools/keymatrix/check.py new file mode 100644 index 0000000..80ac116 --- /dev/null +++ b/soleprint/station/tools/keymatrix/check.py @@ -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 diff --git a/soleprint/station/tools/keymatrix/chord.py b/soleprint/station/tools/keymatrix/chord.py new file mode 100644 index 0000000..eb9c4e6 --- /dev/null +++ b/soleprint/station/tools/keymatrix/chord.py @@ -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 c 1 Page_Up + AHK ^#c ^!1 #3 + AutoKey ["",""] + "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 -- `c` and `1` are both live in the +same session -- so even comparing GNOME to GNOME needs one. + + Chord.parse_gnome("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. +# is GNOME's own alias for 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: # `` alone: GNOME's overlay-key + return "+".join(self.mods) + return "+".join((*self.mods, self.key)) + + @property + def is_bare_modifier(self): + """`` 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 -- `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): + """`c`, `1`, ``, `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: `["", ""]` 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 diff --git a/soleprint/station/tools/keymatrix/cli.py b/soleprint/station/tools/keymatrix/cli.py new file mode 100644 index 0000000..b95e7b7 --- /dev/null +++ b/soleprint/station/tools/keymatrix/cli.py @@ -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) diff --git a/soleprint/station/tools/keymatrix/endpoint.py b/soleprint/station/tools/keymatrix/endpoint.py new file mode 100644 index 0000000..2fb5f7a --- /dev/null +++ b/soleprint/station/tools/keymatrix/endpoint.py @@ -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" diff --git a/soleprint/station/tools/keymatrix/probe/__init__.py b/soleprint/station/tools/keymatrix/probe/__init__.py new file mode 100644 index 0000000..f1c11d9 --- /dev/null +++ b/soleprint/station/tools/keymatrix/probe/__init__.py @@ -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 diff --git a/soleprint/station/tools/keymatrix/probe/deskflow.py b/soleprint/station/tools/keymatrix/probe/deskflow.py new file mode 100644 index 0000000..0debf0e --- /dev/null +++ b/soleprint/station/tools/keymatrix/probe/deskflow.py @@ -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 diff --git a/soleprint/station/tools/keymatrix/probe/gnome.py b/soleprint/station/tools/keymatrix/probe/gnome.py new file mode 100644 index 0000000..38e423d --- /dev/null +++ b/soleprint/station/tools/keymatrix/probe/gnome.py @@ -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: `'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): + """`['Page_Up', '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"), + } diff --git a/soleprint/station/tools/keymatrix/probe/keyd.py b/soleprint/station/tools/keymatrix/probe/keyd.py new file mode 100644 index 0000000..85e7047 --- /dev/null +++ b/soleprint/station/tools/keymatrix/probe/keyd.py @@ -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 diff --git a/soleprint/station/tools/keymatrix/probe/xkb.py b/soleprint/station/tools/keymatrix/probe/xkb.py new file mode 100644 index 0000000..e580e9e --- /dev/null +++ b/soleprint/station/tools/keymatrix/probe/xkb.py @@ -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 diff --git a/soleprint/station/tools/keymatrix/shadow.py b/soleprint/station/tools/keymatrix/shadow.py new file mode 100644 index 0000000..f7b51bd --- /dev/null +++ b/soleprint/station/tools/keymatrix/shadow.py @@ -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 -- `l`, `8`, `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 " + "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 diff --git a/soleprint/station/tools/shuntgen/__main__.py b/soleprint/station/tools/shuntgen/__main__.py index 24fb0bc..685e96c 100644 --- a/soleprint/station/tools/shuntgen/__main__.py +++ b/soleprint/station/tools/shuntgen/__main__.py @@ -6,12 +6,16 @@ Shuntgen CLI. python -m station.tools.shuntgen list 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 ..common.cli import COMMON, Tool, fail, refuse_to_clobber from .emit import SPR_ROOT, ShuntEmitter @@ -25,33 +29,16 @@ def _name_for(args, source: Path) -> str: 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): """Build a shunt from an OpenAPI 3.x / Swagger 2.0 document.""" from ..modelgen.loader.extract.openapi import OpenAPIExtractor - spec_path = Path(args.spec) + spec_path = Path(args.source) if not spec_path.exists(): - print(f"Error: Spec not found: {spec_path}", file=sys.stderr) - sys.exit(1) + fail(f"spec not found: {spec_path}") output = Path(args.output) - _refuse_to_clobber(output, args.force) + refuse_to_clobber(output, args.force, "depot/spec.json", "shuntgen") extractor = OpenAPIExtractor(spec_path) print(f"Reading spec: {spec_path}") @@ -59,15 +46,10 @@ def cmd_from_openapi(args): models, enums = extractor.extract() endpoints = extractor.endpoints() except (RuntimeError, ValueError) as e: - print(f"Error: {e}", file=sys.stderr) - sys.exit(1) + fail(str(e)) if not endpoints: - print( - "Error: the spec declares no operations, so there is nothing to serve.", - file=sys.stderr, - ) - sys.exit(1) + fail("the spec declares no operations, so there is nothing to serve.") name = _name_for(args, spec_path) print(f"Found {len(models)} models, {len(endpoints)} endpoints") @@ -94,19 +76,17 @@ def cmd_from_tabular(args): source_path = Path(args.source) if not source_path.exists(): - print(f"Error: Source not found: {source_path}", file=sys.stderr) - sys.exit(1) + fail(f"source not found: {source_path}") output = Path(args.output) - _refuse_to_clobber(output, args.force) + refuse_to_clobber(output, args.force, "depot/spec.json", "shuntgen") extractor = TabularExtractor(source_path) print(f"Reading sheets: {source_path}") try: models, enums = extractor.extract() except ValueError as e: - print(f"Error: {e}", file=sys.stderr) - sys.exit(1) + fail(str(e)) datasets = extractor.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}/") -def _add_common(parser, port_default: int) -> None: - parser.add_argument( - "--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) +def _add_common(tool, verb, port_default: int) -> None: + """The flags every generating verb shares. - openapi_parser = subparsers.add_parser( - "from-openapi", help="Generate a shunt from an OpenAPI / Swagger document" - ) - openapi_parser.add_argument( - "--spec", "-s", type=str, required=True, - help="Path to the spec (.json, .yaml or .yml)", + Spellings come from COMMON (cli.py:67) rather than being restated here. + --output is required in this tool, which is the kind of per-tool decision + the overrides exist for. + """ + tool.common( + verb, "output", "name", "force", + output=dict(required=True, + help="Where to write the shunt (e.g. artery/shunts/petstore)"), + 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"), ) - _add_common(openapi_parser, 8099) - openapi_parser.set_defaults(func=cmd_from_openapi) - - 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) - tabular_parser.set_defaults(func=cmd_from_tabular) - - list_parser = subparsers.add_parser("list", help="List shunts under artery/shunts/") - list_parser.set_defaults(func=cmd_list) - - args = parser.parse_args() - args.func(args) + tool.argument(verb, "--title", default=None, + help="Display title for the config UI (default: from the name)") + tool.argument(verb, "--port", "-p", type=int, default=port_default, + help=f"Default port for run.py (default: {port_default})") + tool.argument(verb, "--cabinet", default=None, + help="Comma-separated dependency containers (e.g. postgres,redis)") + + +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__": diff --git a/soleprint/station/tools/tester/__init__.py b/soleprint/station/tools/tester/__init__.py index 119f348..6f4da89 100644 --- a/soleprint/station/tools/tester/__init__.py +++ b/soleprint/station/tools/tester/__init__.py @@ -3,9 +3,39 @@ Tester - HTTP contract test runner with web UI. Discovers and runs contract tests from tests/ directory. Tests can be symlinked from production repos or copied during deployment. -""" -from .api import router -from .core import discover_tests, start_test_run, get_run_status +## Why the re-exports are lazy + +`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"] + +_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)) diff --git a/soleprint/station/tools/tester/cli.py b/soleprint/station/tools/tester/cli.py index 0df1435..e601f71 100644 --- a/soleprint/station/tools/tester/cli.py +++ b/soleprint/station/tools/tester/cli.py @@ -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 time -from .config import config -from .core import discover_tests, start_test_run, get_run_status +from ..common.cli import Tool, fail def cmd_discover(args): """List discovered tests.""" + from .core import 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: import json @@ -44,7 +58,10 @@ def cmd_discover(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() @@ -57,8 +74,7 @@ def cmd_run(args): if args.pattern.lower() in t.id.lower() ] if not test_ids: - print(f"No tests matching pattern: {args.pattern}") - return 1 + fail(f"no tests match the pattern: {args.pattern}") print(f"Running {len(test_ids)} tests matching '{args.pattern}'") else: print("Running all tests") @@ -68,23 +84,26 @@ def cmd_run(args): # Start run run_id = start_test_run(test_ids) - # Poll until complete - while True: - status = get_run_status(run_id) - if not status: - print("Error: Run not found") - return 1 + # 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: + status = get_run_status(run_id) + if not status: + fail(f"run not found: {run_id}") - # Print progress - if status.current_test: - sys.stdout.write(f"\r Running: {status.current_test[:60]}...") - sys.stdout.flush() + if status.current_test: + sys.stdout.write(f"\r Running: {status.current_test[:60]}...") + sys.stdout.flush() - if status.status in ("completed", "failed"): - sys.stdout.write("\r" + " " * 80 + "\r") # Clear line - break + if status.status in ("completed", "failed"): + break - time.sleep(0.5) + time.sleep(0.5) + finally: + sys.stdout.write("\r" + " " * 80 + "\r") + sys.stdout.flush() # Print results 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 -def main(args=None): - parser = argparse.ArgumentParser( - description="Contract HTTP Tests - Pure HTTP test runner" - ) - subparsers = parser.add_subparsers(dest="command", help="Available commands") - - # discover command - discover_parser = subparsers.add_parser("discover", help="List discovered tests") - discover_parser.add_argument("--json", action="store_true", help="Output as JSON") +def main(argv=None): + # `Tool` puts required=True on the subparsers. Without it, a bare + # `python -m tester` printed help and exited 0: a no-op reporting success. + tool = Tool("tester", __doc__, package=__package__) - # run command - run_parser = subparsers.add_parser("run", help="Run tests") - run_parser.add_argument("pattern", nargs="?", help="Filter tests by pattern (e.g. 'customers', 'invoices')") + tool.command("discover", cmd_discover, "List discovered tests") + tool.argument("discover", "--json", action="store_true", help="Output as JSON") - args = parser.parse_args(args) + tool.command("run", cmd_run, "Run tests") + tool.argument("run", "pattern", nargs="?", + help="Filter tests by pattern (e.g. 'customers', 'invoices')") - if args.command == "discover": - cmd_discover(args) - elif args.command == "run": - sys.exit(cmd_run(args)) - else: - parser.print_help() + # 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:". + tool.run(argv) if __name__ == "__main__":