250 lines
8.7 KiB
Python
250 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = REPO_ROOT / "skills" / "builder" / "scripts" / "wait_ci.sh"
|
|
|
|
|
|
def git(cwd: Path, *args: str) -> str:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=cwd,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
env={
|
|
**os.environ,
|
|
"GIT_AUTHOR_NAME": "Test",
|
|
"GIT_AUTHOR_EMAIL": "test@example.com",
|
|
"GIT_COMMITTER_NAME": "Test",
|
|
"GIT_COMMITTER_EMAIL": "test@example.com",
|
|
},
|
|
)
|
|
if result.returncode != 0:
|
|
raise AssertionError(f"git {args} failed: {result.stderr}")
|
|
return result.stdout.strip()
|
|
|
|
|
|
class BuilderCiTests(unittest.TestCase):
|
|
def make_repo(self, root: Path, origin: str) -> str:
|
|
git(root, "init", "-b", "main")
|
|
git(root, "config", "user.email", "test@example.com")
|
|
git(root, "config", "user.name", "Test")
|
|
(root / "README").write_text("x\n", encoding="utf-8")
|
|
git(root, "add", "README")
|
|
git(root, "commit", "-m", "init")
|
|
git(root, "remote", "add", "origin", origin)
|
|
return git(root, "rev-parse", "HEAD")
|
|
|
|
def fake_command(self, bin_dir: Path, name: str, body: str) -> None:
|
|
command = bin_dir / name
|
|
command.write_text(f"#!/bin/sh\n{body}\n", encoding="utf-8")
|
|
command.chmod(0o755)
|
|
|
|
def run_wait(
|
|
self,
|
|
root: Path,
|
|
sha: str,
|
|
bin_dir: Path,
|
|
extra_env: dict[str, str] | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
env = {
|
|
**os.environ,
|
|
"PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
|
|
"BUILDER_CI_POLL_SEC": "0",
|
|
"BUILDER_CI_APPEAR_SEC": "0",
|
|
"BUILDER_CI_FORGE": "gitea",
|
|
}
|
|
if extra_env:
|
|
env.update(extra_env)
|
|
return subprocess.run(
|
|
["bash", str(SCRIPT), "--sha", sha, "--root", str(root)],
|
|
cwd=root,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_gitea_success_reads_exact_run_and_jobs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
sha = self.make_repo(root, "https://gitea.example.test/acme/widget.git")
|
|
bin_dir = root / "bin"
|
|
bin_dir.mkdir()
|
|
self.fake_command(
|
|
bin_dir,
|
|
"curl",
|
|
"""
|
|
case "$*" in
|
|
*"/jobs?"*) cat "$BUILDER_CI_FIXTURE/jobs.json" ;;
|
|
*"/actions/runs?"*) cat "$BUILDER_CI_FIXTURE/runs.json" ;;
|
|
*) exit 9 ;;
|
|
esac
|
|
""",
|
|
)
|
|
fixture = root / "fixture"
|
|
fixture.mkdir()
|
|
(fixture / "runs.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"workflow_runs": [
|
|
{
|
|
"id": 9,
|
|
"head_sha": "0" * 40,
|
|
"status": "completed",
|
|
"conclusion": "success",
|
|
"html_url": "https://gitea.example.test/wrong",
|
|
},
|
|
{
|
|
"id": 12,
|
|
"head_sha": sha,
|
|
"status": "completed",
|
|
"conclusion": "success",
|
|
"html_url": "https://gitea.example.test/run/12",
|
|
},
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(fixture / "jobs.json").write_text(
|
|
json.dumps(
|
|
{"jobs": [{"id": 1, "name": "build", "head_sha": sha, "status": "completed", "conclusion": "success"}]}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
result = self.run_wait(
|
|
root,
|
|
sha,
|
|
bin_dir,
|
|
{"BUILDER_CI_FIXTURE": str(fixture)},
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn("OK pipeline=https://gitea.example.test/run/12", result.stdout)
|
|
self.assertIn("id=12", result.stdout)
|
|
self.assertIn("jobs=build=success", result.stdout)
|
|
self.assertNotIn("wrong", result.stdout + result.stderr)
|
|
|
|
def test_gitea_failed_job_is_failure(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
sha = self.make_repo(root, "git@gitea.example.test:acme/widget.git")
|
|
bin_dir = root / "bin"
|
|
bin_dir.mkdir()
|
|
fixture = root / "fixture"
|
|
fixture.mkdir()
|
|
self.fake_command(
|
|
bin_dir,
|
|
"curl",
|
|
"""
|
|
case "$*" in
|
|
*"/jobs?"*) cat "$BUILDER_CI_FIXTURE/jobs.json" ;;
|
|
*"/actions/runs?"*) cat "$BUILDER_CI_FIXTURE/runs.json" ;;
|
|
*) exit 9 ;;
|
|
esac
|
|
""",
|
|
)
|
|
(fixture / "runs.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"workflow_runs": [
|
|
{
|
|
"id": 21,
|
|
"head_sha": sha,
|
|
"status": "completed",
|
|
"conclusion": "failure",
|
|
"html_url": "https://gitea.example.test/run/21",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(fixture / "jobs.json").write_text(
|
|
json.dumps(
|
|
{"jobs": [{"id": 2, "name": "test", "head_sha": sha, "status": "completed", "conclusion": "failure"}]}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
result = self.run_wait(
|
|
root,
|
|
sha,
|
|
bin_dir,
|
|
{"BUILDER_CI_FIXTURE": str(fixture)},
|
|
)
|
|
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
|
self.assertIn("reason=job_failed", result.stderr)
|
|
self.assertIn("test=failure", result.stderr)
|
|
|
|
def test_gitea_does_not_treat_other_sha_as_pipeline(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
sha = self.make_repo(root, "https://gitea.example.test/acme/widget.git")
|
|
bin_dir = root / "bin"
|
|
bin_dir.mkdir()
|
|
fixture = root / "fixture"
|
|
fixture.mkdir()
|
|
self.fake_command(
|
|
bin_dir,
|
|
"curl",
|
|
'cat "$BUILDER_CI_FIXTURE/runs.json"',
|
|
)
|
|
(fixture / "runs.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"workflow_runs": [
|
|
{
|
|
"id": 99,
|
|
"head_sha": "f" * 40,
|
|
"status": "completed",
|
|
"conclusion": "success",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
result = self.run_wait(
|
|
root,
|
|
sha,
|
|
bin_dir,
|
|
{"BUILDER_CI_FIXTURE": str(fixture)},
|
|
)
|
|
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
|
self.assertIn("reason=no_pipeline", result.stderr)
|
|
|
|
def test_github_and_gitlab_branches_still_use_their_cli(self) -> None:
|
|
for forge, command, output in (
|
|
("github", "gh", "printf '%s\\n' '[]'"),
|
|
("gitlab", "glab", "printf '%s\\n' '[]'"),
|
|
):
|
|
with self.subTest(forge=forge), tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
sha = self.make_repo(root, f"https://{forge}.example.test/acme/widget.git")
|
|
bin_dir = root / "bin"
|
|
bin_dir.mkdir()
|
|
self.fake_command(bin_dir, command, output)
|
|
result = self.run_wait(
|
|
root,
|
|
sha,
|
|
bin_dir,
|
|
{
|
|
"BUILDER_CI_FORGE": forge,
|
|
"BUILDER_CI_APPEAR_SEC": "0",
|
|
},
|
|
)
|
|
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
|
self.assertIn("reason=no_pipeline", result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|