vumast3r
111a6c63a0
Remaining findings from the repo audit (1, 2 and 4 landed in df04c56).
restore_baseline now undoes two more leaks, both of which cost real debugging
time this week:
- apiserver (Q123 renames the command to "kube-apiserverr"). This runs FIRST
and over podman/ssh rather than kubectl, because it is exactly the failure
that makes kubectl unusable - every later step in the block depends on the
API being up. It waits for /readyz before continuing.
- staticPodPath (Q103 relocates it to /etc/kubernetes/static on the tier=edge
node). The existing kubelet repair only fired when
/var/lib/kubelet/config.yaml.orig existed, and nothing ever creates that
backup, so a relocated path survived every later session and silently broke
Q05, whose check hardcodes /etc/kubernetes/manifests. Normalised
unconditionally, restarting the kubelet only when the file actually changed.
Verified against the live cluster: with both breaks applied the API was
unreachable, and one restore_baseline run brought all four nodes Ready, restored
the manifest, reset staticPodPath and let the kubelet re-adopt the Q05 mirror
pod. Re-running takes 1.1s versus 14.6s when repairing, confirming it is
idempotent and does not restart anything it has not changed.
etcd (Q102) and certs (Q122) are deliberately left alone, now documented in the
block: neither setup damages anything, so any damage comes from a half-finished
candidate attempt, and "mid-attempt" is indistinguishable from "done correctly"
without guessing - guessing wrong rewrites etcd's data-dir or re-issues certs on
a healthy cluster. proxy-rules (Q302) needs no repair at all; its "breaks" key
exists for no_conflicts() bookkeeping, not because it touches shared state.
Also:
- S[30], S[42], S[108] referenced params that do not exist (%{svc}/%{app},
%{np}/%{port}, %{rev}), so they rendered placeholders verbatim. Rewritten
against the real task text - S[42] was missing the DNS egress rule entirely
and S[108] stopped before the scale/pause/set-image steps, so both would
have failed 2 of 3 checks if followed literally. A param-reference scan over
the whole bank is now clean.
- Q124's check and S[124] used the deprecated v1 Endpoints API; switched to
EndpointSlice, verified both forms return 10.89.1.3:6443 on this cluster.
- q_by_id printed "Valid: 1-101" for non-contiguous ids; it now derives the
real ranges (1-43, 101-124, 201-217, 301-317).
- Untracked workspace/q03. Both q03 and q21 setups start with rm -rf and
rewrite every file from heredocs, so they are generated output, not seed
templates: tracking them only produced phantom diffs, since a re-seed
reverted edits to base/ and deleted anything left in overlays/. With nothing
tracked there, teardown.sh can go back to wiping workspace/ wholesale.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Fqcio9TXkEXH35UHebTgoL
1238 lines
61 KiB
Python
Executable File
1238 lines
61 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""exam - Kubernetes certification exam simulator (CKA / CKAD / CKS tracks).
|
|
Hands-on tasks graded per-check against the live cluster, exam clock, attempt history,
|
|
weak-spot log, drill mode, and a browser front end modelled on the real exam."""
|
|
import http.server
|
|
import importlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
from collections.abc import Callable
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
VERSION = "2026.09.07.1"
|
|
ROOT = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, ROOT)
|
|
|
|
# ------------------------------------------------------------ track ----
|
|
ARGV = sys.argv[1:]
|
|
TRACKS = ["cka", "ckad", "cks"]
|
|
TRACK_ID = ""
|
|
TRACK = {}
|
|
QUESTIONS = []
|
|
DOMAINS = {}
|
|
QUOTA = {}
|
|
QUOTA_HARD = {}
|
|
QBYID = {}
|
|
SETS = {}
|
|
POOL = []
|
|
EXAM_SECONDS = 0
|
|
PASS_PCT = 0
|
|
STATE_FILE = ""
|
|
HISTORY_FILE = ""
|
|
WEAK_FILE = ""
|
|
|
|
|
|
def load_track(track_id):
|
|
"""Bind every track-dependent global. Called at startup and when the UI switches track."""
|
|
try:
|
|
mod = importlib.import_module(f"tracks.{track_id}.questions")
|
|
except ModuleNotFoundError:
|
|
print(f"unknown track '{track_id}' (no tracks/{track_id}/questions.py)"); sys.exit(1)
|
|
importlib.reload(mod)
|
|
track, questions = mod.TRACK, list(mod.QUESTIONS)
|
|
for q in questions:
|
|
q["weight"] = sum(c["pts"] for c in q["checks"])
|
|
q.setdefault("difficulty", "normal")
|
|
q.setdefault("set", None)
|
|
g = globals()
|
|
g.update(
|
|
TRACK_ID=track_id, TRACK=track, QUESTIONS=questions, DOMAINS=track["domains"],
|
|
QUOTA=track.get("quota", {}), QUOTA_HARD=track.get("quota_hard", track.get("quota", {})),
|
|
QBYID={q["id"]: q for q in questions},
|
|
SETS=track.get("sets", {}),
|
|
POOL=[q for q in questions if not q.get("set")],
|
|
EXAM_SECONDS=track["minutes"] * 60, PASS_PCT=track["pass_pct"],
|
|
STATE_FILE=os.path.join(ROOT, ".state", f"state-{track_id}.json"),
|
|
HISTORY_FILE=os.path.join(ROOT, f"history-{track_id}.jsonl"),
|
|
WEAK_FILE=os.path.join(ROOT, f"weak-spots-{track_id}.md"),
|
|
)
|
|
|
|
|
|
def set_questions(name):
|
|
return [q for q in QUESTIONS if q.get("set") == name]
|
|
|
|
|
|
STATE_DIR = os.path.join(ROOT, ".state")
|
|
WORKSPACE = os.path.join(ROOT, "workspace")
|
|
# How {WS} is shown in task text and solutions. Relative to the repo root by default, since that
|
|
# is where you are working; graders always use the absolute $WS internally.
|
|
WS_DISPLAY = os.environ.get("CKA_WS_DISPLAY") or os.path.relpath(WORKSPACE, ROOT)
|
|
DRILL_SECONDS_PER_TASK = 6 * 60
|
|
DRILL_SECONDS_PER_TASK_HARD = 5 * 60
|
|
DRILL_SIZE = 5
|
|
import random
|
|
NAMESPACES_TO_SETTLE = ["apps", "netpol", "svc", "ingress-app", "dns-check", "storage"]
|
|
|
|
# --- cluster targeting (override with env vars to use a non-kind cluster) ---
|
|
CONTEXT = os.environ.get("CKA_CONTEXT", "kind-cka-sim")
|
|
def _default_access():
|
|
try:
|
|
return open(os.path.join(STATE_DIR, "runtime")).read().strip() # written by setup.sh
|
|
except OSError:
|
|
return "podman"
|
|
NODE_ACCESS = os.environ.get("CKA_NODE_ACCESS", _default_access()) # podman | docker | ssh | vagrant
|
|
NODES = {
|
|
"CP": os.environ.get("CKA_CP", "cka-sim-control-plane"),
|
|
"W1": os.environ.get("CKA_W1", "cka-sim-worker"),
|
|
"W2": os.environ.get("CKA_W2", "cka-sim-worker2"),
|
|
"W3": os.environ.get("CKA_W3", "cka-sim-worker3"),
|
|
}
|
|
def shell_hint(node):
|
|
return {"podman": f"podman exec -it {node} bash",
|
|
"docker": f"docker exec -it {node} bash",
|
|
"ssh": f"ssh {node}",
|
|
"vagrant": f"vagrant ssh {node}"}.get(NODE_ACCESS, f"ssh {node}")
|
|
|
|
PRELUDE = r"""
|
|
set -uo pipefail
|
|
K="kubectl --context ${CKA_CONTEXT:-kind-cka-sim}" # grading never depends on the candidate's current context
|
|
WS="$WORKSPACE"
|
|
# j [kubectl get args...] '{jsonpath}' -> prints the jsonpath result ("" on error)
|
|
j() { local n=$#; local path="${!n}"; $K get "${@:1:n-1}" -o "jsonpath=$path" 2>/dev/null; }
|
|
eq() { [ "$1" = "$2" ]; }
|
|
CP="${CKA_CP:-cka-sim-control-plane}"; W1="${CKA_W1:-cka-sim-worker}"
|
|
W2="${CKA_W2:-cka-sim-worker2}"; W3="${CKA_W3:-cka-sim-worker3}"
|
|
# nx NODE 'command' -> run a shell command on a node (stdin is forwarded)
|
|
nx() {
|
|
local n="$1"; shift
|
|
case "${CKA_NODE_ACCESS:-podman}" in
|
|
ssh) ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new "$n" "$@" ;;
|
|
vagrant) vagrant ssh "$n" -c "$*" -- -o LogLevel=QUIET ;;
|
|
docker) docker exec -i "$n" sh -c "$*" ;;
|
|
*) podman exec -i "$n" sh -c "$*" ;;
|
|
esac
|
|
}
|
|
# ensure_etcd_tools -> etcdctl and etcdutl on the control-plane node, as on the real exam.
|
|
# The etcd image is distroless, so they cannot be copied out of the pod; etcd-tools.sh takes
|
|
# them from the image in the node's containerd (and falls back to a download / apt).
|
|
ensure_etcd_tools() {
|
|
nx "$CP" 'command -v etcdctl >/dev/null 2>&1 && command -v etcdutl >/dev/null 2>&1' </dev/null && return 0
|
|
bash "${CKA_ROOT:-.}/etcd-tools.sh" --emit | nx "$CP" 'sh -s'
|
|
}
|
|
"""
|
|
|
|
def _pick_track_arg(argv):
|
|
t = os.environ.get("CKA_TRACK", "cka")
|
|
if "--track" in argv:
|
|
i = argv.index("--track"); t = argv[i + 1]; del argv[i:i + 2]
|
|
return t
|
|
|
|
|
|
load_track(_pick_track_arg(ARGV))
|
|
|
|
TTY = sys.stdout.isatty()
|
|
def c(code, s): return f"\033[{code}m{s}\033[0m" if TTY else s
|
|
def green(s): return c("32", s)
|
|
def red(s): return c("31", s)
|
|
def yellow(s): return c("33", s)
|
|
def bold(s): return c("1", s)
|
|
def dim(s): return c("2", s)
|
|
def die(msg, code=1):
|
|
print(red(msg)); sys.exit(code)
|
|
|
|
# ------------------------------------------------------------- state ----
|
|
def load_state() -> dict[str, Any] | None:
|
|
if not os.path.exists(STATE_FILE):
|
|
return None
|
|
with open(STATE_FILE) as f:
|
|
return json.load(f)
|
|
|
|
def save_state(st: dict[str, Any]) -> None:
|
|
os.makedirs(STATE_DIR, exist_ok=True)
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump(st, f, indent=2)
|
|
|
|
def paused_total(st: dict[str, Any] | None) -> float:
|
|
"""Seconds spent paused so far, including the current pause if any."""
|
|
if not st:
|
|
return 0.0
|
|
t = float(st.get("paused_total", 0))
|
|
if st.get("paused_at"):
|
|
t += time.time() - float(st["paused_at"])
|
|
return t
|
|
|
|
def untimed(st: dict[str, Any] | None) -> bool:
|
|
return bool(st and st.get("untimed"))
|
|
|
|
def remaining(st: dict[str, Any] | None) -> float:
|
|
if not st:
|
|
return 0.0
|
|
if untimed(st):
|
|
return float("inf")
|
|
return float(st.get("started_at", 0)) + float(st.get("duration", EXAM_SECONDS)) + paused_total(st) - time.time()
|
|
|
|
def set_paused(st: dict[str, Any], on: bool) -> None:
|
|
if on and not st.get("paused_at"):
|
|
st["paused_at"] = time.time(); st["pauses"] = int(st.get("pauses", 0)) + 1
|
|
elif not on and st.get("paused_at"):
|
|
st["paused_total"] = float(st.get("paused_total", 0)) + (time.time() - float(st["paused_at"])); st["paused_at"] = None
|
|
save_state(st)
|
|
|
|
def fmt(secs: float | int) -> str:
|
|
secs_int = max(0, int(secs))
|
|
return f"{secs_int // 3600:02d}:{(secs_int % 3600) // 60:02d}:{secs_int % 60:02d}"
|
|
|
|
def require_started() -> dict[str, Any]:
|
|
st = load_state()
|
|
if not st:
|
|
die("No session in progress. Run: ./exam start (or ./exam drill weak)")
|
|
assert st is not None
|
|
return st
|
|
|
|
def session_questions(st: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
if not st:
|
|
return list(QUESTIONS)
|
|
ids = st.get("ids") or [q["id"] for q in QUESTIONS]
|
|
return [q for q in QUESTIONS if q["id"] in ids]
|
|
|
|
def q_by_id(n: int) -> dict[str, Any]:
|
|
for q in QUESTIONS:
|
|
if q["id"] == n:
|
|
return q
|
|
# ids are not contiguous (1-43 base, 101-124 hard, 201-217 / 301-317 sim sets), so print the
|
|
# real ranges rather than "1-<count>", which named ids that do not exist.
|
|
ids = sorted(q["id"] for q in QUESTIONS)
|
|
ranges, start, prev = [], ids[0], ids[0]
|
|
for i in ids[1:] + [None]:
|
|
if i != prev + 1:
|
|
ranges.append(str(start) if start == prev else f"{start}-{prev}")
|
|
start = i
|
|
prev = i
|
|
die(f"No question {n}. Valid: {', '.join(ranges)}")
|
|
return {}
|
|
|
|
def render(text, params):
|
|
"""Substitute %{param}, {WS}, {CP}.. and {SHELL_CP}.. placeholders."""
|
|
t = text.replace("{WS}", WS_DISPLAY)
|
|
for k, v in (params or {}).items():
|
|
t = t.replace("%{" + k + "}", str(v))
|
|
for k, v in NODES.items():
|
|
t = t.replace("{" + k + "}", v).replace("{SHELL_" + k + "}", shell_hint(v))
|
|
return t
|
|
|
|
def session_params(st, q):
|
|
"""Params drawn for this session; outside a session, a fixed sample so text still reads naturally."""
|
|
p = (st or {}).get("params", {}).get(str(q["id"]))
|
|
return p if p is not None else make_params(q, random.Random(q["id"]))
|
|
|
|
def task_text(q, params=None):
|
|
return render(q["task"].strip(), params if params is not None else session_params(load_state(), q))
|
|
|
|
def make_params(q, rng):
|
|
fn = q.get("params")
|
|
return fn(rng) if fn else {}
|
|
|
|
# -------------------------------------------------------------- shell ----
|
|
def run(script, timeout=180, params=None):
|
|
env = dict(os.environ, WORKSPACE=WORKSPACE, CKA_ROOT=ROOT, CKA_NODE_ACCESS=NODE_ACCESS, CKA_CONTEXT=CONTEXT,
|
|
**{f"CKA_{k}": v for k, v in NODES.items()})
|
|
p = subprocess.run(["bash", "-c", PRELUDE + render(script, params)], env=env, capture_output=True, text=True,
|
|
timeout=timeout, stdin=subprocess.DEVNULL)
|
|
return p.returncode, (p.stdout + p.stderr).strip()
|
|
|
|
def kubectl_ok():
|
|
try:
|
|
return subprocess.run(["kubectl", "--context", CONTEXT, "get", "--raw=/version"],
|
|
capture_output=True, text=True, timeout=15).returncode == 0
|
|
except Exception:
|
|
return False
|
|
|
|
def grade_one(q, params=None):
|
|
"""Run every check independently. Returns a result dict with partial credit."""
|
|
if params is None:
|
|
params = session_params(load_state(), q)
|
|
checks = []
|
|
for ch in q["checks"]:
|
|
try:
|
|
rc, out = run(ch["script"], timeout=240, params=params)
|
|
except subprocess.TimeoutExpired:
|
|
rc, out = 1, "timed out"
|
|
checks.append({"name": render(ch["name"], params), "pts": ch["pts"], "pass": rc == 0, "out": out[-400:]})
|
|
earned = sum(ch["pts"] for ch in checks if ch["pass"])
|
|
return {"pass": earned == q["weight"], "earned": earned, "total": q["weight"],
|
|
"checks": checks, "at": time.time(), "difficulty": q["difficulty"]}
|
|
|
|
def result_detail(r):
|
|
return "\n".join((" ✓ " if ch["pass"] else " ✗ ") + ch["name"] + f" ({ch['pts']} pt)" for ch in r["checks"])
|
|
|
|
def score(st, results=None):
|
|
results = results if results is not None else st["results"]
|
|
qs = session_questions(st)
|
|
total = sum(q["weight"] for q in qs)
|
|
earned = sum(results.get(str(q["id"]), {}).get("earned", 0) for q in qs)
|
|
return earned, total, (round(100 * earned / total, 1) if total else 0)
|
|
|
|
# ----------------------------------------------------------- history ----
|
|
def load_history():
|
|
if not os.path.exists(HISTORY_FILE):
|
|
return []
|
|
with open(HISTORY_FILE) as f:
|
|
return [json.loads(l) for l in f if l.strip()]
|
|
|
|
def record_attempt(st):
|
|
earned, total, pct = score(st)
|
|
entry = {"at": datetime.now().isoformat(timespec="minutes"), "mode": st.get("mode", "exam"), "hard": st.get("hard", False),
|
|
"ids": [q["id"] for q in session_questions(st)], "score_pct": pct, "earned": earned, "total": total,
|
|
"time_used": int(st["ended_at"] - st["started_at"] - paused_total(st)),
|
|
"pauses": st.get("pauses", 0), "paused_seconds": int(paused_total(st)),
|
|
"results": {k: {"pass": v["pass"], "earned": v["earned"], "total": v["total"],
|
|
"failed": [ch["name"] for ch in v["checks"] if not ch["pass"]]}
|
|
for k, v in st["results"].items()}}
|
|
with open(HISTORY_FILE, "a") as f:
|
|
f.write(json.dumps(entry) + "\n")
|
|
write_weak_spots()
|
|
return entry
|
|
|
|
class AggEntry:
|
|
def __init__(self) -> None:
|
|
self.attempts: int = 0
|
|
self.fails: int = 0
|
|
self.last: Any = None
|
|
self.last_fail_at: str = ""
|
|
self.failed: list[str] = []
|
|
self.last_earned: str = ""
|
|
|
|
def __getitem__(self, item: str) -> Any:
|
|
return getattr(self, item)
|
|
|
|
def __setitem__(self, item: str, value: Any) -> None:
|
|
setattr(self, item, value)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"attempts": self.attempts,
|
|
"fails": self.fails,
|
|
"last": self.last,
|
|
"last_fail_at": self.last_fail_at,
|
|
"failed": self.failed,
|
|
"last_earned": self.last_earned,
|
|
}
|
|
|
|
|
|
def stats() -> dict[int, AggEntry]:
|
|
agg: dict[int, AggEntry] = {int(q["id"]): AggEntry() for q in QUESTIONS}
|
|
for e in load_history():
|
|
for k, r in e.get("results", {}).items():
|
|
a = agg.get(int(k))
|
|
if a is None:
|
|
continue
|
|
a.attempts += 1
|
|
a.last = r.get("pass")
|
|
a.last_earned = f"{r.get('earned', 0)}/{r.get('total', 0)}"
|
|
if not r.get("pass"):
|
|
a.fails += 1
|
|
a.last_fail_at = str(e.get("at", ""))
|
|
a.failed = list(r.get("failed", []))
|
|
return agg
|
|
|
|
def last_seen():
|
|
"""question id -> index of the most recent session it appeared in (higher = more recent)."""
|
|
seen = {}
|
|
for i, e in enumerate(load_history()):
|
|
for k in e["results"]:
|
|
seen[int(k)] = i
|
|
return seen
|
|
|
|
def drilled_domains():
|
|
out = {d: 0 for d in DOMAINS}
|
|
for e in load_history():
|
|
if e["mode"] != "drill":
|
|
continue
|
|
for k in e["results"]:
|
|
q = QBYID.get(int(k))
|
|
if q:
|
|
out[q["domain"]] += 1
|
|
return out
|
|
|
|
def no_conflicts(cands, n, taken=()):
|
|
"""Take up to n tasks, never two that break the same thing (or clash with tasks already taken)."""
|
|
used = {q.get("breaks") for q in taken if q.get("breaks")}
|
|
out = []
|
|
for q in cands:
|
|
b = q.get("breaks")
|
|
if b and b in used:
|
|
continue
|
|
out.append(q)
|
|
if b:
|
|
used.add(b)
|
|
if len(out) == n:
|
|
break
|
|
return out
|
|
|
|
def pick(domain, n, hard, rng, exclude=(), taken=(), unseen_only=False):
|
|
"""n tasks from a domain: unseen first, then least recently seen; hard sessions prefer hard tasks."""
|
|
seen = last_seen()
|
|
pool = [q for q in POOL if q["domain"] == domain and q["id"] not in exclude
|
|
and (hard or q["difficulty"] == "normal") and not (unseen_only and q["id"] in seen)]
|
|
rng.shuffle(pool)
|
|
pool.sort(key=lambda q: (0 if (hard and q["difficulty"] == "hard") else 1, seen.get(q["id"], -1)))
|
|
return no_conflicts(pool, n, taken)
|
|
|
|
def compose_exam(hard, rng):
|
|
qs = []
|
|
for d, n in (QUOTA_HARD if hard else QUOTA).items():
|
|
qs += pick(d, n, hard, rng, taken=qs)
|
|
rng.shuffle(qs)
|
|
return qs
|
|
|
|
def campaign_batch(hard, rng):
|
|
"""'drill all': DRILL_SIZE tasks you have never attempted, round-robin across domains,
|
|
hard tasks included when --hard. When everything has been seen once, falls back to weak tasks."""
|
|
seen = last_seen()
|
|
pool = [q for q in POOL if q["id"] not in seen and (hard or q["difficulty"] == "normal")]
|
|
if not pool:
|
|
return None, []
|
|
qs, domains = [], list(DOMAINS)
|
|
rng.shuffle(pool)
|
|
pool.sort(key=lambda q: 0 if (hard and q["difficulty"] == "hard") else 1)
|
|
while len(qs) < DRILL_SIZE and pool:
|
|
progressed = False
|
|
for d in domains:
|
|
cand = [q for q in pool if q["domain"] == d and q not in qs]
|
|
got = no_conflicts(cand, 1, qs)
|
|
if got:
|
|
qs += got; progressed = True
|
|
if len(qs) == DRILL_SIZE:
|
|
break
|
|
if not progressed:
|
|
break
|
|
total = len([q for q in POOL if hard or q["difficulty"] == "normal"])
|
|
return f"{total - len(pool)}/{total} attempted", qs
|
|
|
|
def solution_allowed(st: dict[str, Any] | None, qid: int | str) -> bool:
|
|
"""Study: always. Drill: once the task has been graded. Exam: only after submit."""
|
|
if not st or st.get("ended_at") or untimed(st):
|
|
return True
|
|
return bool(st.get("mode") == "drill" and str(qid) in st.get("results", {}))
|
|
|
|
def cmd_solution(args):
|
|
if not args:
|
|
die("usage: ./exam solution N (drills: after grading the task; exams: after submit; --force to peek anyway)")
|
|
force = "--force" in args; args = [a for a in args if a != "--force"]
|
|
q = q_by_id(int(args[0])); st = load_state()
|
|
if not solution_allowed(st, int(q["id"])) and not force:
|
|
die("Not yet: in a drill, grade the task first (./exam grade N); in an exam, submit first. --force overrides.")
|
|
print(bold(f"Solution - Q{q['id']} {q['title']}")); print("-" * 64)
|
|
sol = str(q.get("solution", ""))
|
|
print(render(sol, session_params(st, q))); print("-" * 64)
|
|
print(dim(f"One correct way, not the only one. $do = --dry-run=client -o yaml, k = kubectl. Paths are relative to {ROOT}"))
|
|
|
|
def cmd_progress(args):
|
|
seen = last_seen(); agg = stats()
|
|
print(bold(f"{'Domain':<8}{'attempted':>11}{'passed':>9} {'hard attempted':>15}{'hard passed':>13}"))
|
|
for d in DOMAINS:
|
|
dq = [q for q in QUESTIONS if q["domain"] == d]
|
|
n = [q for q in dq if q["difficulty"] == "normal"]; h = [q for q in dq if q["difficulty"] == "hard"]
|
|
att = lambda L: sum(1 for q in L if q["id"] in seen); ok = lambda L: sum(1 for q in L if agg.get(int(q["id"])) and agg[int(q["id"])].last is True)
|
|
print(f"{d:<8}{att(n):>6}/{len(n):<4}{ok(n):>6}/{len(n):<4} {att(h):>9}/{len(h):<4}{ok(h):>8}/{len(h):<4}")
|
|
left = [q["id"] for q in QUESTIONS if q["id"] not in seen]
|
|
print(dim(f"\n{len(QUESTIONS) - len(left)}/{len(QUESTIONS)} tasks attempted at least once." + (f" Never seen: Q{', Q'.join(map(str, left))}" if left and len(left) <= 25 else "")))
|
|
print(dim("passed = latest attempt was a full pass. ./exam drill all [--hard] works through what's left."))
|
|
|
|
def weak_ids(agg=None):
|
|
agg = agg or stats()
|
|
return [q["id"] for q in QUESTIONS
|
|
if (agg[q["id"]]["last"] is False or agg[q["id"]]["fails"] >= 2)
|
|
or (agg[q["id"]]["attempts"] == 0 and not q.get("set"))]
|
|
|
|
def write_weak_spots():
|
|
agg = stats(); hist = load_history()
|
|
L = [f"# {TRACK_ID.upper()} weak spots", "", f"_Generated by exam from {len(hist)} attempt(s); regenerated on every `end`._", ""]
|
|
exams = [e for e in hist if e["mode"] == "exam"]
|
|
if exams:
|
|
e = exams[-1]
|
|
L += [f"## Domain scores (latest full exam, {e['at']}, {e['score_pct']}%)", ""]
|
|
for d, name in DOMAINS.items():
|
|
dq = [q for q in QUESTIONS if q["domain"] == d]
|
|
dt = sum(q["weight"] for q in dq); de = sum(e["results"].get(str(q["id"]), {}).get("earned", 0) for q in dq)
|
|
L.append(f"- **{d}** {de}/{dt} — {name}")
|
|
L.append("")
|
|
L += ["## Drill next (`./exam drill weak`)", ""]
|
|
weak = [q for q in QUESTIONS if q["id"] in weak_ids(agg) and agg[q["id"]]["attempts"] > 0]
|
|
if not weak:
|
|
L.append("Nothing failing right now.")
|
|
for q in weak:
|
|
a = agg[q["id"]]
|
|
L.append(f"### Q{q['id']:02d} · {q['domain']} · {q['title']} ({a['fails']}/{a['attempts']} failed, last {a['last_earned']} pts, {a['last_fail_at']})")
|
|
failed_list = list(a["failed"]) if isinstance(a["failed"], (list, tuple)) else []
|
|
L += [f"- ✗ {n}" for n in failed_list]
|
|
L.append("")
|
|
L += ["## Attempt matrix", "", "| Q | Domain | Title | " + " | ".join(f"{e['mode'][0].upper()}{i + 1}" for i, e in enumerate(hist)) + " |",
|
|
"|---|---|---|" + "---|" * len(hist)]
|
|
for q in QUESTIONS:
|
|
cells = []
|
|
for e in hist:
|
|
r = e["results"].get(str(q["id"]))
|
|
cells.append("·" if r is None else ("✓" if r["pass"] else f"{r['earned']}/{r['total']}"))
|
|
L.append(f"| {q['id']} | {q['domain']} | {q['title']} | " + " | ".join(cells) + " |")
|
|
L += ["", "_E = full exam, D = drill, · = not in that session, n/m = partial credit._", ""]
|
|
with open(WEAK_FILE, "w") as f:
|
|
f.write("\n".join(L))
|
|
|
|
# ------------------------------------------------------------ session ----
|
|
def preflight():
|
|
out = subprocess.run(["kubectl", "get", "nodes", "-o", "jsonpath={.items[*].metadata.name}"],
|
|
capture_output=True, text=True).stdout.split()
|
|
missing = [v for v in NODES.values() if v not in out]
|
|
if missing:
|
|
die(f"Nodes not found in cluster: {missing}\n cluster has: {out}\n"
|
|
" Set CKA_CP, CKA_W1, CKA_W2, CKA_W3 to your node names (needs 1 control-plane + 3 workers).")
|
|
rc, _ = run('nx "$CP" "true" </dev/null && nx "$W2" "true" </dev/null && nx "$W3" "true" </dev/null', timeout=60)
|
|
if rc != 0:
|
|
die(f"Cannot run commands on nodes with CKA_NODE_ACCESS={NODE_ACCESS}.\n"
|
|
" podman/docker -> node names must be kind container names (kind get nodes --name cka-sim)\n"
|
|
" ssh -> passwordless ssh to each node name must work\n"
|
|
" vagrant -> run from the Vagrantfile directory; names are vagrant machine names")
|
|
print(dim(f" track: {TRACK_ID} cluster: {CONTEXT} node access: {NODE_ACCESS} nodes: {', '.join(NODES.values())}"))
|
|
for tool, why in (("helm", "Helm tasks (the real exam has helm on the terminal)"), ("openssl", "certificate tasks"), ("curl", "tasks that fetch manifests")):
|
|
if not shutil.which(str(tool)):
|
|
print(yellow(f" warning: '{tool}' not found on this machine - needed for {why}"))
|
|
|
|
HARD_CONTEXTS = {"k8s-c1-H": "default", "k8s-c2-AC": "%{ns}", "k8s-c3-CCC": "kube-system"}
|
|
|
|
def context_namespace(ctx, template, params_by_id):
|
|
"""Default namespace for one exam context.
|
|
|
|
Several tasks can name the same context while working in different namespaces, but a
|
|
context has only one default namespace - so it belongs to whichever task actually reads it
|
|
("ctx_ns": True, i.e. the task says to work in the context's namespace without -n). Every
|
|
other task states its namespace in its own text, so the value does not matter to them.
|
|
"""
|
|
if "%{ns}" not in template:
|
|
return template
|
|
def ns_of(pred):
|
|
return [p["ns"] for i, p in params_by_id.items()
|
|
if p.get("ctx") == ctx and p.get("ns") and pred(QBYID.get(int(i)) or {})]
|
|
return (ns_of(lambda q: q.get("ctx_ns")) or ns_of(lambda q: True)
|
|
or [p["ns"] for p in params_by_id.values() if p.get("ns")] or ["default"])[0]
|
|
|
|
def setup_contexts(hard, params_by_id):
|
|
"""Exam-style contexts on the same cluster with different default namespaces.
|
|
|
|
Created whenever a task in the session asks for one - 24 tasks open with
|
|
`kubectl config use-context k8s-c?-?` and are unfollowable without it - not only when the
|
|
whole session was started with --hard. A drill that names hard task ids explicitly
|
|
('./exam drill 121', './exam drill weak') pulls those tasks in with hard=False.
|
|
"""
|
|
cluster = subprocess.run(["kubectl", "config", "view", "-o", f"jsonpath={{.contexts[?(@.name=='{CONTEXT}')].context.cluster}}"],
|
|
capture_output=True, text=True).stdout.strip()
|
|
user = subprocess.run(["kubectl", "config", "view", "-o", f"jsonpath={{.contexts[?(@.name=='{CONTEXT}')].context.user}}"],
|
|
capture_output=True, text=True).stdout.strip()
|
|
for name in HARD_CONTEXTS:
|
|
subprocess.run(["kubectl", "config", "delete-context", name], capture_output=True)
|
|
wanted = set(HARD_CONTEXTS) if hard else {p.get("ctx") for p in params_by_id.values()} & set(HARD_CONTEXTS)
|
|
if not wanted:
|
|
subprocess.run(["kubectl", "config", "use-context", CONTEXT], capture_output=True)
|
|
return
|
|
for name in sorted(wanted):
|
|
subprocess.run(["kubectl", "config", "set-context", name, f"--cluster={cluster}", f"--user={user}",
|
|
f"--namespace={context_namespace(name, HARD_CONTEXTS[name], params_by_id)}"], capture_output=True)
|
|
# Hard mode starts you in c1 like the real exam; a mixed drill stays on the normal context
|
|
# so its non-hard tasks behave as usual and the hard ones still switch as instructed.
|
|
start = "k8s-c1-H" if hard and "k8s-c1-H" in wanted else CONTEXT
|
|
subprocess.run(["kubectl", "config", "use-context", start], capture_output=True)
|
|
|
|
def _log_msg(*args: Any, **kwargs: Any) -> None:
|
|
print(*args, **kwargs)
|
|
|
|
|
|
# Disruptive tasks (see "breaks": ... in the question bank) corrupt shared control-plane /
|
|
# node state. A session's setup only touches the questions it actually contains, so damage
|
|
# from a task in an earlier session lingers if that task isn't picked again - e.g. Q39 wrecks
|
|
# W1's kubelet config, and a later session with different questions never undoes it. This
|
|
# restores every known "breaks" target to a good baseline before each session seeds, so
|
|
# leftover damage never leaks across sessions regardless of which questions get sampled.
|
|
RESTORE_BASELINE = r"""
|
|
# apiserver FIRST: Q123 renames the command to "kube-apiserverr", which takes the whole API
|
|
# down - every kubectl-based step below would fail until this is undone. The repair goes over
|
|
# podman/ssh (nx), not kubectl, so it works precisely when the cluster is unreachable.
|
|
if nx "$CP" 'grep -q "^ *- kube-apiserverr$" /etc/kubernetes/manifests/kube-apiserver.yaml' </dev/null 2>/dev/null; then
|
|
nx "$CP" 'sed -i "s|^\( *\)- kube-apiserverr$|\1- kube-apiserver|" /etc/kubernetes/manifests/kube-apiserver.yaml' </dev/null
|
|
# the kubelet re-reads the manifest on its own; wait for the API to answer again
|
|
for _ in $(seq 1 30); do kubectl --context "${CKA_CONTEXT:-kind-cka-sim}" get --raw=/readyz >/dev/null 2>&1 && break; sleep 2; done
|
|
fi
|
|
|
|
# scheduler: undo the "kube-scheduler" -> "kube-schedulerr" command typo (Q20/Q118)
|
|
nx "$CP" '
|
|
f=/etc/kubernetes/manifests/kube-scheduler.yaml
|
|
sed -i "s|^\( *\)- kube-schedulerr$|\1- kube-scheduler|" $f 2>/dev/null
|
|
' </dev/null
|
|
|
|
# controller-manager: undo Q38's injected flag and Q118's broken image tag
|
|
nx "$CP" '
|
|
f=/etc/kubernetes/manifests/kube-controller-manager.yaml
|
|
sed -i "/--leader-elect-lease-duration=1s/d" $f 2>/dev/null
|
|
sed -i "s|\(image: registry.k8s.io/kube-controller-manager:v[0-9.]*\)-broken|\1|" $f 2>/dev/null
|
|
' </dev/null
|
|
|
|
# coredns: restore the default upstream forward (Q37/Q114 point it somewhere bogus)
|
|
cf=$($K -n kube-system get cm coredns -o jsonpath='{.data.Corefile}' 2>/dev/null)
|
|
if [ -n "$cf" ] && ! printf '%s' "$cf" | grep -Eq '^[[:space:]]*forward \. /etc/resolv\.conf'; then
|
|
printf '%s' "$cf" | sed -E 's|^([[:space:]]*forward \. )[^[:space:]{]+|\1/etc/resolv.conf|' > /tmp/Corefile.restore
|
|
$K -n kube-system create cm coredns --from-file=Corefile=/tmp/Corefile.restore --dry-run=client -o yaml | $K apply -f - >/dev/null 2>&1
|
|
$K -n kube-system rollout restart deploy coredns >/dev/null 2>&1
|
|
fi
|
|
|
|
# w1/w2/w3 kubelet: undo Q17/Q39-style tampering, then make sure it's enabled and running.
|
|
# staticPodPath is normalised unconditionally: Q103 relocates it to /etc/kubernetes/static, and
|
|
# the config.yaml.orig copy below only helps if something made that backup first - nothing does,
|
|
# so a relocated path used to survive every later session and silently break Q05 (whose check
|
|
# looks in /etc/kubernetes/manifests). Only restart the kubelet if we actually changed the file.
|
|
for n in "$W1" "$W2" "$W3"; do
|
|
nx "$n" '
|
|
[ -f /var/lib/kubelet/config.yaml.orig ] && cp -f /var/lib/kubelet/config.yaml.orig /var/lib/kubelet/config.yaml
|
|
if grep -q "^staticPodPath:" /var/lib/kubelet/config.yaml 2>/dev/null \
|
|
&& ! grep -q "^staticPodPath: /etc/kubernetes/manifests$" /var/lib/kubelet/config.yaml; then
|
|
sed -i "s|^staticPodPath: .*|staticPodPath: /etc/kubernetes/manifests|" /var/lib/kubelet/config.yaml
|
|
systemctl restart kubelet >/dev/null 2>&1
|
|
fi
|
|
systemctl enable kubelet >/dev/null 2>&1
|
|
systemctl start kubelet >/dev/null 2>&1
|
|
' </dev/null
|
|
done
|
|
|
|
# w1-taint: clear every randomized demo taint key the bank can apply. Q25 draws from
|
|
# dedicated/team/workload, Q107 from pool/lane, and Q307 pins maintenance=inprogress - miss one
|
|
# and that node silently stops accepting pods for every later session.
|
|
$K taint node "$W1" dedicated- workload- team- pool- lane- maintenance- >/dev/null 2>&1 || true
|
|
|
|
# w1-block/w2-drain/w3-drain: Q307 cordons W1 and Q117 cordons W2, and Q208/Q309 have the
|
|
# candidate drain W2/W3. A cordon left behind fails any later task that counts pods per node
|
|
# (Q26/Q211 DaemonSet coverage, Q212's one-pod-per-node spread).
|
|
for n in "$W1" "$W2" "$W3"; do $K uncordon "$n" >/dev/null 2>&1 || true; done
|
|
|
|
# Deliberately NOT restored, and why:
|
|
# etcd (Q102) and certs (Q122). Neither setup damages anything - Q102 only takes a snapshot
|
|
# and deletes a ConfigMap, Q122 only records the current notBefore. Any damage comes from a
|
|
# half-finished candidate attempt (etcd.yaml left pointing at a restored data-dir, static pod
|
|
# manifests moved to /tmp and never moved back), which is not deterministic: there is no way
|
|
# to tell "mid-attempt" from "correctly completed" without guessing, and guessing wrong here
|
|
# rewrites etcd's data-dir or re-issues certs on a healthy cluster. Re-run ./setup.sh (or
|
|
# ./teardown.sh && ./setup.sh) if one of those leaves the control plane wedged.
|
|
#
|
|
# proxy-rules (Q302) needs nothing: its setup only creates a namespace and deletes a pod and
|
|
# service. The "breaks" key is there for no_conflicts() bookkeeping, not because it damages
|
|
# shared state.
|
|
"""
|
|
|
|
def restore_baseline(log: Callable[..., Any] = _log_msg):
|
|
log(dim(" resetting control-plane / node state to a clean baseline..."))
|
|
rc, out = run(RESTORE_BASELINE, timeout=90)
|
|
if rc != 0:
|
|
log(yellow(f" warning: baseline restore had issues (continuing anyway):\n{dim(out[-500:])}"))
|
|
|
|
|
|
def seed(qs, log: Callable[..., Any] = _log_msg, params_by_id=None):
|
|
params_by_id = params_by_id or {}
|
|
normal = [q for q in qs if not q.get("disruptive")]
|
|
disruptive = [q for q in qs if q.get("disruptive")]
|
|
def one(q):
|
|
try:
|
|
rc, out = run(q["setup"], params=params_by_id.get(str(q["id"])), timeout=240)
|
|
except subprocess.TimeoutExpired:
|
|
rc, out = 1, "setup timed out after 240s"
|
|
log(f" {'✓' if rc == 0 else '!'} Q{q['id']:02d} {q['title']}" + ("" if rc == 0 else f"\n{dim(out[-300:])}"))
|
|
for q in normal:
|
|
one(q)
|
|
if disruptive:
|
|
log(dim(" waiting up to 45s for seeded pods to be scheduled..."))
|
|
for _ in range(15):
|
|
pending = subprocess.run(["kubectl", "--context", CONTEXT, "get", "pods", "-A", "--field-selector=spec.nodeName=",
|
|
"-o", "name"], capture_output=True, text=True).stdout.strip()
|
|
if not pending:
|
|
break
|
|
time.sleep(3)
|
|
for q in disruptive:
|
|
one(q)
|
|
|
|
def begin_session(qs, mode, duration, force, log: Callable[..., Any] = _log_msg, hard=False, untimed_session=False):
|
|
st = load_state()
|
|
if st and not st.get("ended_at") and not force:
|
|
die("A session is already in progress. './exam end' it first, or add --force.")
|
|
if not kubectl_ok():
|
|
die(f"Cannot reach the cluster through kubectl context '{CONTEXT}'. Run ./setup.sh (kind) or set CKA_CONTEXT for another cluster.")
|
|
if not qs:
|
|
die(f"Track '{TRACK_ID}' has no questions yet.")
|
|
preflight()
|
|
os.makedirs(WORKSPACE, exist_ok=True)
|
|
rng = random.Random()
|
|
params_by_id = {str(q["id"]): make_params(q, rng) for q in qs}
|
|
setup_contexts(hard, params_by_id)
|
|
restore_baseline(log)
|
|
log(bold(f"Seeding {len(qs)} scenario(s){' [HARD]' if hard else ''}..."))
|
|
seed(qs, log, params_by_id)
|
|
st = {"mode": mode, "track": TRACK_ID, "hard": hard, "untimed": untimed_session, "ids": [q["id"] for q in qs],
|
|
"params": params_by_id, "duration": duration, "started_at": time.time(), "ended_at": None,
|
|
"results": {}, "flags": [], "notes": ""}
|
|
save_state(st)
|
|
log("")
|
|
log(bold("=" * 64))
|
|
log(bold(f" {TRACK_ID.upper()} {'EXAM' if mode == 'exam' else 'DRILL'}{' · HARD' if hard else ''} - clock started"))
|
|
if hard:
|
|
log(yellow(" Contexts: k8s-c1-H (default) · k8s-c2-AC · k8s-c3-CCC - each task tells you which to use; you are on k8s-c1-H now"))
|
|
log(f" Ends : {datetime.fromtimestamp(st['started_at'] + duration).strftime('%H:%M:%S')} ({duration // 60} min)")
|
|
log(f" Tasks : {', '.join('Q%02d' % q['id'] for q in qs) if mode == 'drill' else len(qs)} Pass mark: {PASS_PCT}%")
|
|
log(bold("=" * 64))
|
|
log(" ./exam web exam-style browser UI (task pane + terminal + clock)\n"
|
|
" ./exam list / show N / grade N / timer / end terminal workflow\n"
|
|
+ (" Tip: read ALL tasks first. Some are cluster-wide failures that block others.\n" if mode == "exam" else "")
|
|
+ f" Files the tasks ask for go under {WS_DISPLAY}/ (relative to this directory)\n"
|
|
+ " Your own terminal: source ./examrc (fish: source ./examrc.fish) -> k, $do, $now, completion")
|
|
return st
|
|
|
|
def finish_session(st, log: Callable[..., Any] = _log_msg):
|
|
qs = session_questions(st)
|
|
over = remaining(st) <= 0
|
|
set_paused(st, False)
|
|
log(bold(f"Submitting - grading {len(qs)} task(s)..."))
|
|
for q in qs:
|
|
r = grade_one(q)
|
|
st["results"][str(q["id"])] = r
|
|
tag = green("PASS") if r["pass"] else (yellow(f"{r['earned']}/{r['total']}") if r["earned"] else red("FAIL"))
|
|
log(f"Q{q['id']:02d} {tag} {q['title']}" + ("" if r["pass"] else f"\n{dim(result_detail(r))}"))
|
|
st["ended_at"] = time.time()
|
|
st["over_time"] = over
|
|
save_state(st)
|
|
setup_contexts(False, {})
|
|
if untimed(st):
|
|
return None
|
|
return record_attempt(st)
|
|
|
|
# ----------------------------------------------------------- commands ----
|
|
def cmd_start(args):
|
|
hard = "--hard" in args; force = "--force" in args
|
|
undrilled = [d for d, n in drilled_domains().items() if n == 0]
|
|
if undrilled and not force:
|
|
die("Drill every domain at least once before a full exam (that is the point of the drills).\n"
|
|
f" Not yet drilled: {', '.join(undrilled)}\n Run: ./exam drill (picks the next domain)\n"
|
|
" or: ./exam start --force to skip the gate.")
|
|
begin_session(compose_exam(hard, random.Random()), "exam", EXAM_SECONDS, force, hard=hard)
|
|
|
|
def next_drill(hard, rng):
|
|
"""No-arg drill: the least-drilled domain, DRILL_SIZE tasks, least-recently-seen first."""
|
|
cov = drilled_domains()
|
|
domain = min(DOMAINS, key=lambda d: (cov[d], list(DOMAINS).index(d)))
|
|
return domain, pick(domain, DRILL_SIZE, hard, rng)
|
|
|
|
def cmd_drill(args):
|
|
force = "--force" in args; hard = "--hard" in args
|
|
args = [a for a in args if a not in ("--force", "--hard")]
|
|
minutes = None
|
|
if "--time" in args:
|
|
i = args.index("--time"); minutes = int(args[i + 1]); del args[i:i + 2]
|
|
per_task = DRILL_SECONDS_PER_TASK_HARD if hard else DRILL_SECONDS_PER_TASK
|
|
if not args:
|
|
domain, qs = next_drill(hard, random.Random())
|
|
cov = drilled_domains()
|
|
print(bold(f"Drill: {DOMAINS[domain]}") + dim(f" ({sum(1 for v in cov.values() if v)}/{len(DOMAINS)} domains drilled so far)"))
|
|
begin_session(qs, "drill", (minutes * 60) if minutes else per_task * len(qs), force, hard=hard)
|
|
return
|
|
if args[0] == "all":
|
|
label, qs = campaign_batch(hard, random.Random())
|
|
if not qs:
|
|
print(bold("Every task in the bank has been attempted." ) + " Switching to weak tasks.")
|
|
qs = [q for q in QUESTIONS if q["id"] in weak_ids()]
|
|
qs = no_conflicts(qs, DRILL_SIZE)
|
|
if not qs:
|
|
die("Nothing weak either - you have passed everything on its latest attempt. Time for ./exam start --hard.")
|
|
else:
|
|
print(bold(f"Campaign: {label}") + dim(f" this batch: Q{', Q'.join(str(q['id']) for q in qs)}"))
|
|
begin_session(qs, "drill", (minutes * 60) if minutes else per_task * len(qs), force, hard=hard)
|
|
return
|
|
if args[0] == "weak":
|
|
ids = weak_ids()
|
|
if not ids:
|
|
die("Nothing marked weak - finish a session with './exam end' first, or name tasks: ./exam drill 2 17")
|
|
else:
|
|
ids = []
|
|
for a in args:
|
|
if a.lower().replace("set", "") in SETS and a.lower().startswith("set"):
|
|
ids += [q["id"] for q in set_questions(a.lower().replace("set", ""))]
|
|
elif a.upper() in DOMAINS:
|
|
ids += [q["id"] for q in QUESTIONS if q["domain"] == a.upper()]
|
|
else:
|
|
ids.append(q_by_id(int(a))["id"])
|
|
ids = sorted(set(ids))
|
|
qs = [q for q in QUESTIONS if q["id"] in ids]
|
|
kept = no_conflicts(qs, len(qs))
|
|
if len(kept) < len(qs):
|
|
dropped = [q["id"] for q in qs if q not in kept]
|
|
print(yellow(f"Dropped Q{', Q'.join(map(str, dropped))}: they break the same thing as another task in this set. Drill them separately."))
|
|
begin_session(kept, "drill", (minutes * 60) if minutes else per_task * len(kept), force, hard=hard)
|
|
|
|
def cmd_study(args):
|
|
"""No clock, solutions unlocked from the start, nothing written to history."""
|
|
hard = "--hard" in args; force = "--force" in args
|
|
rest = [a for a in args if not a.startswith("--")]
|
|
n = DRILL_SIZE
|
|
if "--count" in args:
|
|
n = int(args[args.index("--count") + 1]); rest = [a for a in rest if a != str(n)]
|
|
rng = random.Random()
|
|
if rest and rest[0].upper() in DOMAINS:
|
|
domain = rest[0].upper()
|
|
qs = pick(domain, n, hard, rng)
|
|
print(bold(f"Study: {DOMAINS[domain]}"))
|
|
elif rest and rest[0] == "weak":
|
|
qs = no_conflicts([q for q in QUESTIONS if q["id"] in weak_ids()], n)
|
|
print(bold("Study: tasks you have been failing"))
|
|
elif rest:
|
|
qs = no_conflicts([q_by_id(int(a)) for a in rest], len(rest))
|
|
print(bold("Study: " + ", ".join(f"Q{q['id']}" for q in qs)))
|
|
else:
|
|
domain, qs = next_drill(hard, rng)
|
|
print(bold(f"Study: {DOMAINS[domain]}"))
|
|
begin_session(qs, "drill", EXAM_SECONDS, force, hard=hard, untimed_session=True)
|
|
|
|
def cmd_sim(args):
|
|
"""Fixed 17-task simulator set, full exam rules and clock."""
|
|
force = "--force" in args
|
|
names = [a.lower() for a in args if not a.startswith("--")]
|
|
if not names or names[0] not in SETS:
|
|
die("usage: ./exam sim a|b [--force]\n " + "\n ".join(f"{k}: {v} ({len(set_questions(k))} tasks)" for k, v in SETS.items()))
|
|
name = names[0]
|
|
qs = set_questions(name)
|
|
print(bold(f"{SETS[name]} - {len(qs)} tasks, {TRACK['minutes']} minutes, exam rules (no feedback until you submit)"))
|
|
begin_session(qs, "exam", EXAM_SECONDS, force, hard=True)
|
|
|
|
def cmd_list(args):
|
|
st = load_state()
|
|
results = st["results"] if st and "results" in st else {}
|
|
qs = session_questions(st) if st and not st.get("ended_at") else list(QUESTIONS)
|
|
agg = stats()
|
|
print(bold(f"{'#':>3} {'Domain':<8}{'Pts':>4} {'Now':<8} {'History':<9} Title"))
|
|
for q in qs:
|
|
r = results.get(str(q["id"]))
|
|
now = "-" if r is None else (green("PASS") if r.get("pass") else yellow(f"{r.get('earned')}/{r.get('total')}"))
|
|
a = agg.get(int(q["id"])) or AggEntry()
|
|
hist = "" if a.attempts == 0 else f"{a.attempts - a.fails}/{a.attempts}"
|
|
if a.attempts and a.last is False:
|
|
hist = red(hist + " ✗")
|
|
print(f"{q['id']:>3} {q['domain']:<8}{q['weight']:>4} {now:<17} {hist:<18} {q['title']}{' [hard]' if q['difficulty'] == 'hard' else ''}")
|
|
print(dim("\nHistory = full passes/attempts across sessions, ✗ = failed most recently."))
|
|
if st and not st.get("ended_at"):
|
|
print(status_line(st))
|
|
|
|
def cmd_show(args):
|
|
if not args:
|
|
die("usage: ./exam show N")
|
|
q = q_by_id(int(args[0]))
|
|
st0 = load_state()
|
|
sq = session_questions(st0) if st0 and not st0.get("ended_at") else []
|
|
pos = f"{sq.index(q) + 1} of {len(sq)} | " if q in sq else ""
|
|
print(bold(f"{pos}Q{q['id']} | {DOMAINS[q['domain']]} | {q['weight']} pts{' | HARD' if q['difficulty'] == 'hard' else ''}"))
|
|
print(bold(q["title"])); print("-" * 64); print(task_text(q)); print("-" * 64)
|
|
st = load_state()
|
|
if st and not st.get("ended_at"):
|
|
print(status_line(st))
|
|
|
|
def cmd_grade(args):
|
|
st = require_started()
|
|
if not args:
|
|
die("usage: ./exam grade N | all")
|
|
ids = [q["id"] for q in session_questions(st)] if args[0] == "all" else [int(a) for a in args]
|
|
for n in ids:
|
|
q = q_by_id(n); r = grade_one(q)
|
|
st["results"][str(n)] = r
|
|
tag = green("PASS") if r.get("pass") else yellow(f"{r.get('earned')}/{r.get('total')}")
|
|
print(f"Q{n:02d} {tag} {q['title']}"); print(dim(result_detail(r)))
|
|
save_state(st)
|
|
if args[0] == "all":
|
|
print_score(st)
|
|
|
|
def print_score(st):
|
|
qs = session_questions(st)
|
|
earned, total, pct = score(st)
|
|
print(); print(bold("Per-domain:"))
|
|
for d, name in DOMAINS.items():
|
|
dq = [q for q in qs if q["domain"] == d]
|
|
if not dq:
|
|
continue
|
|
dt = sum(q["weight"] for q in dq); de = sum(st["results"].get(str(q["id"]), {}).get("earned", 0) for q in dq)
|
|
n = int(20 * de / dt)
|
|
print(f" {d:<7} {'█' * n}{'░' * (20 - n)} {de:>3}/{dt:<3} {dim(name)}")
|
|
verdict = green("PASS") if pct >= PASS_PCT else red("FAIL")
|
|
print(f"\n{bold('Score:')} {earned}/{total} = {pct}% -> {verdict} (pass mark {PASS_PCT}%)")
|
|
|
|
def status_line(st):
|
|
if untimed(st):
|
|
return green("study mode - no clock") + dim(f" [{len(session_questions(st))} tasks, nothing recorded]")
|
|
r = remaining(st)
|
|
if st.get("paused_at"):
|
|
return yellow(f"⏸ PAUSED - {fmt(r)} remaining") + dim(" ./exam resume")
|
|
if r <= 0:
|
|
return red("⏰ TIME IS UP - run ./exam end")
|
|
col = green if r > 30 * 60 else (yellow if r > 10 * 60 else red)
|
|
return col(f"⏱ {fmt(r)} remaining") + dim(f" [{st.get('mode', 'exam')}]")
|
|
|
|
def cmd_status(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
print("Session ended."); print_score(st); return
|
|
print(status_line(st))
|
|
print(dim(f"{sum(1 for v in st['results'].values() if v['pass'])} passed / {len(st['results'])} graded / {len(session_questions(st))} in session"))
|
|
|
|
def cmd_timer(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
die("Session already ended.")
|
|
if untimed(st):
|
|
die("This is a study session - there is no clock.")
|
|
try:
|
|
while True:
|
|
st = load_state(); r = remaining(st)
|
|
if st.get("paused_at"):
|
|
sys.stdout.write("\r" + yellow(bold(f"⏸ {fmt(r)} PAUSED ")) + dim(" ./exam resume ")); sys.stdout.flush(); time.sleep(1); continue
|
|
if r <= 0:
|
|
print("\r" + red(bold("⏰ 00:00:00 TIME IS UP - run ./exam end "))); break
|
|
col = green if r > 30 * 60 else (yellow if r > 10 * 60 else red)
|
|
sys.stdout.write("\r" + col(bold(f"⏱ {fmt(r)} remaining")) + dim(" (Ctrl-C to hide)")); sys.stdout.flush()
|
|
if int(r) in (30 * 60, 15 * 60, 5 * 60) and TTY:
|
|
sys.stdout.write("\a")
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
print()
|
|
|
|
def cmd_pause(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
die("Session already ended.")
|
|
set_paused(st, True); print(status_line(st))
|
|
|
|
def cmd_resume(args):
|
|
st = require_started()
|
|
set_paused(st, False); print(status_line(st))
|
|
|
|
def cmd_end(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
die("Session already ended and recorded.")
|
|
if remaining(st) <= 0:
|
|
print(yellow("Note: time expired before submission."))
|
|
entry = finish_session(st)
|
|
print_score(st)
|
|
if entry is None:
|
|
print(dim("Study session - not added to history or weak-spots.")); return
|
|
print(dim(f"Time used: {fmt(st['ended_at'] - st['started_at'])}" + (" (over time)" if st.get("over_time") else "")))
|
|
failed = sorted(int(k) for k, v in entry["results"].items() if not v["pass"])
|
|
print(dim(f"Attempt #{len(load_history())} logged to {os.path.basename(HISTORY_FILE)}; {os.path.basename(WEAK_FILE)} updated."))
|
|
if failed:
|
|
print(bold("\nTo drill what you missed: ./exam drill weak") + dim(f" (Q{', Q'.join(map(str, failed))})"))
|
|
|
|
def cmd_abort(args):
|
|
"""End a session without grading it and restore the kubectl context."""
|
|
st = load_state()
|
|
if not st or st.get("ended_at"):
|
|
die("No session in progress.")
|
|
assert st is not None
|
|
if "--yes" not in args:
|
|
die(f"This discards the {st.get('mode', 'exam')} in progress without grading or recording it.\n"
|
|
" Confirm: ./exam abort --yes (to grade and keep it instead: ./exam end)")
|
|
st["ended_at"] = time.time(); st["aborted"] = True
|
|
save_state(st)
|
|
setup_contexts(False, {})
|
|
print(yellow("Session discarded. Nothing was graded or written to history."))
|
|
print(dim(f" kubectl context restored to {CONTEXT}. Seeded objects are still on the cluster;"))
|
|
print(dim(" './exam drill ...' or './exam start' re-seeds, './teardown.sh' removes the cluster."))
|
|
|
|
def cmd_weak(args):
|
|
agg = stats(); hist = load_history()
|
|
if not hist:
|
|
die("No history yet - finish a session with './exam end'.")
|
|
ids = [i for i in weak_ids(agg) if agg[i]["attempts"] > 0]
|
|
print(bold(f"Weak spots after {len(hist)} attempt(s):"))
|
|
if not ids:
|
|
print(green(" none - everything attempted has passed on its latest try"))
|
|
for i in ids:
|
|
q = q_by_id(i); a = agg[i]
|
|
print(f"\n{red('Q%02d' % i)} {q['domain']:<7} {q['title']} " + dim(f"{a['fails']}/{a['attempts']} failed, last {a['last_earned']} pts, {a['last_fail_at']}"))
|
|
for n in a["failed"]:
|
|
print(dim(" ✗ " + n))
|
|
never = [q["id"] for q in QUESTIONS if agg[q["id"]]["attempts"] == 0]
|
|
if never:
|
|
print(dim(f"\nNever attempted: Q{', Q'.join(map(str, never))}"))
|
|
print(dim(f"\nFull log: {WEAK_FILE} Drill them: ./exam drill weak"))
|
|
|
|
def cmd_history(args):
|
|
hist = load_history()
|
|
if not hist:
|
|
die("No history yet.")
|
|
print(bold(f"{'#':>3} {'when':<17}{'mode':<7}{'score':>7} {'time':<9} tasks"))
|
|
for i, e in enumerate(hist, 1):
|
|
col = green if e["score_pct"] >= PASS_PCT else red
|
|
tasks = f"all {len(QUESTIONS)}" if len(e["ids"]) == len(QUESTIONS) else "Q" + ",".join(map(str, e["ids"]))
|
|
print(f"{i:>3} {e['at']:<17}{(e['mode'] + ('*' if e.get('hard') else '')):<7}{col('%5.1f%%' % e['score_pct']):>16} {fmt(e['time_used']):<9} {tasks}")
|
|
exams = [e for e in hist if e["mode"] == "exam"]
|
|
print(dim("* = hard"))
|
|
if len(exams) >= 2:
|
|
print(bold("\nDomain trend (full exams, % of points):"))
|
|
for d in DOMAINS:
|
|
dq = [q for q in QUESTIONS if q["domain"] == d]; dt = sum(q["weight"] for q in dq)
|
|
pts = [round(100 * sum(e["results"].get(str(q["id"]), {}).get("earned", 0) for q in dq) / dt) for e in exams]
|
|
print(f" {d:<7} " + " → ".join(f"{p:>3}%" for p in pts))
|
|
|
|
def cmd_reset(args):
|
|
if not args:
|
|
die("usage: ./exam reset N (re-seed one task's scenario)")
|
|
q = q_by_id(int(args[0])); rc, out = run(str(q.get("setup", "")), params=session_params(load_state(), q))
|
|
print(("re-seeded " if rc == 0 else "problem re-seeding ") + f"Q{q['id']:02d}: {q['title']}")
|
|
if out:
|
|
print(dim(out))
|
|
|
|
# --------------------------------------------------------------- web ----
|
|
class Api(http.server.BaseHTTPRequestHandler):
|
|
"""Serves web/index.html plus a small JSON API. State lives in the same files the CLI uses."""
|
|
lock = threading.Lock()
|
|
log_lines = []
|
|
seeding = False
|
|
seed_error = None
|
|
ttyd_port = 7681
|
|
shell = "bash"
|
|
|
|
def log_message(self, format: str, *args: Any) -> None: # quiet
|
|
pass
|
|
|
|
def _json(self, obj, code=200):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code); self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
|
|
|
|
def _read_json(self):
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
return json.loads(self.rfile.read(n) or b"{}")
|
|
|
|
def do_GET(self):
|
|
p = urllib.parse.urlparse(self.path)
|
|
if p.path in ("/", "/index.html"):
|
|
with open(os.path.join(ROOT, "web", "index.html"), "rb") as f:
|
|
body = f.read()
|
|
self.send_response(200); self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body); return
|
|
if p.path == "/api/session":
|
|
with Api.lock:
|
|
return self._json(self.session_payload())
|
|
if p.path == "/api/history":
|
|
return self._json({"history": load_history(), "weak": weak_ids(), "stats": {str(k): v.to_dict() for k, v in stats().items()}})
|
|
self._json({"error": "not found"}, 404)
|
|
|
|
def do_POST(self):
|
|
p = urllib.parse.urlparse(self.path); body = self._read_json()
|
|
with Api.lock:
|
|
st = load_state()
|
|
if p.path == "/api/track":
|
|
t = body.get("track")
|
|
if t not in TRACKS:
|
|
return self._json({"error": f"unknown track {t}"}, 400)
|
|
if Api.seeding:
|
|
return self._json({"error": "a session is being seeded"}, 409)
|
|
load_track(t)
|
|
Api.log_lines = []
|
|
return self._json(self.session_payload())
|
|
if p.path == "/api/start":
|
|
if Api.seeding:
|
|
return self._json({"error": "already seeding"}, 409)
|
|
mode = body.get("mode", "exam"); hard = bool(body.get("hard")); rng = random.Random()
|
|
Api.log_lines = []; Api.seed_error = None; Api.seeding = True
|
|
ansi = __import__("re").compile(r"\x1b\[[0-9;]*m")
|
|
ulog = lambda line: Api.log_lines.append(ansi.sub("", line))
|
|
def work():
|
|
try:
|
|
if body.get("sim") in SETS:
|
|
begin_session(set_questions(body["sim"]), "exam", EXAM_SECONDS, True, log=ulog, hard=True)
|
|
elif mode == "exam":
|
|
begin_session(compose_exam(hard, rng), "exam", EXAM_SECONDS, True, log=ulog, hard=hard)
|
|
else:
|
|
if body.get("domain") in DOMAINS:
|
|
qs = pick(body["domain"], DRILL_SIZE, hard, rng)
|
|
elif body.get("campaign"):
|
|
_, qs = campaign_batch(hard, rng)
|
|
if not qs:
|
|
qs = no_conflicts([q for q in QUESTIONS if q["id"] in weak_ids()], DRILL_SIZE)
|
|
elif body.get("next"):
|
|
_, qs = next_drill(hard, rng)
|
|
else:
|
|
ids = body.get("ids") or weak_ids()
|
|
qs = no_conflicts([q for q in QUESTIONS if q["id"] in ids], len(ids))
|
|
per = DRILL_SECONDS_PER_TASK_HARD if hard else DRILL_SECONDS_PER_TASK
|
|
begin_session(qs, "drill", per * len(qs), True, log=ulog, hard=hard,
|
|
untimed_session=bool(body.get("study")))
|
|
except SystemExit:
|
|
Api.seed_error = "\n".join(l for l in Api.log_lines if l.strip()) or "start failed - see the terminal running ./exam web"
|
|
except Exception as e: # noqa: BLE001
|
|
Api.seed_error = f"{type(e).__name__}: {e}"
|
|
finally:
|
|
Api.seeding = False
|
|
threading.Thread(target=work, daemon=True).start()
|
|
return self._json({"seeding": True})
|
|
if not st:
|
|
return self._json({"error": "no session"}, 400)
|
|
if p.path == "/api/flag":
|
|
qid = int(body["id"]); flags = set(st.get("flags", []))
|
|
flags.symmetric_difference_update({qid}); st["flags"] = sorted(flags); save_state(st)
|
|
elif p.path == "/api/pause":
|
|
set_paused(st, bool(body.get("paused", True)))
|
|
elif p.path == "/api/notes":
|
|
st["notes"] = body.get("notes", ""); save_state(st)
|
|
elif p.path == "/api/grade":
|
|
if st.get("mode") != "drill" and not st.get("ended_at"):
|
|
return self._json({"error": "No feedback during an exam. Submit to see results."}, 403)
|
|
q = q_by_id(int(body["id"]))
|
|
if "results" not in st or not isinstance(st["results"], dict):
|
|
st["results"] = {}
|
|
st["results"][str(q["id"])] = grade_one(q, session_params(st, q)); save_state(st)
|
|
elif p.path == "/api/end":
|
|
if not st.get("ended_at"):
|
|
finish_session(st, log=lambda *_: None)
|
|
elif p.path == "/api/reset":
|
|
q = q_by_id(int(body["id"])); run(str(q.get("setup", "")), params=session_params(st, q))
|
|
else:
|
|
return self._json({"error": "not found"}, 404)
|
|
return self._json(self.session_payload())
|
|
|
|
def session_payload(self) -> dict[str, Any]:
|
|
st = load_state()
|
|
qs = session_questions(st) if st else QUESTIONS
|
|
payload: dict[str, Any] = {
|
|
"track": TRACK, "track_id": TRACK_ID, "workspace": WS_DISPLAY, "version": VERSION, "ttyd_port": Api.ttyd_port,
|
|
"ttyd": shutil.which("ttyd") is not None, "shell": Api.shell, "setup_log": Api.log_lines,
|
|
"questions": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "weight": q["weight"], "difficulty": q["difficulty"],
|
|
"task": task_text(q, session_params(st, q)), "checks": [{"name": c["name"], "pts": c["pts"]} for c in q["checks"]],
|
|
"solution": render(str(q.get("solution", "")), session_params(st, q)) if solution_allowed(st, int(q["id"])) else None}
|
|
for q in qs],
|
|
"bank": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "difficulty": q["difficulty"]} for q in QUESTIONS],
|
|
"tracks": TRACKS, "drilled": drilled_domains(), "sets": {k: {"name": v, "count": len(set_questions(k))} for k, v in SETS.items()},
|
|
"seen": sorted(last_seen()),
|
|
"seeding": Api.seeding, "seed_error": Api.seed_error,
|
|
"session": None,
|
|
}
|
|
if st:
|
|
earned, total, pct = score(st)
|
|
payload["session"] = {"mode": st.get("mode"), "hard": st.get("hard", False), "started_at": st["started_at"], "duration": st.get("duration", EXAM_SECONDS),
|
|
"ends_at": st["started_at"] + st.get("duration", EXAM_SECONDS), "ended_at": st.get("ended_at"),
|
|
"over_time": st.get("over_time", False), "flags": st.get("flags", []),
|
|
"paused": bool(st.get("paused_at")), "paused_total": paused_total(st), "pauses": st.get("pauses", 0),
|
|
"untimed": untimed(st), "notes": st.get("notes", ""),
|
|
"results": st["results"], "score": {"earned": earned, "total": total, "pct": pct, "pass_pct": PASS_PCT},
|
|
"domains": self.domain_scores(st), "server_time": time.time()}
|
|
return payload
|
|
|
|
def domain_scores(self, st: dict[str, Any]) -> dict[str, Any]:
|
|
qs = session_questions(st); out = {}
|
|
st_results: dict[str, Any] = st.get("results", {})
|
|
for d in DOMAINS:
|
|
dq = [q for q in qs if q["domain"] == d]
|
|
if dq:
|
|
out[d] = {"earned": sum(st_results.get(str(q["id"]), {}).get("earned", 0) for q in dq),
|
|
"total": sum(q["weight"] for q in dq)}
|
|
return out
|
|
|
|
def cmd_web(args):
|
|
port = 8080
|
|
if "--port" in args:
|
|
port = int(args[args.index("--port") + 1])
|
|
Api.shell = os.environ.get("CKA_SHELL", "bash")
|
|
if "--shell" in args:
|
|
Api.shell = args[args.index("--shell") + 1]
|
|
os.makedirs(WORKSPACE, exist_ok=True)
|
|
import socket
|
|
def port_free(pt):
|
|
with socket.socket() as sk:
|
|
sk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
sk.bind(("127.0.0.1", pt)); return True
|
|
except OSError:
|
|
return False
|
|
if not port_free(port):
|
|
die(f"Port {port} is already in use - probably another './exam web' still running.\n"
|
|
f" Find it: ss -ltnp | grep :{port} or start on another port: ./exam web --port {port + 10}")
|
|
if shutil.which("ttyd") and not port_free(Api.ttyd_port):
|
|
print(yellow(f" ttyd port {Api.ttyd_port} already in use - reusing the existing terminal server."))
|
|
ttyd = None
|
|
if shutil.which("ttyd") and port_free(Api.ttyd_port):
|
|
env = dict(os.environ, WORKSPACE=WORKSPACE, KUBECONFIG=os.environ.get("KUBECONFIG", os.path.expanduser("~/.kube/config")))
|
|
rc = os.path.join(ROOT, "examrc.fish" if "fish" in Api.shell else "examrc")
|
|
if "fish" in Api.shell:
|
|
shell_cmd = [Api.shell, "-C", f"source {rc}"]
|
|
else:
|
|
shell_cmd = [Api.shell, "--rcfile", rc, "-i"]
|
|
# -W makes the terminal writable, so it must never be reachable off this machine: it is an
|
|
# unauthenticated root-equivalent shell with the cluster kubeconfig. ttyd binds every
|
|
# interface by default, so pin it to loopback (macOS needs lo0; override if yours differs).
|
|
iface = os.environ.get("CKA_TTYD_INTERFACE", "lo")
|
|
ttyd = subprocess.Popen(["ttyd", "-p", str(Api.ttyd_port), "-i", iface, "-W",
|
|
"-t", "fontSize=14", "-t", "theme={\"background\":\"#1a1b1e\"}",
|
|
"--cwd", ROOT] + shell_cmd, env=env,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
print(dim(f" terminal: ttyd on {iface}:{Api.ttyd_port} running {Api.shell} in {ROOT} with k=kubectl, $do and $now preset"))
|
|
else:
|
|
print(yellow(" ttyd not found - the terminal pane will show install instructions.\n"
|
|
" (dnf/apt/pacman/brew install ttyd, or https://github.com/tsl0922/ttyd/releases)"))
|
|
http.server.ThreadingHTTPServer.allow_reuse_address = True
|
|
srv = http.server.ThreadingHTTPServer(("127.0.0.1", port), Api) # type: ignore[arg-type]
|
|
print(bold(f"\n {TRACK_ID.upper()} exam UI: http://127.0.0.1:{port} (exam {VERSION})\n") + dim(" Ctrl-C stops the UI (not the exam clock)."))
|
|
def stop(*_):
|
|
if ttyd:
|
|
ttyd.terminate()
|
|
srv.shutdown()
|
|
signal.signal(signal.SIGINT, lambda *_: threading.Thread(target=stop).start())
|
|
signal.signal(signal.SIGTERM, lambda *_: threading.Thread(target=stop).start())
|
|
try:
|
|
srv.serve_forever()
|
|
finally:
|
|
if ttyd:
|
|
ttyd.terminate()
|
|
|
|
def cmd_version(args):
|
|
print(f"exam {VERSION} track={TRACK_ID} bank={len(QUESTIONS)} tasks python={sys.version.split()[0]}")
|
|
|
|
def cmd_help(args):
|
|
print(__doc__ + f"\nversion {VERSION}")
|
|
print(f"""usage: ./exam [--track cka|ckad|cks] <command> (current track: {TRACK_ID}, {len(QUESTIONS)} tasks)
|
|
|
|
web [--port 8080] [--shell bash] exam-style browser UI: tasks, terminal, clock, submit
|
|
study [DOMAIN|weak|N..] no clock, solutions unlocked, nothing recorded (--count N, default {DRILL_SIZE})
|
|
drill next domain you have drilled least: {DRILL_SIZE} tasks, 6 min each (--hard: 5 min, hard tasks)
|
|
drill all [--hard] campaign: {DRILL_SIZE} never-attempted tasks per session until the bank is done, then weak ones
|
|
drill weak|N..|DOMAIN.. chosen tasks (--time MIN, --force, --hard)
|
|
progress attempted/passed per domain, normal and hard
|
|
solution N reference solution (drills: after grading N; exams: after submit)
|
|
sim a|b fixed 17-task simulator set, exam rules, {TRACK['minutes']} min
|
|
start [--hard] full exam: {sum(QUOTA.values()) or '?'} tasks sampled by domain ({sum(QUOTA_HARD.values()) or '?'} in hard mode), {TRACK['minutes']} min.
|
|
Requires every domain drilled once first (--force skips). Hard mode adds traps,
|
|
exact-output tasks, compound failures, collateral checks and exam-style contexts.
|
|
list tasks in this session, with current + historical results
|
|
show N display task N
|
|
grade N|all grade one task (or all) per check - drills only, in the UI
|
|
status / timer time remaining (timer = live countdown)
|
|
pause / resume stop the clock (recorded on the attempt - the real exam has no pause)
|
|
end submit: grade per check, score, log attempt, update weak-spots
|
|
weak what you keep getting wrong, with the failing checks
|
|
history every attempt, scores, and per-domain trend
|
|
reset N re-seed the scenario for task N
|
|
""")
|
|
|
|
COMMANDS = {"start": cmd_start, "drill": cmd_drill, "list": cmd_list, "show": cmd_show, "grade": cmd_grade,
|
|
"status": cmd_status, "timer": cmd_timer, "pause": cmd_pause, "resume": cmd_resume, "end": cmd_end, "weak": cmd_weak, "history": cmd_history,
|
|
"reset": cmd_reset, "web": cmd_web, "abort": cmd_abort, "progress": cmd_progress, "solution": cmd_solution, "sim": cmd_sim, "study": cmd_study, "version": cmd_version, "help": cmd_help}
|
|
|
|
if __name__ == "__main__":
|
|
cmd = ARGV[0] if ARGV else "help"
|
|
COMMANDS.get(cmd, cmd_help)(ARGV[1:])
|