casebench
test cases straight onto the board.
git clone https://owenrusk.dev/casebench.git
casebench / casebench/cli.py -rw-r--r-- · 6822 bytes
1 import argparse
2 import getpass
3 import os
4 import sys
5 from pathlib import Path
6
7 from . import casefile, lock
8 from .store import BENCH, Placed, Player, Store
9
10
11 class Fail(Exception):
12 pass
13
14
15 def parser() -> argparse.ArgumentParser:
16 top = argparse.ArgumentParser(prog="casebench")
17 sub = top.add_subparsers(dest="command", required=True)
18
19 put = sub.add_parser("put", help="put a case on a board")
20 put.add_argument("folder", type=Path)
21 boards(put)
22 put.set_defaults(run=run_put)
23
24 note = sub.add_parser("note", help="write a note on a case")
25 note.add_argument("case")
26 words = note.add_mutually_exclusive_group(required=True)
27 words.add_argument("text", nargs="?")
28 words.add_argument("-f", "--file", help="read the note from a file; - for stdin")
29 note.add_argument("--item", help="the item to write on; the case's title if left out")
30 boards(note)
31 note.set_defaults(run=run_note)
32
33 locker = sub.add_parser("lock", help="lock a case behind one or more answers")
34 locker.add_argument("case")
35 locker.add_argument("--answer", action="append", help="one per line of the lock; asked for if left out")
36 locker.add_argument(
37 "--rest",
38 help="tries/minutes: after that many wrong tries in a row, take nothing for that long"
39 " (default: $CASEBENCH_REST, or 3/60)",
40 )
41 locker.add_argument("--text", default="locked.", help="what the lock says (default: locked.)")
42 boards(locker)
43 locker.set_defaults(run=run_lock)
44
45 unlocker = sub.add_parser("unlock", help="take the lock off a case")
46 unlocker.add_argument("case")
47 boards(unlocker)
48 unlocker.set_defaults(run=run_unlock)
49
50 lister = sub.add_parser("list", help="what the bench has put where")
51 lister.add_argument("--player", action="append", help="a player's handle; every player if left out")
52 lister.set_defaults(run=run_list)
53
54 puller = sub.add_parser("pull", help="take a test case off a board")
55 puller.add_argument("case")
56 boards(puller)
57 puller.set_defaults(run=run_pull)
58 return top
59
60
61 def boards(command: argparse.ArgumentParser) -> None:
62 which = command.add_mutually_exclusive_group(required=True)
63 which.add_argument("--player", action="append", help="a player's handle; more than once is fine")
64 which.add_argument("--all", action="store_true", help="every player's board")
65
66
67 def main(argv: list[str] | None = None, store: Store | None = None) -> int:
68 args = parser().parse_args(argv)
69 opened = store is None
70 if store is None:
71 dsn = os.environ.get("CASEBENCH_DSN")
72 if not dsn:
73 print("casebench: set CASEBENCH_DSN", file=sys.stderr)
74 return 2
75 from .pgstore import PgStore # psycopg only when there's a database to talk to
76
77 store = PgStore(dsn)
78 try:
79 args.run(store, args)
80 except (Fail, ValueError) as e:
81 print(f"casebench: {e}", file=sys.stderr)
82 return 1
83 finally:
84 if opened:
85 store.close()
86 return 0
87
88
89 def targets(store: Store, args: argparse.Namespace) -> list[Player]:
90 if getattr(args, "all", False) or not args.player:
91 return store.players()
92 found = []
93 for handle in args.player:
94 who = store.player(handle)
95 if who is None:
96 raise Fail(f"no player {handle}")
97 found.append(who)
98 return found
99
100
101 def on_board(store: Store, who: Player, case_id: str) -> Placed:
102 placed = store.placed(who, case_id)
103 if placed is None:
104 raise Fail(f"{who.handle} doesn't have {case_id}")
105 return placed
106
107
108 def run_put(store: Store, args: argparse.Namespace) -> None:
109 case = casefile.load(args.folder)
110 players = targets(store, args)
111 for who in players:
112 placed = store.placed(who, case.id)
113 if placed is not None and placed.source != BENCH:
114 raise Fail(f"{who.handle} already has {case.id} from the runtime")
115 for who in players:
116 store.put(who, case)
117 print(f"{who.handle}: put {case.id}")
118
119
120 def run_note(store: Store, args: argparse.Namespace) -> None:
121 if args.file is None:
122 text = args.text
123 elif args.file == "-":
124 text = sys.stdin.read()
125 else:
126 text = Path(args.file).read_text(encoding="utf-8")
127 text = text.rstrip()
128 if not text:
129 raise Fail("nothing to write")
130 players = targets(store, args)
131 for who in players:
132 on_board(store, who, args.case)
133 if args.item and args.item not in store.items(who, args.case):
134 raise Fail(f"{who.handle}: {args.case} has no item {args.item}")
135 where = args.case + (f" / {args.item}" if args.item else "")
136 for who in players:
137 store.note(who, args.case, args.item, text)
138 print(f"{who.handle}: noted {where}")
139
140
141 def run_lock(store: Store, args: argparse.Namespace) -> None:
142 rest = lock.parse_rest(args.rest or os.environ.get("CASEBENCH_REST", "3/60"))
143 players = targets(store, args)
144 for who in players:
145 on_board(store, who, args.case)
146 new = lock.make(args.answer or ask(), rest, args.text)
147 for who in players:
148 store.lock(who, args.case, new)
149 print(f"{who.handle}: locked {args.case}")
150
151
152 def ask() -> list[str]:
153 # answers typed here stay out of the shell's history. an empty line ends them.
154 answers = []
155 while answer := getpass.getpass(f"line {len(answers) + 1}: "):
156 answers.append(answer)
157 return answers
158
159
160 def run_unlock(store: Store, args: argparse.Namespace) -> None:
161 unlocked = 0
162 for who in targets(store, args):
163 if store.unlock(who, args.case):
164 unlocked += 1
165 print(f"{who.handle}: unlocked {args.case}")
166 else:
167 print(f"{who.handle}: {args.case} wasn't locked")
168 if not unlocked:
169 raise Fail(f"{args.case} wasn't locked anywhere")
170
171
172 def run_list(store: Store, args: argparse.Namespace) -> None:
173 for who in targets(store, args):
174 for placed in store.board(who):
175 if placed.source != BENCH and placed.lock is None:
176 continue # the runtime's own, untouched
177 marks = [placed.source]
178 if placed.lock:
179 marks.append(placed.lock)
180 if placed.notes:
181 marks.append(f"{placed.notes} note" + ("" if placed.notes == 1 else "s"))
182 print("\t".join([who.handle, placed.case_id, placed.title, ", ".join(marks)]))
183
184
185 def run_pull(store: Store, args: argparse.Namespace) -> None:
186 players = targets(store, args)
187 for who in players:
188 if on_board(store, who, args.case).source != BENCH:
189 raise Fail(f"{who.handle}: {args.case} came from the runtime; the bench only pulls what it put there")
190 for who in players:
191 store.pull(who, args.case)
192 print(f"{who.handle}: pulled {args.case}")