Compare commits
1 Commits
step35/443
...
step35/482
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1b97aa395d |
140
bin/enable-branch-protection.py
Executable file
140
bin/enable-branch-protection.py
Executable file
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
enable-branch-protection — Apply branch protection to Timmy_Foundation repos.
|
||||
|
||||
Enforces issue #482: agents must not merge before review.
|
||||
Applies identical protection rules to the main branch of core repos:
|
||||
- required_approvals: 1 (at least one human review)
|
||||
- require_status_checks: true (CI must pass)
|
||||
- restrict_merge: true (only admins + designated reviewers can merge)
|
||||
|
||||
Exit codes:
|
||||
0 — all requested branches protected (or already protected)
|
||||
1 — at least one branch could not be protected
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
from gitea_client import GiteaClient
|
||||
|
||||
# Core Timmy Foundation repositories
|
||||
CORE_REPOS = [
|
||||
"Timmy_Foundation/the-nexus",
|
||||
"Timmy_Foundation/timmy-home",
|
||||
"Timmy_Foundation/timmy-config",
|
||||
"Timmy_Foundation/hermes-agent",
|
||||
"Timmy_Foundation/.profile",
|
||||
"Timmy_Foundation/the-door",
|
||||
"Timmy_Foundation/turboquant",
|
||||
]
|
||||
|
||||
PROTECTION_RULE = {
|
||||
"required_approvals": 1,
|
||||
"require_status_checks": True,
|
||||
"restrict_merge": True,
|
||||
"dismiss_stale_reviews": False,
|
||||
"require_owner_approval": False,
|
||||
"allow_force_pushes": False,
|
||||
"allow_deletions": False,
|
||||
}
|
||||
|
||||
|
||||
def parse_args(argv: list[str]) -> argparse.Namespace:
|
||||
p = argparse.ArgumentParser(
|
||||
description="Apply branch protection to core repos (Closes #482)"
|
||||
)
|
||||
p.add_argument(
|
||||
"--branch",
|
||||
default="main",
|
||||
help="Branch to protect (default: main)",
|
||||
)
|
||||
p.add_argument(
|
||||
"--repo",
|
||||
action="append",
|
||||
help="Limit to specific repo(s); may be repeated. Overrides CORE_REPOS.",
|
||||
)
|
||||
p.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Show what would change without making API calls",
|
||||
)
|
||||
p.add_argument(
|
||||
"--verbose",
|
||||
action="store_true",
|
||||
help="Print detailed progress",
|
||||
)
|
||||
return p.parse_args(argv)
|
||||
|
||||
|
||||
def protection_matches(current: dict[str, object], desired: dict[str, object]) -> bool:
|
||||
for key, val in desired.items():
|
||||
if current.get(key) != val:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def protect_branch(
|
||||
client: GiteaClient,
|
||||
repo: str,
|
||||
branch: str,
|
||||
dry_run: bool = False,
|
||||
verbose: bool = False,
|
||||
) -> tuple[bool, str]:
|
||||
try:
|
||||
current = client.get_branch_protection(repo, branch)
|
||||
if current:
|
||||
if protection_matches(current, PROTECTION_RULE):
|
||||
return True, f"✓ {repo}/{branch} already protected"
|
||||
if dry_run:
|
||||
return True, f"⚠ {repo}/{branch} would be updated (rule differs)"
|
||||
client.set_branch_protection(repo, branch, **PROTECTION_RULE)
|
||||
return True, f"✓ {repo}/{branch} protection updated"
|
||||
else:
|
||||
if dry_run:
|
||||
return True, f"+ {repo}/{branch} would be protected"
|
||||
client.set_branch_protection(repo, branch, **PROTECTION_RULE)
|
||||
return True, f"✓ {repo}/{branch} now protected"
|
||||
except Exception as e:
|
||||
return False, f"✗ {repo}/{branch}: {e}"
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv or sys.argv[1:])
|
||||
repos = args.repo if args.repo else CORE_REPOS
|
||||
branch = args.branch
|
||||
|
||||
client = GiteaClient()
|
||||
|
||||
if not args.dry_run:
|
||||
try:
|
||||
client._get("/version")
|
||||
except Exception as e:
|
||||
print(f"ERROR: Cannot reach Gitea: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
all_ok = True
|
||||
for repo in repos:
|
||||
ok, msg = protect_branch(client, repo, branch, dry_run=args.dry_run, verbose=args.verbose)
|
||||
print(msg)
|
||||
if not ok:
|
||||
all_ok = False
|
||||
|
||||
if all_ok:
|
||||
print(f"\nDone. {len(repos)} branch{'s' if len(repos)!=1 else ''} checked.")
|
||||
return 0
|
||||
else:
|
||||
print("\nSome branches failed. Review errors above.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -515,6 +515,67 @@ class GiteaClient:
|
||||
return pr
|
||||
return None
|
||||
|
||||
# -- Branch Protection ----------------------------------------------------
|
||||
|
||||
def get_branch_protection(
|
||||
self, repo: str, branch: str
|
||||
) -> Optional[dict[str, Any]]:
|
||||
"""Get branch protection settings for a branch. Returns None if no protection."""
|
||||
try:
|
||||
return self._get(
|
||||
f"{self._repo_path(repo)}/branch_protection/{urllib.parse.quote(branch)}"
|
||||
)
|
||||
except GiteaError as e:
|
||||
if e.status == 404:
|
||||
return None
|
||||
raise
|
||||
|
||||
def set_branch_protection(
|
||||
self,
|
||||
repo: str,
|
||||
branch: str,
|
||||
required_approvals: int = 1,
|
||||
require_status_checks: bool = True,
|
||||
restrict_merge: bool = True,
|
||||
dismiss_stale_reviews: bool = False,
|
||||
require_owner_approval: bool = False,
|
||||
allow_force_pushes: bool = False,
|
||||
allow_deletions: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Create or update branch protection for a branch.
|
||||
|
||||
Enforces:
|
||||
- At least 1 approval before merge
|
||||
- CI/status checks must pass before merge
|
||||
|
||||
Returns the protection rule dict from Gitea.
|
||||
"""
|
||||
data: dict[str, Any] = {
|
||||
"required_approvals": required_approvals,
|
||||
"require_status_checks": require_status_checks,
|
||||
"restrict_merge": restrict_merge,
|
||||
"dismiss_stale_reviews": dismiss_stale_reviews,
|
||||
"require_owner_approval": require_owner_approval,
|
||||
"allow_force_pushes": allow_force_pushes,
|
||||
"allow_deletions": allow_deletions,
|
||||
}
|
||||
return self._post(
|
||||
f"{self._repo_path(repo)}/branch_protection/{urllib.parse.quote(branch)}",
|
||||
data,
|
||||
)
|
||||
|
||||
def delete_branch_protection(self, repo: str, branch: str) -> bool:
|
||||
"""Remove branch protection from a branch. Returns True if deleted."""
|
||||
try:
|
||||
self._delete(
|
||||
f"{self._repo_path(repo)}/branch_protection/{urllib.parse.quote(branch)}"
|
||||
)
|
||||
return True
|
||||
except GiteaError as e:
|
||||
if e.status == 404:
|
||||
return False
|
||||
raise
|
||||
|
||||
# -- Convenience ---------------------------------------------------------
|
||||
|
||||
def get_issue_with_comments(
|
||||
|
||||
@@ -316,3 +316,136 @@ class TestFindAgentIssues:
|
||||
result = client.find_agent_issues("repo", "claude")
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
# BRANCH PROTECTION — Issue #482
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestBranchProtection:
|
||||
"""Tests for branch protection enforcement (Closes #482)."""
|
||||
|
||||
def _mock_protection(
|
||||
self,
|
||||
monkeypatch,
|
||||
client,
|
||||
response: dict | None,
|
||||
status: int = 200,
|
||||
):
|
||||
"""Helper to mock _get/_post/_delete for branch protection endpoints."""
|
||||
def fake_request(
|
||||
method: str,
|
||||
path: str,
|
||||
data: dict | None = None,
|
||||
params: dict | None = None,
|
||||
retries: int = 3,
|
||||
backoff: float = 1.5,
|
||||
):
|
||||
if method == "GET" and "/branch_protection/" in path:
|
||||
if status == 404:
|
||||
from gitea_client import GiteaError
|
||||
raise GiteaError(404, "not found", path)
|
||||
return response if response is not None else {}
|
||||
if method == "POST" and "/branch_protection/" in path:
|
||||
return {"status": "ok", "protected": True, **data}
|
||||
if method == "DELETE" and "/branch_protection/" in path:
|
||||
from gitea_client import GiteaError
|
||||
if status == 404:
|
||||
raise GiteaError(404, "not found", path)
|
||||
return {"status": "ok"}
|
||||
# Fall back to original for other endpoints
|
||||
raise AssertionError(f"Unexpected call: {method} {path}")
|
||||
|
||||
monkeypatch.setattr(client, "_request", fake_request)
|
||||
|
||||
def test_get_branch_protection_returns_none_when_no_protection(
|
||||
self, monkeypatch
|
||||
):
|
||||
"""404 → None means branch is unprotected."""
|
||||
client = GiteaClient.__new__(GiteaClient)
|
||||
self._mock_protection(monkeypatch, client, None, status=404)
|
||||
result = client.get_branch_protection("owner/repo", "main")
|
||||
assert result is None
|
||||
|
||||
def test_get_branch_protection_returns_settings(self, monkeypatch):
|
||||
"""200 with JSON dict returns those settings."""
|
||||
client = GiteaClient.__new__(GiteaClient)
|
||||
protection = {
|
||||
"required_approvals": 2,
|
||||
"require_status_checks": True,
|
||||
"restrict_merge": False,
|
||||
}
|
||||
self._mock_protection(monkeypatch, client, protection)
|
||||
result = client.get_branch_protection("owner/repo", "main")
|
||||
assert result == protection
|
||||
|
||||
def test_set_branch_protection_posts_correct_payload(self, monkeypatch):
|
||||
"""POST sends all protection fields to Gitea."""
|
||||
client = GiteaClient.__new__(GiteaClient)
|
||||
posted: dict[str, object] = {}
|
||||
|
||||
def capture_post(method, path, data=None, **kw):
|
||||
if method == "POST":
|
||||
posted.update(data or {})
|
||||
return {"status": "ok", "protected": True}
|
||||
|
||||
monkeypatch.setattr(client, "_request", capture_post)
|
||||
result = client.set_branch_protection(
|
||||
"Timmy_Foundation/timmy-config",
|
||||
"main",
|
||||
required_approvals=1,
|
||||
require_status_checks=True,
|
||||
restrict_merge=True,
|
||||
)
|
||||
|
||||
assert posted["required_approvals"] == 1
|
||||
assert posted["require_status_checks"] is True
|
||||
assert posted["restrict_merge"] is True
|
||||
# Defaults should be present
|
||||
assert posted["allow_force_pushes"] is False
|
||||
assert posted["allow_deletions"] is False
|
||||
|
||||
def test_delete_branch_protection(self, monkeypatch):
|
||||
"""DELETE removes protection; 404 is already-gone."""
|
||||
client = GiteaClient.__new__(GiteaClient)
|
||||
calls = []
|
||||
|
||||
def fake_delete(method, path, **kw):
|
||||
if method == "DELETE" and "/branch_protection/" in path:
|
||||
calls.append(path)
|
||||
return {"status": "ok"}
|
||||
raise AssertionError(path)
|
||||
|
||||
monkeypatch.setattr(client, "_request", fake_delete)
|
||||
ok = client.delete_branch_protection("owner/repo", "main")
|
||||
assert ok is True
|
||||
assert any("branch_protection/main" in c for c in calls)
|
||||
|
||||
def test_delete_branch_protection_idempotent(self, monkeypatch):
|
||||
"""404 on delete is treated as already-gone and returns False."""
|
||||
from gitea_client import GiteaError
|
||||
|
||||
client = GiteaClient.__new__(GiteaClient)
|
||||
|
||||
def raise_404(*args, **kw):
|
||||
raise GiteaError(404, "not found", "")
|
||||
|
||||
monkeypatch.setattr(client, "_request", raise_404)
|
||||
ok = client.delete_branch_protection("owner/repo", "main")
|
||||
assert ok is False
|
||||
|
||||
def test_set_branch_protection_url_encodes_branch(self, monkeypatch):
|
||||
"""Branch names with special chars are URL-encoded."""
|
||||
client = GiteaClient.__new__(GiteaClient)
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
def capture_post(method, path, data=None, **kw):
|
||||
if method == "POST":
|
||||
captured["path"] = path
|
||||
captured["data"] = data
|
||||
return {"protected": True}
|
||||
|
||||
monkeypatch.setattr(client, "_request", capture_post)
|
||||
client.set_branch_protection("o/r", "feature/foo-bar")
|
||||
|
||||
assert "/branch_protection/feature%2Ffoo-bar" in captured["path"]
|
||||
|
||||
Reference in New Issue
Block a user