137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Detect drift between manifest.yaml-rendered pipelines and each target
|
|
repo's HEAD `.woodpecker.yaml` (oleks/ci-scripts#13).
|
|
|
|
Unlike `generate.py --check` (which compares against local `../building/*`
|
|
checkouts), this fetches each repo's live HEAD file over the Gitea API, so it
|
|
works from a fresh CI clone of ci-scripts alone. On drift it opens or updates
|
|
a single "CI drift detected" issue in oleks/ci-scripts; when drift clears, it
|
|
closes that issue if still open.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
import yaml
|
|
|
|
import generate
|
|
|
|
GITEA_HOST = "git.oleks.space"
|
|
OWNER = "oleks"
|
|
ISSUE_TITLE = "CI drift detected"
|
|
HERE = pathlib.Path(__file__).parent
|
|
|
|
|
|
def api(method: str, path: str, token: str, body: dict | None = None) -> dict:
|
|
url = f"https://{GITEA_HOST}/api/v1{path}"
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Authorization", f"token {token}")
|
|
if data is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else {}
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
return {}
|
|
raise
|
|
|
|
|
|
def fetch_head_file(repo: str, token: str) -> str:
|
|
data = api("GET", f"/repos/{OWNER}/{repo}/contents/.woodpecker.yaml", token)
|
|
if not data or "content" not in data:
|
|
return ""
|
|
return base64.b64decode(data["content"]).decode()
|
|
|
|
|
|
def find_drift_issue(token: str) -> dict | None:
|
|
issues = api(
|
|
"GET",
|
|
f"/repos/{OWNER}/ci-scripts/issues?state=open&type=issues&q={ISSUE_TITLE.replace(' ', '+')}",
|
|
token,
|
|
)
|
|
for issue in issues or []:
|
|
if issue.get("title") == ISSUE_TITLE:
|
|
return issue
|
|
return None
|
|
|
|
|
|
def repo_name(dir_: str) -> str:
|
|
return dir_.rsplit("/", 1)[-1]
|
|
|
|
|
|
def main() -> int:
|
|
token = os.environ.get("CI_DRIFT_TOKEN")
|
|
if not token:
|
|
print("CI_DRIFT_TOKEN not set", file=sys.stderr)
|
|
return 1
|
|
|
|
manifest = yaml.safe_load((HERE / "manifest.yaml").read_text())
|
|
jobs = [(r, generate.render) for r in manifest["repos"]] + [
|
|
(r, generate.render_thin) for r in manifest.get("thin_repos", [])
|
|
]
|
|
|
|
drifted = []
|
|
for repo, render_fn in jobs:
|
|
name = repo_name(repo["dir"])
|
|
expected = render_fn(repo)
|
|
actual = fetch_head_file(name, token)
|
|
if actual != expected:
|
|
drifted.append(name)
|
|
|
|
existing_issue = find_drift_issue(token)
|
|
|
|
if drifted:
|
|
body_lines = [
|
|
"Drift detected between `manifest.yaml`-rendered pipelines and "
|
|
"the repo's live `.woodpecker.yaml`:",
|
|
"",
|
|
] + [f"- {OWNER}/{name}" for name in sorted(drifted)]
|
|
body_lines += [
|
|
"",
|
|
"Re-render with `python3 generate.py` in oleks/ci-scripts and "
|
|
"push the update to the drifted repo(s), or reconcile the "
|
|
"manifest if the hand-edit was intentional.",
|
|
]
|
|
body = "\n".join(body_lines)
|
|
if existing_issue:
|
|
api(
|
|
"PATCH",
|
|
f"/repos/{OWNER}/ci-scripts/issues/{existing_issue['number']}",
|
|
token,
|
|
{"body": body},
|
|
)
|
|
print(f"updated issue #{existing_issue['number']}")
|
|
else:
|
|
api(
|
|
"POST",
|
|
f"/repos/{OWNER}/ci-scripts/issues",
|
|
token,
|
|
{"title": ISSUE_TITLE, "body": body},
|
|
)
|
|
print("opened drift issue")
|
|
print(f"drift found in: {', '.join(sorted(drifted))}")
|
|
return 1
|
|
|
|
print("no drift")
|
|
if existing_issue:
|
|
api(
|
|
"PATCH",
|
|
f"/repos/{OWNER}/ci-scripts/issues/{existing_issue['number']}",
|
|
token,
|
|
{"state": "closed", "body": "Drift cleared on next scheduled check."},
|
|
)
|
|
print(f"closed issue #{existing_issue['number']}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|