vumast3r
23a9774a9f
The Q02/Q102 seeds tried to copy etcdctl out of the running etcd pod with `kubectl exec -- cat`, which cannot work: registry.k8s.io/etcd is distroless (no cat, no sh, no tar). The copy failed, stderr went to /dev/null and seeding still printed a green tick, so the snapshot task had no etcdctl on the node. Q102 was worse: it also ran `exec -- sh -c 'etcdctl ... snapshot save'` inside that container, so its /opt/cka/backup.db was never created at all. - etcd-tools.sh installs both binaries on the control-plane node, taking them from the etcd image already in the node's containerd (offline, exactly the server version), falling back to the upstream release tarball, then apt. - setup.sh runs it at bootstrap; ensure_etcd_tools in the exam prelude installs them through CKA_NODE_ACCESS when a task is seeded, so ssh/vagrant clusters get them too, and both etcd setups now fail with a reason instead of hiding it. - Q102 takes its snapshot with the node's etcdctl; Q02 no longer documents a kubectl cp fallback that cannot work on a distroless image. Co-authored-by: Junie <junie@jetbrains.com>
1112 lines
53 KiB
Python
Executable File
1112 lines
53 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
|
|
die(f"No question {n}. Valid: 1-{len(QUESTIONS)}")
|
|
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 setup_contexts(hard, params_by_id):
|
|
"""Hard mode: several contexts on the same cluster with different default namespaces, like the real exam."""
|
|
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)
|
|
if not hard:
|
|
subprocess.run(["kubectl", "config", "use-context", CONTEXT], capture_output=True)
|
|
return
|
|
ns_ac = next((p.get("ns") for p in params_by_id.values() if p.get("ns")), "default")
|
|
for name, ns in HARD_CONTEXTS.items():
|
|
subprocess.run(["kubectl", "config", "set-context", name, f"--cluster={cluster}", f"--user={user}",
|
|
f"--namespace={ns.replace('%{ns}', ns_ac)}"], capture_output=True)
|
|
subprocess.run(["kubectl", "config", "use-context", "k8s-c1-H"], capture_output=True)
|
|
|
|
def _log_msg(*args: Any, **kwargs: Any) -> None:
|
|
print(*args, **kwargs)
|
|
|
|
|
|
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)
|
|
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"]
|
|
ttyd = subprocess.Popen(["ttyd", "-p", str(Api.ttyd_port), "-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 :{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:])
|