83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
|
"""CLI: python -m rtut {list,validate,show,render} ..."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from . import pack as P
|
||
|
|
|
||
|
|
|
||
|
|
def _packs(root: str) -> list[P.Pack]:
|
||
|
|
return [P.load(p) for p in P.discover(Path(root))]
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv: list[str] | None = None) -> int:
|
||
|
|
ap = argparse.ArgumentParser(prog="rtut")
|
||
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
|
for name in ("list", "validate"):
|
||
|
|
sp = sub.add_parser(name)
|
||
|
|
sp.add_argument("root", nargs="?", default=".")
|
||
|
|
sp = sub.add_parser("launch"); sp.add_argument("manifest")
|
||
|
|
sp.add_argument("--timeout", type=float, default=30.0)
|
||
|
|
sp = sub.add_parser("show"); sp.add_argument("manifest")
|
||
|
|
sp = sub.add_parser("render"); sp.add_argument("manifest"); sp.add_argument("step")
|
||
|
|
sp.add_argument("--param", "-p", action="append", default=[], help="name=value")
|
||
|
|
a = ap.parse_args(argv)
|
||
|
|
|
||
|
|
if a.cmd in ("list", "validate"):
|
||
|
|
packs = _packs(a.root)
|
||
|
|
if not packs:
|
||
|
|
print("no packs found", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
bad = 0
|
||
|
|
for p in packs:
|
||
|
|
d = p.data
|
||
|
|
if a.cmd == "list":
|
||
|
|
print(f"{p.id}\t{d.get('engine')}\t{d.get('exercise_status')}\t{d.get('title')}")
|
||
|
|
else:
|
||
|
|
print(f"{'ok ' if p.ok else 'FAIL'} {p.id}")
|
||
|
|
for e in p.errors:
|
||
|
|
print(f" - {e}")
|
||
|
|
bad += not p.ok
|
||
|
|
return 1 if bad and a.cmd == "validate" else 0
|
||
|
|
pk = P.load(Path(a.manifest))
|
||
|
|
if not pk.ok:
|
||
|
|
print("\n".join(pk.errors), file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
if a.cmd == "launch":
|
||
|
|
from . import legacy
|
||
|
|
try:
|
||
|
|
proc = legacy.launch(pk, a.timeout)
|
||
|
|
except (ValueError, TimeoutError) as e:
|
||
|
|
print(f"error: {e}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
print(f"running: {pk.data['legacy'].get('url', '(no url)')} (Ctrl-C stops it)")
|
||
|
|
try:
|
||
|
|
return proc.wait()
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
proc.terminate()
|
||
|
|
return 0
|
||
|
|
if a.cmd == "show":
|
||
|
|
d = pk.data
|
||
|
|
print(f"{d['title']} [{d['exercise_status']}] engine={d['engine']}")
|
||
|
|
for i, s in enumerate(d.get("steps", []), 1):
|
||
|
|
print(f"{i}. [{s['owner']}] {s['title']} ({s.get('risk', 'read-only')})")
|
||
|
|
print(f" done when: {s['verify']['done_when']}")
|
||
|
|
return 0
|
||
|
|
values = dict(kv.split("=", 1) for kv in a.param)
|
||
|
|
step = next((s for s in pk.data.get("steps", []) if s["id"] == a.step), None)
|
||
|
|
if not step or "command" not in step:
|
||
|
|
print("no such step or step has no command", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
try:
|
||
|
|
print(P.render(step["command"], values, pk))
|
||
|
|
except (KeyError, ValueError) as e:
|
||
|
|
print(f"error: {e}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|