70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from .models import ApprovalContext, Review
|
|
|
|
|
|
class ValidationDenied(RuntimeError):
|
|
def __init__(self, code: str) -> None:
|
|
super().__init__(code)
|
|
self.code = code
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidationResult:
|
|
owner_review: Review
|
|
machine_contexts: tuple[str, ...]
|
|
|
|
|
|
class SingleOwnerApprovalValidator:
|
|
"""Pure candidate validator. It has no Gitea or production write path."""
|
|
|
|
def validate(self, context: ApprovalContext) -> ValidationResult:
|
|
if context.now >= context.expires_at:
|
|
raise ValidationDenied("APPROVAL_EXPIRED")
|
|
if context.wip:
|
|
raise ValidationDenied("WIP_BLOCKED")
|
|
if context.requester == context.owner:
|
|
raise ValidationDenied("REQUESTER_SELF_APPROVAL")
|
|
if context.executor == context.owner:
|
|
raise ValidationDenied("EXECUTOR_SELF_APPROVAL")
|
|
|
|
approved = [review for review in context.reviews if review.state == "APPROVED"]
|
|
owner_reviews = [
|
|
review for review in approved
|
|
if review.reviewer == context.owner and review.reviewer_id == context.owner_id
|
|
]
|
|
if len(owner_reviews) != 1:
|
|
raise ValidationDenied("OWNER_APPROVAL_REQUIRED")
|
|
owner_review = owner_reviews[0]
|
|
if owner_review.stale:
|
|
raise ValidationDenied("STALE_OWNER_APPROVAL")
|
|
if owner_review.dismissed:
|
|
raise ValidationDenied("DISMISSED_OWNER_APPROVAL")
|
|
if owner_review.head_oid != context.current_head_oid:
|
|
raise ValidationDenied("OWNER_APPROVAL_NOT_LATEST_HEAD")
|
|
if owner_review.reviewed_at >= context.expires_at:
|
|
raise ValidationDenied("OWNER_APPROVAL_EXPIRED")
|
|
if any(
|
|
review.state == "APPROVED"
|
|
and review.reviewer != context.owner
|
|
and review.reviewer_id != context.owner_id
|
|
for review in context.reviews
|
|
):
|
|
# Non-owner approvals may exist for legacy mechanics, but never
|
|
# satisfy or replace the human authorization decision.
|
|
pass
|
|
|
|
gates = {gate.context: gate for gate in context.machine_gates}
|
|
if set(gates) != set(context.required_machine_contexts):
|
|
raise ValidationDenied("MACHINE_GATE_SET_MISMATCH")
|
|
for gate in gates.values():
|
|
if gate.state != "success":
|
|
raise ValidationDenied("MACHINE_GATE_FAILED")
|
|
if gate.head_oid != context.current_head_oid:
|
|
raise ValidationDenied("MACHINE_GATE_NOT_LATEST_HEAD")
|
|
if gate.observed_at >= context.expires_at:
|
|
raise ValidationDenied("MACHINE_GATE_EXPIRED")
|
|
return ValidationResult(owner_review, tuple(sorted(gates)))
|