269 lines
11 KiB
Python
269 lines
11 KiB
Python
"""Future release endpoint: contract tests for cloud/release_gate.
|
|
|
|
These tests pin the CONTRACT, not an endpoint: there is no live
|
|
release endpoint in this codebase (building one here would be
|
|
the "new live endpoint touching production" this track forbids),
|
|
so every test calls ``evaluate_release`` directly with a local
|
|
SQLite chain. Nothing here deploys, publishes, or touches a
|
|
production flow; the chain lives in a TemporaryDirectory and is
|
|
deleted with the test.
|
|
|
|
What is pinned (and nowhere else):
|
|
|
|
- A blocked audit pushback blocks release (409, audit code),
|
|
even when the fresh hard gate passes.
|
|
- A passing re-audit unblocks release -- but only with a FRESH
|
|
passing hard gate supplied by the caller; a merge-time 200 is
|
|
never an input, and a missing hard gate blocks fail-closed.
|
|
- A chain-read failure blocks fail-closed (state unavailable),
|
|
distinguishable from an audit block and from a hard-gate block.
|
|
- A hard-gate failure blocks release even when the chain is
|
|
clear, and both sources are listed when both block.
|
|
- CORRECTED-to-pass does NOT unblock: a meta-review correction
|
|
to ``pass`` with no new passing conclusion stays blocked; only
|
|
a fresh passing verdict unblocks (the #247/#244 semantic,
|
|
re-checked at release time through the same reader).
|
|
- A fix alone does not unblock.
|
|
- A merge-time pass is not reused: a chain that was clear at
|
|
"merge time" and is blocked by the time release is evaluated
|
|
blocks release (the evaluator re-reads the chain every call).
|
|
- The evaluator is read-only: evaluating never moves the chain
|
|
head (no row is appended by a check).
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.audit_pushback import GATE_CHECKS # noqa: F401
|
|
from cloud.audit_wiring import pushback_package_from_gate
|
|
from cloud.audit_wiring import record_audit_verdict
|
|
from cloud.audit_wiring import record_fix
|
|
from cloud.audit_wiring import record_meta_review
|
|
from cloud.audit_wiring import record_pushback
|
|
from cloud.audit_wiring import record_reaudit
|
|
from cloud.ledger import LedgerStore
|
|
from cloud.merge_gate_hard import evaluate_hard_merge_gate
|
|
from cloud.release_gate import evaluate_release
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.audit_pushback import GATE_CHECKS # noqa: F401
|
|
from cloud.audit_wiring import pushback_package_from_gate
|
|
from cloud.audit_wiring import record_audit_verdict
|
|
from cloud.audit_wiring import record_fix
|
|
from cloud.audit_wiring import record_meta_review
|
|
from cloud.audit_wiring import record_pushback
|
|
from cloud.audit_wiring import record_reaudit
|
|
from cloud.ledger import LedgerStore
|
|
from cloud.merge_gate_hard import evaluate_hard_merge_gate
|
|
from cloud.release_gate import evaluate_release
|
|
|
|
MODULE = "apps/web"
|
|
CONTRACT = "v2"
|
|
|
|
|
|
def _passing_gate():
|
|
return evaluate_hard_merge_gate(
|
|
files_changed=["apps/web/a.py"], owned_dir="apps/web",
|
|
protection="protected", ci_status="success",
|
|
ci_checks=[{"name": "ci", "status": "success"}],
|
|
required_checks=["ci"], contract_version="v2",
|
|
contract_frozen=True, coordinator_approved=True,
|
|
unresolved_review_comments=0)
|
|
|
|
|
|
def _failing_gate():
|
|
return evaluate_hard_merge_gate(
|
|
files_changed=["apps/web/a.py"], owned_dir="apps/web",
|
|
protection="protected", ci_status="pending",
|
|
ci_checks=[{"name": "ci", "status": "pending"}],
|
|
required_checks=["ci"], contract_version="v2",
|
|
contract_frozen=True, coordinator_approved=True,
|
|
unresolved_review_comments=0)
|
|
|
|
|
|
class ReleaseGateTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.store = LedgerStore(self.db_path)
|
|
self.task = "task-1"
|
|
self.project = "proj-1"
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.store.close()
|
|
except Exception:
|
|
pass
|
|
self.tmp.cleanup()
|
|
|
|
def _conn(self):
|
|
return sqlite3.connect(self.db_path)
|
|
|
|
def _decide(self, hard_gate=None, task_status="merged"):
|
|
conn = self._conn()
|
|
try:
|
|
return evaluate_release(
|
|
self.task, conn, hard_gate=hard_gate,
|
|
task_status=task_status)
|
|
finally:
|
|
conn.close()
|
|
|
|
def _block_task(self):
|
|
record_audit_verdict(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
"fail", "audit", failures=["ci_green"])
|
|
package = pushback_package_from_gate(
|
|
_failing_gate(), self.task, 0)
|
|
record_pushback(self.store, self.task, self.project, MODULE,
|
|
CONTRACT, 1, package["failed_checks"], package,
|
|
failure_source="audit")
|
|
return package
|
|
|
|
def _head(self):
|
|
return self.store.verify()["head_hash"]
|
|
|
|
def test_audit_pushback_blocks_release_even_with_passing_gate(self):
|
|
self._block_task()
|
|
decision = self._decide(hard_gate=_passing_gate())
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["http_status"], 409)
|
|
self.assertEqual(decision["code"],
|
|
"release_blocked_audit_pushback")
|
|
self.assertEqual(decision["blocked_by"], "audit_pushback")
|
|
self.assertIn("release_blocked_audit_pushback",
|
|
decision["blocked_sources"])
|
|
self.assertIn("ci_green", decision["failed_checks"])
|
|
self.assertEqual(decision["round"], 1)
|
|
|
|
def test_reaudit_pass_allows_release_with_fresh_passing_gate(self):
|
|
package = self._block_task()
|
|
record_reaudit(self.store, self.task, self.project, MODULE,
|
|
CONTRACT, 1, {"verdict": "pass"},
|
|
package=package)
|
|
decision = self._decide(hard_gate=_passing_gate())
|
|
self.assertTrue(decision["allowed"])
|
|
self.assertEqual(decision["http_status"], 200)
|
|
self.assertEqual(decision["code"], "release_allowed")
|
|
self.assertIsNone(decision["blocked_by"])
|
|
self.assertEqual(decision["blocked_sources"], [])
|
|
|
|
def test_corrected_to_pass_still_blocks_until_fresh_pass(self):
|
|
# One failing verdict is the ONLY block here, so the
|
|
# correction below is the only thing that could unblock
|
|
# -- if it did, this test would catch it. (A second fail
|
|
# or an open pushback would mask a wrong unblock.)
|
|
verdict = record_audit_verdict(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
"fail", "audit", failures=["ci_green"])
|
|
self.assertFalse(
|
|
self._decide(hard_gate=_passing_gate())["allowed"])
|
|
# Audit-of-audit correction: the original fail was wrong,
|
|
# corrected to pass. That schedules a re-audit; it does
|
|
# NOT release the gate by itself.
|
|
record_meta_review(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
verdict["id"], "fail", "pass", True, reviewer="auditor-b")
|
|
decision = self._decide(hard_gate=_passing_gate())
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["code"],
|
|
"release_blocked_audit_pushback")
|
|
# Only a fresh passing verdict unblocks.
|
|
record_audit_verdict(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
"pass", "audit")
|
|
decision = self._decide(hard_gate=_passing_gate())
|
|
self.assertTrue(decision["allowed"], decision)
|
|
|
|
def test_hard_gate_blocks_release_when_chain_is_clear(self):
|
|
record_audit_verdict(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
"pass", "audit")
|
|
decision = self._decide(hard_gate=_failing_gate())
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["http_status"], 409)
|
|
self.assertEqual(decision["code"], "release_blocked_hard_gate")
|
|
self.assertEqual(decision["blocked_by"], "hard_gate")
|
|
self.assertTrue(decision["blocks"])
|
|
self.assertNotIn("release_blocked_audit_pushback",
|
|
decision["blocked_sources"])
|
|
|
|
def test_both_sources_listed_when_both_block(self):
|
|
self._block_task()
|
|
decision = self._decide(hard_gate=_failing_gate())
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertIn("release_blocked_hard_gate",
|
|
decision["blocked_sources"])
|
|
self.assertIn("release_blocked_audit_pushback",
|
|
decision["blocked_sources"])
|
|
self.assertTrue(decision["blocks"])
|
|
self.assertTrue(decision["failed_checks"])
|
|
|
|
def test_chain_read_failure_fails_closed(self):
|
|
conn = self._conn()
|
|
conn.close()
|
|
decision = evaluate_release(
|
|
self.task, conn, hard_gate=_passing_gate(),
|
|
task_status="merged")
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["http_status"], 409)
|
|
self.assertEqual(decision["code"],
|
|
"release_blocked_state_unavailable")
|
|
self.assertEqual(
|
|
decision["blocked_by"],
|
|
"audit_pushback_state_unavailable")
|
|
|
|
def test_missing_hard_gate_evidence_blocks(self):
|
|
record_audit_verdict(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
"pass", "audit")
|
|
decision = self._decide(hard_gate=None)
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["code"],
|
|
"release_hard_gate_evidence_missing")
|
|
|
|
def test_unmerged_task_cannot_release(self):
|
|
record_audit_verdict(
|
|
self.store, self.task, self.project, MODULE, CONTRACT,
|
|
"pass", "audit")
|
|
decision = self._decide(hard_gate=_passing_gate(),
|
|
task_status="completed")
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["code"], "release_requires_merged")
|
|
|
|
def test_merge_time_pass_is_not_reused_at_release(self):
|
|
# "Merge time": chain clear, fresh gate passes.
|
|
first = self._decide(hard_gate=_passing_gate())
|
|
self.assertTrue(first["allowed"], first)
|
|
# Between merge and release, a pushback lands on the chain.
|
|
self._block_task()
|
|
second = self._decide(hard_gate=_passing_gate())
|
|
self.assertFalse(second["allowed"])
|
|
self.assertEqual(second["code"],
|
|
"release_blocked_audit_pushback")
|
|
|
|
def test_fix_alone_does_not_unblock_release(self):
|
|
package = self._block_task()
|
|
record_fix(self.store, self.task, self.project, MODULE,
|
|
CONTRACT, 1,
|
|
{"summary": "fix ci", "diff_sha256": "b" * 64,
|
|
"files_changed": ["apps/web/a.py"]}, package)
|
|
decision = self._decide(hard_gate=_passing_gate())
|
|
self.assertFalse(decision["allowed"])
|
|
|
|
def test_evaluator_is_read_only(self):
|
|
self._block_task()
|
|
before = self._head()
|
|
self._decide(hard_gate=_passing_gate())
|
|
self._decide(hard_gate=_failing_gate())
|
|
self.assertEqual(self._head(), before)
|
|
self.assertTrue(self.store.verify()["ok"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|