"""Release gate: contract tests for cloud/release_gate (adapter). These tests pin the CONTRACT, not the endpoint: the live endpoint (``POST /tasks/{id}/release``) calls the wiring evaluator directly, and every test here calls the ``release_gate`` adapter ``evaluate_release`` directly with a local SQLite chain to prove the adapter translates the single wiring implementation into the legacy shape without a second evaluator. 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()