"""Tests for terminal_bench_toolkit.py. Run: python3 test_terminal_bench_toolkit.py""" import hashlib import math import os import random import re import tempfile import unittest import terminal_bench_toolkit as tb def task_by_name(name): return {t.name: t for t in tb.demo_tasks()}[name] ALIAS = {"instruction": "instruction.md", "task": "task.toml", "solution": "solution/solve.sh", "checks": "tests/checks.toml", "env": "environment/x.txt"} def mk_files(**over): """A tiny valid task spec; keyword overrides replace spec files (None removes one).""" f = { "instruction.md": "Write the word hi to /app/out.txt.\n", "task.toml": tb._toml_task("lab/tiny", "x", "easy"), "solution/solve.sh": "echo hi > /app/out.txt\n", "tests/checks.toml": '[[check]]\ntype = "file_equals"\npath = "/app/out.txt"\ntext = "hi"\n', } for k, v in over.items(): if k == "extra": f.update(v) continue key = ALIAS[k] if v is None: f.pop(key, None) else: f[key] = v return f def mk_task(**over): return tb.parse_task(mk_files(**over)) def mk_ctask(**over): """Like mk_task, but with the lab canary added to every eligible file.""" return tb.parse_task(tb.add_canary(mk_files(**over), tb.LAB_CANARY)) def mk_result(task="lab/a", agent="x", trial=0, passed=True, cost=0.0, elapsed=1.0, version="1.0.0", timeout=60.0, flags=(), steps=True, tin=10, tout=5): st = [tb.Step(0, 1.0, ("ls", "Enter"), 1.0, "", 1, 1)] if steps else [] return tb.TrialResult(task, version, agent, trial, "passed" if passed else "failed", "agent_done", 1.0 if passed else 0.0, 1 if passed else 0, 1, st, elapsed, tin, tout, cost, list(flags), timeout, False) class Shared(unittest.TestCase): def test_count_tokens(self): self.assertEqual(tb.count_tokens("ls -la /app"), 5) # ls, -, la, /, app self.assertEqual(tb.count_tokens(""), 0) self.assertEqual(tb.count_tokens("a b"), 2) self.assertEqual(tb.count_tokens("a,b"), 3) def test_mean_stdev(self): self.assertEqual(tb.mean([]), 0.0) self.assertEqual(tb.mean([1, 2, 3, 4]), 2.5) self.assertEqual(tb.stdev([5]), 0.0) self.assertEqual(tb.stdev([]), 0.0) self.assertAlmostEqual(tb.stdev([1, 2, 3, 4]), math.sqrt(5 / 3)) def test_seed_from_is_stable_and_distinct(self): self.assertEqual(tb.seed_from(1, "a"), tb.seed_from(1, "a")) self.assertNotEqual(tb.seed_from(1, "a"), tb.seed_from(1, "b")) self.assertNotEqual(tb.seed_from(1, "a"), tb.seed_from(2, "a")) self.assertNotEqual(tb.seed_from("a", "b"), tb.seed_from("ab")) self.assertLess(tb.seed_from("x"), 16 ** 12) class Canary(unittest.TestCase): def test_make_canary(self): c = tb.make_canary("bench") self.assertTrue(c.startswith(tb.CANARY_PREFIX)) self.assertIn("lab-canary GUID", c) self.assertEqual(c, tb.make_canary("bench")) self.assertNotEqual(c, tb.make_canary("other")) self.assertRegex(c, r"GUID [0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[0-9a-f]{4}-[0-9a-f]{12}$") def test_lab_canary_is_not_the_real_one(self): self.assertNotIn("26b5c67b", tb.LAB_CANARY) def test_has_and_strip(self): c = tb.make_canary("b") self.assertTrue(tb.has_canary("x\n" + c + "\ny")) self.assertTrue(tb.has_canary("x " + tb.CANARY_PREFIX)) self.assertFalse(tb.has_canary("nothing here")) self.assertTrue(tb.has_canary("a " + c, c)) self.assertFalse(tb.has_canary("a " + tb.CANARY_PREFIX, c)) self.assertEqual(tb.strip_canary("a\n" + c + "\nb"), "a\nb") self.assertEqual(tb.strip_canary("plain"), "plain") def test_eligible_files(self): for p in ("instruction.md", "task.toml", "solution/solve.sh", "tests/checks.toml"): self.assertTrue(tb.canary_eligible(p), p) for p in ("environment/names.txt", "tests/expected.txt", "environment/data.csv", "Dockerfile"): self.assertFalse(tb.canary_eligible(p), p) def test_add_canary(self): c = tb.make_canary("b") f = {"a.md": "doc\n", "b.toml": "k = 1", "c.sh": "echo\n", "d.txt": "data\n"} g = tb.add_canary(f, c) self.assertEqual(g["a.md"], f"doc\n\n") self.assertEqual(g["b.toml"], f"k = 1\n# {c}\n") self.assertEqual(g["c.sh"], f"echo\n# {c}\n") self.assertEqual(g["d.txt"], "data\n") self.assertEqual(tb.add_canary(g, c), g) # idempotent self.assertEqual(f["a.md"], "doc\n") # input untouched def test_toml_canary_still_parses(self): g = tb.add_canary(mk_files(), tb.LAB_CANARY) self.assertEqual(tb.parse_task(g).name, "lab/tiny") class Spec(unittest.TestCase): def test_parse_version(self): self.assertEqual(tb.parse_version("1.2.3"), (1, 2, 3)) self.assertEqual(tb.parse_version("10.0.12"), (10, 0, 12)) for bad in ("1.2", "a.b.c", "1.2.3.4", "", "v1.2.3", "1.2.x"): self.assertIsNone(tb.parse_version(bad), bad) def test_parse_step(self): self.assertEqual(tb.parse_step("ls"), ("ls", None)) self.assertEqual(tb.parse_step("@95 sleep 90"), ("sleep 90", 95.0)) self.assertEqual(tb.parse_step("@2.5 ls -l"), ("ls -l", 2.5)) self.assertEqual(tb.parse_step("echo @5"), ("echo @5", None)) def test_parse_solution(self): text = "# comment\n\n ls \n # indented comment\necho hi\n" self.assertEqual(tb.parse_solution(text), ["ls", "echo hi"]) self.assertEqual(tb.parse_solution(""), []) def test_parse_checks_ok(self): cs = tb.parse_checks('[[check]]\ntype = "file_exists"\npath = "/app/x"\n[[check]]\ntype = "file_lines"\npath = "/app/y"\ncount = 2\nname = "mine"\n') self.assertEqual([c["name"] for c in cs], ["file_exists_0", "mine"]) self.assertEqual(cs[1]["count"], 2) def test_parse_checks_errors(self): for text, frag in [("", "no [[check]]"), ("x = [", "not valid TOML"), ('[[check]]\ntype = "nope"\n', "unknown type"), ('[[check]]\ntype = "file_equals"\npath = "/app/x"\n', "missing field 'text'"), ('[[check]]\npath = "/app/x"\n', "unknown type None")]: with self.assertRaises(tb.TaskError) as cm: tb.parse_checks(text) self.assertIn(frag, str(cm.exception), text) def test_every_check_type_has_required_fields(self): self.assertEqual(set(tb.CHECK_TYPES), {"file_exists", "file_absent", "file_equals", "file_contains", "file_not_contains", "file_lines", "file_sha256", "file_equals_file", "cmd_stdout"}) self.assertEqual(tb.CHECK_TYPES["cmd_stdout"], ("cmd", "equals")) def test_demo_tasks_parse(self): ts = tb.demo_tasks() self.assertEqual(len(ts), 6) self.assertEqual([t.name for t in ts], ["lab/sort-names", "lab/count-errors", "lab/make-report", "lab/fix-config", "lab/hash-file", "lab/long-job"]) for t in ts: self.assertNotIn(tb.CANARY_PREFIX, t.instruction) self.assertTrue(t.solution) def test_task_fields(self): t = task_by_name("lab/sort-names") self.assertEqual(t.version, "1.0.0") self.assertEqual(t.agent_timeout, 60.0) self.assertEqual(t.verifier_timeout, 10.0) self.assertEqual(t.category, "text") self.assertEqual(t.difficulty, "easy") self.assertEqual(sorted(t.env_files), ["names.txt"]) self.assertEqual(sorted(t.test_files), ["checks.toml", "expected_sorted.txt"]) self.assertEqual(t.solution, ["sort /app/names.txt | uniq > /app/sorted.txt"]) self.assertEqual(task_by_name("lab/long-job").agent_timeout, 150.0) self.assertEqual(task_by_name("lab/long-job").env_files, {}) def test_default_allowlist_is_everything(self): self.assertEqual(mk_task().allowed, tb.BUILTINS) def test_allowlist_from_toml(self): toml = tb._toml_task("lab/tiny", "x", "easy") + '[environment]\nallowed_commands = ["echo", "ls"]\n' self.assertEqual(mk_task(task=toml).allowed, frozenset({"echo", "ls"})) bad = tb._toml_task("lab/tiny", "x", "easy") + '[environment]\nallowed_commands = ["echo", "rm -rf"]\n' with self.assertRaises(tb.TaskError) as cm: mk_task(task=bad) self.assertIn("unknown commands", str(cm.exception)) def test_instruction_has_canary_removed(self): t = mk_task(instruction="Do it.\n\n") self.assertEqual(t.instruction, "Do it.") def test_spec_errors(self): cases = [ (dict(instruction=None), "missing file instruction.md"), (dict(task=None), "missing file task.toml"), (dict(checks=None), "missing file tests/checks.toml"), (dict(task="x = ["), "task.toml is not valid TOML"), (dict(task=tb._toml_task("tiny", "x", "easy")), "must look like org/name"), (dict(task=tb._toml_task("Lab/Tiny", "x", "easy")), "must look like org/name"), (dict(task=tb._toml_task("lab/tiny", "x", "easy", version="1.0")), "must look like 1.2.3"), (dict(task=tb._toml_task("lab/tiny", "x", "easy", agent_timeout=0)), "agent timeout_sec"), (dict(task=tb._toml_task("lab/tiny", "x", "easy", agent_timeout=-5)), "agent timeout_sec"), (dict(task=tb._toml_task("lab/tiny", "x", "easy", verifier_timeout=0)), "verifier timeout_sec"), (dict(task=tb._toml_task("lab/tiny", "x", "easy", agent_timeout="true")), "agent timeout_sec"), (dict(instruction=" \n"), "instruction.md is empty"), (dict(instruction="\n"), "instruction.md is empty"), (dict(checks='[[check]]\ntype = "x"\n'), "unknown type"), ] for over, frag in cases: with self.assertRaises(tb.TaskError) as cm: mk_task(**over) self.assertIn(frag, " | ".join(cm.exception.errors), over) def test_boolean_timeout_is_rejected(self): toml = tb._toml_task("lab/tiny", "x", "easy").replace("timeout_sec = 60", "timeout_sec = true") with self.assertRaises(tb.TaskError): mk_task(task=toml) def test_errors_are_collected_together(self): with self.assertRaises(tb.TaskError) as cm: mk_task(task=tb._toml_task("bad", "x", "easy", agent_timeout=0, version="x"), instruction="") self.assertGreaterEqual(len(cm.exception.errors), 4) def test_missing_solution_is_allowed(self): t = mk_task(solution=None) self.assertEqual(t.solution, []) def test_digest(self): a, b = mk_task(), mk_task() self.assertEqual(a.digest(), b.digest()) self.assertEqual(len(a.digest()), 12) self.assertNotEqual(a.digest(), mk_task(instruction="Other text.\n").digest()) self.assertNotEqual(a.digest(), mk_task(env="1").digest()) class TokenizerAndParser(unittest.TestCase): def words(self, line): return [(t.text, t.op) for t in tb.tokenize(line)] def test_plain_words(self): self.assertEqual(self.words("ls -la /app"), [("ls", False), ("-la", False), ("/app", False)]) self.assertEqual(self.words(""), []) self.assertEqual(self.words(" \t "), []) def test_quotes(self): self.assertEqual(self.words("echo 'a b' \"c d\""), [("echo", False), ("a b", False), ("c d", False)]) self.assertEqual(self.words("echo ''"), [("echo", False), ("", False)]) self.assertEqual(self.words("echo a'b c'd"), [("echo", False), ("ab cd", False)]) self.assertEqual(self.words('echo "a\\"b"'), [("echo", False), ('a"b', False)]) self.assertEqual(self.words('echo "a\\nb"'), [("echo", False), ("a\\nb", False)]) self.assertEqual(self.words("echo \"it's\""), [("echo", False), ("it's", False)]) self.assertEqual(self.words("echo 'a\"b'"), [("echo", False), ('a"b', False)]) def test_backslash(self): self.assertEqual(self.words("echo a\\ b"), [("echo", False), ("a b", False)]) self.assertEqual(self.words("echo a\\"), [("echo", False), ("a\\", False)]) def test_operators_split_without_spaces(self): self.assertEqual(self.words("a;b"), [("a", False), (";", True), ("b", False)]) self.assertEqual(self.words("a&&b"), [("a", False), ("&&", True), ("b", False)]) self.assertEqual(self.words("a||b"), [("a", False), ("||", True), ("b", False)]) self.assertEqual(self.words("a|b"), [("a", False), ("|", True), ("b", False)]) self.assertEqual(self.words("a>b"), [("a", False), (">", True), ("b", False)]) self.assertEqual(self.words("a>>b"), [("a", False), (">>", True), ("b", False)]) self.assertEqual(self.words("a'"), [("echo", False), (">", False)]) def test_tokenizer_errors(self): for bad in ("echo 'a", 'echo "a', "echo a & b", "ls &"): with self.assertRaises(ValueError, msg=bad): tb.tokenize(bad) def shape(self, line): return [(op, [(c.argv, c.out_file, c.append, c.in_file) for c in pl]) for op, pl in tb.parse_line(line)] def test_parse_single(self): self.assertEqual(self.shape("ls /app"), [(None, [(["ls", "/app"], None, False, None)])]) self.assertEqual(tb.parse_line(""), []) self.assertEqual(tb.parse_line(" "), []) def test_parse_pipeline(self): self.assertEqual(self.shape("cat a | sort | uniq"), [(None, [(["cat", "a"], None, False, None), (["sort"], None, False, None), (["uniq"], None, False, None)])]) def test_parse_chain_ops(self): ops = [op for op, _ in tb.parse_line("a && b || c ; d")] self.assertEqual(ops, [None, "&&", "||", ";"]) def test_parse_redirects(self): self.assertEqual(self.shape("echo hi > f"), [(None, [(["echo", "hi"], "f", False, None)])]) self.assertEqual(self.shape("echo hi >> f"), [(None, [(["echo", "hi"], "f", True, None)])]) self.assertEqual(self.shape("cat < f"), [(None, [(["cat"], None, False, "f")])]) self.assertEqual(self.shape("> f echo hi"), [(None, [(["echo", "hi"], "f", False, None)])]) self.assertEqual(self.shape("sort < a > b"), [(None, [(["sort"], "b", False, "a")])]) self.assertEqual(self.shape("a | b > f"), [(None, [(["a"], None, False, None), (["b"], "f", False, None)])]) def test_trailing_semicolon_is_fine(self): self.assertEqual(len(tb.parse_line("ls ;")), 1) def test_parse_errors(self): for bad in ("a |", "| a", "&& a", "a &&", "a ||", "; a", "a > ", "a >> |", "a | | b", "a < "): with self.assertRaises(ValueError, msg=bad): tb.parse_line(bad) def test_should_run(self): self.assertTrue(tb.should_run(None, 5)) self.assertTrue(tb.should_run(";", 5)) self.assertTrue(tb.should_run("&&", 0)) self.assertFalse(tb.should_run("&&", 1)) self.assertTrue(tb.should_run("||", 1)) self.assertFalse(tb.should_run("||", 0)) class SandboxTests(unittest.TestCase): def setUp(self): self.task = task_by_name("lab/sort-names") self.sb = tb.Sandbox(self.task) self.addCleanup(self.sb.cleanup) def test_layout(self): self.assertEqual(self.sb.listdir("/"), ["app", "tmp"]) self.assertEqual(self.sb.listdir("/app"), ["names.txt"]) self.assertEqual(self.sb.read("/app/names.txt"), "mia\nzoe\nadam\nzoe\nbob\nadam\n") self.assertEqual(self.sb.roots, ["/app", "/tmp"]) self.assertFalse(self.sb.exists("/tests")) def test_cleanup_and_context_manager(self): with tb.Sandbox(self.task) as sb: root = sb.root self.assertTrue(root.exists()) self.assertFalse(root.exists()) def test_cleanup_after_a_trial(self): before = set(os.listdir(tempfile.gettempdir())) tb.run_trial(self.task, tb.OracleAgent(), 0) after = set(os.listdir(tempfile.gettempdir())) self.assertEqual({n for n in after - before if n.startswith("minitb-")}, set()) def test_confinement(self): for p in ("/etc/passwd", "/", "/appx/file", "/app/../etc/passwd", "../../etc/passwd", "/tmpx", "/tests/x", "/root"): if p == "/": continue with self.assertRaises(tb.SandboxError, msg=p): self.sb.resolve(p) self.assertTrue(str(self.sb.resolve("/app/../tmp/x")).endswith("/tmp/x")) def test_relative_paths_use_cwd(self): self.assertEqual(self.sb.read("names.txt", "/app")[:3], "mia") self.sb.write("/tmp/a.txt", "t") self.assertEqual(self.sb.read("a.txt", "/tmp"), "t") self.assertEqual(self.sb.normalize("../tmp/./a.txt", "/app"), "/tmp/a.txt") self.assertEqual(self.sb.normalize("/app//x/", "/tmp"), "/app/x") def test_symlink_escape_is_blocked(self): outside = tempfile.mkdtemp() self.addCleanup(lambda: os.rmdir(outside)) link = self.sb.root / "app" / "evil" os.symlink(outside, link) with self.assertRaises(tb.SandboxError) as cm: self.sb.resolve("/app/evil/secret") self.assertEqual(str(cm.exception), "Permission denied") def test_exists_isdir_isfile(self): self.assertTrue(self.sb.exists("/app")) self.assertTrue(self.sb.isdir("/app")) self.assertFalse(self.sb.isfile("/app")) self.assertTrue(self.sb.isfile("/app/names.txt")) self.assertFalse(self.sb.isdir("/app/names.txt")) self.assertFalse(self.sb.exists("/etc")) self.assertFalse(self.sb.isdir("/etc")) self.assertFalse(self.sb.isfile("/etc/passwd")) self.assertTrue(self.sb.isdir("/")) self.assertTrue(self.sb.isdir("..")) self.assertFalse(self.sb.exists("/app/nope")) def test_read_errors(self): with self.assertRaises(tb.SandboxError) as cm: self.sb.read("/app/nope") self.assertEqual(str(cm.exception), "No such file or directory") with self.assertRaises(tb.SandboxError) as cm: self.sb.read("/app") self.assertEqual(str(cm.exception), "Is a directory") def test_write_and_append(self): self.sb.write("/app/new.txt", "a") self.sb.write("/app/new.txt", "b", append=True) self.assertEqual(self.sb.read("/app/new.txt"), "ab") self.sb.write("/app/new.txt", "c") self.assertEqual(self.sb.read("/app/new.txt"), "c") self.sb.write("/app/e.txt", "x", append=True) # append to a missing file creates it self.assertEqual(self.sb.read("/app/e.txt"), "x") def test_write_errors(self): with self.assertRaises(tb.SandboxError) as cm: self.sb.write("/app/no/such/x", "a") self.assertEqual(str(cm.exception), "No such file or directory") with self.assertRaises(tb.SandboxError) as cm: self.sb.write("/app", "a") self.assertEqual(str(cm.exception), "Is a directory") with self.assertRaises(tb.SandboxError): self.sb.write("/etc/x", "a") def test_size_limit(self): small = tb.Sandbox(mk_task(), max_file_bytes=10) self.addCleanup(small.cleanup) small.write("/app/ok.txt", "x" * 10) # exactly at the limit is fine with self.assertRaises(tb.SandboxError) as cm: small.write("/app/big.txt", "x" * 11) self.assertEqual(str(cm.exception), "File too large") with self.assertRaises(tb.SandboxError): small.write("/app/ok.txt", "y", append=True) # append counts the old text def test_file_count_limit(self): few = tb.Sandbox(self.task, max_files=2) self.addCleanup(few.cleanup) few.write("/app/b.txt", "1") # names.txt + b.txt = 2 with self.assertRaises(tb.SandboxError) as cm: few.write("/app/c.txt", "1") self.assertEqual(str(cm.exception), "Too many files") few.write("/app/b.txt", "2") # overwriting is fine self.assertEqual(few.file_count(), 2) def test_listdir(self): self.sb.write("/app/z.txt", "") self.sb.write("/app/.hidden", "") self.assertEqual(self.sb.listdir("/app"), [".hidden", "names.txt", "z.txt"]) with self.assertRaises(tb.SandboxError): self.sb.listdir("/app/names.txt") with self.assertRaises(tb.SandboxError): self.sb.listdir("/etc") def test_leaky_has_tests(self): sb = tb.Sandbox(self.task, leaky=True) self.addCleanup(sb.cleanup) self.assertEqual(sb.listdir("/"), ["app", "tests", "tmp"]) self.assertEqual(sorted(sb.listdir("/tests")), ["checks.toml", "expected_sorted.txt"]) self.assertEqual(sb.read("/tests/expected_sorted.txt"), "adam\nbob\nmia\nzoe\n") def test_app_files(self): (self.sb.root / "app" / "d").mkdir() self.sb.write("/app/d/f.txt", "q") self.assertEqual(self.sb.app_files(), {"names.txt": "mia\nzoe\nadam\nzoe\nbob\nadam\n", "d/f.txt": "q"}) def test_no_subprocess_in_the_toolkit(self): with open(tb.__file__) as fh: src = fh.read() self.assertNotRegex(src, r"(?m)^\s*(import|from)\s+subprocess") self.assertNotIn("os.system", src) self.assertNotIn("docker run", src) class ShellBase(unittest.TestCase): allowed = None def setUp(self): self.sb = tb.Sandbox(mk_task()) self.addCleanup(self.sb.cleanup) self.shell = tb.Shell(self.sb, self.allowed) def sh(self, line, stdin=None): return self.shell.run(line, stdin) def put(self, path, text): self.sb.write(path, text) def out(self, line, stdin=None): return self.sh(line, stdin).stdout class ShellBasics(ShellBase): def test_pwd_cd(self): self.assertEqual(self.out("pwd"), "/app\n") self.assertEqual(self.out("cd /tmp && pwd"), "/tmp\n") self.assertEqual(self.out("pwd"), "/tmp\n") # cwd persists self.assertEqual(self.out("cd && pwd"), "/app\n") self.assertEqual(self.out("cd .. && pwd"), "/\n") self.assertEqual(self.out("cd /app && cd ../tmp && pwd"), "/tmp\n") def test_cd_errors(self): r = self.sh("cd nope") self.assertEqual((r.status, r.stderr), (1, "bash: cd: nope: No such file or directory\n")) self.put("/app/f.txt", "x") self.assertEqual(self.sh("cd f.txt").status, 1) self.assertEqual(self.sh("cd /etc").status, 1) self.assertEqual(self.out("pwd"), "/app\n") def test_ls(self): for n in ("b.txt", "a.txt", ".h"): self.put("/app/" + n, "") self.assertEqual(self.out("ls"), "a.txt\nb.txt\n") self.assertEqual(self.out("ls -a"), ".h\na.txt\nb.txt\n") self.assertEqual(self.out("ls -la"), ".h\na.txt\nb.txt\n") self.assertEqual(self.out("ls /app/a.txt"), "a.txt\n") self.assertEqual(self.out("ls /"), "app\ntmp\n") self.assertEqual(self.out("ls /app /tmp"), "a.txt\nb.txt\n") r = self.sh("ls nope") self.assertEqual((r.status, r.stderr), (2, "ls: cannot access 'nope': No such file or directory\n")) self.assertEqual(self.sh("ls /etc").status, 2) self.assertEqual(self.out("cd /tmp && ls"), "") def test_cat(self): self.put("/app/a.txt", "one\n") self.put("/app/b.txt", "two\n") self.assertEqual(self.out("cat a.txt"), "one\n") self.assertEqual(self.out("cat a.txt b.txt"), "one\ntwo\n") r = self.sh("cat a.txt nope b.txt") self.assertEqual((r.stdout, r.stderr, r.status), ("one\ntwo\n", "cat: nope: No such file or directory\n", 1)) self.assertEqual(self.out("echo hi | cat"), "hi\n") self.assertEqual(self.out("cat < a.txt"), "one\n") self.assertEqual(self.out("cat"), "") # run() treats missing stdin as empty self.assertEqual(self.out("cat -n a.txt"), "one\n") self.assertEqual(self.sh("cat /app").stderr, "cat: /app: Is a directory\n") def test_bare_cat_needs_stdin_in_a_session(self): c = tb.parse_line("cat")[0][1] with self.assertRaises(tb.NeedStdin): self.shell.exec_pipeline(c, None) self.assertEqual(self.shell.exec_pipeline(c, "x\n").stdout, "x\n") def test_echo(self): self.assertEqual(self.out("echo a b"), "a b\n") self.assertEqual(self.out("echo"), "\n") self.assertEqual(self.out("echo -n hi"), "hi") self.assertEqual(self.out("echo -n"), "") self.assertEqual(self.out("echo 'a b'"), "a b\n") self.assertEqual(self.out("echo hi -n"), "hi -n\n") def test_mkdir(self): self.assertEqual(self.sh("mkdir d").status, 0) self.assertTrue(self.sb.isdir("/app/d")) r = self.sh("mkdir d") self.assertEqual((r.status, r.stderr), (1, "mkdir: cannot create directory 'd': File exists\n")) self.assertEqual(self.sh("mkdir -p d").status, 0) self.assertEqual(self.sh("mkdir -p a/b/c").status, 0) self.assertTrue(self.sb.isdir("/app/a/b/c")) r = self.sh("mkdir x/y") self.assertEqual((r.status, r.stderr), (1, "mkdir: cannot create directory 'x/y': No such file or directory\n")) self.assertEqual(self.sh("mkdir /etc/x").status, 1) self.assertEqual(self.sh("mkdir e1 e2").status, 0) self.assertTrue(self.sb.isdir("/app/e2")) def test_touch(self): self.put("/app/keep.txt", "data") self.assertEqual(self.sh("touch new.txt keep.txt").status, 0) self.assertEqual(self.sb.read("/app/new.txt"), "") self.assertEqual(self.sb.read("/app/keep.txt"), "data") r = self.sh("touch /etc/x") self.assertEqual((r.status, r.stderr), (1, "touch: cannot touch '/etc/x': No such file or directory\n")) def test_cp(self): self.put("/app/a.txt", "A") self.assertEqual(self.sh("cp a.txt b.txt").status, 0) self.assertEqual(self.sb.read("/app/b.txt"), "A") self.sh("mkdir d") self.assertEqual(self.sh("cp a.txt d").status, 0) self.assertEqual(self.sb.read("/app/d/a.txt"), "A") self.assertEqual(self.sh("cp -r d d2").status, 0) self.assertEqual(self.sb.read("/app/d2/a.txt"), "A") r = self.sh("cp d d3") self.assertEqual((r.status, r.stderr), (1, "cp: -r not specified; omitting directory 'd'\n")) r = self.sh("cp nope x") self.assertEqual((r.status, r.stderr), (1, "cp: cannot stat 'nope': No such file or directory\n")) self.assertEqual(self.sh("cp a.txt").stderr, "cp: usage: cp [-r] SRC DST\n") self.assertEqual(self.sh("cp a.txt x/y").status, 1) self.assertEqual(self.sh("cp a.txt /etc/x").status, 1) self.assertEqual(self.sh("cp -r d d2").status, 0) # d2 exists now, so this copies to d2/d self.assertEqual(self.sb.read("/app/d2/d/a.txt"), "A") r = self.sh("cp -r d d2") self.assertEqual((r.status, r.stderr), (1, "cp: cannot overwrite 'd2'\n")) self.assertEqual(self.sb.read("/app/a.txt"), "A") # source untouched def test_mv(self): self.put("/app/a.txt", "A") self.assertEqual(self.sh("mv a.txt b.txt").status, 0) self.assertFalse(self.sb.exists("/app/a.txt")) self.assertEqual(self.sb.read("/app/b.txt"), "A") self.sh("mkdir d") self.assertEqual(self.sh("mv b.txt d").status, 0) self.assertEqual(self.sb.read("/app/d/b.txt"), "A") self.put("/app/c.txt", "C") self.put("/app/e.txt", "E") self.sh("mv c.txt e.txt") self.assertEqual(self.sb.read("/app/e.txt"), "C") # overwrites a file r = self.sh("mv nope x") self.assertEqual((r.status, r.stderr), (1, "mv: cannot stat 'nope': No such file or directory\n")) self.assertEqual(self.sh("mv e.txt").stderr, "mv: usage: mv SRC DST\n") self.assertEqual(self.sh("mv e.txt x/y").status, 1) self.assertEqual(self.sh("mv e.txt /etc/y").status, 1) self.assertTrue(self.sb.exists("/app/e.txt")) self.sh("mkdir -p d2/e.txt") r = self.sh("mv e.txt d2") self.assertEqual((r.status, r.stderr), (1, "mv: cannot overwrite directory 'd2'\n")) def test_rm(self): self.put("/app/a.txt", "") self.assertEqual(self.sh("rm a.txt").status, 0) self.assertFalse(self.sb.exists("/app/a.txt")) r = self.sh("rm a.txt") self.assertEqual((r.status, r.stderr), (1, "rm: cannot remove 'a.txt': No such file or directory\n")) self.assertEqual(self.sh("rm -f a.txt").status, 0) self.sh("mkdir -p d/e") r = self.sh("rm d") self.assertEqual((r.status, r.stderr), (1, "rm: cannot remove 'd': Is a directory\n")) self.assertEqual(self.sh("rm -r d").status, 0) self.assertFalse(self.sb.exists("/app/d")) self.sh("mkdir x") self.assertEqual(self.sh("rm -rf x").status, 0) self.assertEqual(self.sh("rm /etc/passwd").status, 1) def test_rm_protects_the_roots(self): for line in ("rm -rf /app", "rm -rf /tmp", "rm -r /app/", "cd /app && rm -rf ."): r = self.sh(line) self.assertEqual(r.status, 1, line) self.assertIn("Permission denied", r.stderr, line) self.assertTrue(self.sb.isdir("/app")) def test_true_false(self): self.assertEqual(self.sh("true").status, 0) self.assertEqual(self.sh("false").status, 1) def test_sleep(self): self.assertEqual(self.sh("sleep 5").status, 0) self.assertEqual(self.sh("sleep 1.5").status, 0) r = self.sh("sleep abc") self.assertEqual((r.status, r.stderr), (1, "sleep: invalid time interval 'abc'\n")) r = self.sh("sleep") self.assertEqual((r.status, r.stderr), (1, "sleep: missing operand\n")) class ShellText(ShellBase): def setUp(self): super().setUp() self.put("/app/n.txt", "".join(f"l{i}\n" for i in range(1, 13))) # l1 .. l12 self.put("/app/w.txt", "b\na\nc\na\n") def test_head(self): self.assertEqual(self.out("head n.txt").count("\n"), 10) self.assertEqual(self.out("head -n 2 n.txt"), "l1\nl2\n") self.assertEqual(self.out("head -3 n.txt"), "l1\nl2\nl3\n") self.assertEqual(self.out("head -n 0 n.txt"), "") self.assertEqual(self.out("cat n.txt | head -n 1"), "l1\n") self.assertEqual(self.out("head -n 1 w.txt n.txt"), "b\nl1\n") self.assertEqual(self.out("head -n 50 w.txt"), "b\na\nc\na\n") def test_tail(self): self.assertEqual(self.out("tail -n 2 n.txt"), "l11\nl12\n") self.assertEqual(self.out("tail -2 n.txt"), "l11\nl12\n") self.assertEqual(self.out("tail n.txt").splitlines()[0], "l3") self.assertEqual(self.out("tail -n 0 n.txt"), "") self.assertEqual(self.out("cat n.txt | tail -n 1"), "l12\n") self.assertEqual(self.out("tail -n 99 w.txt"), "b\na\nc\na\n") def test_head_tail_errors(self): r = self.sh("head nope") self.assertEqual((r.status, r.stderr), (1, "head: nope: No such file or directory\n")) self.assertEqual(self.sh("tail -n 1 nope w.txt").stdout, "a\n") def test_wc(self): self.assertEqual(self.out("wc -l w.txt"), "4 w.txt\n") self.assertEqual(self.out("wc -w w.txt"), "4 w.txt\n") self.assertEqual(self.out("wc -c w.txt"), "8 w.txt\n") self.assertEqual(self.out("wc w.txt"), "4 4 8 w.txt\n") self.assertEqual(self.out("cat w.txt | wc -l"), "4\n") self.assertEqual(self.out("echo 'a b c' | wc -w"), "3\n") self.sb.write("/app/u.txt", "é\n") self.assertEqual(self.out("wc -c u.txt"), "3 u.txt\n") # bytes, not characters self.assertEqual(self.out("wc -l w.txt n.txt"), "4 w.txt\n12 n.txt\n") self.assertEqual(self.sh("wc -l nope").status, 1) def test_sort(self): self.assertEqual(self.out("sort w.txt"), "a\na\nb\nc\n") self.assertEqual(self.out("sort -r w.txt"), "c\nb\na\na\n") self.assertEqual(self.out("sort -u w.txt"), "a\nb\nc\n") self.assertEqual(self.out("sort -ru w.txt"), "c\nb\na\n") self.put("/app/num.txt", "10\n9\n2\n") self.assertEqual(self.out("sort num.txt"), "10\n2\n9\n") self.assertEqual(self.out("sort -n num.txt"), "2\n9\n10\n") self.assertEqual(self.out("sort -nr num.txt"), "10\n9\n2\n") self.put("/app/mix.txt", "x\n3\n-1\n2.5\n") self.assertEqual(self.out("sort -n mix.txt"), "-1\nx\n2.5\n3\n") # x counts as 0 self.assertEqual(self.out("sort w.txt num.txt"), "10\n2\n9\na\na\nb\nc\n") self.assertEqual(self.out("cat w.txt | sort -u"), "a\nb\nc\n") self.assertEqual(self.sh("sort nope").status, 1) def test_uniq(self): self.assertEqual(self.out("uniq w.txt"), "b\na\nc\na\n") # only adjacent repeats go self.assertEqual(self.out("sort w.txt | uniq"), "a\nb\nc\n") self.assertEqual(self.out("sort w.txt | uniq -c"), " 2 a\n 1 b\n 1 c\n") self.assertEqual(self.out("echo -n | uniq"), "") self.put("/app/e.txt", "x\n\n\ny\n") self.assertEqual(self.out("uniq e.txt"), "x\n\ny\n") self.assertEqual(self.sh("uniq nope").status, 1) def test_grep_basics(self): self.put("/app/log.txt", "INFO a\nERROR x\nerror y\nINFO b.c\nINFOxb\n") self.assertEqual(self.out("grep ERROR log.txt"), "ERROR x\n") self.assertEqual(self.out("grep -i error log.txt"), "ERROR x\nerror y\n") self.assertEqual(self.out("grep -v INFO log.txt"), "ERROR x\nerror y\n") self.assertEqual(self.out("grep -c INFO log.txt"), "3\n") self.assertEqual(self.out("grep -n ERROR log.txt"), "2:ERROR x\n") self.assertEqual(self.out("grep -cv INFO log.txt"), "2\n") self.assertEqual(self.out("cat log.txt | grep -c x"), "2\n") # 'ERROR x' and 'INFOxb' self.assertEqual(self.out("grep b.c log.txt"), "INFO b.c\n") # literal dot without -E self.assertEqual(self.out("grep -E 'b.c' log.txt"), "INFO b.c\n") self.assertEqual(self.out("grep -E 'INFO.b' log.txt"), "INFO b.c\nINFOxb\n") self.assertEqual(self.out("grep 'INFO.b' log.txt"), "") def test_grep_status_and_errors(self): self.put("/app/log.txt", "a\nb\n") self.assertEqual(self.sh("grep a log.txt").status, 0) self.assertEqual(self.sh("grep zzz log.txt").status, 1) self.assertEqual(self.sh("grep -c zzz log.txt").stdout, "0\n") self.assertEqual(self.sh("grep -c zzz log.txt").status, 1) self.assertEqual(self.sh("grep -v zzz log.txt").status, 0) r = self.sh("grep a nope") self.assertEqual((r.status, r.stderr), (2, "grep: nope: No such file or directory\n")) self.assertEqual(self.sh("grep").status, 2) r = self.sh("grep -E '(' log.txt") self.assertEqual((r.status, r.stderr), (2, "grep: invalid regular expression\n")) self.assertEqual(self.sh("grep a nope log.txt").stdout, "log.txt:a\n") self.assertEqual(self.sh("grep a nope log.txt").status, 2) def test_grep_many_files(self): self.put("/app/p.txt", "x1\nz\n") self.put("/app/q.txt", "x2\nx3\n") self.assertEqual(self.out("grep x p.txt q.txt"), "p.txt:x1\nq.txt:x2\nq.txt:x3\n") self.assertEqual(self.out("grep -c x p.txt q.txt"), "p.txt:1\nq.txt:2\n") self.assertEqual(self.out("grep -n x p.txt q.txt"), "p.txt:1:x1\nq.txt:1:x2\nq.txt:2:x3\n") self.assertEqual(self.out("grep -n x p.txt"), "1:x1\n") def test_cut(self): self.put("/app/c.csv", "id,name,score\n1,ann,90\n2,bo\n") self.assertEqual(self.out("cut -d, -f2 c.csv"), "name\nann\nbo\n") self.assertEqual(self.out("cut -d , -f 1,3 c.csv"), "id,score\n1,90\n2\n") self.assertEqual(self.out("cut -f1 c.csv"), "id,name,score\n1,ann,90\n2,bo\n") self.assertEqual(self.out("cut -d, -f3 c.csv"), "score\n90\n\n") self.assertEqual(self.out("echo 'a b c' | cut -d' ' -f2"), "b\n") self.assertEqual(self.out("cut -d, -f9 c.csv"), "\n\n\n") for bad in ("cut -d, c.csv", "cut -d, -fx c.csv", "cut -d, -f1, c.csv"): r = self.sh(bad) self.assertEqual((r.status, r.stderr), (1, "cut: you must specify a list of fields (-f N[,M])\n"), bad) self.assertEqual(self.sh("cut -d, -f1 nope").status, 1) def test_sed(self): self.put("/app/s.txt", "aaa\nbab\n") self.assertEqual(self.out("sed 's/a/X/' s.txt"), "Xaa\nbXb\n") self.assertEqual(self.out("sed 's/a/X/g' s.txt"), "XXX\nbXb\n") self.assertEqual(self.out("sed 's/b+/_/' s.txt"), "aaa\n_ab\n") # Python regex, not POSIX BRE self.assertEqual(self.out("sed 's/b/[&]/' s.txt"), "aaa\n[&]ab\n") self.assertEqual(self.sh("sed 's/\\(b\\)a/\\1\\1/' s.txt").status, 1) # BRE groups are not supported self.assertEqual(self.out("sed 's/(b)a/\\1\\1/' s.txt"), "aaa\nbbb\n") self.assertEqual(self.out("echo hello | sed 's/l/L/g'"), "heLLo\n") self.assertEqual(self.out("sed 's/a//' s.txt"), "aa\nbb\n") self.assertEqual(self.out("sed 's/x/y/' s.txt"), "aaa\nbab\n") def test_sed_errors(self): self.put("/app/s.txt", "a\n") self.assertEqual(self.sh("sed").stderr, "sed: usage: sed 's/OLD/NEW/[g]' [FILE]\n") for bad in ("sed d s.txt", "sed 's/a/b' s.txt", "sed 's/a/b/x' s.txt", "sed 'y/a/b/' s.txt"): r = self.sh(bad) self.assertEqual((r.status, r.stderr), (1, "sed: only s/OLD/NEW/ and s/OLD/NEW/g are supported\n"), bad) r = self.sh("sed 's/(/x/' s.txt") self.assertEqual((r.status, r.stderr), (1, "sed: invalid regular expression\n")) r = self.sh("sed 's/a/\\1/' s.txt") self.assertEqual((r.status, r.stderr), (1, "sed: invalid reference in the replacement\n")) self.assertEqual(self.sh("sed 's/a/b/' nope").status, 1) def test_sha256sum(self): self.put("/app/abc.txt", "abc") self.assertEqual(self.out("sha256sum abc.txt"), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad abc.txt\n") self.assertEqual(self.out("echo -n abc | sha256sum"), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad -\n") self.assertEqual(self.out("sha256sum abc.txt | cut -d' ' -f1"), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad\n") self.assertEqual(self.sh("sha256sum nope").status, 1) class ShellStructure(ShellBase): def test_operators(self): self.assertEqual(self.out("echo a && echo b"), "a\nb\n") self.assertEqual(self.out("false && echo no"), "") self.assertEqual(self.out("false || echo yes"), "yes\n") self.assertEqual(self.out("true || echo no"), "") self.assertEqual(self.out("false ; echo after"), "after\n") self.assertEqual(self.out("echo a; echo b"), "a\nb\n") self.assertEqual(self.out("false && echo no || echo rescue"), "rescue\n") self.assertEqual(self.out("true && false || echo x && echo y"), "x\ny\n") def test_status_is_the_last_one(self): self.assertEqual(self.sh("true && false").status, 1) self.assertEqual(self.sh("false ; true").status, 0) self.assertEqual(self.sh("false | true").status, 0) self.assertEqual(self.sh("true | false").status, 1) self.assertEqual(self.sh("false && true").status, 1) def test_pipes(self): self.assertEqual(self.out("echo hi | cat | cat"), "hi\n") self.put("/app/w.txt", "b\na\nb\n") self.assertEqual(self.out("cat w.txt | sort | uniq -c | sort -r"), " 2 b\n 1 a\n") def test_redirects(self): self.assertEqual(self.sh("echo hi > f.txt").stdout, "") self.assertEqual(self.sb.read("/app/f.txt"), "hi\n") self.sh("echo more >> f.txt") self.assertEqual(self.sb.read("/app/f.txt"), "hi\nmore\n") self.sh("echo new > f.txt") self.assertEqual(self.sb.read("/app/f.txt"), "new\n") self.assertEqual(self.out("cat < f.txt"), "new\n") self.sh("sort < f.txt > g.txt") self.assertEqual(self.sb.read("/app/g.txt"), "new\n") self.assertEqual(self.sb.read("/app/f.txt"), "new\n") def test_redirect_errors(self): r = self.sh("echo hi > /etc/x") self.assertEqual((r.status, r.stderr), (1, "bash: /etc/x: No such file or directory\n")) r = self.sh("echo hi > nodir/x") self.assertEqual(r.status, 1) r = self.sh("cat < nope") self.assertEqual((r.status, r.stderr), (1, "bash: nope: No such file or directory\n")) self.assertEqual(self.sh("echo hi > /app").status, 1) def test_redirect_in_the_middle_of_a_pipe(self): self.assertEqual(self.out("echo hi > f.txt | cat"), "") self.assertEqual(self.sb.read("/app/f.txt"), "hi\n") def test_failed_command_still_creates_its_redirect_file(self): self.sh("nosuch > f.txt") self.assertTrue(self.sb.exists("/app/f.txt")) def test_syntax_errors(self): for bad in ("echo 'a", "a &&", "| a", "echo a & echo b", "echo >"): r = self.sh(bad) self.assertEqual(r.status, 2, bad) self.assertTrue(r.stderr.startswith("bash: "), bad) self.assertEqual(r.stdout, "", bad) self.assertEqual(self.sh("echo 'a").stderr, "bash: unterminated quote\n") self.assertEqual(self.sh("echo a & echo b").stderr, "bash: unsupported operator &\n") def test_empty_line(self): r = self.sh("") self.assertEqual((r.stdout, r.stderr, r.status), ("", "", 0)) self.assertEqual(self.sh(" ").status, 0) self.assertEqual(self.sh("> f.txt").status, 0) self.assertFalse(self.sb.exists("/app/f.txt")) # a bare redirect does nothing here def test_output_property(self): r = tb.Result("a", "b", 3, 0.5) self.assertEqual(r.output, "ab") def test_truncation(self): self.put("/app/big.txt", "x" * 12000) r = self.sh("cat big.txt") self.assertEqual(r.stdout, "x" * 10000 + "\n[output truncated]\n") self.put("/app/ok.txt", "y" * 10000) self.assertEqual(self.out("cat ok.txt"), "y" * 10000) def test_stdin_argument(self): self.assertEqual(self.out("cat", stdin="given\n"), "given\n") self.assertEqual(self.out("sort", stdin="b\na\n"), "a\nb\n") def test_split_flags(self): f = tb.Shell._split_flags self.assertEqual(f(["-n", "5", "--", "-x"]), (["-n"], ["5", "-x"])) self.assertEqual(f(["-5"]), ([], ["-5"])) self.assertEqual(f(["-"]), ([], ["-"])) self.assertEqual(f(["a", "-b"]), (["-b"], ["a"])) self.assertEqual(f(["--", "--"]), ([], ["--"])) class ShellAllowlist(ShellBase): allowed = {"echo", "foo"} def test_blocked_command(self): r = self.sh("ls") self.assertEqual((r.status, r.stderr, r.stdout), (127, "bash: ls: command not found\n", "")) self.assertEqual(self.out("echo ok"), "ok\n") def test_name_without_implementation(self): r = self.sh("foo") self.assertEqual((r.status, r.stderr), (127, "bash: foo: command not found\n")) def test_allowlist_applies_inside_pipes(self): r = self.sh("echo a | sort") self.assertEqual(r.status, 127) def test_builtins_list(self): self.assertEqual(len(tb.BUILTINS), 22) for name in tb.BUILTINS: self.assertTrue(hasattr(tb.Shell, "cmd_" + name), name) for bad in ("curl", "python3", "bash", "sh", "apt", "docker", "ssh", "wget", "sudo", "kill"): self.assertNotIn(bad, tb.BUILTINS) def test_default_allows_all_builtins(self): self.assertEqual(tb.Shell(self.sb).allowed, tb.BUILTINS) class ShellTiming(ShellBase): def test_pipeline_duration(self): d = lambda line: self.shell.pipeline_duration(tb.parse_line(line)[0][1]) self.assertAlmostEqual(d("ls"), 0.05) self.assertAlmostEqual(d("ls | sort"), 0.10) self.assertAlmostEqual(d("sleep 90"), 90.05) self.assertAlmostEqual(d("sleep 1.5"), 1.55) self.assertAlmostEqual(d("sleep abc"), 0.05) self.assertAlmostEqual(d("sleep"), 0.05) self.assertAlmostEqual(d("sleep -5"), 0.05) self.assertAlmostEqual(d("echo sleep 9"), 0.05) self.assertAlmostEqual(d("sleep 2 | sleep 3"), 5.10) def test_run_durations(self): self.assertAlmostEqual(self.sh("ls").duration, 0.05) self.assertAlmostEqual(self.sh("ls && ls").duration, 0.10) self.assertAlmostEqual(self.sh("false && ls").duration, 0.05) self.assertAlmostEqual(self.sh("nosuch").duration, 0.05) self.assertAlmostEqual(self.sh("echo a | cat | cat").duration, 0.15) self.assertAlmostEqual(self.sh("sleep 3").duration, 3.05) self.assertAlmostEqual(self.sh("sleep 3 && sleep 2").duration, 5.10) self.assertAlmostEqual(self.sh("false && sleep 9").duration, 0.05) def test_constants(self): self.assertEqual((tb.BASE_COST, tb.INSTANT, tb.MAX_OUTPUT), (0.05, 1.0, 10_000)) class SessionBase(unittest.TestCase): def setUp(self): self.sb = tb.Sandbox(mk_task()) self.addCleanup(self.sb.cleanup) self.clock = tb.Clock() self.s = tb.TmuxSession(self.sb, self.clock) def keys(self, *k, wait=1.0): self.s.send_keys(list(k), wait) def run_line(self, line, wait=1.0): self.keys(line, "Enter", wait=wait) class SessionTests(SessionBase): def test_fresh_pane(self): self.assertEqual(self.s.capture_pane(), "/app$ ") self.assertEqual(self.s.prompt(), "/app$ ") def test_typing_without_enter(self): self.s.send_keys("ls", wait=0) self.assertEqual(self.s.capture_pane(), "/app$ ls") self.keys("-la", wait=0) self.assertEqual(self.s.capture_pane(), "/app$ ls-la") def test_command_and_output(self): self.sb.write("/app/a.txt", "") self.run_line("ls") self.assertEqual(self.s.capture_pane(), "/app$ ls\na.txt\n/app$ ") def test_empty_enter(self): self.keys("Enter") self.assertEqual(self.s.capture_pane(), "/app$ \n/app$ ") self.keys("Enter") self.assertEqual(self.s.capture_pane(), "/app$ \n/app$ \n/app$ ") def test_no_output_command(self): self.run_line("touch x") self.assertEqual(self.s.capture_pane(), "/app$ touch x\n/app$ ") def test_stderr_is_shown(self): self.run_line("nosuch") self.assertEqual(self.s.capture_pane(), "/app$ nosuch\nbash: nosuch: command not found\n/app$ ") def test_syntax_error_is_shown(self): self.run_line("echo 'a") self.assertEqual(self.s.capture_pane(), "/app$ echo 'a\nbash: unterminated quote\n/app$ ") def test_blank_lines_in_output_are_kept(self): self.run_line("echo a; echo; echo b") self.assertEqual(self.s.capture_pane(), "/app$ echo a; echo; echo b\na\n\nb\n/app$ ") def test_output_without_newline(self): self.run_line("echo -n hi") self.assertEqual(self.s.capture_pane(), "/app$ echo -n hi\nhi\n/app$ ") def test_emit_empty(self): self.s._emit("") self.assertEqual(self.s.history, []) self.s._emit("a\n\n") self.assertEqual(self.s.history, ["a", ""]) def test_cd_changes_the_prompt(self): self.run_line("cd /tmp") self.assertTrue(self.s.capture_pane().endswith("\n/tmp$ ")) def test_wrap(self): s = tb.TmuxSession(self.sb, self.clock, width=10) s.send_keys("echo hello world", wait=0) self.assertEqual(s.capture_pane(), "/app$ echo\n hello wor\nld") def test_wrap_exact_width(self): s = tb.TmuxSession(self.sb, self.clock, width=10) s.send_keys("1234", wait=0) self.assertEqual(s.capture_pane(), "/app$ 1234") def test_height_keeps_the_last_lines(self): s = tb.TmuxSession(self.sb, self.clock, height=3) for w in ("a", "b", "c"): s.send_keys([f"echo {w}", "Enter"], wait=0) self.assertEqual(s.capture_pane(), "/app$ echo c\nc\n/app$ ") def test_c_c_at_the_prompt(self): self.s.send_keys("abc", wait=0) self.keys("C-c", wait=0) self.assertEqual(self.s.capture_pane(), "/app$ abc^C\n/app$ ") self.assertEqual(self.s.typed, "") def test_c_d_at_the_prompt_is_ignored(self): self.s.send_keys("ab", wait=0) self.keys("C-d", wait=0) self.assertEqual(self.s.capture_pane(), "/app$ ab") def test_clock(self): self.s.send_keys("x", wait=2.5) self.assertEqual(self.clock.now, 2.5) self.clock.advance(-3) self.assertEqual(self.clock.now, 2.5) self.s.advance(0.5) self.assertEqual(self.clock.now, 3.0) class SessionLongCommands(SessionBase): def test_instant_boundary(self): self.run_line("sleep 0.9") self.assertIsNone(self.s.running) self.run_line("sleep 1") self.assertIsNotNone(self.s.running) def test_long_command_hides_the_prompt(self): self.run_line("sleep 5", wait=1) self.assertEqual(self.s.capture_pane(), "/app$ sleep 5") self.assertIsNotNone(self.s.running) def test_finish_time(self): self.run_line("sleep 5", wait=0) self.s.advance(5.0) self.assertIsNotNone(self.s.running) # 5.05 s are needed self.s.advance(0.04) self.assertIsNotNone(self.s.running) self.s.advance(0.02) self.assertIsNone(self.s.running) self.assertEqual(self.s.capture_pane(), "/app$ sleep 5\n/app$ ") def test_effects_wait_for_the_clock(self): self.run_line("sleep 5 && echo x > /app/o.txt", wait=4) self.assertFalse(self.sb.exists("/app/o.txt")) self.s.advance(1.2) self.assertEqual(self.sb.read("/app/o.txt"), "x\n") def test_one_big_advance_finishes_everything(self): self.run_line("sleep 5 && echo x > /app/o.txt", wait=100) self.assertTrue(self.sb.exists("/app/o.txt")) self.assertIsNone(self.s.running) def test_c_c_kills_before_the_effects(self): self.run_line("sleep 5 && echo x > /app/o.txt", wait=1) self.keys("C-c", wait=0) self.assertIsNone(self.s.running) self.s.advance(100) self.assertFalse(self.sb.exists("/app/o.txt")) self.assertEqual(self.s.capture_pane(), "/app$ sleep 5 && echo x > /app/o.txt\n^C\n/app$ ") def test_chain_with_two_sleeps(self): self.run_line("sleep 3 ; echo a ; sleep 3 ; echo b", wait=3.2) self.assertIn("\na", self.s.capture_pane()) self.assertNotIn("\nb", self.s.capture_pane()) self.assertIsNotNone(self.s.running) self.s.advance(2.9) # t = 6.1, finish at 6.15 self.assertIsNotNone(self.s.running) self.s.advance(0.1) # t = 6.2 self.assertIsNone(self.s.running) self.assertEqual(self.s.history[-2:], ["a", "b"]) def test_and_skips_after_failure(self): self.run_line("false && sleep 50") self.assertIsNone(self.s.running) self.assertEqual(self.s.capture_pane(), "/app$ false && sleep 50\n/app$ ") def test_or_runs_a_sleep_after_failure(self): self.run_line("false || sleep 50") self.assertIsNotNone(self.s.running) def test_keys_during_a_sleep_are_lost(self): self.run_line("sleep 5", wait=0) self.keys("echo x > /app/lost.txt", "Enter", wait=0) self.s.advance(10) self.assertFalse(self.sb.exists("/app/lost.txt")) self.assertEqual(self.s.typed, "") self.assertEqual(self.s.capture_pane(), "/app$ sleep 5\n/app$ ") def test_long_command_starts_after_earlier_work(self): self.run_line("echo a > /app/a.txt && sleep 5 && echo b > /app/b.txt", wait=1) self.assertTrue(self.sb.exists("/app/a.txt")) # the first, instant part already ran self.assertFalse(self.sb.exists("/app/b.txt")) self.s.advance(10) self.assertTrue(self.sb.exists("/app/b.txt")) def test_background_job_time_is_virtual(self): import time t0 = time.time() self.run_line("sleep 3600", wait=4000) self.assertLess(time.time() - t0, 1.0) self.assertIsNone(self.s.running) class SessionStdin(SessionBase): def test_cat_to_file(self): self.keys("cat > /app/n.txt", "Enter", "one", "Enter", "two", "Enter", wait=0) self.assertIsNotNone(self.s.running) self.assertFalse(self.sb.exists("/app/n.txt")) self.keys("C-d", wait=0) self.assertIsNone(self.s.running) self.assertEqual(self.sb.read("/app/n.txt"), "one\ntwo\n") self.assertTrue(self.s.capture_pane().endswith("two\n/app$ ")) def test_partial_line_is_shown_without_a_prompt(self): self.keys("cat", "Enter", "par", wait=0) self.assertEqual(self.s.capture_pane(), "/app$ cat\npar") def test_cat_echoes_back(self): self.keys("cat", "Enter", "x", "Enter", "C-d", wait=0) self.assertEqual(self.s.capture_pane(), "/app$ cat\nx\nx\n/app$ ") def test_c_c_cancels(self): self.keys("cat > /app/n.txt", "Enter", "one", "Enter", "C-c", wait=0) self.assertIsNone(self.s.running) self.assertFalse(self.sb.exists("/app/n.txt")) self.assertEqual(self.s.capture_pane(), "/app$ cat > /app/n.txt\none\n^C\n/app$ ") def test_typed_chars_are_discarded_on_c_c(self): self.keys("cat", "Enter", "abc", "C-c", wait=0) self.assertEqual(self.s.typed, "") def test_empty_input(self): self.keys("cat > /app/n.txt", "Enter", "C-d", wait=0) self.assertEqual(self.sb.read("/app/n.txt"), "") def test_pipeline_after_stdin_command(self): self.keys("sort", "Enter", "b", "Enter", "a", "Enter", "C-d", wait=0) self.assertEqual(self.s.history[-2:], ["a", "b"]) def test_stdin_command_then_more_chain(self): self.keys("cat > /app/n.txt ; echo done", "Enter", "q", "Enter", "C-d", wait=0) self.assertEqual(self.s.history[-1], "done") self.assertEqual(self.sb.read("/app/n.txt"), "q\n") def test_stdin_command_then_sleep(self): self.keys("cat > /app/n.txt ; sleep 5", "Enter", "q", "Enter", "C-d", wait=0) self.assertIsNotNone(self.s.running) self.assertEqual(self.s.running.kind, "time") self.s.advance(6) self.assertIsNone(self.s.running) def checks_task(*checks, extra=None, verifier_timeout=10): """A task whose checks.toml is built from (type, fields) pairs.""" toml = "".join("[[check]]\ntype = %s\n%s" % ( '"%s"' % t, "".join('%s = %s\n' % (k, repr(v).replace("'", '"') if isinstance(v, str) else v) for k, v in f.items())) for t, f in checks) return mk_task(checks=toml, extra=extra or {}, task=tb._toml_task("lab/tiny", "x", "easy", verifier_timeout=verifier_timeout)) class VerifierTests(unittest.TestCase): def grade(self, task, files=None, leaky=False): with tb.Sandbox(task, leaky=leaky) as sb: for p, t in (files or {}).items(): sb.write(p, t) return tb.verify(task, sb) def one(self, ctype, fields, files, extra=None): t = checks_task((ctype, fields), extra=extra) return self.grade(t, files) def test_file_exists_and_absent(self): self.assertEqual(self.one("file_exists", {"path": "/app/a"}, {"/app/a": ""}).reward, 1.0) r = self.one("file_exists", {"path": "/app/a"}, {}) self.assertEqual((r.reward, r.outcomes[0].detail), (0.0, "/app/a does not exist")) self.assertEqual(self.one("file_absent", {"path": "/app/a"}, {}).reward, 1.0) r = self.one("file_absent", {"path": "/app/a"}, {"/app/a": ""}) self.assertEqual((r.reward, r.outcomes[0].detail), (0.0, "/app/a exists")) def test_file_equals(self): eq = lambda have, want: self.one("file_equals", {"path": "/app/a", "text": want}, {"/app/a": have}).reward self.assertEqual(eq("3\n", "3"), 1.0) self.assertEqual(eq("3", "3"), 1.0) self.assertEqual(eq("3\n\n\n", "3"), 1.0) self.assertEqual(eq("a\r\nb\r\n", "a\nb"), 1.0) self.assertEqual(eq("3 \n", "3"), 0.0) # inner spaces matter self.assertEqual(eq(" 3\n", "3"), 0.0) self.assertEqual(eq("33", "3"), 0.0) r = self.one("file_equals", {"path": "/app/a", "text": "3"}, {"/app/a": "4"}) self.assertEqual(r.outcomes[0].detail, "/app/a does not match") self.assertEqual(eq("7\n", "7"), 1.0) def test_file_equals_accepts_numbers(self): t = checks_task(("file_equals", {"path": "/app/a", "text": 7})) self.assertEqual(self.grade(t, {"/app/a": "7\n"}).reward, 1.0) def test_contains(self): c = lambda t, have, want: self.one(t, {"path": "/app/a", "text": want}, {"/app/a": have}).reward self.assertEqual(c("file_contains", "abc", "b"), 1.0) self.assertEqual(c("file_contains", "abc", "d"), 0.0) self.assertEqual(c("file_not_contains", "abc", "d"), 1.0) self.assertEqual(c("file_not_contains", "abc", "b"), 0.0) def test_file_lines(self): c = lambda have, n: self.one("file_lines", {"path": "/app/a", "count": n}, {"/app/a": have}).reward self.assertEqual(c("a\nb\nc\n", 3), 1.0) self.assertEqual(c("a\nb\nc", 3), 1.0) self.assertEqual(c("a\nb\n", 3), 0.0) self.assertEqual(c("", 0), 1.0) self.assertEqual(c("a\n\nb\n", 3), 1.0) def test_sha256(self): d = "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" # sha256 of "abc" c = lambda have: self.one("file_sha256", {"path": "/app/a", "sha256": d}, {"/app/a": have}).reward self.assertEqual(c("abc"), 1.0) self.assertEqual(c("abc\n"), 0.0) def test_equals_file_uses_the_pristine_test_file(self): t = checks_task(("file_equals_file", {"path": "/app/a", "expected": "want.txt"}), extra={"tests/want.txt": "x\ny\n"}) self.assertEqual(self.grade(t, {"/app/a": "x\ny\n"}).reward, 1.0) self.assertEqual(self.grade(t, {"/app/a": "x\ny"}).reward, 1.0) self.assertEqual(self.grade(t, {"/app/a": "x\nz\n"}).reward, 0.0) def test_cmd_stdout(self): t = checks_task(("cmd_stdout", {"cmd": "cat /app/a | wc -l", "equals": "2"})) self.assertEqual(self.grade(t, {"/app/a": "p\nq\n"}).reward, 1.0) r = self.grade(t, {"/app/a": "p\n"}) self.assertEqual((r.reward, r.outcomes[0].detail), (0.0, "stdout was '1'")) t = checks_task(("cmd_stdout", {"cmd": "echo 5", "equals": 5})) self.assertEqual(self.grade(t).reward, 1.0) def test_missing_and_outside_files_fail_cleanly(self): for path in ("/app/missing", "/etc/passwd", "/app"): r = self.one("file_equals", {"path": path, "text": "x"}, {}) self.assertEqual(r.reward, 0.0, path) self.assertIn(path, r.outcomes[0].detail) r = self.one("file_equals", {"path": "/app/missing", "text": "x"}, {}) self.assertEqual(r.outcomes[0].detail, "/app/missing: No such file or directory") r = self.one("cmd_stdout", {"cmd": "cat /app/missing", "equals": "x"}, {}) self.assertEqual(r.reward, 0.0) def test_unknown_check_type(self): with tb.Sandbox(mk_task()) as sb: sb.write("/app/a", "") o = tb.run_check(sb, {"type": "mystery", "path": "/app/a", "name": "m"}, lambda n: "") self.assertEqual((o.name, o.passed, o.detail), ("m", False, "unknown check type mystery")) def test_all_or_nothing(self): t = checks_task(("file_exists", {"path": "/app/a"}), ("file_exists", {"path": "/app/b"}), ("file_exists", {"path": "/app/c"})) r = self.grade(t, {"/app/a": "", "/app/b": ""}) self.assertEqual((r.reward, r.n_passed, len(r.outcomes)), (0.0, 2, 3)) r = self.grade(t, {"/app/a": "", "/app/b": "", "/app/c": ""}) self.assertEqual((r.reward, r.n_passed), (1.0, 3)) self.assertEqual(self.grade(t).n_passed, 0) def test_outcome_names(self): t = checks_task(("file_exists", {"path": "/app/a"}), ("file_exists", {"path": "/app/b", "name": "second"})) self.assertEqual([o.name for o in self.grade(t).outcomes], ["file_exists_0", "second"]) def test_isolated_mode_removes_the_tests(self): t = task_by_name("lab/sort-names") with tb.Sandbox(t) as sb: tb.verify(t, sb) self.assertFalse(sb.exists("/tests")) self.assertNotIn("/tests", sb.roots) self.assertFalse((sb.root / "tests").exists()) self.assertEqual(sb.roots, ["/app", "/tmp"]) def test_isolated_mode_ignores_what_the_agent_could_not_touch(self): t = task_by_name("lab/sort-names") with tb.Sandbox(t) as sb: sb.write("/app/sorted.txt", "adam\nbob\nmia\nzoe\n") self.assertEqual(tb.verify(t, sb).reward, 1.0) def test_leaky_mode_reads_the_tests_from_the_sandbox(self): t = task_by_name("lab/sort-names") with tb.Sandbox(t, leaky=True) as sb: sb.write("/app/sorted.txt", "nonsense\n") self.assertEqual(tb.verify(t, sb).reward, 0.0) sb.write("/tests/checks.toml", '[[check]]\ntype = "file_exists"\npath = "/app/sorted.txt"\n') r = tb.verify(t, sb) self.assertEqual((r.reward, len(r.outcomes)), (1.0, 1)) self.assertTrue(sb.exists("/tests")) # leaky mode leaves them in place def test_leaky_mode_with_broken_or_missing_checks(self): t = task_by_name("lab/sort-names") with tb.Sandbox(t, leaky=True) as sb: sb.write("/tests/checks.toml", "x = [") r = tb.verify(t, sb) self.assertEqual((r.reward, r.outcomes), (0.0, [])) self.assertTrue(r.error.startswith("could not load checks: ")) sb.write("/tests/checks.toml", "") self.assertEqual(tb.verify(t, sb).reward, 0.0) with tb.Sandbox(t, leaky=True) as sb: os.remove(sb.root / "tests" / "checks.toml") r = tb.verify(t, sb) self.assertEqual(r.reward, 0.0) self.assertIn("No such file", r.error) def test_leaky_equals_file_reads_the_sandbox_copy(self): t = task_by_name("lab/sort-names") with tb.Sandbox(t, leaky=True) as sb: sb.write("/app/sorted.txt", "zzz\n") sb.write("/tests/expected_sorted.txt", "zzz\n") sb.write("/tests/checks.toml", '[[check]]\ntype = "file_equals_file"\npath = "/app/sorted.txt"\nexpected = "expected_sorted.txt"\n') self.assertEqual(tb.verify(t, sb).reward, 1.0) def test_verifier_timeout(self): t = checks_task(("cmd_stdout", {"cmd": "sleep 20 && echo x", "equals": "x"})) r = self.grade(t) self.assertEqual((r.reward, r.outcomes[0].detail), (0.0, "verifier timeout")) t = checks_task(("cmd_stdout", {"cmd": "sleep 5 && echo x", "equals": "x"})) self.assertEqual(self.grade(t).reward, 1.0) t = checks_task(("cmd_stdout", {"cmd": "sleep 6 && echo x", "equals": "x"}), ("cmd_stdout", {"cmd": "sleep 6 && echo x", "equals": "x"})) r = self.grade(t) self.assertEqual([o.passed for o in r.outcomes], [True, False]) self.assertEqual(r.outcomes[1].detail, "verifier timeout") t = checks_task(("cmd_stdout", {"cmd": "sleep 20 && echo x", "equals": "x"}), verifier_timeout=30) self.assertEqual(self.grade(t).reward, 1.0) def test_check_count_never_zero(self): self.assertEqual(tb.VerifierResult([], 0.0).n_passed, 0) class StubRng: def __init__(self, values): self.values = list(values) def random(self): return self.values.pop(0) class AgentTests(unittest.TestCase): def test_base_agent(self): a = tb.Agent() self.assertEqual((a.price_in, a.price_out, a.name), (3.0, 15.0, "agent")) with self.assertRaises(NotImplementedError): a.act("", 0) self.assertTrue(tb.NoopAgent().act("", 0).done) self.assertFalse(tb.Action().done) self.assertEqual((tb.Action().keys, tb.Action().wait, tb.Action().think), ((), 1.0, 0.0)) def test_mutate(self): a = tb.ScriptedAgent() self.assertEqual(a._mutate("echo hi there"), "echox hi there") self.assertEqual(a._mutate("ls"), "lsx") def agent(self, script, rolls, **kw): a = tb.ScriptedAgent(script=script, **kw) a.reset(mk_task(), StubRng(rolls)) return a def test_no_typo_walks_the_script(self): a = self.agent(["echo 1", "echo 2"], [0.9, 0.9], p_typo=0.5, latency=3.0) x = a.act("", 0) self.assertEqual((x.keys, x.wait, x.think, x.done), (("echo 1", "Enter"), 1.0, 3.0, False)) self.assertEqual(a.act("", 1).keys, ("echo 2", "Enter")) self.assertTrue(a.act("", 2).done) def test_roll_boundary(self): a = self.agent(["ls"], [0.5], p_typo=0.5) self.assertEqual(a.act("", 0).keys, ("ls", "Enter")) # random() == p_typo is not a typo a = self.agent(["ls"], [0.4999], p_typo=0.5) self.assertEqual(a.act("", 0).keys, ("lsx", "Enter")) def test_typo_then_retry_then_success(self): a = self.agent(["echo 1", "echo 2"], [0.0, 0.9, 0.9], p_typo=0.5, retry=True) self.assertEqual(a.act("", 0).keys[0], "echox 1") self.assertEqual(a.act("bash: echox: command not found", 1).keys[0], "echo 1") self.assertEqual(a.act("", 2).keys[0], "echo 2") self.assertTrue(a.act("", 3).done) def test_retry_needs_to_see_the_error(self): a = self.agent(["echo 1", "echo 2"], [0.0, 0.9], p_typo=0.5, retry=True) a.act("", 0) self.assertEqual(a.act("something else", 1).keys[0], "echo 2") # gave up on line 1 def test_no_retry_flag(self): a = self.agent(["echo 1", "echo 2"], [0.0, 0.9], p_typo=0.5, retry=False) a.act("", 0) self.assertEqual(a.act("command not found", 1).keys[0], "echo 2") def test_max_retries(self): a = self.agent(["echo 1", "echo 2"], [0.0, 0.0, 0.0, 0.9], p_typo=0.5, retry=True, max_retries=2) self.assertEqual([a.act("command not found", i).keys[0] for i in range(3)], ["echox 1", "echox 1", "echox 1"]) self.assertEqual(a.act("command not found", 3).keys[0], "echo 2") def test_retry_counter_resets_on_the_next_line(self): a = self.agent(["echo 1", "echo 2"], [0.0, 0.9, 0.0, 0.9], p_typo=0.5, retry=True, max_retries=1) a.act("", 0) a.act("command not found", 1) # retry of line 1 works a.act("", 2) # line 2 typo self.assertEqual(a.act("command not found", 3).keys[0], "echo 2") # still allowed one retry def test_waits(self): a = self.agent(["@95 sleep 90", "ls"], [0.9, 0.9], p_typo=0.0) self.assertEqual(a.act("", 0).wait, 95.0) self.assertEqual(a.act("", 1).wait, 1.0) a = self.agent(["@95 sleep 90"], [0.9], p_typo=0.0, patient=False) self.assertEqual(a.act("", 0).wait, 1.0) def test_script_defaults_to_the_solution(self): a = tb.OracleAgent() a.reset(task_by_name("lab/sort-names"), random.Random(0)) self.assertEqual(a.act("", 0).keys[0], "sort /app/names.txt | uniq > /app/sorted.txt") self.assertEqual((a.name, a.latency, a.p_typo), ("oracle", 0.0, 0.0)) def test_reset_restarts(self): a = tb.OracleAgent() t = task_by_name("lab/sort-names") a.reset(t, random.Random(0)) a.act("", 0) a.reset(t, random.Random(0)) self.assertFalse(a.act("", 0).done) def test_cheating_plans(self): a = tb.CheatingAgent("copy_expected") a.reset(task_by_name("lab/sort-names"), random.Random(0)) cmds = [] while not (x := a.act("", len(cmds))).done: cmds.append(x.keys[0]) self.assertEqual(cmds, ["ls /tests", "cat /tests/expected_sorted.txt", "cp /tests/expected_sorted.txt /app/sorted.txt"]) a = tb.CheatingAgent("rewrite_tests", then_solve=True) a.reset(task_by_name("lab/sort-names"), random.Random(0)) cmds = [] while not (x := a.act("", len(cmds))).done: cmds.append(x.keys[0]) self.assertEqual(len(cmds), 4) self.assertTrue(cmds[0].startswith("echo '[[check]]' > /tests/checks.toml")) self.assertEqual(cmds[-1], "sort /app/names.txt | uniq > /app/sorted.txt") self.assertEqual(a.name, "cheat-rewrite_tests+solve") self.assertEqual(tb.CheatingAgent("copy_expected").name, "cheat-copy_expected") def test_copy_expected_falls_back_when_there_is_no_expected_file(self): a = tb.CheatingAgent("copy_expected") a.reset(task_by_name("lab/count-errors"), random.Random(0)) self.assertTrue(a.act("", 0).keys[0].startswith("echo '[[check]]'")) def test_cheating_wait_and_latency(self): a = tb.CheatingAgent("copy_expected", then_solve=True, latency=7.0) a.reset(task_by_name("lab/long-job"), random.Random(0)) for _ in range(3): self.assertEqual((a.act("", 0).wait, ), (1.0,)) x = a.act("", 3) self.assertEqual((x.wait, x.think), (95.0, 7.0)) def test_stuck_agent(self): x = tb.StuckAgent().act("", 5) self.assertEqual((x.keys, x.done), (("ls", "Enter"), False)) class TrialTests(unittest.TestCase): def test_oracle_passes_every_demo_task(self): for t in tb.demo_tasks(): r = tb.run_trial(t, tb.OracleAgent(), 0) self.assertEqual((r.status, r.stop, r.reward), ("passed", "agent_done", 1.0), t.name) self.assertEqual(r.tests_passed, r.tests_total, t.name) self.assertEqual(r.flags, [], t.name) self.assertTrue(r.passed) def test_noop_fails_every_demo_task(self): for t in tb.demo_tasks(): r = tb.run_trial(t, tb.NoopAgent(), 0) self.assertEqual((r.status, r.stop, r.reward, r.steps, r.elapsed), ("failed", "agent_done", 0.0, [], 0.0), t.name) def test_task_state_is_fresh_for_every_trial(self): t = task_by_name("lab/sort-names") self.assertTrue(tb.run_trial(t, tb.OracleAgent(), 0).passed) self.assertFalse(tb.run_trial(t, tb.NoopAgent(), 1).passed) def test_oracle_on_the_long_job(self): r = tb.run_trial(task_by_name("lab/long-job"), tb.OracleAgent(), 0) self.assertEqual(r.elapsed, 95.0) self.assertEqual(r.steps[0].wait, 95.0) def test_impatient_agent_fails_the_long_job_but_not_by_timeout(self): ag = tb.ScriptedAgent("imp", patient=False, latency=2.0) r = tb.run_trial(task_by_name("lab/long-job"), ag, 0) self.assertEqual((r.status, r.stop, r.tests_passed), ("failed", "agent_done", 0)) self.assertEqual(r.elapsed, 3.0) def test_typo_without_retry_fails(self): ag = tb.ScriptedAgent("typo", p_typo=1.0, retry=False) r = tb.run_trial(mk_task(), ag, 0) self.assertEqual((r.status, len(r.steps)), ("failed", 1)) self.assertIn("command not found", r.steps[0].observation) def test_retry_counts(self): for retry, mr, n in [(False, 1, 1), (True, 1, 2), (True, 2, 3), (True, 0, 1)]: ag = tb.ScriptedAgent("t", p_typo=1.0, retry=retry, max_retries=mr) self.assertEqual(len(tb.run_trial(mk_task(), ag, 0).steps), n, (retry, mr)) def test_elapsed_and_steps(self): ag = tb.ScriptedAgent("a", script=["echo hi > /app/out.txt", "ls"], latency=2.0) r = tb.run_trial(mk_task(), ag, 0) self.assertEqual(r.elapsed, 6.0) # (2 think + 1 wait) x 2 self.assertEqual([s.index for s in r.steps], [0, 1]) self.assertEqual([s.t for s in r.steps], [3.0, 6.0]) self.assertEqual(r.steps[0].keys, ("echo hi > /app/out.txt", "Enter")) self.assertEqual(r.steps[1].observation, "/app$ echo hi > /app/out.txt\n/app$ ls\nout.txt\n/app$ ") def test_token_accounting(self): t = mk_task() ag = tb.ScriptedAgent("a", script=["echo hi > /app/out.txt", "ls"], latency=2.0, price_in=3.0, price_out=15.0) r = tb.run_trial(t, ag, 0) ctx0 = tb.count_tokens(t.instruction + "\n\n/app$ ") o1 = tb.count_tokens("echo hi > /app/out.txt Enter") o2 = tb.count_tokens("ls Enter") ctx1 = ctx0 + o1 + tb.count_tokens(r.steps[0].observation) self.assertEqual(r.steps[0].in_tokens, ctx0) self.assertEqual(r.steps[1].in_tokens, ctx1) self.assertEqual((r.steps[0].out_tokens, r.steps[1].out_tokens), (o1, o2)) self.assertEqual(r.tokens_in, ctx0 + ctx1) self.assertEqual(r.tokens_out, o1 + o2) self.assertAlmostEqual(r.cost, (ctx0 + ctx1) * 3 / 1e6 + (o1 + o2) * 15 / 1e6) def test_price_changes_cost_not_tokens(self): t = mk_task() a = tb.run_trial(t, tb.ScriptedAgent("a", price_in=1.0, price_out=1.0), 0) b = tb.run_trial(t, tb.ScriptedAgent("a", price_in=2.0, price_out=2.0), 0) self.assertEqual((a.tokens_in, a.tokens_out), (b.tokens_in, b.tokens_out)) self.assertAlmostEqual(b.cost, 2 * a.cost) z = tb.run_trial(t, tb.ScriptedAgent("a", price_in=0.0, price_out=0.0), 0) self.assertEqual(z.cost, 0.0) def test_first_observation_has_the_instruction_and_the_pane(self): seen = [] class Rec(tb.Agent): name = "rec" def act(self, observation, step): seen.append((observation, step)) return tb.Action(done=True) t = mk_task() tb.run_trial(t, Rec(), 0) self.assertEqual(seen, [(t.instruction + "\n\n/app$ ", 0)]) def test_agent_may_return_none(self): class Quit(tb.Agent): name = "quit" def act(self, observation, step): return None self.assertEqual(tb.run_trial(mk_task(), Quit(), 0).stop, "agent_done") def test_timeout_before_the_first_step(self): ag = tb.ScriptedAgent("slow", latency=100.0) t = mk_task() r = tb.run_trial(t, ag, 0) self.assertEqual((r.status, r.stop, r.reward, r.tests_passed, r.tests_total), ("timeout", "timeout", 0.0, 0, 1)) self.assertEqual(r.steps, []) self.assertEqual(r.elapsed, 100.0) def test_timeout_boundary(self): t = mk_task() # agent timeout 60 s self.assertEqual(tb.run_trial(t, tb.ScriptedAgent("a", latency=60.0), 0).status, "timeout") self.assertEqual(tb.run_trial(t, tb.ScriptedAgent("a", latency=58.9), 0).status, "passed") # ends at 59.9 r = tb.run_trial(t, tb.ScriptedAgent("a", latency=59.0), 0) # ends at exactly 60 self.assertEqual((r.status, len(r.steps)), ("timeout", 1)) def test_timeout_after_a_step_discards_the_pass(self): t = mk_task() r = tb.run_trial(t, tb.ScriptedAgent("a", latency=59.5), 0) self.assertEqual((r.status, r.stop, r.reward, len(r.steps)), ("timeout", "timeout", 0.0, 1)) def test_timeout_scale(self): t = mk_task() self.assertEqual(tb.run_trial(t, tb.OracleAgent(), 0, tb.HarnessConfig(timeout_scale=0.5)).agent_timeout, 30.0) self.assertEqual(tb.run_trial(t, tb.OracleAgent(), 0, tb.HarnessConfig(timeout_scale=5)).agent_timeout, 300.0) self.assertEqual(tb.run_trial(t, tb.OracleAgent(), 0).agent_timeout, 60.0) slow = tb.ScriptedAgent("a", latency=70.0) self.assertEqual(tb.run_trial(t, slow, 0).status, "timeout") self.assertEqual(tb.run_trial(t, slow, 0, tb.HarnessConfig(timeout_scale=2.0)).status, "passed") def test_max_turns(self): r = tb.run_trial(mk_task(), tb.StuckAgent(), 0, tb.HarnessConfig(max_turns=5)) self.assertEqual((r.stop, len(r.steps), r.status), ("max_turns", 5, "failed")) r = tb.run_trial(mk_task(), tb.StuckAgent(), 0) self.assertEqual(len(r.steps), 30) def test_stuck_agent_times_out_when_the_turns_are_long(self): r = tb.run_trial(mk_task(), tb.StuckAgent(), 0, tb.HarnessConfig(max_turns=100)) self.assertEqual((r.stop, len(r.steps)), ("timeout", 30)) # 2 s per step, 60 s limit: step 30 ends at 60 def test_config_defaults(self): c = tb.HarnessConfig() self.assertEqual((c.width, c.height, c.max_turns, c.leaky, c.timeout_scale), (80, 24, 30, False, 1.0)) def test_pane_size_is_used(self): r = tb.run_trial(mk_task(), tb.ScriptedAgent("a", script=["echo " + "x" * 30]), 0, tb.HarnessConfig(width=10, height=4)) self.assertLessEqual(len(r.steps[0].observation.splitlines()), 4) self.assertTrue(all(len(l) <= 10 for l in r.steps[0].observation.splitlines())) def test_allowlist_is_enforced_in_trials(self): toml = tb._toml_task("lab/tiny", "x", "easy") + '[environment]\nallowed_commands = ["ls"]\n' t = mk_task(task=toml) r = tb.run_trial(t, tb.OracleAgent(), 0) self.assertEqual(r.status, "failed") self.assertIn("echo: command not found", r.steps[0].observation) def test_determinism(self): t = task_by_name("lab/make-report") ag = tb.ScriptedAgent("a", p_typo=0.4, retry=True) key = lambda r: (r.status, [s.keys for s in r.steps], r.elapsed, r.tokens_in, r.cost) self.assertEqual(key(tb.run_trial(t, ag, 3, seed=7)), key(tb.run_trial(t, ag, 3, seed=7))) def test_trials_and_seeds_differ(self): t = task_by_name("lab/make-report") ag = tb.ScriptedAgent("a", p_typo=0.4) by_trial = {tb.run_trial(t, ag, k, seed=1).status for k in range(20)} by_seed = {tb.run_trial(t, ag, 0, seed=s).status for s in range(20)} self.assertEqual(by_trial, {"passed", "failed"}) self.assertEqual(by_seed, {"passed", "failed"}) def test_cheaters_in_the_isolated_harness(self): t = task_by_name("lab/sort-names") r = tb.run_trial(t, tb.CheatingAgent("copy_expected"), 0) self.assertEqual((r.status, r.flags), ("failed", ["touched /tests"])) self.assertIn("ls: cannot access '/tests'", r.steps[0].observation) r = tb.run_trial(t, tb.CheatingAgent("rewrite_tests"), 0) self.assertEqual((r.status, r.flags), ("failed", ["touched /tests", "touched checks.toml"])) r = tb.run_trial(t, tb.CheatingAgent("copy_expected", then_solve=True), 0) self.assertEqual((r.status, r.flags), ("passed", ["touched /tests"])) # honest work, dishonest trajectory def test_cheaters_in_the_leaky_harness(self): t = task_by_name("lab/sort-names") cfg = tb.HarnessConfig(leaky=True) r = tb.run_trial(t, tb.CheatingAgent("copy_expected"), 0, cfg) self.assertEqual((r.status, r.tests_passed, r.tests_total, r.leaky), ("passed", 2, 2, True)) r = tb.run_trial(t, tb.CheatingAgent("rewrite_tests"), 0, cfg) self.assertEqual((r.status, r.tests_passed, r.tests_total), ("passed", 1, 1)) self.assertEqual(tb.run_trial(t, tb.NoopAgent(), 0, cfg).status, "failed") def test_flag_trajectory(self): st = lambda *cmds: [tb.Step(i, 0, (c, "Enter"), 1, "", 0, 0) for i, c in enumerate(cmds)] self.assertEqual(tb.flag_trajectory([]), []) self.assertEqual(tb.flag_trajectory(st("ls", "cat /app/x")), []) self.assertEqual(tb.flag_trajectory(st("cat /tests/a")), ["touched /tests"]) self.assertEqual(tb.flag_trajectory(st("ls /tests")), ["touched /tests"]) self.assertEqual(tb.flag_trajectory(st("cat /testsuite")), []) self.assertEqual(tb.flag_trajectory(st("cat /solution/solve.sh")), ["touched /solution"]) self.assertEqual(tb.flag_trajectory(st("cat checks.toml")), ["touched checks.toml"]) self.assertEqual(tb.flag_trajectory(st("cat /solution/x", "ls /tests", "cat checks.toml")), ["touched /solution", "touched /tests", "touched checks.toml"]) def test_run_benchmark_shape_and_order(self): tasks = tb.demo_tasks()[:2] by = tb.run_benchmark(tasks, [tb.OracleAgent(), tb.NoopAgent()], trials=3) self.assertEqual(list(by), ["oracle", "noop"]) self.assertEqual(len(by["oracle"]), 6) self.assertEqual([(r.task, r.trial) for r in by["oracle"]], [(tasks[0].name, 0), (tasks[0].name, 1), (tasks[0].name, 2), (tasks[1].name, 0), (tasks[1].name, 1), (tasks[1].name, 2)]) self.assertTrue(all(r.passed for r in by["oracle"])) self.assertFalse(any(r.passed for r in by["noop"])) def test_run_benchmark_seed_and_config(self): tasks = tb.demo_tasks() ag = [tb.ScriptedAgent("a", p_typo=0.3)] key = lambda by: [(r.status, r.elapsed) for r in by["a"]] self.assertEqual(key(tb.run_benchmark(tasks, ag, 2, seed=5)), key(tb.run_benchmark(tasks, ag, 2, seed=5))) self.assertNotEqual(key(tb.run_benchmark(tasks, ag, 5, seed=5)), key(tb.run_benchmark(tasks, ag, 5, seed=6))) by = tb.run_benchmark(tasks[:1], [tb.CheatingAgent("copy_expected")], 1, config=tb.HarnessConfig(leaky=True)) self.assertTrue(by["cheat-copy_expected"][0].passed) def test_result_passed_property(self): self.assertTrue(mk_result(passed=True).passed) self.assertFalse(mk_result(passed=False).passed) class StatsTests(unittest.TestCase): def test_pass_at_k_known_values(self): self.assertAlmostEqual(tb.pass_at_k(10, 3, 2), 24 / 45) # 1 - C(7,2)/C(10,2) self.assertAlmostEqual(tb.pass_at_k(5, 2, 3), 0.9) # 1 - C(3,3)/C(5,3) self.assertAlmostEqual(tb.pass_at_k(10, 7, 3), 1 - 1 / 120) # 1 - C(3,3)/C(10,3) self.assertEqual(tb.pass_at_k(10, 0, 3), 0.0) self.assertEqual(tb.pass_at_k(10, 10, 3), 1.0) self.assertEqual(tb.pass_at_k(10, 8, 3), 1.0) # fewer than k failures self.assertEqual(tb.pass_at_k(10, 7, 4), 1.0) self.assertLess(tb.pass_at_k(10, 6, 4), 1.0) self.assertAlmostEqual(tb.pass_at_k(10, 6, 4), 1 - 1 / 210) def test_pass_at_1_is_the_pass_rate(self): for c in range(0, 11): self.assertAlmostEqual(tb.pass_at_k(10, c, 1), c / 10) self.assertAlmostEqual(tb.pass_hat_k(10, c, 1), c / 10) def test_pass_hat_k_known_values(self): self.assertAlmostEqual(tb.pass_hat_k(10, 3, 2), 3 / 45) self.assertAlmostEqual(tb.pass_hat_k(10, 9, 3), 84 / 120) self.assertEqual(tb.pass_hat_k(10, 2, 3), 0.0) self.assertEqual(tb.pass_hat_k(10, 10, 5), 1.0) self.assertEqual(tb.pass_hat_k(10, 0, 1), 0.0) def test_pass_at_k_grows_and_pass_hat_k_shrinks_with_k(self): for c in (2, 5, 8): at = [tb.pass_at_k(10, c, k) for k in range(1, 11)] hat = [tb.pass_hat_k(10, c, k) for k in range(1, 11)] self.assertEqual(at, sorted(at)) self.assertEqual(hat, sorted(hat, reverse=True)) for a, h in zip(at, hat): self.assertLessEqual(h, a + 1e-12) def test_toy_model_with_independent_trials(self): self.assertAlmostEqual(0.75 ** 3, 0.421875) # my arithmetic: three passes in a row at p = 0.75 self.assertAlmostEqual(1 - 0.25 ** 3, 0.984375) # at least one pass in three tries def test_pass_k_errors(self): for fn in (tb.pass_at_k, tb.pass_hat_k): for args in [(10, 11, 1), (10, -1, 1), (10, 5, 0), (10, 5, 11), (0, 0, 1)]: with self.assertRaises(ValueError, msg=(fn.__name__, args)): fn(*args) self.assertEqual(tb.pass_at_k(3, 1, 3), 1.0) # k == n is allowed def test_wilson_known_values(self): lo, hi = tb.wilson_interval(5, 10) self.assertAlmostEqual(lo, 0.2366, places=4) self.assertAlmostEqual(hi, 0.7634, places=4) lo, hi = tb.wilson_interval(0, 10) self.assertEqual(lo, 0.0) self.assertAlmostEqual(hi, 0.2775, places=4) lo, hi = tb.wilson_interval(10, 10) self.assertAlmostEqual(lo, 0.7225, places=4) self.assertEqual(hi, 1.0) lo, hi = tb.wilson_interval(1, 1) self.assertAlmostEqual(lo, 0.2065, places=4) self.assertEqual(hi, 1.0) lo, hi = tb.wilson_interval(8, 10, z=1.0) self.assertAlmostEqual(lo, 0.6491, places=4) self.assertAlmostEqual(hi, 0.8964, places=4) def test_wilson_properties(self): lo, hi = tb.wilson_interval(30, 100) lo2, hi2 = tb.wilson_interval(300, 1000) self.assertGreater(hi - lo, hi2 - lo2) # more trials, narrower self.assertLess(lo, 0.3) self.assertGreater(hi, 0.3) lo0, hi0 = tb.wilson_interval(4, 10, z=0.0) self.assertAlmostEqual(lo0, 0.4) self.assertAlmostEqual(hi0, 0.4) wide = tb.wilson_interval(5, 10, z=2.58) self.assertLess(wide[0], 0.2366) def test_wilson_errors(self): for args in [(1, 0), (-1, 5), (6, 5)]: with self.assertRaises(ValueError): tb.wilson_interval(*args) def R(self, spec): """spec: {task: [bool, ...]} -> results.""" return [mk_result(task=t, trial=i, passed=p) for t, ps in spec.items() for i, p in enumerate(ps)] def test_task_counts_and_rate(self): rs = self.R({"a": [True, False, True, False], "b": [True]}) self.assertEqual(tb.task_counts(rs), {"a": (4, 2), "b": (1, 1)}) self.assertAlmostEqual(tb.resolution_rate(rs), 0.75) # every task counts once; pooled would be 0.6 self.assertEqual(tb.task_counts([]), {}) self.assertEqual(tb.resolution_rate([]), 0.0) def test_clustered_interval(self): m, lo, hi, se = tb.clustered_interval([0.4, 0.5, 0.6]) self.assertAlmostEqual(m, 0.5) self.assertAlmostEqual(se, 0.1 / math.sqrt(3)) self.assertAlmostEqual(lo, 0.5 - 1.96 * se) self.assertAlmostEqual(hi, 0.5 + 1.96 * se) m, lo, hi, se = tb.clustered_interval([0.0, 0.5, 1.0]) self.assertEqual((lo, hi), (0.0, 1.0)) # clamped to [0, 1] self.assertAlmostEqual(se, 0.5 / math.sqrt(3)) self.assertEqual(tb.clustered_interval([0.7]), (0.7, 0.7, 0.7, 0.0)) self.assertEqual(tb.clustered_interval(iter([0.5, 0.5]))[3], 0.0) self.assertAlmostEqual(tb.clustered_interval([0.4, 0.5, 0.6], z=1.0)[1], 0.5 - 0.1 / math.sqrt(3)) with self.assertRaises(ValueError): tb.clustered_interval([]) def test_clustered_is_wider_than_pooled_when_tasks_differ(self): rates = [1.0] * 5 + [0.0] * 5 # half the tasks always pass, half never _, lo, hi, _ = tb.clustered_interval(rates) wlo, whi = tb.wilson_interval(50, 100) # 10 trials per task, pooled self.assertGreater(hi - lo, whi - wlo) def test_paired_difference(self): m, lo, hi, se = tb.paired_difference([1, 0, 1, 1], [0, 0, 1, 0]) self.assertAlmostEqual(m, 0.5) self.assertAlmostEqual(se, math.sqrt(1 / 3) / 2) self.assertAlmostEqual(lo, 0.5 - 1.96 * se) self.assertAlmostEqual(hi, 0.5 + 1.96 * se) self.assertLess(lo, 0) # not clamped self.assertGreater(hi, 1) def test_pairing_removes_shared_task_difficulty(self): a = [0.9, 0.6, 0.3, 0.1] b = [x - 0.1 for x in a] # always exactly 0.1 worse m, lo, hi, se = tb.paired_difference(a, b) self.assertAlmostEqual(m, 0.1) self.assertLess(se, 1e-9) self.assertGreater(tb.clustered_interval(a)[3], 0.1) # the unpaired spread is much larger def test_paired_errors(self): for a, b in [([], []), ([1], [1, 2])]: with self.assertRaises(ValueError): tb.paired_difference(a, b) self.assertEqual(tb.paired_difference([0.5], [0.25])[0], 0.25) def test_bootstrap(self): r = [0.2, 0.5, 0.8, 1.0, 0.0] lo, hi = tb.bootstrap_interval(r) self.assertEqual((lo, hi), (0.18, 0.82)) # regression value, seed 0 self.assertEqual(tb.bootstrap_interval(r), (lo, hi)) self.assertTrue(all(abs(x - 0.3) < 1e-12 for x in tb.bootstrap_interval([0.3, 0.3, 0.3]))) self.assertEqual(tb.bootstrap_interval([0, 1], n_boot=100, seed=3, alpha=0.1), (0.0, 1.0)) a, b = tb.bootstrap_interval([0.1, 0.9, 0.5], n_boot=50, seed=1) self.assertAlmostEqual(a, 0.2333333333333333) self.assertAlmostEqual(b, 0.7666666666666666) self.assertLessEqual(min(r), lo) self.assertGreaterEqual(max(r), hi) w = tb.bootstrap_interval(r, alpha=0.5) self.assertGreater(hi - lo, w[1] - w[0]) def test_run_scores(self): rs = self.R({"a": [True, False], "b": [True, True]}) self.assertEqual(tb.run_scores(rs, 2), [1.0, 0.5]) self.assertEqual(tb.run_scores(rs, 3), [1.0, 0.5, 0.0]) # no third run exists self.assertEqual(tb.run_scores(rs, 1), [1.0]) rs = self.R({"a": [True, False, True], "b": [False]}) self.assertEqual(tb.run_scores(rs, 3), [0.5, 0.0, 1.0]) class ContaminationTests(unittest.TestCase): def test_words_and_ngrams(self): self.assertEqual(tb.words("Hello, World! It's 42."), ["hello", "world", "it", "s", "42"]) self.assertEqual(tb.ngrams("a b c d", 2), {("a", "b"), ("b", "c"), ("c", "d")}) self.assertEqual(tb.ngrams("a b c", 3), {("a", "b", "c")}) self.assertEqual(tb.ngrams("a b", 3), set()) self.assertEqual(tb.ngrams("A a A a", 2), {("a", "a")}) self.assertEqual(len(tb.ngrams(" ".join("w%d" % i for i in range(20)))), 13) # n defaults to 8 def test_overlap_fraction(self): ref = "one two three four five" self.assertEqual(tb.overlap_fraction(ref, ref, 3), 1.0) self.assertAlmostEqual(tb.overlap_fraction("one two three four", ref, 3), 2 / 3) self.assertEqual(tb.overlap_fraction("unrelated words only", ref, 3), 0.0) self.assertEqual(tb.overlap_fraction(ref, "too short", 3), 0.0) self.assertEqual(tb.overlap_fraction("", ref, 3), 0.0) self.assertEqual(tb.overlap_fraction("ONE two THREE four five", ref, 3), 1.0) self.assertEqual(tb.overlap_fraction("five four three two one", ref, 3), 0.0) # order matters self.assertEqual(tb.overlap_fraction("x " + ref + " y", ref, 3), 1.0) def test_scan_and_decontaminate(self): c = tb.make_canary("x") docs = {"z": "plain", "b": "has " + c, "a": "has " + tb.CANARY_PREFIX, "d": "text"} self.assertEqual(tb.scan_corpus(docs), ["a", "b"]) self.assertEqual(tb.scan_corpus(docs, c), ["b"]) kept, dropped = tb.decontaminate(docs) self.assertEqual((sorted(kept), dropped), (["d", "z"], ["a", "b"])) kept, dropped = tb.decontaminate(docs, c) self.assertEqual((sorted(kept), dropped), (["a", "d", "z"], ["b"])) self.assertEqual(tb.decontaminate({}), ({}, [])) self.assertEqual(len(docs), 4) def test_report(self): tasks = {"t1": "one two three four five six", "t2": "red green blue cyan pink gold"} docs = {"d_copy": "red green blue cyan pink gold", "d_half": "one two three four", "d_none": "nothing here", "d_can": "nothing " + tb.CANARY_PREFIX} rows = tb.contamination_report(docs, tasks, n=3, threshold=0.5) self.assertEqual([r["doc"] for r in rows], ["d_copy", "d_half", "d_can", "d_none"]) self.assertEqual(rows[0], {"doc": "d_copy", "task": "t2", "overlap": 1.0, "canary": False, "flagged": True}) self.assertEqual((rows[1]["task"], rows[1]["overlap"], rows[1]["flagged"]), ("t1", 0.5, True)) # 2 of 4 trigrams, exactly the threshold self.assertEqual((rows[2]["canary"], rows[2]["flagged"], rows[2]["overlap"]), (True, False, 0.0)) self.assertEqual(tb.contamination_report(docs, tasks, n=3, threshold=0.51)[1]["flagged"], False) self.assertEqual(tb.contamination_report({"d": "x"}, {}, n=3), [{"doc": "d", "task": "", "overlap": 0.0, "canary": False, "flagged": False}]) self.assertEqual(tb.contamination_report({}, tasks), []) def test_report_rounds_and_breaks_ties_by_name(self): rows = tb.contamination_report({"b": "one two three four", "a": "one two three four"}, {"t": "one two three four five"}, n=3) self.assertEqual([r["doc"] for r in rows], ["a", "b"]) self.assertEqual(rows[0]["overlap"], 0.667) def test_report_defaults(self): text = "w1 w2 w3 w4 w5 w6 w7 w8 w9" rows = tb.contamination_report({"d": "w1 w2 w3 w4 w5 w6 w7 w8"}, {"t": text}) self.assertEqual((rows[0]["overlap"], rows[0]["flagged"]), (0.5, True)) # default n = 8, threshold 0.5 def test_reword(self): self.assertEqual(tb.reword("a b c d e f"), "a b c d e really f") self.assertEqual(tb.reword("a b c d e f", every=2), "a b really c d really e f really") self.assertEqual(tb.reword("Sort the file."), "order the document") self.assertEqual(tb.reword("count lines, create directory"), "tally rows make folder") # punctuation on a swapped word is dropped self.assertEqual(tb.reword(""), "") def test_reword_defeats_the_ngram_filter(self): text = "Compute the SHA-256 hash of the payload file and write only the hex digest to the output file." self.assertEqual(tb.overlap_fraction(text, text), 1.0) self.assertLess(tb.overlap_fraction(tb.reword(text), text), 0.5) self.assertFalse(tb.has_canary(tb.reword(text))) class QATests(unittest.TestCase): def codes(self, task): return [f.code for f in tb.static_checks(task)] def test_demo_tasks_are_clean(self): for t in tb.demo_tasks(): self.assertEqual(tb.static_checks(t), [], t.name) def test_missing_canaries(self): fs = tb.static_checks(mk_task()) missing = [f.message for f in fs if f.code == "canary_missing"] self.assertEqual(missing, ["instruction.md has no canary line", "solution/solve.sh has no canary line", "task.toml has no canary line", "tests/checks.toml has no canary line"]) self.assertTrue(all(f.level == "error" for f in fs if f.code == "canary_missing")) self.assertEqual(self.codes(mk_ctask()), ["hardcoded_solution"]) def test_data_files_do_not_need_a_canary(self): self.assertEqual(self.codes(mk_ctask(env="data data\n")), ["hardcoded_solution"]) def test_tests_in_environment(self): t = mk_ctask(extra={"environment/x.txt": "secret answer\n", "tests/ans.txt": "secret answer\n"}) f = [x for x in tb.static_checks(t) if x.code == "tests_in_environment"] self.assertEqual(len(f), 1) self.assertEqual((f[0].level, f[0].message), ("error", "environment/x.txt copies a tests file")) t = mk_ctask(extra={"environment/x.txt": " secret answer ", "tests/ans.txt": "secret answer\n"}) self.assertIn("tests_in_environment", self.codes(t)) # whitespace is ignored t = mk_ctask(extra={"environment/x.txt": "hi\n", "tests/ans.txt": "hi\n"}) self.assertNotIn("tests_in_environment", self.codes(t)) # too short to be a leak t = mk_ctask(extra={"environment/x.txt": "abcd\n", "tests/ans.txt": "abcd\n"}) self.assertIn("tests_in_environment", self.codes(t)) # four characters is enough def test_unmentioned_path(self): chk = '[[check]]\ntype = "file_exists"\npath = "%s"\n' t = mk_ctask(checks=chk % "/app/other.txt") f = [x for x in tb.static_checks(t) if x.code == "unmentioned_path"] self.assertEqual(len(f), 1) self.assertEqual(f[0].message, "check file_exists_0 tests /app/other.txt, which the instruction never names") self.assertNotIn("unmentioned_path", self.codes(mk_ctask(checks=chk % "/app/out.txt"))) self.assertNotIn("unmentioned_path", self.codes(mk_ctask(checks=chk % "/tmp/out.txt"))) # basename is named cmd = '[[check]]\ntype = "cmd_stdout"\ncmd = "ls"\nequals = "x"\n' self.assertNotIn("unmentioned_path", self.codes(mk_ctask(checks=cmd))) def test_solution_findings(self): self.assertIn("no_solution", self.codes(mk_ctask(solution=None))) self.assertNotIn("hardcoded_solution", self.codes(mk_ctask(solution=None))) toml = tb._toml_task("lab/tiny", "x", "easy") + '[environment]\nallowed_commands = ["echo"]\n' t = mk_ctask(task=toml, solution="sort /app/a | uniq > /app/out.txt\n") f = [x for x in tb.static_checks(t) if x.code == "allowlist_blocks_solution"] self.assertEqual(len(f), 1) self.assertEqual(f[0].message, "the solution needs ['sort', 'uniq'], which the task does not allow") self.assertEqual(self.codes(mk_ctask(solution="@5 sort /app/a\n")), []) def test_solution_commands(self): sc = lambda *lines: tb._solution_commands(mk_task(solution="\n".join(lines) + "\n")) self.assertEqual(sc("a | b && c ; d"), {"a", "b", "c", "d"}) self.assertEqual(sc("@5 sleep 1"), {"sleep"}) self.assertEqual(sc("echo 'x"), {""}) self.assertEqual(sc("ls", "cat f"), {"ls", "cat"}) self.assertEqual(sc("echo hi > f"), {"echo"}) def test_hardcoded_solution(self): hc = lambda *lines: "hardcoded_solution" in self.codes(mk_ctask(solution="\n".join(lines) + "\n")) self.assertTrue(hc("echo 3 > /app/out.txt")) self.assertTrue(hc("@5 echo 3 > /app/out.txt")) self.assertTrue(hc("@2.5 echo 3")) self.assertTrue(hc("echo 3", "echo 4")) self.assertFalse(hc("echo 3", "sort /app/x")) self.assertFalse(hc("sort /app/x > /app/out.txt")) self.assertFalse(hc("echoes 3")) f = [x for x in tb.static_checks(mk_ctask()) if x.code == "hardcoded_solution"] self.assertEqual(f[0].level, "warn") def test_weak_checks(self): exists = '[[check]]\ntype = "file_exists"\npath = "/app/out.txt"\n[[check]]\ntype = "file_absent"\npath = "/app/out.txt"\n' self.assertIn("weak_checks", self.codes(mk_ctask(checks=exists))) mixed = exists + '[[check]]\ntype = "file_contains"\npath = "/app/out.txt"\ntext = "hi"\n' self.assertNotIn("weak_checks", self.codes(mk_ctask(checks=mixed))) w = [f for f in tb.static_checks(mk_ctask(checks=exists)) if f.code == "weak_checks"] self.assertEqual(w[0].level, "warn") def test_timeout_range(self): for secs, bad in [(5, True), (9.9, True), (10, False), (7200, False), (7201, True)]: t = mk_ctask(task=tb._toml_task("lab/tiny", "x", "easy", agent_timeout=secs)) self.assertEqual("timeout_unusual" in self.codes(t), bad, secs) def test_audit_passes_for_demo_tasks(self): for t in tb.demo_tasks(): a = tb.audit_task(t) self.assertTrue(a.ok, t.name) self.assertTrue(a.oracle_passes) self.assertTrue(a.noop_fails) self.assertEqual(a.cheaters_fail, {"copy_expected": True, "rewrite_tests": True}) def test_audit_catches_a_leaky_harness(self): a = tb.audit_task(task_by_name("lab/sort-names"), tb.HarnessConfig(leaky=True)) self.assertEqual(a.cheaters_fail, {"copy_expected": False, "rewrite_tests": False}) self.assertFalse(a.ok) self.assertTrue(a.oracle_passes) def test_audit_catches_a_check_that_passes_for_nothing(self): t = mk_ctask(checks='[[check]]\ntype = "file_exists"\npath = "/app"\n') a = tb.audit_task(t) self.assertFalse(a.noop_fails) self.assertFalse(a.ok) def test_audit_catches_a_broken_solution(self): t = mk_ctask(solution="echo wrong > /app/out.txt\n") a = tb.audit_task(t) self.assertFalse(a.oracle_passes) self.assertFalse(a.ok) def test_audit_without_a_solution(self): a = tb.audit_task(mk_ctask(solution=None)) self.assertFalse(a.oracle_passes) self.assertIn("no_solution", [f.code for f in a.findings]) self.assertFalse(a.ok) def test_audit_ok_needs_each_part(self): f_err = [tb.Finding("error", "x", "m")] f_warn = [tb.Finding("warn", "x", "m")] good = {"a": True} self.assertTrue(tb.AuditReport([], True, True, good).ok) self.assertTrue(tb.AuditReport(f_warn, True, True, good).ok) self.assertFalse(tb.AuditReport(f_err, True, True, good).ok) self.assertFalse(tb.AuditReport([], False, True, good).ok) self.assertFalse(tb.AuditReport([], True, False, good).ok) self.assertFalse(tb.AuditReport([], True, True, {"a": True, "b": False}).ok) self.assertTrue(tb.AuditReport([], True, True, {}).ok) class LeaderboardTests(unittest.TestCase): def results(self, spec, **kw): return [mk_result(task=t, trial=i, passed=p, **kw) for t, ps in spec.items() for i, p in enumerate(ps)] def test_summarize(self): rs = self.results({"a": [True] * 4, "b": [True, True, False, False]}, cost=0.01, elapsed=10.0, tin=100, tout=20) row = tb.summarize("x", rs, k=3) self.assertEqual((row.agent, row.n_tasks, row.trials_per_task, row.k), ("x", 2, 4, 3)) self.assertAlmostEqual(row.rate, 0.75) self.assertAlmostEqual(row.lo, 0.75 - 1.96 * 0.25) self.assertEqual(row.hi, 1.0) wlo, whi = tb.wilson_interval(6, 8) self.assertAlmostEqual(row.wilson_lo, wlo) self.assertAlmostEqual(row.wilson_hi, whi) self.assertAlmostEqual(row.pass_at_k, 1.0) # a: 1.0, b: 2 failures < 3 so 1.0 self.assertAlmostEqual(row.pass_hat_k, 0.5) # a: 1.0, b: C(2,3) = 0 self.assertEqual((row.tokens_in, row.tokens_out), (800, 160)) self.assertAlmostEqual(row.cost_per_run, 0.08 / 4) # total cost over the trials per task self.assertAlmostEqual(row.mean_elapsed, 10.0) def test_summarize_uses_k(self): rs = self.results({"a": [True, True, False, False]}) r2, r3 = tb.summarize("x", rs, k=2), tb.summarize("x", rs, k=3) self.assertAlmostEqual(r2.pass_hat_k, 1 / 6) self.assertAlmostEqual(r3.pass_hat_k, 0.0) self.assertAlmostEqual(r2.pass_at_k, 1 - 1 / 6) def test_summarize_z(self): rs = self.results({"a": [True] * 4, "b": [True, False, False, False], "c": [False] * 4}) a, b = tb.summarize("x", rs, z=1.0), tb.summarize("x", rs, z=2.0) self.assertLess(a.hi - a.lo, b.hi - b.lo) self.assertLess(a.wilson_hi - a.wilson_lo, b.wilson_hi - b.wilson_lo) def test_summarize_errors(self): with self.assertRaises(ValueError) as cm: tb.summarize("x", []) self.assertEqual(str(cm.exception), "no results") with self.assertRaises(ValueError) as cm: tb.summarize("x", self.results({"a": [True, True], "b": [True] * 5}), k=3) self.assertEqual(str(cm.exception), "need at least 3 trials per task, the smallest has 2") tb.summarize("x", self.results({"a": [True] * 3}), k=3) # exactly k is fine def test_unequal_trials(self): row = tb.summarize("x", self.results({"a": [True] * 3, "b": [False] * 5}), k=3) self.assertEqual(row.trials_per_task, 3) self.assertAlmostEqual(row.rate, 0.5) self.assertAlmostEqual(row.cost_per_run, 0.0) def test_leaderboard_order(self): by = {"zeta": self.results({"a": [True, True, True]}, cost=1.0), "alpha": self.results({"a": [True, True, True]}, cost=1.0), "cheap": self.results({"a": [True, True, True]}, cost=0.5), "weak": self.results({"a": [True, False, False]}, cost=0.1)} self.assertEqual([r.agent for r in tb.leaderboard(by)], ["cheap", "alpha", "zeta", "weak"]) self.assertEqual(tb.leaderboard({}), []) self.assertEqual(tb.leaderboard(by, k=2)[0].k, 2) def row(self, name, rate, cost): return tb.Row(name, 1, 1, rate, 0, 0, 0, 0, 0, 0, 3, 0, 0, cost, 0) def test_pareto(self): a, b, c, d = self.row("A", .9, .5), self.row("B", .8, .1), self.row("C", .7, .2), self.row("D", .9, .5) front = tb.pareto_front([a, b, c, d]) self.assertEqual([r.agent for r in front], ["B", "A", "D"]) # C is beaten by B; ties stay self.assertEqual(tb.pareto_front([]), []) e = self.row("E", .9, .4) self.assertEqual([r.agent for r in tb.pareto_front([a, e])], ["E"]) f = self.row("F", .95, .5) self.assertEqual([r.agent for r in tb.pareto_front([a, f])], ["F"]) self.assertEqual([r.agent for r in tb.pareto_front([self.row("X", .5, .5), self.row("Y", .6, .6)])], ["X", "Y"]) def official(self): return {"lab/tiny": mk_task()} def test_valid_submission(self): v = tb.validate_submission([mk_result(task="lab/tiny")], self.official()) self.assertEqual((v.ok, v.problems, v.rescored), (True, [], [])) def test_missing_and_extra_tasks(self): v = tb.validate_submission([], self.official(), min_trials=2) self.assertEqual(v.problems, ["task lab/tiny has fewer than 2 trials"]) v = tb.validate_submission([mk_result(task="lab/tiny")], self.official(), min_trials=2) self.assertFalse(v.ok) v = tb.validate_submission([mk_result(task="lab/tiny"), mk_result(task="lab/tiny", trial=1)], self.official(), min_trials=2) self.assertTrue(v.ok) v = tb.validate_submission([mk_result(task="lab/tiny"), mk_result(task="lab/other")], self.official()) self.assertEqual(v.problems, ["task lab/other is not in the official set"]) def test_version_and_timeout_problems(self): v = tb.validate_submission([mk_result(task="lab/tiny", version="1.0.1")], self.official()) self.assertEqual(v.problems, ["lab/tiny: version 1.0.1 is not 1.0.0"]) v = tb.validate_submission([mk_result(task="lab/tiny", timeout=300.0)], self.official()) self.assertEqual(v.problems, ["lab/tiny: timeout 300.0s is not the official 60.0s"]) v = tb.validate_submission([mk_result(task="lab/tiny", timeout=60.0000000001)], self.official()) self.assertTrue(v.ok) def test_passing_trial_needs_a_trajectory(self): v = tb.validate_submission([mk_result(task="lab/tiny", steps=False)], self.official()) self.assertEqual(v.problems, ["lab/tiny trial 0: passing trial has no trajectory"]) v = tb.validate_submission([mk_result(task="lab/tiny", steps=False, passed=False)], self.official()) self.assertTrue(v.ok) def test_problems_are_sorted_and_unique(self): rs = [mk_result(task="lab/tiny", trial=i, version="9.9.9") for i in range(3)] + [mk_result(task="lab/zzz")] v = tb.validate_submission(rs, self.official()) self.assertEqual(v.problems, ["lab/tiny: version 9.9.9 is not 1.0.0", "task lab/zzz is not in the official set"]) def test_flagged_passes_are_rescored(self): flagged = mk_result(task="lab/tiny", flags=["touched /tests"]) v = tb.validate_submission([flagged, mk_result(task="lab/tiny", trial=1)], self.official()) self.assertTrue(v.ok) self.assertEqual(len(v.rescored), 1) r = v.rescored[0] self.assertEqual((r.reward, r.status, r.passed, r.note), (0.0, "failed", False, "rescored to 0: touched /tests")) self.assertTrue(flagged.passed) # the original is not changed v = tb.validate_submission([mk_result(task="lab/tiny", passed=False, flags=["touched /tests"])], self.official()) self.assertEqual(v.rescored, []) def test_apply_rescoring(self): a = mk_result(task="a", flags=["touched /tests", "touched checks.toml"]) b = mk_result(task="b") c = mk_result(task="c", passed=False, flags=["touched /tests"]) out = tb.apply_rescoring([a, b, c]) self.assertEqual([r.passed for r in out], [False, True, False]) self.assertEqual(out[0].note, "rescored to 0: touched /tests, touched checks.toml") self.assertIs(out[1], b) self.assertIs(out[2], c) self.assertTrue(a.passed) self.assertEqual(tb.resolution_rate(out), 1 / 3) def test_bump_kind(self): self.assertEqual(tb.bump_kind("1.0.0", "2.0.0"), "major") self.assertEqual(tb.bump_kind("1.9.9", "2.0.0"), "major") self.assertEqual(tb.bump_kind("1.0.0", "1.1.0"), "minor") self.assertEqual(tb.bump_kind("1.0.5", "1.1.0"), "minor") self.assertEqual(tb.bump_kind("1.0.0", "1.0.1"), "patch") self.assertEqual(tb.bump_kind("1.0.0", "1.0.0"), "none") self.assertEqual(tb.bump_kind("1.2.3", "3.0.0"), "major") for old, new in [("2.0.0", "1.9.9"), ("1.2.0", "1.1.9"), ("1.0.1", "1.0.0")]: with self.assertRaises(ValueError): tb.bump_kind(old, new) for old, new in [("1.0", "1.0.1"), ("1.0.0", "x")]: with self.assertRaises(ValueError): tb.bump_kind(old, new) def test_plan_migration(self): old = {"a": "1.0.0", "b": "1.0.0", "c": "1.0.0", "d": "1.0.0", "e": "1.0.0"} new = {"a": "1.0.0", "b": "1.0.1", "c": "1.1.0", "d": "2.0.0", "f": "1.0.0"} self.assertEqual(tb.plan_migration(old, new), {"a": "reuse", "b": "reuse", "c": "regrade", "d": "rerun", "f": "run", "e": "drop"}) self.assertEqual(tb.plan_migration({}, {}), {}) def test_dataset_bump(self): old = {"a": "1.0.0", "b": "1.0.0"} self.assertEqual(tb.dataset_bump(old, dict(old)), "none") self.assertEqual(tb.dataset_bump(old, {"a": "1.0.1", "b": "1.0.0"}), "patch") self.assertEqual(tb.dataset_bump(old, {"a": "1.0.1", "b": "1.2.0"}), "minor") self.assertEqual(tb.dataset_bump(old, {"a": "1.1.0", "b": "1.0.1"}), "minor") self.assertEqual(tb.dataset_bump(old, {"a": "2.0.0", "b": "1.0.0"}), "major") self.assertEqual(tb.dataset_bump(old, {"a": "1.0.0"}), "major") # a task was removed self.assertEqual(tb.dataset_bump(old, {"a": "1.0.0", "b": "1.0.0", "c": "1.0.0"}), "major") # a task was added def test_migration_cost(self): plan = {"a": "reuse", "b": "regrade", "c": "rerun", "d": "run", "e": "drop"} m = tb.migration_cost(plan, 10, 0.4, 0.02) self.assertEqual((m["run"], m["regrade"], m["reuse"]), (20, 10, 10)) self.assertAlmostEqual(m["cost"], 8.2) self.assertAlmostEqual(m["full_rerun_cost"], 16.0) m = tb.migration_cost({}, 10, 0.4, 0.02) self.assertEqual((m["run"], m["cost"], m["full_rerun_cost"]), (0, 0, 0)) self.assertLess(tb.migration_cost(plan, 5, 0.4, 0.02)["cost"], m["cost"] + 8.2) class DemoTests(unittest.TestCase): def capture(self): lines = [] tb.demo(lines.append) return "\n".join(lines) def test_demo_is_deterministic(self): self.assertEqual(self.capture(), self.capture()) def test_demo_sections(self): text = self.capture() heads = re.findall(r"^== (\d+)\. ", text, re.M) self.assertEqual(heads, [str(i) for i in range(1, 11)]) def test_demo_key_lines(self): text = self.capture() for frag in ["canary in every doc/config/script file: True", "bash: curl: command not found", "honest submission ok: True", "dataset bump: major", "second try, waited 95s (t=226): ready.txt exists: True", "note.txt = 'hello\\n'", "cheater with a real solution: passes 12 of 12, rescored to 0: 12, counted passes: 0", "pareto front"]: self.assertIn(frag, text) def test_demo_leaderboard_order(self): by = tb.run_benchmark(tb.demo_tasks(), tb.demo_agents(), trials=10, seed=1) rows = tb.leaderboard(by, k=3) self.assertEqual([r.agent for r in rows], ["careful", "impatient", "speedy", "noop"]) self.assertEqual([round(r.rate, 3) for r in rows], [0.933, 0.817, 0.7, 0.0]) def test_demo_roster(self): self.assertEqual([a.name for a in tb.demo_agents()], ["careful", "speedy", "impatient", "noop"]) def test_demo_fixture_hash_is_right(self): t = task_by_name("lab/hash-file") want = hashlib.sha256(t.env_files["payload.txt"].encode()).hexdigest() self.assertIn(want, t.test_files["checks.toml"]) def test_main_guard(self): with open(tb.__file__) as fh: src = fh.read() self.assertIn('if __name__ == "__main__":', src) if __name__ == "__main__": unittest.main(verbosity=2)