How to Build a Branch Protection Audit Mock API in Python

import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

REPOSITORIES = {
    "alpha": {
        "default_branch": "main",
        "branches": ["main", "develop", "feature-x"],
        "protection_rules": {
            "main": {"required_reviews": 2, "dismiss_stale": True, "required_checks": ["CI"]},
            "develop": {"required_reviews": 1, "dismiss_stale": False, "required_checks": []},
            "feature-x": {"required_reviews": 0, "dismiss_stale": False, "required_checks": []}
        },
        "collaborators": ["alice", "bob", "carol"]
    },
    "beta": {
        "default_branch": "trunk",
        "branches": ["trunk", "release"],
        "protection_rules": {
            "trunk": {"required_reviews": 3, "dismiss_stale": True, "required_checks": ["lint", "test"]},
            "release": {"required_reviews": 1, "dismiss_stale": True, "required_checks": ["build"]}
        },
        "collaborators": ["dave", "erin"]
    }
}


class AuditHandler(BaseHTTPRequestHandler):

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path.strip("/")
        query = parse_qs(parsed.query)

        if not path:
            self._send_json(200, {"repositories": list(REPOSITORIES.keys())})
            return

        parts = path.split("/")
        if len(parts) == 1 and parts[0] in REPOSITORIES:
            repo_name = parts[0]
            branch_filter = query.get("branch", [None])[0]
            repo = REPOSITORIES[repo_name]
            if branch_filter:
                if branch_filter not in repo["protection_rules"]:
                    self._send_json(404, {"error": f"Branch '{branch_filter}' not found"})
                    return
                rules = {branch_filter: repo["protection_rules"][branch_filter]}
            else:
                rules = repo["protection_rules"]
            self._send_json(200, {"repository": repo_name, "protection_rules": rules})
            return

        if len(parts) == 2 and parts[0] == "audit":
            repo_name = parts[1]
            if repo_name not in REPOSITORIES:
                self._send_json(404, {"error": f"Repository '{repo_name}' not found"})
                return
            repo = REPOSITORIES[repo_name]
            audit_results = {}
            for branch, rules in repo["protection_rules"].items():
                issues = []
                if branch == repo["default_branch"]:
                    if rules["required_reviews"] < 2:
                        issues.append("default branch requires at least 2 reviews")
                    if not rules["required_checks"]:
                        issues.append("default branch missing required checks")
                else:
                    if not rules["required_checks"] and len(repo["branches"]) > 2:
                        issues.append("non-default branch lacks required checks")
                if rules["required_reviews"] == 0:
                    issues.append("no review requirement")
                audit_results[branch] = {
                    "status": "compliant" if not issues else "non-compliant",
                    "issues": issues,
                    "protection": rules
                }
            self._send_json(200, {"repository": repo_name, "audit_results": audit_results})
            return

        self._send_json(404, {"error": "Endpoint not found"})

    def _send_json(self, status, data):
        body = json.dumps(data, indent=2).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


def run_server(port=8000):
    server = HTTPServer(("127.0.0.1", port), AuditHandler)
    print(f"Mock API running on http://127.0.0.1:{port}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    run_server()