236 lines
10 KiB
Python
236 lines
10 KiB
Python
"""CI 等待（scripts/ci_wait.py）：临时 Git 仓库 + 进程内 Gitea 合并状态桩 + 注入的时钟与 sleep。
|
||
|
||
全部离线运行：不联网、不读取真实 ~/.secrets（HOME 指向临时目录），token 为合成值。
|
||
"""
|
||
import contextlib
|
||
import http.server
|
||
import io
|
||
import json
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
SCRIPT = SOURCE / "scripts" / "ci_wait.py"
|
||
sys.path.insert(0, str(SOURCE / "scripts"))
|
||
|
||
import ci_wait # noqa: E402
|
||
|
||
TOKEN = "synthetic-gitea-token-not-real"
|
||
|
||
|
||
def status(context, state, url=None):
|
||
return {"context": context, "status": state, "target_url": url or f"https://ci.invalid/{context}"}
|
||
|
||
|
||
class FakeGitea:
|
||
"""Gitea 合并状态桩（127.0.0.1）：校验 token；每次 GET 依次返回脚本中的下一份 statuses（末项重复）。"""
|
||
|
||
def __init__(self, token=TOKEN):
|
||
self.token = token
|
||
self.script = []
|
||
self.requests = []
|
||
stub = self
|
||
|
||
class Handler(http.server.BaseHTTPRequestHandler):
|
||
def log_message(self, *args):
|
||
pass
|
||
|
||
def _send(self, code, obj):
|
||
body = json.dumps(obj).encode()
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self):
|
||
if self.headers.get("Authorization") != f"token {stub.token}":
|
||
return self._send(401, {"message": "unauthorized"})
|
||
parts = self.path.split("?", 1)[0].split("/")
|
||
if parts[1:4] != ["api", "v1", "repos"] or len(parts) != 9 or parts[6] != "commits" \
|
||
or parts[8] != "status":
|
||
return self._send(404, {"message": "not found"})
|
||
stub.requests.append("/".join(parts[4:6]) + "@" + parts[7])
|
||
statuses = stub.script.pop(0) if len(stub.script) > 1 else (stub.script or [[]])[0]
|
||
return self._send(200, {"sha": parts[7], "statuses": statuses})
|
||
|
||
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
self.url = f"http://127.0.0.1:{self.server.server_address[1]}/api/v1"
|
||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||
|
||
def close(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
|
||
|
||
def closed_port_url():
|
||
with socket.socket() as sock:
|
||
sock.bind(("127.0.0.1", 0))
|
||
port = sock.getsockname()[1]
|
||
return f"http://127.0.0.1:{port}/api/v1"
|
||
|
||
|
||
class FakeClock:
|
||
def __init__(self):
|
||
self.now = 0.0
|
||
self.sleeps = []
|
||
|
||
def clock(self):
|
||
return self.now
|
||
|
||
def sleep(self, seconds):
|
||
self.sleeps.append(seconds)
|
||
self.now += seconds
|
||
|
||
|
||
class CiWaitTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory(prefix="ci-wait-")
|
||
self.addCleanup(self.tmp.cleanup)
|
||
root = Path(self.tmp.name)
|
||
self.repo = root / "repo"
|
||
env = {k: v for k, v in os.environ.items()
|
||
if not k.startswith("GIT_") and k not in {"GITEA_API_URL", "GITEA_REPO", "GITEA_TOKEN"}}
|
||
env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_TERMINAL_PROMPT": "0",
|
||
"GIT_AUTHOR_NAME": "CI Wait Test", "GIT_AUTHOR_EMAIL": "ci-wait@example.invalid",
|
||
"GIT_COMMITTER_NAME": "CI Wait Test", "GIT_COMMITTER_EMAIL": "ci-wait@example.invalid",
|
||
"HOME": str(root / "no-home"),
|
||
"NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost",
|
||
"GITEA_TOKEN": TOKEN})
|
||
self.gitea = FakeGitea()
|
||
self.addCleanup(self.gitea.close)
|
||
env["GITEA_API_URL"] = self.gitea.url
|
||
self.env = env
|
||
env_patch = mock.patch.dict(os.environ, env, clear=True)
|
||
env_patch.start()
|
||
self.addCleanup(env_patch.stop)
|
||
self.git(root, "init", "--initial-branch=main", str(self.repo))
|
||
(self.repo / "README").write_text("fixture\n")
|
||
self.git(self.repo, "add", "README")
|
||
self.git(self.repo, "commit", "-m", "Create fixture")
|
||
self.git(self.repo, "remote", "add", "origin", "ssh://git@git-ssh.example.invalid:31411/fixture/repo.git")
|
||
self.sha = self.git(self.repo, "rev-parse", "HEAD")
|
||
cwd = os.getcwd()
|
||
os.chdir(self.repo)
|
||
self.addCleanup(os.chdir, cwd)
|
||
|
||
def git(self, cwd, *args):
|
||
return subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, check=True).stdout.strip()
|
||
|
||
def run_wait(self, *args, env=None):
|
||
clock = FakeClock()
|
||
out = io.StringIO()
|
||
with mock.patch.dict(os.environ, env or {}), contextlib.redirect_stdout(out), \
|
||
contextlib.redirect_stderr(out):
|
||
code = ci_wait.main(list(args), sleep=clock.sleep, clock=clock.clock)
|
||
text = out.getvalue()
|
||
self.assertNotIn(TOKEN, text)
|
||
return code, text, clock
|
||
|
||
def test_success_requires_two_consecutive_all_green_polls(self):
|
||
self.gitea.script = [[status("ci/test", "pending")], [status("ci/test", "pending")],
|
||
[status("ci/test", "success")], [status("ci/test", "success")]]
|
||
code, out, clock = self.run_wait("--timeout", "300")
|
||
self.assertEqual(code, 0, out)
|
||
self.assertIn(f"[verdict] CI_SUCCESS sha={self.sha}", out)
|
||
self.assertEqual(len(self.gitea.requests), 4, out)
|
||
self.assertEqual(self.gitea.requests[0], f"fixture/repo@{self.sha}")
|
||
self.assertEqual(clock.sleeps, [15.0, 15.0, 15.0], out)
|
||
# 状态行只在变化时打印：pending 与 success 各一行。
|
||
lines = [line for line in out.splitlines() if line.startswith("[ci-wait] sha=")]
|
||
self.assertEqual(len(lines), 2, out)
|
||
|
||
def test_late_registering_job_does_not_succeed_early(self):
|
||
self.gitea.script = [[status("ci/test", "success")],
|
||
[status("ci/test", "success"), status("ci/build", "pending")],
|
||
[status("ci/test", "success"), status("ci/build", "success")],
|
||
[status("ci/test", "success"), status("ci/build", "success")]]
|
||
code, out, _ = self.run_wait("--timeout", "300")
|
||
self.assertEqual(code, 0, out)
|
||
self.assertEqual(len(self.gitea.requests), 4, out)
|
||
self.assertIn("ci/build=pending", out)
|
||
|
||
def test_failure_exits_fast_with_target_url(self):
|
||
url = "https://ci.invalid/runs/42"
|
||
self.gitea.script = [[status("ci/test", "pending"), status("ci/build", "success")],
|
||
[status("ci/test", "failure", url), status("ci/build", "success")]]
|
||
code, out, _ = self.run_wait("--sha", self.sha, "--timeout", "300")
|
||
self.assertEqual(code, 1, out)
|
||
self.assertEqual(len(self.gitea.requests), 2, out)
|
||
self.assertIn(f"[ci-wait] FAIL ci/test {url}", out)
|
||
self.assertNotIn("FAIL ci/build", out)
|
||
self.assertIn(f"[verdict] CI_FAILED sha={self.sha}", out)
|
||
|
||
def test_missing_required_context_waits_until_timeout(self):
|
||
self.gitea.script = [[status("ci/test", "success")]]
|
||
code, out, clock = self.run_wait("--contexts", "ci/test,e2e", "--interval", "5", "--timeout", "30")
|
||
self.assertEqual(code, 2, out)
|
||
self.assertEqual(clock.now, 30.0)
|
||
self.assertIn("last observed: ci/test=success", out)
|
||
self.assertIn("missing contexts: e2e", out)
|
||
self.assertIn(f"[verdict] CI_TIMEOUT sha={self.sha}", out)
|
||
|
||
def test_skipped_and_warning_contexts_count_as_done_not_failed(self):
|
||
self.gitea.script = [[status("ci/test", "pending"), status("ci/test-pg", "skipped")],
|
||
[status("ci/test", "success"), status("ci/test-pg", "skipped"),
|
||
status("ci/svc-realcombo", "warning")]]
|
||
code, out, _ = self.run_wait("--contexts", "test-pg", "--timeout", "300")
|
||
self.assertEqual(code, 0, out)
|
||
self.assertEqual(len(self.gitea.requests), 3, out)
|
||
self.assertNotIn("FAIL", out)
|
||
self.assertIn(f"[verdict] CI_SUCCESS sha={self.sha}", out)
|
||
|
||
def test_skipped_does_not_mask_a_pending_context(self):
|
||
self.gitea.script = [[status("ci/test", "pending"), status("ci/test-pg", "skipped")]]
|
||
code, out, _ = self.run_wait("--interval", "5", "--timeout", "30")
|
||
self.assertEqual(code, 2, out)
|
||
self.assertIn(f"[verdict] CI_TIMEOUT sha={self.sha}", out)
|
||
|
||
def test_context_names_match_as_substrings(self):
|
||
self.gitea.script = [[status("CI / test (pull_request)", "success")]]
|
||
code, out, _ = self.run_wait("--contexts", "test", "--timeout", "60")
|
||
self.assertEqual(code, 0, out)
|
||
|
||
def test_unreachable_api_exits_6_after_retrying_until_deadline(self):
|
||
code, out, clock = self.run_wait("--timeout", "60", env={"GITEA_API_URL": closed_port_url()})
|
||
self.assertEqual(code, 6, out)
|
||
self.assertEqual(clock.now, 60.0)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", out)
|
||
|
||
def test_unauthorized_or_missing_token_exits_6_immediately(self):
|
||
code, out, clock = self.run_wait("--timeout", "60", env={"GITEA_TOKEN": "wrong-token"})
|
||
self.assertEqual(code, 6, out)
|
||
self.assertEqual(clock.sleeps, [], out)
|
||
self.assertIn("HTTP 401", out)
|
||
with mock.patch.dict(os.environ):
|
||
del os.environ["GITEA_TOKEN"]
|
||
code, out, _ = self.run_wait("--timeout", "60")
|
||
self.assertEqual(code, 6, out)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", out)
|
||
|
||
def test_sha_must_resolve_to_full_sha(self):
|
||
code, out, _ = self.run_wait("--sha", "not-a-ref")
|
||
self.assertEqual(code, 2, out)
|
||
self.assertIn("full 40-hex SHA", out)
|
||
self.assertEqual(self.gitea.requests, [])
|
||
|
||
def test_cli_entry_point_reports_remote_unverified(self):
|
||
env = self.env | {"GITEA_API_URL": closed_port_url()}
|
||
result = subprocess.run([sys.executable, str(SCRIPT), "--interval", "5", "--timeout", "0.2"],
|
||
cwd=self.repo, env=env, capture_output=True, text=True, timeout=60)
|
||
out = result.stdout + result.stderr
|
||
self.assertEqual(result.returncode, 6, out)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", out)
|
||
self.assertNotIn(TOKEN, out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|