host
f73c3f7a05
此前 /pr 的子进程调用没有超时：远端通路卡死时 handler 线程永久挂住，客户端断开也收不回来。_run 真实路径一律 带 timeout（默认 120 秒，CLOUD_PR_SUBPROCESS_TIMEOUT 可改，非法值回落默认），超时杀子进程、返回 124，调用方 按原有失败路径处理（push/pr_create/commit 抛 PrStepError， 探针按跳过处理）。App 路径的 GitHub API 调用已有 30 秒 超时，本次不动。测试：真实 sleep 超时 2 个、环境变量 2 项、超时映射 push/commit 2 个
134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""pr.py 子进程超时单测。
|
||
|
||
真实路径：sleep 命令实测超时真的杀进程、真的按时返回
|
||
（不是等命令跑完）；环境变量改超时、非法值回落默认。
|
||
映射路径：假 runner 回 124 超时结果时，push/commit 按原有
|
||
失败路径抛 PrStepError（step 对得上），不挂住、不误报成功。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
try:
|
||
from cloud.pr import (
|
||
PrStepError,
|
||
_pr_subprocess_timeout,
|
||
_run,
|
||
create_task_pr,
|
||
)
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.pr import (
|
||
PrStepError,
|
||
_pr_subprocess_timeout,
|
||
_run,
|
||
create_task_pr,
|
||
)
|
||
|
||
|
||
def _result(returncode=0, stdout="", stderr=""):
|
||
return SimpleNamespace(returncode=returncode, stdout=stdout,
|
||
stderr=stderr)
|
||
|
||
|
||
class _Git:
|
||
"""假 git：push/commit 的失败模式可配，其余按脚本回。"""
|
||
|
||
def __init__(self, push="ok", commit="nothing"):
|
||
self.push = push
|
||
self.commit = commit
|
||
|
||
def __call__(self, cmd, cwd):
|
||
if "remote" in cmd:
|
||
return _result(0, "https://github.com/octo/a.git\n")
|
||
if "branch" in cmd:
|
||
return _result(0, "host/t1\n")
|
||
if "add" in cmd:
|
||
return _result(0, "")
|
||
if "commit" in cmd:
|
||
if self.commit == "timeout":
|
||
return _result(124, "", "command timed out")
|
||
return _result(1, "",
|
||
"nothing to commit, working tree clean")
|
||
if "push" in cmd:
|
||
if self.push == "timeout":
|
||
return _result(124, "", "command timed out")
|
||
return _result(0, "")
|
||
return _result(0, "")
|
||
|
||
|
||
def _gh(cmd, cwd):
|
||
if cmd[:2] == ["gh", "auth"]:
|
||
return _result(0, "", "")
|
||
if cmd[:2] == ["gh", "pr"]:
|
||
return _result(0, "https://github.com/octo/a/pull/9\n")
|
||
return _result(0, "")
|
||
|
||
|
||
class RunTimeoutTestCase(unittest.TestCase):
|
||
def test_real_sleep_is_killed_on_timeout(self):
|
||
start = time.monotonic()
|
||
r = _run(["sleep", "3"], timeout=1)
|
||
elapsed = time.monotonic() - start
|
||
self.assertEqual(r.returncode, 124)
|
||
self.assertIn("timed out", r.stderr)
|
||
self.assertLess(elapsed, 2.5)
|
||
|
||
def test_fast_command_ok_with_default_timeout(self):
|
||
r = _run(["true"])
|
||
self.assertEqual(r.returncode, 0)
|
||
|
||
def test_env_override_and_invalid_fallback(self):
|
||
old = os.environ.get("CLOUD_PR_SUBPROCESS_TIMEOUT")
|
||
try:
|
||
os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = "1"
|
||
self.assertEqual(_pr_subprocess_timeout(), 1)
|
||
start = time.monotonic()
|
||
r = _run(["sleep", "3"])
|
||
self.assertEqual(r.returncode, 124)
|
||
self.assertLess(time.monotonic() - start, 2.5)
|
||
os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = "abc"
|
||
self.assertEqual(_pr_subprocess_timeout(), 120)
|
||
os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = "0"
|
||
self.assertEqual(_pr_subprocess_timeout(), 120)
|
||
finally:
|
||
if old is None:
|
||
os.environ.pop("CLOUD_PR_SUBPROCESS_TIMEOUT", None)
|
||
else:
|
||
os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = old
|
||
|
||
|
||
class TimeoutMappingTestCase(unittest.TestCase):
|
||
def _worktree(self, tmp):
|
||
d = os.path.join(tmp, "t1")
|
||
os.makedirs(d, exist_ok=True)
|
||
return d
|
||
|
||
def test_push_timeout_is_pr_step_error(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._worktree(tmp)
|
||
with self.assertRaises(PrStepError) as ctx:
|
||
create_task_pr("t1", 1, worktrees_root=tmp,
|
||
git_runner=_Git(push="timeout"),
|
||
gh_runner=_gh, gh_which="/usr/bin/gh")
|
||
self.assertEqual(ctx.exception.step, "push")
|
||
self.assertIn("timed out", ctx.exception.message)
|
||
|
||
def test_commit_timeout_is_pr_step_error(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._worktree(tmp)
|
||
with self.assertRaises(PrStepError) as ctx:
|
||
create_task_pr("t1", 1, worktrees_root=tmp,
|
||
git_runner=_Git(commit="timeout"),
|
||
gh_runner=_gh, gh_which="/usr/bin/gh")
|
||
self.assertEqual(ctx.exception.step, "commit")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|