129 lines
7.8 KiB
Python
129 lines
7.8 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
from werkrevisor import codex as codex_eval
|
|
|
|
|
|
class CodexTest(unittest.TestCase):
|
|
def test_pr_url_gives_server_owner_repo_and_number(self):
|
|
self.assertEqual(codex_eval.parse_pr_url("https://dev.example.net/team/hs.app/pulls/306"),
|
|
("https://dev.example.net", "team", "hs.app", 306))
|
|
with self.assertRaises(ValueError):
|
|
codex_eval.parse_pr_url("https://dev.example.net/team/hs.app/issues/306")
|
|
|
|
def test_token_use_is_summed_over_all_turns(self):
|
|
lines = [json.dumps({"type": "turn.completed", "usage": {"input_tokens": 10, "cached_input_tokens": 4,
|
|
"output_tokens": 2, "reasoning_output_tokens": 1}}),
|
|
"not json",
|
|
json.dumps({"type": "item.completed", "item": {"type": "command_execution", "command": "git diff"}}),
|
|
json.dumps({"type": "item.completed", "item": {"type": "agent_message", "text": "{}"}}),
|
|
json.dumps({"type": "turn.completed", "usage": {"input_tokens": 5, "output_tokens": 3}}),
|
|
json.dumps({"type": "turn.failed", "error": {"message": "quota"}})]
|
|
usage, commands, errors = codex_eval.summarize_events(lines)
|
|
self.assertEqual(usage, {"input_tokens": 15, "cached_input_tokens": 4, "output_tokens": 5,
|
|
"reasoning_output_tokens": 1})
|
|
self.assertEqual(commands, ["git diff"])
|
|
self.assertEqual(len(errors), 1)
|
|
self.assertIn("quota", errors[0])
|
|
|
|
def test_prompt_names_the_merge_base_and_the_description(self):
|
|
prompt = codex_eval.prompt_for({"number": 7, "title": "Scope", "body": " Adds a scope. "}, "abc123")
|
|
self.assertIn('#7 "Scope"', prompt)
|
|
self.assertIn("git diff abc123..HEAD", prompt)
|
|
self.assertIn("Adds a scope.", prompt)
|
|
self.assertIn("(none)", codex_eval.prompt_for({"number": 7, "title": "Scope", "body": None}, "abc123"))
|
|
|
|
def test_prompt_excludes_generated_sections(self):
|
|
prompt = codex_eval.prompt_for({"number": 7, "title": "Scope", "body": ""}, "abc123")
|
|
self.assertIn("`<!-- generated by … -->` up to the next heading;\n report nothing about such a section.", prompt)
|
|
|
|
def test_schema_requires_every_field(self):
|
|
finding = codex_eval.SCHEMA["properties"]["findings"]["items"]
|
|
self.assertEqual(sorted(finding["required"]), sorted(finding["properties"]))
|
|
self.assertEqual(sorted(codex_eval.SCHEMA["required"]), sorted(codex_eval.SCHEMA["properties"]))
|
|
|
|
def test_only_a_complete_answer_counts_as_a_review(self):
|
|
finding = {"severity": "high", "file": "a.py", "line": 3, "title": "t", "explanation": "e"}
|
|
self.assertTrue(codex_eval.valid_answer({"critical": False, "critical_reason": "", "findings": [finding]}))
|
|
for broken in (None, {"unparsed": "text"}, {"critical": "no", "findings": []}, {"critical": True},
|
|
{"critical": True, "findings": [{**finding, "severity": "blocker"}]},
|
|
{"critical": True, "critical_reason": "", "findings": [{**finding, "line": True}]},
|
|
{"critical": True, "findings": []},
|
|
{"critical": True, "critical_reason": "", "findings": [{**finding, "title": None}]},
|
|
{"critical": True, "critical_reason": "", "findings": [
|
|
{k: v for k, v in finding.items() if k != "explanation"}]}):
|
|
with self.subTest(broken=broken):
|
|
self.assertFalse(codex_eval.valid_answer(broken))
|
|
|
|
def test_the_login_values_are_secrets(self):
|
|
with tempfile.TemporaryDirectory() as home:
|
|
with open(os.path.join(home, "auth.json"), "w") as f:
|
|
json.dump({"auth_mode": "chatgpt", "OPENAI_API_KEY": None,
|
|
"tokens": {"id_token": "x" * 40, "refresh_token": "r" * 20, "account_id": "a" * 10,
|
|
"kind": "short"}}, f)
|
|
self.assertEqual(sorted(codex_eval.auth_secrets(home)), ["a" * 10, "r" * 20, "x" * 40])
|
|
self.assertEqual(codex_eval.auth_secrets(os.path.join(home, "missing")), [])
|
|
|
|
def test_the_jail_hides_the_home_except_the_login_and_the_programs(self):
|
|
with tempfile.TemporaryDirectory() as home:
|
|
programs, outside, tmp = os.path.join(home, ".nvm", "bin"), tempfile.gettempdir(), os.path.join(home, "t")
|
|
os.makedirs(programs)
|
|
argv = codex_eval.jail(["codex", "exec"], home, os.path.join(home, ".codex"), [], [tmp],
|
|
os.pathsep.join([programs, "/usr/bin"]))
|
|
self.assertEqual(argv[:9], ["bwrap", "--ro-bind", "/", "/", "--tmpfs", "/tmp", "--tmpfs", home, "--dev"])
|
|
pairs = [tuple(argv[i:i + 3]) for i in range(len(argv) - 2)]
|
|
self.assertIn(("--bind", os.path.join(home, ".codex"), os.path.join(home, ".codex")), pairs)
|
|
self.assertIn(("--ro-bind", programs, programs), pairs)
|
|
self.assertIn(("--bind", tmp, tmp), pairs)
|
|
self.assertNotIn("/usr/bin", argv)
|
|
self.assertNotIn(outside, argv[9:])
|
|
self.assertEqual(argv[-2:], ["codex", "exec"])
|
|
|
|
def test_the_sandbox_check_runs_a_bwrap_inside_the_jail(self):
|
|
calls = []
|
|
|
|
def run(command, **kwargs):
|
|
calls.append(command)
|
|
return unittest.mock.Mock(returncode=1, stderr="bwrap: setting up uid map: Permission denied\n")
|
|
with unittest.mock.patch.object(codex_eval.shutil, "which", return_value="/usr/bin/bwrap"), \
|
|
unittest.mock.patch.object(codex_eval.subprocess, "run", run):
|
|
self.assertEqual(codex_eval.check_sandbox(), "bwrap: setting up uid map: Permission denied")
|
|
[command] = calls
|
|
self.assertEqual(command[0], "bwrap")
|
|
self.assertEqual(command[-5:], ["bwrap", "--ro-bind", "/", "/", "true"])
|
|
self.assertIn("--die-with-parent", command)
|
|
with unittest.mock.patch.object(codex_eval.shutil, "which", return_value="/usr/bin/bwrap"), \
|
|
unittest.mock.patch.object(codex_eval.subprocess, "run", return_value=unittest.mock.Mock(returncode=0)):
|
|
self.assertIsNone(codex_eval.check_sandbox())
|
|
with unittest.mock.patch.object(codex_eval.shutil, "which", return_value=None):
|
|
self.assertEqual(codex_eval.check_sandbox(), "bwrap is missing")
|
|
|
|
def test_only_a_failed_bwrap_is_a_jail_fault(self):
|
|
uid_map = "bwrap: setting up uid map: Permission denied"
|
|
self.assertTrue(codex_eval.jail_failed({"exit_code": 1, "errors": ["quota", uid_map]}))
|
|
self.assertFalse(codex_eval.jail_failed({"exit_code": 1, "errors": ["quota"]}))
|
|
self.assertFalse(codex_eval.jail_failed({"exit_code": 0, "errors": [uid_map]}))
|
|
|
|
def test_a_token_reaches_only_its_own_server(self):
|
|
calls = []
|
|
|
|
def git(*args, cwd=None, env=None):
|
|
calls.append((args, cwd, (env or {}).get("GIT_CONFIG_VALUE_0")))
|
|
if args[0] == "clone":
|
|
os.makedirs(args[-1])
|
|
if args[0] == "worktree":
|
|
os.makedirs(args[4])
|
|
return ""
|
|
with tempfile.TemporaryDirectory() as root, unittest.mock.patch.object(codex_eval, "_git", git):
|
|
for server, token in (("https://a.example.org", "token-a"), ("https://b.example.org", "token-b")):
|
|
codex_eval.checkout(server, "team", "app", 7, "h" * 40, "b" * 40, token, root)
|
|
fetches = [(args, cwd, header) for args, cwd, header in calls if args[0] in ("clone", "fetch")]
|
|
for args, cwd, header in fetches:
|
|
server = "https://a.example.org" if header.endswith("token-a") else "https://b.example.org"
|
|
self.assertIn(f"{server}/team/app.git", args, f"{header} went to {args}")
|
|
mirrors = {cwd for args, cwd, _ in fetches if args[0] == "fetch"}
|
|
self.assertEqual(len(mirrors), 2)
|