"""Supplement Gitea native protection when review identity is not expressible.""" from __future__ import annotations from dataclasses import dataclass from typing import Any, Mapping, Sequence class MergeGateDenied(RuntimeError): pass @dataclass(frozen=True) class MergeGateDecision: allowed: bool owner_review_id: int required_contexts: tuple[str, ...] def validate_protected_merge( pull_request: Mapping[str, Any], reviews: Sequence[Mapping[str, Any]], statuses: Mapping[str, str], *, required_contexts: Sequence[str], unresolved_conversations: int, owner_login: str = "owner", owner_id: int = 2, ) -> MergeGateDecision: if pull_request.get("state") != "open" or pull_request.get("mergeable") is not True: raise MergeGateDenied("pull_request_not_mergeable") author = (pull_request.get("user") or {}).get("login") owner_review = None for review in reviews: user = review.get("user") or {} if review.get("state") == "APPROVED" and user.get("login") == owner_login and user.get("id") == owner_id: if author == owner_login or (pull_request.get("author_id") == owner_id): raise MergeGateDenied("pr_author_self_approval_denied") owner_review = review if owner_review is None: raise MergeGateDenied("owner_review_required") if unresolved_conversations != 0: raise MergeGateDenied("unresolved_conversations_block_merge") missing = [context for context in required_contexts if statuses.get(context) != "success"] if missing: raise MergeGateDenied("required_checks_missing:" + ",".join(missing)) return MergeGateDecision(True, int(owner_review.get("id", 0)), tuple(required_contexts))