#!/usr/bin/env python3 """Автопроверка практических задач диагностики. Использование (из каталога diag): python3 grade.py # проверить все задачи (с санитайзерами) python3 grade.py 01 04 07 # только указанные python3 grade.py --fast # без ASAN/UBSAN/TSAN (быстрее, слабее) python3 grade.py --submit # после проверки скопировать ответы в submissions/ Что требуется от решающего: файл solution.cpp / solution.sh / crash.c + answer.txt, каждый в своём каталоге tasks//. Результат: results_tasks.json (по задаче: pass, сколько проверок, время, лог). """ import json import os import shutil import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) TASKS = os.path.join(HERE, "tasks") SUB = os.path.join(HERE, "submissions") CXX = os.environ.get("CXX", "g++") SAN = "-fsanitize=address,undefined -fno-omit-frame-pointer" FAST = False # имя задачи: (тип, что запускать, таймаут, флаги компиляции) SPEC = { "01_bits": ("cpp", "test_bits.cpp", 180), "02_list": ("cpp", "test_list.cpp", 180), "03_ring": ("cpp", "test_ring.cpp", 180), "04_ipv4": ("cpp", "test_ipv4.cpp", 180), "05_lis": ("cpp", "test_lis.cpp", 300), "06_threads": ("cpp", "test_threads.cpp", 600), # собирается с TSan, не с ASAN "07_epoll": ("bin", "test_epoll.py", 300), "08_bash": ("bash", "test_bash.py", 300), "09_gdb": ("py", "test_gdb.py", 300), "10_hash": ("cpp", "test_hash.cpp", 300), "11_heap": ("cpp", "test_heap.cpp", 600), "12_dijkstra":("cpp", "test_dijkstra.cpp",600), "13_subnet": ("cpp", "test_subnet.cpp", 120), } def sh(cmd, cwd, timeout): t0 = time.time() try: r = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout + r.stderr, time.time() - t0, False except subprocess.TimeoutExpired as e: out = (e.stdout or b"").decode(errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "") return None, out + "\n[TIMEOUT] превышено время %s c" % timeout, time.time() - t0, True def run_task(name, root): d = os.path.join(root, name) kind, test, timeout = SPEC[name] log = [] rc = None timed_out = False if kind == "cpp": flags = ["-std=c++17", "-O2", "-Wall", "-Wextra", "-g"] extra = [] if name == "06_threads": flags = ["-std=c++17", "-O1", "-Wall", "-Wextra", "-g", "-pthread"] extra = [] if FAST else ["-fsanitize=thread"] elif not FAST: extra = SAN.split() cmd = [CXX] + flags + (extra or []) + [test, "-o", "t_" + name] rc, out, secs, timed_out = sh(cmd, d, timeout) log.append("$ " + " ".join(cmd) + "\n" + out) if rc != 0: return False, log, secs, timed_out rc, out, secs2, timed_out = sh(["./t_" + name], d, timeout) log.append("$ ./t_" + name + "\n" + out) secs += secs2 if "unsupported VMA" in out and not FAST: # ThreadSanitizer не поддерживает этот размер адресного пространства # (Raspberry Pi / ядро 6.18, 47-бит VMA) — откатываемся на ASAN. cmd2 = [CXX, "-std=c++17", "-O1", "-Wall", "-Wextra", "-g", "-pthread"] \ + SAN.split() + [test, "-o", "t_" + name] rc, out2, s, to = sh(cmd2, d, timeout) log.append("$ " + " ".join(cmd2) + "\n# TSan недоступен на этой платформе, откат на ASAN\n" + out2) if rc == 0: rc, out2, s2, to = sh(["./t_" + name], d, timeout) log.append("$ ./t_" + name + "\n" + out2) secs += s2 elif kind == "bin": # epoll-сервер: solution.cpp -> бинарь solution, затем python-тест cmd = [CXX, "-std=c++17", "-O2", "-Wall", "-Wextra", "-g", "solution.cpp", "-o", "solution"] rc, out, secs, timed_out = sh(cmd, d, timeout) log.append("$ " + " ".join(cmd) + "\n" + out) if rc != 0: return False, log, secs, timed_out rc, out, secs2, timed_out = sh([sys.executable, test], d, timeout) log.append("$ python3 " + test + "\n" + out) secs += secs2 else: # bash / py — тест сам всё делает if name == "08_bash": os.chmod(os.path.join(d, "solution.sh"), 0o755) rc, out, secs, timed_out = sh([sys.executable, test], d, timeout) log.append("$ python3 " + test + "\n" + out) body = log[-1] passed = (rc == 0) and ("FAIL" not in body) and ("ALL PASS" in body) return passed, log, secs, timed_out def main(): global FAST args = [a for a in sys.argv[1:]] FAST = "--fast" in args do_submit = "--submit" in args names = [a for a in args if a.isdigit() or a in SPEC] if names: names = [n for n in SPEC if n.split("_")[0] in names or n in names] else: names = list(SPEC) results = {} print(f"{'задача':12s} {'итог':7s} {'проверок':9s} {'FAIL':5s} {'время':8s}") for n in names: ok, log, secs, to = run_task(n, TASKS) body = log[-1] n_ok = body.count("ok ") n_bad = body.count("FAIL ") results[n] = { "pass": ok, "checks_ok": n_ok, "checks_fail": n_bad, "seconds": round(secs, 2), "timeout": to, "log": log, } print(f"{n:12s} {'PASS' if ok else 'FAIL':7s} {n_ok:<9d} {n_bad:<5d} {secs:8.2f}") total = len(names) done = sum(1 for n in names if results[n]["pass"]) with open(os.path.join(HERE, "results_tasks.json"), "w", encoding="utf-8") as f: json.dump({"passed": done, "total": total, "tasks": results}, f, ensure_ascii=False, indent=2) print(f"\nИТОГ ЗАДАЧ: {done}/{total} PASS. Подробности: results_tasks.json") if do_submit: os.makedirs(SUB, exist_ok=True) for fn in ("results_quiz.json", "results_tasks.json"): p = os.path.join(HERE, fn) if os.path.exists(p): shutil.copy2(p, os.path.join(SUB, fn)) for n in SPEC: d = os.path.join(TASKS, n) for fn in ("solution.cpp", "solution.sh", "crash.c", "answer.txt"): p = os.path.join(d, fn) if os.path.exists(p): shutil.copy2(p, os.path.join(SUB, f"{n}__{fn}")) for fn in os.listdir(SUB): os.chmod(os.path.join(SUB, fn), 0o644) print(f"Ответы скопированы в {SUB}") print("Пакет: 01-09 диагностика, 10-13 добор по слабым доменам (см. WEEK7.md).") sys.exit(0 if done == total else 1) if __name__ == "__main__": main()