| """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
|
| 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
|
|
|
