codex-app-server/cloud/tests/test_audit_wiring.py

"""Audit-gap-1 wiring: track 3 (pushback) <-> track 1 (hard gate)
<-> track 4 (hash-chain ledger), through cloud/audit_wiring.py.

What is pinned here (and nowhere else):

- CONDITIONAL maps to fail at the wiring boundary, original kept.
- Hard-gate 409 blocks codes become pushback-package failed
  checks (branch protection travels as its own extra, never
  dropped, never disguised as one of the ten gate checks).
- Every pushback / fix / re-audit / rebuild / escalation step
  lands on the hash chain, and the chain still verifies.
- The server ANDs the chain's pushback block with the hard
  gate's five checks: audit fail -> merge 409, status
  unchanged, chain intact -> fix -> re-audit pass -> merge 200.
  This chain runs against the real HTTP endpoint, not just
  pure functions.
- The QA full chain is covered separately: a package carrying
  QA failed links cannot be re-audited to a pass until exactly
  those links have been re-run passing.
- The resolved meta-review tension: a correction to fail
  re-blocks at once, but a correction to pass does NOT unblock
  by itself (track 3: schedule a re-audit, not release the
  gate); only a fresh passing verdict / re-audit unblocks.
  Track 4's failure counts still use effective verdicts, so
  the two views may disagree while a re-audit is pending --
  that disagreement is asserted, not papered over.
"""

import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import unittest
from pathlib import Path

try:
    from cloud.audit_pushback import GATE_CHECKS
    from cloud.audit_pushback import decide_pushback
    from cloud.audit_wiring import gate_failed_checks
    from cloud.audit_wiring import normalize_re_review_result
    from cloud.audit_wiring import normalize_verdict
    from cloud.audit_wiring import pushback_block_for_task
    from cloud.audit_wiring import pushback_package_from_gate
    from cloud.audit_wiring import record_audit_verdict
    from cloud.audit_wiring import record_escalation
    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.audit_wiring import record_rebuild
    from cloud.audit_wiring import record_third_failure_assessment
    from cloud.auth import AuthStore
    from cloud.ledger import LedgerStore
    from cloud.ledger import audit_timeline
    from cloud.merge_gate_hard import evaluate_hard_merge_gate
    from cloud.server import create_server
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.audit_pushback import GATE_CHECKS
    from cloud.audit_pushback import decide_pushback
    from cloud.audit_wiring import gate_failed_checks
    from cloud.audit_wiring import normalize_re_review_result
    from cloud.audit_wiring import normalize_verdict
    from cloud.audit_wiring import pushback_block_for_task
    from cloud.audit_wiring import pushback_package_from_gate
    from cloud.audit_wiring import record_audit_verdict
    from cloud.audit_wiring import record_escalation
    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.audit_wiring import record_rebuild
    from cloud.audit_wiring import record_third_failure_assessment
    from cloud.auth import AuthStore
    from cloud.ledger import LedgerStore
    from cloud.ledger import audit_timeline
    from cloud.merge_gate_hard import evaluate_hard_merge_gate
    from cloud.server import create_server

MODULE = "apps/web"
CONTRACT = "v2"


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)


def _full_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_audit():
    checks = {name: True for name in GATE_CHECKS}
    checks["ci_green"] = False
    return {
        "task_id": "task-1",
        "verdict": "fail",
        "verdict_text": "audit fail: ci not green",
        "signer": "auditor-a",
        "executor": "executor-e",
        "coordinator_approver": "coordinator-c",
        "qa_tester": "qa-q",
        "security_verdict": "pass",
        "checks": checks,
        "failed_items": [],
        "deviations": [],
        "regression": {"items": [
            {"path": "apps/web/app.py", "sha256_manifest": "a" * 64,
             "sha256_recomputed": "a" * 64, "source": "exact"},
        ]},
        "demo": {"verified_runnable": True, "runs_app_entry": True,
                 "command": "python3 apps/web/app.py"},
    }


class VerdictNormalizationTestCase(unittest.TestCase):
    def test_conditional_maps_to_fail_and_keeps_original(self):
        normalized, original = normalize_verdict("conditional")
        self.assertEqual(normalized, "fail")
        self.assertEqual(original, "conditional")
        normalized, original = normalize_verdict("pass")
        self.assertEqual((normalized, original), ("pass", "pass"))
        normalized, original = normalize_verdict(None)
        self.assertEqual(normalized, "fail")
        self.assertIsNone(original)

    def test_re_review_conditional_maps_to_fail(self):
        result = normalize_re_review_result({"verdict": "conditional"})
        self.assertEqual(result["verdict"], "fail")
        self.assertEqual(result["original_verdict"], "conditional")
        result = normalize_re_review_result({"verdict": "pass"})
        self.assertEqual(result["verdict"], "pass")
        self.assertNotIn("original_verdict", result)


class GatePackageTestCase(unittest.TestCase):
    def test_all_block_families_map_and_branch_protection_survives(self):
        gate = evaluate_hard_merge_gate(
            files_changed=["apps/other/x.py"], owned_dir="apps/web",
            protection="unknown", ci_status="pending",
            ci_checks=[{"name": "ci", "status": "failure"}],
            required_checks=["ci"], contract_version=None,
            contract_frozen=False, coordinator_approved=None,
            unresolved_review_comments=["c1"])
        self.assertFalse(gate["allowed"])
        checks, extras = gate_failed_checks(gate)
        for name in ("ci_green", "directory_isolation",
                     "contract_version", "coordinator_approval",
                     "unresolved_review_comments"):
            self.assertIn(name, checks)
        self.assertIn("branch_protection", extras)
        package = pushback_package_from_gate(gate, "task-1", 0)
        self.assertIn("branch_protection", package["failed_checks"])
        self.assertIn("ci_green", package["failed_checks"])
        # Full re-audit criteria, not failed-items-only.
        self.assertEqual(package["reaudit_criteria"],
                         list(GATE_CHECKS))
        # No block code is lost in translation.
        self.assertTrue(package["failed_items"])

    def test_qa_links_carried_onto_package(self):
        links = [{"link": "checkout-flow", "case": "pay",
                  "failure": "500", "evidence": "log",
                  "reproduction": ["run checkout"]}]
        package = pushback_package_from_gate(
            _failing_gate(), "task-1", 0, qa_failed_links=links)
        self.assertEqual(package["qa_failed_links"], links)
        self.assertEqual(package["reaudit_must_rerun_links"], links)


class LedgerRecordingTestCase(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 _block(self):
        conn = sqlite3.connect(self.db_path)
        try:
            return pushback_block_for_task(conn, self.task)
        finally:
            conn.close()

    def _chain_ok(self):
        result = self.store.verify()
        self.assertTrue(result["ok"], result)
        return result

    def test_conditional_verdict_stored_as_fail_with_original(self):
        entry = record_audit_verdict(
            self.store, self.task, self.project, MODULE, CONTRACT,
            "conditional", "audit", failures=["ci_green"])
        timeline = audit_timeline(self.store.conn, task_id=self.task)
        verdict = [e for e in timeline["entries"]
                   if e["action"] == "audit_verdict"][0]
        self.assertEqual(verdict["parsed"]["verdict"], "fail")
        self.assertEqual(verdict["parsed"]["extra"]["original_verdict"],
                         "conditional")
        self.assertEqual(entry["id"], verdict["id"])
        self._chain_ok()

    def test_pushback_fix_reaudit_cycle_moves_gate(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")
        block = self._block()
        self.assertTrue(block["merge_blocked"])
        self.assertTrue(block["release_blocked"])
        self.assertEqual(block["blocked_by"], "audit_pushback")
        self.assertEqual(block["round"], 1)
        self.assertIn("ci_green", block["failed_checks"])
        # A landed fix alone does not unblock.
        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)
        self.assertTrue(self._block()["merge_blocked"])
        # A passing re-audit unblocks; chain still verifies.
        record_reaudit(self.store, self.task, self.project, MODULE,
                       CONTRACT, 1, {"verdict": "pass"},
                       package=package)
        self.assertFalse(self._block()["merge_blocked"])
        self._chain_ok()

    def test_qa_guard_refuses_unrerun_or_failed_link_pass(self):
        links = [{"link": "checkout-flow", "case": "pay",
                  "failure": "500", "evidence": "log",
                  "reproduction": ["run checkout"]}]
        package = pushback_package_from_gate(
            _failing_gate(), self.task, 0, qa_failed_links=links)
        record_audit_verdict(self.store, self.task, self.project,
                             MODULE, CONTRACT, "fail", "audit",
                             failures=["ci_green"])
        record_pushback(self.store, self.task, self.project, MODULE,
                        CONTRACT, 1, package["failed_checks"], package,
                        failure_source="audit")
        before = self.store.verify()["count"]
        with self.assertRaises(ValueError):
            record_reaudit(self.store, self.task, self.project,
                           MODULE, CONTRACT, 1, {"verdict": "pass"},
                           package=package, rerun_results={})
        with self.assertRaises(ValueError):
            record_reaudit(self.store, self.task, self.project,
                           MODULE, CONTRACT, 1, {"verdict": "pass"},
                           package=package,
                           rerun_results={"checkout-flow": False})
        self.assertEqual(self.store.verify()["count"], before)
        self.assertTrue(self._block()["merge_blocked"])
        record_reaudit(self.store, self.task, self.project, MODULE,
                       CONTRACT, 1, {"verdict": "pass"},
                       package=package,
                       rerun_results={"checkout-flow": True})
        self.assertFalse(self._block()["merge_blocked"])

    def test_rebuild_alone_does_not_unblock(self):
        record_audit_verdict(self.store, self.task, self.project,
                             MODULE, CONTRACT, "fail", "audit",
                             failures=["ci_green"])
        record_rebuild(self.store, self.task, self.project, MODULE,
                       CONTRACT, 1, {"commit": "old123"},
                       root_cause="implementation", status="decided")
        self.assertTrue(self._block()["merge_blocked"])
        record_rebuild(self.store, self.task, self.project, MODULE,
                       CONTRACT, 1, {"commit": "old123"},
                       new_impl_ref={"commit": "new456"},
                       root_cause="implementation", status="rebuilt",
                       retained_assets={"old_tests_status": "pass"})
        self.assertTrue(self._block()["merge_blocked"])
        record_audit_verdict(self.store, self.task, self.project,
                             MODULE, CONTRACT, "pass", "audit")
        self.assertFalse(self._block()["merge_blocked"])
        self._chain_ok()

    def test_escalation_blocks_sample_does_not(self):
        record_escalation(self.store, self.task, self.project,
                          MODULE, CONTRACT, "blocked_quota",
                          target="coordinator",
                          reason="fix exceeds quota", round_no=2,
                          failed_checks=["ci_green"])
        self.assertTrue(self._block()["merge_blocked"])
        record_escalation(self.store, self.task, self.project,
                          MODULE, CONTRACT, "audit_sample",
                          reason="every 10th record")
        self.assertTrue(self._block()["merge_blocked"])
        record_audit_verdict(self.store, self.task, self.project,
                             MODULE, CONTRACT, "pass", "audit")
        self.assertFalse(self._block()["merge_blocked"])

    def test_meta_review_tension_is_pinned(self):
        fail_entry = record_audit_verdict(
            self.store, self.task, self.project, MODULE, CONTRACT,
            "fail", "audit", failures=["ci_green"])
        self.assertTrue(self._block()["merge_blocked"])
        # Wrong block corrected to pass: gate STAYS blocked until
        # a fresh passing re-audit (track 3 restored action).
        record_meta_review(self.store, self.task, self.project,
                           MODULE, CONTRACT, fail_entry["id"],
                           "fail", "pass", True, "reviewer-r",
                           findings=["wrong call"],
                           triggered_process="re-review")
        self.assertTrue(self._block()["merge_blocked"])
        # ...while track 4 counts use the effective verdict.
        counts = self.store.module_failure_counts(MODULE, CONTRACT,
                                                  task_id=self.task)
        self.assertEqual(counts["total_fails"], 0)
        record_audit_verdict(self.store, self.task, self.project,
                             MODULE, CONTRACT, "pass", "audit")
        self.assertFalse(self._block()["merge_blocked"])
        # Wrong pass corrected to fail: re-blocks at once.
        pass_entry = record_audit_verdict(
            self.store, self.task, self.project, MODULE, CONTRACT,
            "pass", "audit")
        self.assertFalse(self._block()["merge_blocked"])
        record_meta_review(self.store, self.task, self.project,
                           MODULE, CONTRACT, pass_entry["id"],
                           "pass", "fail", True, "reviewer-r",
                           findings=["missed ci"],
                           triggered_process="re-review")
        self.assertTrue(self._block()["merge_blocked"])
        self._chain_ok()

    def test_modules_are_scoped_not_global(self):
        record_audit_verdict(self.store, self.task, self.project,
                             "apps/a", CONTRACT, "fail", "audit",
                             failures=["ci_green"])
        record_audit_verdict(self.store, self.task, self.project,
                             "apps/b", CONTRACT, "pass", "audit")
        self.assertTrue(self._block()["merge_blocked"])
        record_audit_verdict(self.store, self.task, self.project,
                             "apps/a", CONTRACT, "pass", "audit")
        self.assertFalse(self._block()["merge_blocked"])

    def test_triage_and_root_cause_mapping(self):
        record_third_failure_assessment(
            self.store, self.task, self.project, MODULE, CONTRACT,
            "targeted_fix",
            {"audit_conclusion": "nearly done",
             "failure_items": ["ci_green"],
             "passed_items": ["directory_isolation"]})
        timeline = audit_timeline(self.store.conn, task_id=self.task)
        assessment = [e for e in timeline["entries"]
                      if e["action"] == "audit_third_failure_assessment"][0]
        self.assertEqual(assessment["parsed"]["determination"],
                         "continuation_round")
        with self.assertRaises(ValueError):
            record_third_failure_assessment(
                self.store, self.task, self.project, MODULE, CONTRACT,
                "early_refreeze", {"audit_conclusion": "contract"})
        with self.assertRaises(ValueError):
            record_rebuild(self.store, self.task, self.project,
                           MODULE, CONTRACT, 1, {"commit": "old"},
                           root_cause="unknown")
        # Design root cause stores as contract_design.
        record_rebuild(self.store, self.task, self.project, MODULE,
                       CONTRACT, 1, {"commit": "old"},
                       root_cause="design", status="decided")
        timeline = audit_timeline(self.store.conn, task_id=self.task)
        rebuild = [e for e in timeline["entries"]
                   if e["action"] == "audit_rebuild"][0]
        self.assertEqual(rebuild["parsed"]["root_cause"],
                         "contract_design")
        self._chain_ok()

    def test_missing_ledger_table_is_unblocked_not_error(self):
        fresh = os.path.join(self.tmp.name, "empty.db")
        conn = sqlite3.connect(fresh)
        try:
            block = pushback_block_for_task(conn, "nope")
        finally:
            conn.close()
        self.assertFalse(block["merge_blocked"])
        self.assertIsNone(block["blocked_by"])

    def test_decide_pushback_round_feeds_ledger_round(self):
        decision = decide_pushback(_failing_audit(), 0)
        self.assertEqual(decision["action"], "pushback")
        self.assertEqual(decision["next_round"], 1)
        record_audit_verdict(self.store, self.task, self.project,
                             MODULE, CONTRACT, "fail", "audit",
                             failures=decision["failed_checks"])
        package = pushback_package_from_gate(
            _failing_gate(), self.task, decision["round"])
        record_pushback(self.store, self.task, self.project, MODULE,
                        CONTRACT, decision["next_round"],
                        package["failed_checks"], package,
                        failure_source="audit")
        block = self._block()
        self.assertTrue(block["merge_blocked"])
        self.assertEqual(block["round"], 1)


class EndpointChainTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        self.auth = AuthStore(self.db_path)
        self.owner_id, self.owner_key = self.auth.create_user("owner")
        self.server = create_server(self.db_path, "127.0.0.1", 0)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()
        status, self.project = self._req(
            "POST", "/projects", {"name": "demo"}, self.owner_key)
        self.assertEqual(status, 201)
        self._saved_env = {k: os.environ.get(k) for k in
                           ("CLOUD_MERGE_GATE_MODE",
                            "CLOUD_BRANCH_PROTECTION")}
        os.environ.pop("CLOUD_MERGE_GATE_MODE", None)
        os.environ.pop("CLOUD_BRANCH_PROTECTION", None)
        self.ledger = LedgerStore(self.db_path)

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)
        for key, value in self._saved_env.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
        try:
            self.ledger.close()
        except Exception:
            pass
        try:
            self.auth.conn.close()
        except Exception:
            pass
        self.tmp.cleanup()

    def _req(self, method, path, body=None, key=None):
        conn = http.client.HTTPConnection(
            "127.0.0.1", self.port, timeout=10)
        hdrs = {}
        payload = None
        if body is not None:
            payload = json.dumps(body).encode("utf-8")
            hdrs["Content-Type"] = "application/json"
        if key is not None:
            hdrs["X-API-Key"] = key
        try:
            conn.request(method, path, body=payload, headers=hdrs)
            resp = conn.getresponse()
            data = resp.read()
        finally:
            conn.close()
        try:
            parsed = json.loads(data.decode("utf-8")) if data else None
        except ValueError:
            parsed = None
        return resp.status, parsed

    def _completed_task(self):
        status, task = self._req(
            "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "x"},
            self.owner_key)
        self.assertEqual(status, 201)
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE tasks SET status = 'completed' WHERE id = ?",
                (task["id"],))
            conn.commit()
        finally:
            conn.close()
        return task

    def _hard_body(self, **overrides):
        body = {
            "files_changed": ["apps/web/a.py"],
            "owned_dir": "apps/web",
            "branch_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,
        }
        body.update(overrides)
        return body

    def _status_of(self, task_id):
        _s, got = self._req("GET", "/tasks/%s" % task_id,
                            key=self.owner_key)
        return got["status"]

    def _block_task(self, task, package=None):
        if package is None:
            package = pushback_package_from_gate(_failing_gate(),
                                                 task["id"], 0)
        record_audit_verdict(self.ledger, task["id"],
                             self.project["id"], MODULE, CONTRACT,
                             "fail", "audit",
                             failures=package["failed_checks"])
        record_pushback(self.ledger, task["id"], self.project["id"],
                        MODULE, CONTRACT, 1, package["failed_checks"],
                        package, failure_source="audit")
        return package

    def test_audit_fail_blocks_merge_fix_reaudit_releases(self):
        task = self._completed_task()
        package = self._block_task(task)
        status, blocked = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 409, blocked)
        self.assertIn("audit_pushback_blocked",
                      blocked["merge_gate"]["blocks"])
        self.assertTrue(
            blocked["merge_gate"]["audit_pushback"]["merge_blocked"])
        self.assertTrue(
            blocked["merge_gate"]["audit_pushback"]["release_blocked"])
        self.assertEqual(
            blocked["merge_gate"]["audit_pushback"]["blocked_by"],
            "audit_pushback")
        self.assertEqual(self._status_of(task["id"]), "completed")
        # Fix + passing re-audit (no QA links on this package).
        record_fix(self.ledger, task["id"], self.project["id"],
                   MODULE, CONTRACT, 1,
                   {"summary": "fix ci", "diff_sha256": "b" * 64,
                    "files_changed": ["apps/web/a.py"]}, package)
        status, still = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 409)
        record_reaudit(self.ledger, task["id"], self.project["id"],
                       MODULE, CONTRACT, 1, {"verdict": "pass"},
                       package=package)
        status, merged = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 200, merged)
        self.assertEqual(self._status_of(task["id"]), "merged")
        self.assertFalse(
            merged["merge_gate"]["audit_pushback"]["merge_blocked"])
        chain = self.ledger.verify()
        self.assertTrue(chain["ok"], chain)

    def test_qa_failed_links_gate_the_release(self):
        task = self._completed_task()
        links = [{"link": "checkout-flow", "case": "pay",
                  "failure": "500", "evidence": "log",
                  "reproduction": ["run checkout"]}]
        package = pushback_package_from_gate(
            _failing_gate(), task["id"], 0, qa_failed_links=links)
        self._block_task(task, package=package)
        status, _blocked = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 409)
        # Claimed pass without a passing re-run of the failed
        # link must not reach the chain, and must not unblock.
        with self.assertRaises(ValueError):
            record_reaudit(self.ledger, task["id"],
                           self.project["id"], MODULE, CONTRACT, 1,
                           {"verdict": "pass"}, package=package,
                           rerun_results={"checkout-flow": False})
        status, _still = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 409)
        record_reaudit(self.ledger, task["id"], self.project["id"],
                       MODULE, CONTRACT, 1, {"verdict": "pass"},
                       package=package,
                       rerun_results={"checkout-flow": True})
        status, merged = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 200, merged)

    def test_pushback_blocks_with_full_hard_evidence(self):
        # A1 改写：旧用例 test_pushback_blocks_even_in_record_mode
        # 已改名。本用例的性质不变——链上推回阻塞时，即使五项
        # 证据全齐也 409；只是不再经 record 模式证明“与模式无关”，
        # 因为服务端已无其它模式可比。
        task = self._completed_task()
        self._block_task(task)
        status, blocked = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(), self.owner_key)
        self.assertEqual(status, 409, blocked)
        self.assertIn("audit_pushback_blocked",
                      blocked["merge_gate"]["blocks"])
        self.assertEqual(self._status_of(task["id"]), "completed")

    def test_hard_gate_failure_without_pushback_has_no_pushback_code(self):
        task = self._completed_task()
        status, blocked = self._req(
            "POST", "/tasks/%s/merge" % task["id"],
            self._hard_body(ci_status="pending"), self.owner_key)
        self.assertEqual(status, 409, blocked)
        self.assertIn("ci_not_green", blocked["merge_gate"]["blocks"])
        self.assertNotIn("audit_pushback_blocked",
                         blocked["merge_gate"]["blocks"])
        self.assertFalse(
            blocked["merge_gate"]["audit_pushback"]["merge_blocked"])
        self.assertEqual(self._status_of(task["id"]), "completed")