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("`` 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)