sysadminandClaude Opus 4.8
eac7afe5cb
`gitea_assess_pr_sync_status` could permanently block a merge-ready PR whose head commit had no status contexts. Gitea's combined commit-status endpoint reports `state: pending` both when a check is executing and when the status-context collection is empty. The wrapper took that `state` verbatim, and `checks_required` defaulted to True with no production path ever setting it, so "no CI configured" was indistinguishable from "CI is running" and waiting could never resolve it. Root cause spans the whole dataflow, not one call site: * the wrapper read only the combined `state` and never counted contexts; its `checks_status="none"` fallback was unreachable for any truthy state * `pr_sync_status.assess_pr_sync_status` defaulted `checks_required=True` * the MCP wrapper exposed no derived `checks_required` and never passed one * the branch-protection reader fetched the payload carrying `enable_status_check` / `status_check_contexts` and discarded both Changes: * `pr_sync_status.py`: add `classify_commit_checks` plus explicit CHECKS_* classifications (success / failure / pending / none / not_required / missing_required / unknown). The combined state is recorded for observability but is never evidence that CI is executing. Aggregation is fail-closed and newest-wins per context. * `pr_sync_status.py`: rewrite the merge_now checks gate to consume the derived `checks_required`, distinguish the new classifications with precise blocker reasons, and fail closed on unrecognized vocabulary. The previous gate let any value outside its fixed vocabulary fall through to merge_now. * `gitea_mcp_server.py`: add `_branch_protection_policy` deriving both the current-base rule and the status-check requirement from one live read; `_branch_protection_requires_current_base` is retained as a thin accessor with unchanged semantics. Add `_commit_checks_snapshot` which returns the context collection alongside the combined state. * `gitea_mcp_server.py`: wire the derived `checks_required` into the production assessment path and report `checks_evidence`. `checks_required` is always derived from live evidence and is deliberately not a caller-supplied input, so no session can declare checks optional without proof. Live classification also outranks a caller-supplied `checks_status`, which can no longer mask a real required-check failure. An unreadable protection policy or status collection stays fail-closed. Approval, current-head, current-base, mergeability, conflict, role, lease and merge-authorization gates are unchanged. Tests: `tests/test_issue_751_checks_assessor.py` (40 tests). Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
603 lines
23 KiB
Python
603 lines
23 KiB
Python
"""Regression coverage for the PR checks assessor defect (#751).
|
|
|
|
Gitea's *combined* commit status reports ``state: pending`` both when a check is
|
|
executing and when the status-context collection is empty. The assessor read
|
|
``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head
|
|
had no status contexts — and never would — was routed to ``blocked`` forever.
|
|
|
|
These tests pin the corrected semantics end to end: live branch protection
|
|
decides whether checks are required, and the actual context collection decides
|
|
what the checks say.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import pr_sync_status # noqa: E402
|
|
from pr_sync_status import ( # noqa: E402
|
|
ACTION_BLOCKED,
|
|
ACTION_MERGE_NOW,
|
|
ACTION_UPDATE_BRANCH_BY_MERGE,
|
|
CHECKS_FAILURE,
|
|
CHECKS_MISSING_REQUIRED,
|
|
CHECKS_NONE,
|
|
CHECKS_NOT_REQUIRED,
|
|
CHECKS_PENDING,
|
|
CHECKS_SUCCESS,
|
|
CHECKS_UNKNOWN,
|
|
assess_pr_sync_status,
|
|
classify_commit_checks,
|
|
)
|
|
|
|
|
|
def _sha(prefix: str) -> str:
|
|
return (prefix + "0" * 40)[:40]
|
|
|
|
|
|
PR_HEAD = _sha("aaaaaaaa")
|
|
BASE_HEAD = _sha("bbbbbbbb")
|
|
|
|
|
|
def _base_kwargs(**overrides):
|
|
data = {
|
|
"host": "gitea.prgs.cc",
|
|
"org": "Scaled-Tech-Consulting",
|
|
"repo": "Gitea-Tools",
|
|
"pr_number": 751,
|
|
"pr_state": "open",
|
|
"source_branch": "fix/issue-751-checks-assessor",
|
|
"pr_head_sha": PR_HEAD,
|
|
"base_head_sha": BASE_HEAD,
|
|
"commits_behind": 0,
|
|
"mergeable": True,
|
|
"has_conflicts": False,
|
|
"branch_protection_requires_current_base": False,
|
|
"approval_at_current_head": True,
|
|
"checks_status": "success",
|
|
"active_author_lock": True,
|
|
"active_reviewer_lease": False,
|
|
"active_merger_lease": False,
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def _reasons(result) -> str:
|
|
return " | ".join(result["reasons"]).lower()
|
|
|
|
|
|
class TestClassifyCommitChecks(unittest.TestCase):
|
|
"""Pure classification from live evidence."""
|
|
|
|
def test_empty_collection_with_combined_pending_is_not_executing_ci(self):
|
|
# The exact PR #750 failure mode at the classification layer.
|
|
result = classify_commit_checks(
|
|
combined_state="pending",
|
|
statuses=[],
|
|
checks_enabled=True,
|
|
required_contexts=[],
|
|
)
|
|
self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
|
|
self.assertEqual(result["checks_status"], CHECKS_NONE)
|
|
self.assertEqual(result["context_count"], 0)
|
|
joined = " ".join(result["reasons"]).lower()
|
|
self.assertIn("empty", joined)
|
|
self.assertIn("does not indicate", joined)
|
|
|
|
def test_protection_disables_status_checks(self):
|
|
result = classify_commit_checks(
|
|
combined_state="pending",
|
|
statuses=[],
|
|
checks_enabled=False,
|
|
required_contexts=[],
|
|
)
|
|
self.assertFalse(result["checks_required"])
|
|
self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED)
|
|
|
|
def test_no_required_contexts_aggregates_reported_contexts(self):
|
|
result = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[{"context": "build", "status": "success"}],
|
|
checks_enabled=True,
|
|
required_contexts=[],
|
|
)
|
|
self.assertTrue(result["checks_required"])
|
|
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
|
|
|
def test_required_checks_pending(self):
|
|
result = classify_commit_checks(
|
|
combined_state="pending",
|
|
statuses=[{"context": "build", "status": "pending"}],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_PENDING)
|
|
|
|
def test_required_checks_failed(self):
|
|
result = classify_commit_checks(
|
|
combined_state="failure",
|
|
statuses=[{"context": "build", "status": "failure"}],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_FAILURE)
|
|
|
|
def test_required_checks_successful(self):
|
|
result = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[{"context": "build", "status": "success"}],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
|
self.assertTrue(result["checks_required"])
|
|
|
|
def test_required_context_configured_with_no_matching_result(self):
|
|
result = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[{"context": "lint", "status": "success"}],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
|
|
self.assertEqual(result["missing_required_contexts"], ["build"])
|
|
|
|
def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self):
|
|
# An unrelated failing context must not fail a satisfied required set.
|
|
result = classify_commit_checks(
|
|
combined_state="failure",
|
|
statuses=[
|
|
{"context": "build", "status": "success"},
|
|
{"context": "optional-scan", "status": "failure"},
|
|
],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
|
# ...and a failing *required* context still fails despite passing extras.
|
|
failing = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[
|
|
{"context": "build", "status": "failure"},
|
|
{"context": "optional-scan", "status": "success"},
|
|
],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(failing["checks_status"], CHECKS_FAILURE)
|
|
|
|
def test_policy_unreadable_fails_closed(self):
|
|
result = classify_commit_checks(
|
|
combined_state=None,
|
|
statuses=[],
|
|
checks_enabled=None,
|
|
required_contexts=[],
|
|
policy_determinable=False,
|
|
)
|
|
self.assertTrue(result["checks_required"])
|
|
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
|
self.assertIn("fail closed", " ".join(result["reasons"]).lower())
|
|
|
|
def test_malformed_policy_indeterminate_flag_fails_closed(self):
|
|
result = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[{"context": "build", "status": "success"}],
|
|
checks_enabled=None,
|
|
required_contexts=[],
|
|
policy_determinable=True,
|
|
)
|
|
self.assertTrue(result["checks_required"])
|
|
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
|
|
|
def test_status_collection_unreadable_fails_closed_when_required(self):
|
|
result = classify_commit_checks(
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
status_determinable=False,
|
|
)
|
|
self.assertTrue(result["checks_required"])
|
|
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
|
|
|
def test_unrecognized_context_state_never_reads_as_success(self):
|
|
result = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[{"context": "build", "status": "banana"}],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
|
|
|
def test_newest_wins_per_context(self):
|
|
# Gitea returns newest-first; the stale failure must not win.
|
|
result = classify_commit_checks(
|
|
combined_state="success",
|
|
statuses=[
|
|
{"context": "build", "status": "success"},
|
|
{"context": "build", "status": "failure"},
|
|
],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
|
|
|
def test_malformed_status_rows_are_discarded_not_treated_as_passing(self):
|
|
result = classify_commit_checks(
|
|
combined_state="pending",
|
|
statuses=[{"context": "build"}, "not-a-dict", None],
|
|
checks_enabled=True,
|
|
required_contexts=["build"],
|
|
)
|
|
self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
|
|
|
|
|
|
class TestChecksGateSemantics(unittest.TestCase):
|
|
"""``assess_pr_sync_status`` routing and blocker reasons."""
|
|
|
|
def test_not_required_allows_merge_now_with_empty_checks(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_NOT_REQUIRED),
|
|
checks_required=False,
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
self.assertTrue(result["approval_valid_for_merge"])
|
|
self.assertFalse(result["checks_required"])
|
|
self.assertIn("does not require status checks", _reasons(result))
|
|
|
|
def test_none_blocks_when_checks_required(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_NONE),
|
|
checks_required=True,
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertIn("no status context", _reasons(result))
|
|
self.assertIn("not executing ci", _reasons(result))
|
|
|
|
def test_missing_required_blocks(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED),
|
|
checks_required=True,
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertIn("no matching status result", _reasons(result))
|
|
|
|
def test_required_pending_blocks(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertIn("not finished", _reasons(result))
|
|
|
|
def test_required_failure_blocks(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertIn("failed", _reasons(result))
|
|
|
|
def test_required_success_merges(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
|
|
def test_unknown_blocks_when_required(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertIn("unknown", _reasons(result))
|
|
|
|
def test_unrecognized_status_does_not_fall_through_to_merge(self):
|
|
# Regression: the previous gate only handled a fixed vocabulary and let
|
|
# anything else reach merge_now.
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status="totally-unexpected"),
|
|
checks_required=True,
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertIn("unrecognized checks status", _reasons(result))
|
|
|
|
def test_checks_gate_does_not_bypass_approval_requirement(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False
|
|
),
|
|
checks_required=False,
|
|
)
|
|
self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
self.assertFalse(result["approval_valid_for_merge"])
|
|
|
|
def test_checks_gate_does_not_bypass_current_base_protection(self):
|
|
# Existing current-base behavior stays intact (#727).
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
checks_status=CHECKS_NOT_REQUIRED,
|
|
commits_behind=3,
|
|
branch_protection_requires_current_base=True,
|
|
),
|
|
checks_required=False,
|
|
)
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
|
|
)
|
|
|
|
def test_checks_gate_does_not_bypass_conflicts(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False),
|
|
checks_required=False,
|
|
)
|
|
self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
|
|
|
|
class TestBranchProtectionPolicy(unittest.TestCase):
|
|
"""Live protection reader derives the status-check requirement."""
|
|
|
|
def setUp(self):
|
|
import gitea_mcp_server as gms
|
|
|
|
self.gms = gms
|
|
|
|
def _policy(self, responses):
|
|
def fake_api_request(method, url, auth, *args, **kwargs):
|
|
for fragment, payload in responses.items():
|
|
if fragment in url:
|
|
if isinstance(payload, Exception):
|
|
raise payload
|
|
return payload
|
|
return None
|
|
|
|
with patch.object(self.gms, "api_request", side_effect=fake_api_request):
|
|
return self.gms._branch_protection_policy(
|
|
"https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master"
|
|
)
|
|
|
|
def test_status_checks_enabled_with_contexts(self):
|
|
policy = self._policy({
|
|
"branch_protections": [{
|
|
"branch_name": "master",
|
|
"block_on_outdated_branch": True,
|
|
"enable_status_check": True,
|
|
"status_check_contexts": ["ci/build", " "],
|
|
}],
|
|
})
|
|
self.assertTrue(policy["determinable"])
|
|
self.assertTrue(policy["checks_enabled"])
|
|
self.assertTrue(policy["requires_current_base"])
|
|
self.assertEqual(policy["required_contexts"], ["ci/build"])
|
|
|
|
def test_status_checks_disabled(self):
|
|
policy = self._policy({
|
|
"branch_protections": [{
|
|
"branch_name": "master",
|
|
"enable_status_check": False,
|
|
}],
|
|
})
|
|
self.assertTrue(policy["determinable"])
|
|
self.assertFalse(policy["checks_enabled"])
|
|
|
|
def test_no_protection_rule_means_checks_not_required(self):
|
|
# PR #750's repository shape: protection list readable but empty.
|
|
policy = self._policy({"branch_protections": [], "branches/master": {}})
|
|
self.assertTrue(policy["determinable"])
|
|
self.assertFalse(policy["protection_found"])
|
|
self.assertFalse(policy["checks_enabled"])
|
|
self.assertIsNone(policy["requires_current_base"])
|
|
|
|
def test_protection_without_status_field_means_not_enabled(self):
|
|
policy = self._policy({
|
|
"branch_protections": [{
|
|
"branch_name": "master",
|
|
"block_on_outdated_branch": True,
|
|
}],
|
|
})
|
|
self.assertTrue(policy["determinable"])
|
|
self.assertFalse(policy["checks_enabled"])
|
|
self.assertTrue(policy["requires_current_base"])
|
|
|
|
def test_api_failure_is_not_determinable(self):
|
|
policy = self._policy({
|
|
"branch_protections": RuntimeError("boom"),
|
|
})
|
|
self.assertFalse(policy["determinable"])
|
|
self.assertIsNone(policy["checks_enabled"])
|
|
|
|
def test_missing_branch_is_not_determinable(self):
|
|
with patch.object(self.gms, "api_request", return_value=[]):
|
|
policy = self.gms._branch_protection_policy(
|
|
"https://example/api/v1/repos/o/r", {}, base_branch=" "
|
|
)
|
|
self.assertFalse(policy["determinable"])
|
|
|
|
def test_requires_current_base_accessor_preserved(self):
|
|
def fake(method, url, auth, *a, **k):
|
|
if "branch_protections" in url:
|
|
return [{"branch_name": "master",
|
|
"block_on_outdated_branch": True}]
|
|
return None
|
|
|
|
with patch.object(self.gms, "api_request", side_effect=fake):
|
|
value = self.gms._branch_protection_requires_current_base(
|
|
"https://example/api/v1/repos/o/r", {}, base_branch="master"
|
|
)
|
|
self.assertTrue(value)
|
|
|
|
|
|
class TestCommitChecksSnapshot(unittest.TestCase):
|
|
def setUp(self):
|
|
import gitea_mcp_server as gms
|
|
|
|
self.gms = gms
|
|
|
|
def test_reads_state_and_context_collection(self):
|
|
payload = {"state": "pending", "statuses": [{"context": "b",
|
|
"status": "pending"}]}
|
|
with patch.object(self.gms, "api_request", return_value=payload):
|
|
snap = self.gms._commit_checks_snapshot(
|
|
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
|
)
|
|
self.assertTrue(snap["determinable"])
|
|
self.assertEqual(snap["combined_state"], "pending")
|
|
self.assertEqual(len(snap["statuses"]), 1)
|
|
|
|
def test_empty_collection_recorded_as_determinable(self):
|
|
with patch.object(
|
|
self.gms, "api_request", return_value={"state": "pending", "statuses": []}
|
|
):
|
|
snap = self.gms._commit_checks_snapshot(
|
|
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
|
)
|
|
self.assertTrue(snap["determinable"])
|
|
self.assertEqual(snap["statuses"], [])
|
|
|
|
def test_api_failure_is_not_determinable(self):
|
|
with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")):
|
|
snap = self.gms._commit_checks_snapshot(
|
|
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
|
)
|
|
self.assertFalse(snap["determinable"])
|
|
|
|
|
|
class TestMcpWrapperForwarding(unittest.TestCase):
|
|
"""The production MCP path must reach the derived ``checks_required``."""
|
|
|
|
def setUp(self):
|
|
import gitea_mcp_server as gms
|
|
|
|
self.gms = gms
|
|
self.base = (
|
|
"https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools"
|
|
)
|
|
|
|
def _patches(self, fake_api_request, spy):
|
|
return [
|
|
patch.object(self.gms, "_profile_operation_gate", return_value=None),
|
|
patch.object(self.gms, "_resolve", return_value=(
|
|
"gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")),
|
|
patch.object(self.gms, "_auth", return_value={"Authorization": "x"}),
|
|
patch.object(self.gms, "repo_api_url", return_value=self.base),
|
|
patch.object(self.gms, "api_request", side_effect=fake_api_request),
|
|
patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={
|
|
"success": True, "approval_at_current_head": True}),
|
|
patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={
|
|
"has_author_lock": True}),
|
|
patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy),
|
|
]
|
|
|
|
def _run(self, *, protections, status_payload, pr_number=750,
|
|
caller_checks_status=None):
|
|
captured = {}
|
|
real_assess = pr_sync_status.assess_pr_sync_status
|
|
|
|
def spy(**kwargs):
|
|
captured.update(kwargs)
|
|
return real_assess(**kwargs)
|
|
|
|
def fake_api_request(method, url, auth, *args, **kwargs):
|
|
if f"/pulls/{pr_number}" in url:
|
|
return {
|
|
"number": pr_number,
|
|
"state": "open",
|
|
"title": "t",
|
|
"body": "b",
|
|
"mergeable": True,
|
|
"head": {"sha": PR_HEAD, "ref": "fix/x"},
|
|
"base": {"sha": BASE_HEAD, "ref": "master"},
|
|
}
|
|
if "/branch_protections" in url:
|
|
if isinstance(protections, Exception):
|
|
raise protections
|
|
return protections
|
|
if "/branches/master" in url:
|
|
return {"commit": {"id": BASE_HEAD}}
|
|
if "/status" in url:
|
|
if isinstance(status_payload, Exception):
|
|
raise status_payload
|
|
return status_payload
|
|
if "/compare/" in url:
|
|
return {"total_commits": 0}
|
|
if "/comments" in url:
|
|
return []
|
|
return None
|
|
|
|
stack = self._patches(fake_api_request, spy)
|
|
for p in stack:
|
|
p.start()
|
|
try:
|
|
result = self.gms.gitea_assess_pr_sync_status(
|
|
pr_number=pr_number, remote="prgs",
|
|
checks_status=caller_checks_status,
|
|
)
|
|
finally:
|
|
for p in reversed(stack):
|
|
p.stop()
|
|
return result, captured
|
|
|
|
def test_derived_checks_required_is_forwarded(self):
|
|
result, captured = self._run(
|
|
protections=[{"branch_name": "master", "enable_status_check": True,
|
|
"status_check_contexts": ["ci/build"]}],
|
|
status_payload={"state": "pending", "statuses": [
|
|
{"context": "ci/build", "status": "pending"}]},
|
|
)
|
|
self.assertIn("checks_required", captured)
|
|
self.assertTrue(captured["checks_required"])
|
|
self.assertEqual(captured["checks_status"], CHECKS_PENDING)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
|
|
def test_pr750_empty_status_with_no_protection_is_merge_ready(self):
|
|
# The exact reported failure: combined pending, zero contexts, and no
|
|
# branch protection requiring checks.
|
|
result, captured = self._run(
|
|
protections=[],
|
|
status_payload={"state": "pending", "statuses": []},
|
|
)
|
|
self.assertFalse(captured["checks_required"])
|
|
self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED)
|
|
self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
|
|
def test_protection_read_failure_blocks(self):
|
|
result, captured = self._run(
|
|
protections=RuntimeError("protection unavailable"),
|
|
status_payload={"state": "success", "statuses": []},
|
|
)
|
|
self.assertTrue(captured["checks_required"])
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
|
|
def test_caller_supplied_status_cannot_mask_live_required_failure(self):
|
|
# A caller-declared "success" must not override live evidence.
|
|
result, captured = self._run(
|
|
protections=[{"branch_name": "master", "enable_status_check": True,
|
|
"status_check_contexts": ["ci/build"]}],
|
|
status_payload={"state": "failure", "statuses": [
|
|
{"context": "ci/build", "status": "failure"}]},
|
|
caller_checks_status="success",
|
|
)
|
|
self.assertEqual(captured["checks_status"], CHECKS_FAILURE)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertEqual(
|
|
result["checks_evidence"]["caller_supplied_checks_status"], "success"
|
|
)
|
|
|
|
def test_checks_evidence_is_reported_without_secrets(self):
|
|
result, _ = self._run(
|
|
protections=[],
|
|
status_payload={"state": "pending", "statuses": []},
|
|
)
|
|
evidence = result["checks_evidence"]
|
|
self.assertEqual(evidence["context_count"], 0)
|
|
self.assertEqual(evidence["combined_state"], "pending")
|
|
self.assertFalse(evidence["protection_found"])
|
|
blob = repr(result).lower()
|
|
self.assertNotIn("authorization", blob)
|
|
self.assertNotIn("token", blob)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|