176 lines
6.7 KiB
Python
176 lines
6.7 KiB
Python
"""Fail-closed server gates for autonomous exact-head PR merging."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
import pytest
|
|
|
|
from packages.connector_kit import GiteaError
|
|
from tests.test_gitea_connector import connector, context
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutate", "code"),
|
|
[
|
|
(lambda fixture: fixture._action_run.update(event="workflow_dispatch"),
|
|
"GITEA_PR_VALIDATION_REQUIRED"),
|
|
(lambda fixture: fixture._action_run.update(head_sha="d" * 40),
|
|
"GITEA_PR_VALIDATION_REQUIRED"),
|
|
(lambda fixture: fixture._statuses[0].update(status="failure"),
|
|
"GITEA_PR_CHECK_FAILED"),
|
|
(lambda fixture: fixture._statuses[0].update(status="pending"),
|
|
"GITEA_PR_CHECK_PENDING"),
|
|
(lambda fixture: fixture._reverse_compare_commits.append({"sha": "b" * 40}),
|
|
"GITEA_BRANCH_OUTDATED"),
|
|
(lambda fixture: fixture._branches[0]["commit"].update(id="b" * 40),
|
|
"GITEA_BASE_CHANGED"),
|
|
(lambda fixture: fixture._reviews.append({
|
|
"id": 12, "state": "REQUEST_CHANGES", "user": {"login": "reviewer"}}),
|
|
"GITEA_BLOCKING_REVIEW"),
|
|
(lambda fixture: fixture._branch_protections["acme/hub"][0].update(
|
|
required_approvals=1), "GITEA_APPROVAL_REQUIRED"),
|
|
(lambda fixture: fixture._branch_protections["acme/hub"][0].update(
|
|
status_check_contexts=[
|
|
"Managed validation / full (pull_request)", "Extra required check"
|
|
]), "GITEA_REQUIRED_CHECK_BLOCKED"),
|
|
(lambda fixture: fixture._branch_protections["acme/hub"][0].update(
|
|
enable_push=True), "GITEA_PROTECTION_DRIFT"),
|
|
(lambda fixture: fixture._branch_protections["acme/hub"][0].update(
|
|
enable_force_push=True), "GITEA_PROTECTION_DRIFT"),
|
|
],
|
|
)
|
|
def test_merge_rejects_specific_blocker(mutate, code: str) -> None:
|
|
connector_instance, fixture = connector()
|
|
mutate(fixture)
|
|
try:
|
|
with pytest.raises(GiteaError) as captured:
|
|
connector_instance.pull_request_merge(
|
|
context(), "hub-repository", 7, "e" * 40, "squash"
|
|
)
|
|
assert captured.value.code == code
|
|
assert fixture._pulls[0].get("merged") is not True
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
def test_zero_required_approvals_and_green_exact_pr_run_merge() -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
merged = connector_instance.pull_request_merge(
|
|
context(), "hub-repository", 7, "e" * 40, "squash"
|
|
)["data"]
|
|
assert merged["validationRunId"] == 42
|
|
assert merged["merged"] is True
|
|
assert merged["resultingMainSha"] == "6" * 40
|
|
assert merged["auditEvidence"]["validationEvent"] == "pull_request"
|
|
assert merged["auditEvidence"]["resultingMainSha"] == "6" * 40
|
|
assert fixture._pulls[0]["merged"] is True
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
def test_additional_required_context_is_checked_on_exact_head() -> None:
|
|
connector_instance, fixture = connector()
|
|
fixture._branch_protections["acme/hub"][0]["status_check_contexts"].append(
|
|
"Extra required check"
|
|
)
|
|
fixture._statuses.append({
|
|
"context": "Extra required check", "status": "success", "target_url": "",
|
|
})
|
|
try:
|
|
merged = connector_instance.pull_request_merge(
|
|
context(), "hub-repository", 7, "e" * 40, "squash"
|
|
)["data"]
|
|
assert merged["merged"] is True
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
def test_change_status_uses_the_same_pr_gate_as_merge() -> None:
|
|
connector_instance, fixture = connector()
|
|
fixture._branches.append(
|
|
{"name": "codex/workflows", "commit": {"id": "e" * 40}, "protected": False}
|
|
)
|
|
fixture._action_run["event"] = "workflow_dispatch"
|
|
try:
|
|
status = connector_instance.change_status(
|
|
context(), "hub-repository", "codex/workflows"
|
|
)["data"]
|
|
assert status["readyToMerge"] is False
|
|
assert status["validationState"] == "pr_check_missing"
|
|
assert status["mergeEligibility"]["reason"] == "GITEA_PR_VALIDATION_REQUIRED"
|
|
assert "ensure-pr-validation" in status["nextActions"]
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
def test_validation_ensure_reuses_successful_exact_pr_check() -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
result = connector_instance.pull_request_validation_ensure(
|
|
context(), "hub-repository", 7, "e" * 40
|
|
)["data"]
|
|
assert result["state"] == "success"
|
|
assert result["runId"] == 42
|
|
assert result["event"] == "pull_request"
|
|
assert result["reused"] is True
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
def test_validation_ensure_reopens_pr_to_create_real_pr_event() -> None:
|
|
connector_instance, fixture = connector()
|
|
fixture._action_run["event"] = "workflow_dispatch"
|
|
try:
|
|
result = connector_instance.pull_request_validation_ensure(
|
|
context(), "hub-repository", 7, "e" * 40
|
|
)["data"]
|
|
assert result["state"] == "pending"
|
|
assert result["triggerMethod"] == "reopen_pull_request"
|
|
assert result["event"] == "pull_request"
|
|
assert fixture._pulls[0]["state"] == "open"
|
|
assert fixture._action_run["event"] == "pull_request"
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
def test_validation_ensure_reruns_failed_pr_run() -> None:
|
|
connector_instance, fixture = connector()
|
|
fixture._action_run.update(status="completed", conclusion="failure")
|
|
try:
|
|
result = connector_instance.pull_request_validation_ensure(
|
|
context(), "hub-repository", 7, "e" * 40
|
|
)["data"]
|
|
assert result["state"] == "pending"
|
|
assert result["triggerMethod"] == "rerun_pull_request"
|
|
assert fixture._action_run["status"] == "queued"
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
@pytest.mark.parametrize("moving_ref", ["head", "base"])
|
|
def test_merge_rechecks_head_and_base_before_write(moving_ref: str) -> None:
|
|
connector_instance, fixture = connector()
|
|
original_pull = connector_instance.client.pull_request
|
|
calls = 0
|
|
|
|
def moving_pull(repository, number):
|
|
nonlocal calls
|
|
calls += 1
|
|
result = copy.deepcopy(original_pull(repository, number))
|
|
if calls == 2:
|
|
result[moving_ref]["sha"] = "f" * 40
|
|
return result
|
|
|
|
connector_instance.client.pull_request = moving_pull
|
|
try:
|
|
with pytest.raises(GiteaError) as captured:
|
|
connector_instance.pull_request_merge(
|
|
context(), "hub-repository", 7, "e" * 40, "squash"
|
|
)
|
|
assert captured.value.code == "GITEA_CONCURRENT_MODIFICATION"
|
|
assert fixture._pulls[0].get("merged") is not True
|
|
finally:
|
|
connector_instance.client.close()
|