107 lines
3.3 KiB
Python
Executable File
107 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Machine campaign approval gate for protected engineering PRs."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
from typing import Any, Mapping
|
|
|
|
from release_campaign_authorization import CampaignAuthorization, CampaignError
|
|
|
|
|
|
class GateError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def git_is_ancestor(repository: Path, ancestor: str, descendant: str) -> bool:
|
|
result = subprocess.run(
|
|
[
|
|
"git",
|
|
f"--git-dir={repository}",
|
|
"merge-base",
|
|
"--is-ancestor",
|
|
ancestor,
|
|
descendant,
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
)
|
|
if result.returncode == 0:
|
|
return True
|
|
if result.returncode == 1:
|
|
return False
|
|
raise GateError("git_ancestor_check_failed")
|
|
|
|
|
|
def evaluate(
|
|
campaign: CampaignAuthorization,
|
|
facts: Mapping[str, Any],
|
|
repository: Path,
|
|
) -> dict[str, Any]:
|
|
required = {
|
|
"repository",
|
|
"pr_number",
|
|
"base_ref",
|
|
"base_oid",
|
|
"head_oid",
|
|
"required_checks_all_success",
|
|
"required_checks_same_head",
|
|
"unresolved_conversations",
|
|
"mergeable",
|
|
"protected_target",
|
|
}
|
|
if set(facts) != required:
|
|
raise GateError("pr_facts_fields_invalid")
|
|
if facts["repository"] != "aios/aios":
|
|
raise GateError("repository_mismatch")
|
|
if facts["base_ref"] != "refs/heads/main" or facts["protected_target"] != "main":
|
|
raise GateError("protected_main_required")
|
|
if facts["required_checks_all_success"] is not True:
|
|
raise GateError("required_checks_not_success")
|
|
if facts["required_checks_same_head"] is not True:
|
|
raise GateError("required_checks_stale")
|
|
if facts["unresolved_conversations"] != 0:
|
|
raise GateError("unresolved_conversations")
|
|
if facts["mergeable"] is not True:
|
|
raise GateError("pr_not_mergeable")
|
|
if not git_is_ancestor(repository, campaign.anchor_oid, str(facts["base_oid"])):
|
|
raise GateError("base_outside_campaign_lineage")
|
|
if not git_is_ancestor(repository, str(facts["base_oid"]), str(facts["head_oid"])):
|
|
raise GateError("head_outside_pr_lineage")
|
|
campaign.assert_operation("protected_merge")
|
|
return {
|
|
"schema": "aios.campaign-approval-gate-result.v1",
|
|
"status": "SUCCESS",
|
|
"campaign_id": campaign.campaign_id,
|
|
"pr_number": facts["pr_number"],
|
|
"head_oid": facts["head_oid"],
|
|
"owner_ui_interaction_required": False,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--campaign", type=Path, required=True)
|
|
p.add_argument("--revocation", type=Path)
|
|
p.add_argument("--pr-facts", type=Path, required=True)
|
|
p.add_argument("--repository", type=Path, required=True)
|
|
a = p.parse_args()
|
|
campaign = CampaignAuthorization.load(
|
|
a.campaign,
|
|
revocation=a.revocation,
|
|
require_active=True,
|
|
)
|
|
facts = json.loads(a.pr_facts.read_text(encoding="utf-8"))
|
|
print(json.dumps(evaluate(campaign, facts, a.repository), sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (CampaignError, GateError) as exc:
|
|
print(json.dumps({"status": "FAILURE", "reason": str(exc)}, sort_keys=True))
|
|
raise SystemExit(2)
|