File size: 3,782 Bytes
67ebcef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Test runner for LiveCodeBench items (stdin programs and LeetCode-style functional solutions).
run_lcb(code, item, tests) -> (passed, total, first failure description); stops at the first failing test."""
import json
import os
import subprocess
import sandbox # noqa: E402 (same folder)
import sys
import tempfile
TMP = os.path.join(tempfile.gettempdir(), "lfm3_lcb")
os.makedirs(TMP, exist_ok=True)
HARNESS = r'''
import sys, json
from typing import *
import collections, heapq, bisect, itertools, functools, math, string, re
from collections import *
from heapq import *
from bisect import *
from itertools import *
from functools import *
from math import *
sys.setrecursionlimit(1 << 20)
__CODE__
_tests = json.loads(sys.stdin.read())
for _k, (_inp, _exp) in enumerate(_tests):
_args = [json.loads(_l) for _l in _inp.strip().split("\n")]
try:
_res = getattr(Solution(), __FUNC__)(*_args)
except Exception as _e:
print(json.dumps({"k": _k, "err": repr(_e)[:300]}))
sys.exit(0)
_exp_v = json.loads(_exp)
_ok = _res == _exp_v or (isinstance(_res, tuple) and list(_res) == _exp_v) or (
isinstance(_res, float) and isinstance(_exp_v, (int, float)) and abs(_res - _exp_v) < 1e-6)
if not _ok:
print(json.dumps({"k": _k, "got": repr(_res)[:300]}))
sys.exit(0)
print(json.dumps({"k": len(_tests)}))
'''
def _same(got, exp):
a, b = got.split(), exp.split()
if len(a) != len(b):
return False
for x, y in zip(a, b):
if x != y:
try:
if abs(float(x) - float(y)) > 1e-6 * max(1.0, abs(float(y))):
return False
except ValueError:
return False
return True
def _run(src, stdin, timeout):
fd, path = tempfile.mkstemp(suffix=".py", dir=TMP)
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(src)
try:
r = sandbox.run([sys.executable, "-I", path], input=stdin.encode(), capture_output=True, timeout=timeout,
cwd=TMP, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
return r.returncode, r.stdout.decode(errors="replace"), r.stderr.decode(errors="replace")
except subprocess.TimeoutExpired:
return -1, "", "TimeoutError: time limit exceeded"
finally:
try:
os.remove(path)
except OSError:
pass
def run_lcb(code, item, tests, per_test_timeout=6):
if item["testtype"] == "functional":
src = HARNESS.replace("__CODE__", code).replace("__FUNC__", repr(item["func_name"]))
rc, so, se = _run(src, json.dumps(tests), per_test_timeout * max(1, min(len(tests), 10)))
try:
res = json.loads(so.strip().splitlines()[-1])
except Exception:
err = "\n".join(se.strip().splitlines()[-5:]) or "TimeoutError: time limit exceeded"
return 0, len(tests), f"Running the solution failed:\n{err}"
k = res["k"]
if k == len(tests):
return k, len(tests), None
inp, exp = tests[k]
got = res.get("got") or ("raised " + res.get("err", ""))
return k, len(tests), f"Input:\n{inp[:800]}\nExpected:\n{exp[:400]}\nYour solution returned:\n{got}"
for k, (inp, exp) in enumerate(tests):
rc, so, se = _run(code, inp, per_test_timeout)
if not (rc == 0 and _same(so, exp)):
got = so.strip()[:400] if rc == 0 else (so.strip()[:200] + "\n" + "\n".join(se.strip().splitlines()[-5:])).strip()
return k, len(tests), f"Input:\n{inp[:800]}\nExpected output:\n{exp[:400]}\nYour program's output:\n{got or '(nothing)'}"
return len(tests), len(tests), None
|