host
4cd9b5eaa4
新增 run_tests.py：三套件统一入口，只加仓库根（deploy 套件 再加 deploy/ 目录在根之前），各套件走独立子进程；verify 作 业改走运行器，coverage 作业因子进程测不到覆盖率保持原 discover 命令。tests/ 13 个文件的 sys.path 前置 import 全部 改回正常 import（含 test_host_trace 的 try/except 回落）； deploy 的 deploy 目录插入与 verify-multiuser 的 spec 加载按 陷阱 2 如实保留（仅删 verify-multiuser 冗余的根插入）。 .coveragerc omit 加 */run_tests.py，使新运行器文件不计入 生产覆盖率。门槛未动（仍 75%），未补新测试。
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
"""run_parallel.py 单测（CI 第二阶段覆盖率：基线 0%）。
|
|
|
|
不跑真实 client.py/codex：只断言 main() 拼出的子进程命令
|
|
（可执行文件、client.py 路径、cwd 解析、sandbox 默认值、
|
|
approval-policy、codex-home 透传）与成功计数/返回码——
|
|
命令构造与汇总正是本文件的全部逻辑。
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from unittest import mock
|
|
import run_parallel as rp_mod
|
|
|
|
|
|
class FakePopen:
|
|
seen = []
|
|
|
|
def __init__(self, cmd):
|
|
self.cmd = cmd
|
|
FakePopen.seen.append(cmd)
|
|
self._rc = 0
|
|
|
|
def wait(self):
|
|
return self._rc
|
|
|
|
|
|
class RunParallelTest(unittest.TestCase):
|
|
def _run(self, jobs, extra_args):
|
|
FakePopen.seen = []
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
jobs_path = os.path.join(tmp, "jobs.json")
|
|
with open(jobs_path, "w", encoding="utf-8") as f:
|
|
json.dump(jobs, f)
|
|
argv = ["run_parallel.py", "--jobs", jobs_path] + extra_args
|
|
with mock.patch.object(rp_mod.subprocess, "Popen", FakePopen), \
|
|
mock.patch.object(sys, "argv", argv):
|
|
out = io.StringIO()
|
|
with redirect_stdout(out):
|
|
rc = rp_mod.main()
|
|
return rc, list(FakePopen.seen), out.getvalue()
|
|
|
|
def test_commands_and_all_success(self):
|
|
jobs = [{"cwd": "job-a", "prompt": "review a"},
|
|
{"cwd": "/tmp/abs-b", "prompt": "review b",
|
|
"sandbox": "workspace-write"}]
|
|
rc, seen, out = self._run(jobs, ["--codex-home", "/tmp/ch"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(len(seen), 2)
|
|
client = os.path.join(rp_mod.HERE, "client.py")
|
|
self.assertEqual(seen[0], [sys.executable, client,
|
|
"--prompt", "review a",
|
|
"--cwd", os.path.join(rp_mod.HERE, "job-a"),
|
|
"--sandbox", "read-only",
|
|
"--approval-policy", "never",
|
|
"--codex-home", "/tmp/ch"])
|
|
self.assertEqual(seen[1][5], "/tmp/abs-b")
|
|
self.assertEqual(seen[1][7], "workspace-write")
|
|
self.assertIn("2/2", out)
|
|
|
|
def test_one_failure_returns_1(self):
|
|
class FailingPopen(FakePopen):
|
|
def __init__(self, cmd):
|
|
super().__init__(cmd)
|
|
self._idx = len(FakePopen.seen)
|
|
|
|
def wait(self):
|
|
return 1 if self._idx == 2 else 0
|
|
|
|
jobs = [{"cwd": "/tmp/a", "prompt": "x"},
|
|
{"cwd": "/tmp/b", "prompt": "y"}]
|
|
FakePopen.seen = []
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
jobs_path = os.path.join(tmp, "jobs.json")
|
|
with open(jobs_path, "w", encoding="utf-8") as f:
|
|
json.dump(jobs, f)
|
|
with mock.patch.object(rp_mod.subprocess, "Popen",
|
|
FailingPopen), \
|
|
mock.patch.object(sys, "argv",
|
|
["run_parallel.py", "--jobs",
|
|
jobs_path]):
|
|
out = io.StringIO()
|
|
with redirect_stdout(out):
|
|
rc = rp_mod.main()
|
|
self.assertEqual(rc, 1)
|
|
self.assertIn("1/2", out.getvalue())
|
|
|
|
def test_empty_jobs_returns_0(self):
|
|
rc, seen, out = self._run([], [])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(seen, [])
|
|
self.assertIn("0/0", out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|