Compare commits
1 Commits
step35/595
...
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 pr
|
||||||
return None
|
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 ---------------------------------------------------------
|
# -- Convenience ---------------------------------------------------------
|
||||||
|
|
||||||
def get_issue_with_comments(
|
def get_issue_with_comments(
|
||||||
|
|||||||
@@ -316,3 +316,136 @@ class TestFindAgentIssues:
|
|||||||
result = client.find_agent_issues("repo", "claude")
|
result = client.find_agent_issues("repo", "claude")
|
||||||
|
|
||||||
assert len(result) == 0
|
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