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

"""gapfill2 H3：_merge_task 成功审计段里 rollback 自身的 except: pass。

上一批唯一未覆盖分支：G7 新段里 ``conn.rollback()`` 本身抛错时的
防御分支（基底行号 server.py:6615-6616）。rollback 从不失败，故
全量覆盖率里该分支一直缺失。本文件只用注入式测试覆盖它，绝不
改动 rollback 语义、绝不把该分支改成可观测行为：

- 注入一个 rollback 会抛错的连接代理（除 rollback 外全部透传
  真实连接），同时让审计失败（沿用 G7 的 RuntimeError 注入），
  使执行流必然走进 ``except Exception: pass``；
- 断言：rollback 的异常被吞掉（原审计错误原样上抛，不被掩盖）、
  任务状态仍是 completed、审计 0 行、原样重试后成功 200。

真实模型 0 轮，不联网；不跑真实 HTTP（handler 直调 +
真实 AuthStore/DB，审计行直接查库验证）。
"""

import os
import sqlite3
import sys
import tempfile
import time
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(REPO_ROOT))

import cloud.server as server_mod  # noqa: E402
from cloud.auth import AuthStore  # noqa: E402
from cloud.server import CloudControlPlaneHandler, init_db  # noqa: E402

NO_BLOCK = {"merge_blocked": False, "release_blocked": False,
            "blocked_by": None, "round": None, "failed_checks": []}

HARD_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,
}


class _RollbackFailingConn:
    """代理真实连接：除 rollback 外全部透传，rollback 计数后抛错。

    execute / commit / close / row_factory 等一律委托给真实连接，
    只有 rollback 被替换为“计数 + 抛错”，精确模拟“rollback 自身
    失败”这一防御分支的前提，不改动其它任何连接语义。
    """

    def __init__(self, real, calls):
        self._real = real
        self._calls = calls

    def rollback(self):
        self._calls["n"] += 1
        raise RuntimeError("rollback down")

    def __getattr__(self, name):
        return getattr(self._real, name)


class RollbackFailureSwallowedTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        init_db(self.db_path)
        self.auth = AuthStore(self.db_path)
        self.owner_id, _key = self.auth.create_user("owner")
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "INSERT INTO projects(id, name, owner_id, created_at,"
                " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
                ("p1", "P", self.owner_id, time.time(), None))
            conn.execute(
                "INSERT INTO tasks(id, project_id, prompt, status,"
                " created_by, created_at, updated_at)"
                " VALUES(?, ?, ?, ?, ?, ?, ?)",
                ("t1", "p1", "do work", "completed", self.owner_id,
                 time.time(), time.time()))
            conn.commit()
        finally:
            conn.close()
        self.auth.create_project("p1", self.owner_id)

    def tearDown(self):
        try:
            self.auth.conn.close()
        except Exception:
            pass
        self.tmp.cleanup()

    def _handler(self):
        h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
        h.server = SimpleNamespace(db_path=self.db_path)
        sent = []
        h._send_json = lambda status, obj, extra_headers=None: sent.append(
            (status, obj))
        h._read_json = lambda: (dict(HARD_BODY), None)
        h.sent = sent
        return h

    def _merge(self, h):
        with mock.patch.object(server_mod, "_pushback_block",
                               return_value=dict(NO_BLOCK)):
            h._merge_task(self.owner_id, "t1")

    def _status(self):
        conn = sqlite3.connect(self.db_path)
        try:
            return conn.execute(
                "SELECT status FROM tasks WHERE id = 't1'").fetchone()[0]
        finally:
            conn.close()

    def _merge_audits(self):
        return [a for a in self.auth.list_audit("p1")
                if a["action"] == "merge_task"]

    def test_rollback_failure_is_swallowed(self):
        # 审计失败 + rollback 自身失败：rollback 的异常必须被吞掉，
        # 原审计错误原样上抛（不被 rollback 错误掩盖），状态回滚
        # 语义不变（completed、审计 0 行），原样重试后成功。
        h = self._handler()
        real_db = CloudControlPlaneHandler._db
        calls = {"n": 0}

        def failing_db(handler_self):
            return _RollbackFailingConn(real_db(handler_self), calls)

        with mock.patch.object(CloudControlPlaneHandler, "_db",
                               failing_db):
            with mock.patch.object(
                    server_mod, "_audit_on_conn",
                    side_effect=RuntimeError("audit down")):
                with self.assertRaises(RuntimeError) as ctx:
                    self._merge(h)
        self.assertEqual(str(ctx.exception), "audit down")
        self.assertEqual(calls["n"], 1)
        self.assertEqual(h.sent, [])
        self.assertEqual(self._status(), "completed")
        self.assertEqual(self._merge_audits(), [])
        # 原样重试（审计恢复、连接恢复后）必须成功：状态仍是
        # completed，不是卡在 merged 导致重复合并 409。
        h2 = self._handler()
        self._merge(h2)
        self.assertEqual(h2.sent[-1][0], 200)
        self.assertEqual(self._status(), "merged")
        self.assertEqual(len(self._merge_audits()), 1)


if __name__ == "__main__":
    unittest.main()