185 lines
7.0 KiB
Python
185 lines
7.0 KiB
Python
"""gapfill G7：_merge_task 状态提交与成功审计同一事务。
|
||
|
||
旧写法（A4 时序残留）：先 commit 状态为 merged、再另开
|
||
AuthStore 连接写 merge_task 审计；审计失败时状态已是
|
||
merged、响应却是 500——“合并了却报失败”，调用方无法靠
|
||
状态判断，也无法安全重试（重复合并 409）。
|
||
|
||
新写法：audit_on_conn 把审计对（audit_log 行 + 链条目）
|
||
写进同一连接、一次 commit；审计失败则整段回滚，状态仍是
|
||
completed、异常上抛后由 _dispatch_safe 转 500 JSON，调用方
|
||
可原样重试合并。链头唯一冲突的重试口径与 AuthStore.audit
|
||
完全一致（5 次、只重试 IntegrityError，其它错误直接抛）。
|
||
成功语义不变：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, audit_on_conn # 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 MergeAtomicTestCase(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_success_unchanged(self):
|
||
# 成功语义与旧写法一致：200、merged、审计有行、门禁在体里。
|
||
h = self._handler()
|
||
self._merge(h)
|
||
status, obj = h.sent[-1]
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(obj["status"], "merged")
|
||
self.assertIn("merge_gate", obj)
|
||
self.assertEqual(self._status(), "merged")
|
||
self.assertEqual(len(self._merge_audits()), 1)
|
||
self.assertEqual(self._merge_audits()[0]["detail"], "t1")
|
||
|
||
def test_audit_failure_rolls_back_status(self):
|
||
# 旧行为是 merged + 500；新行为是 completed + 抛错
|
||
# （_dispatch_safe 转 500 JSON），调用方可原样重试。
|
||
h = self._handler()
|
||
with mock.patch.object(
|
||
server_mod, "_audit_on_conn",
|
||
side_effect=RuntimeError("audit down")):
|
||
with self.assertRaises(RuntimeError):
|
||
self._merge(h)
|
||
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")
|
||
|
||
def test_ledger_conflict_retries_then_succeeds(self):
|
||
# 链头唯一冲突重试整段（与 AuthStore.audit 同口径）：
|
||
# 前两次冲突、第三次成功，整体仍是一次合并、一条审计。
|
||
calls = {"n": 0}
|
||
|
||
def flaky(conn, user_id, project_id, action, detail="",
|
||
created_at=None):
|
||
calls["n"] += 1
|
||
if calls["n"] <= 2:
|
||
raise sqlite3.IntegrityError("chain head race")
|
||
return audit_on_conn(conn, user_id, project_id, action,
|
||
detail=detail, created_at=created_at)
|
||
|
||
h = self._handler()
|
||
with mock.patch.object(server_mod, "_audit_on_conn",
|
||
side_effect=flaky):
|
||
self._merge(h)
|
||
self.assertEqual(h.sent[-1][0], 200)
|
||
self.assertEqual(calls["n"], 3)
|
||
self.assertEqual(self._status(), "merged")
|
||
self.assertEqual(len(self._merge_audits()), 1)
|
||
|
||
def test_ledger_conflict_exhausted_rolls_back(self):
|
||
# 5 次冲突耗尽：抛错、状态回滚、审计无行、不留半截合并。
|
||
calls = {"n": 0}
|
||
|
||
def always_conflict(*args, **kwargs):
|
||
calls["n"] += 1
|
||
raise sqlite3.IntegrityError("chain head race")
|
||
|
||
h = self._handler()
|
||
with mock.patch.object(server_mod, "_audit_on_conn",
|
||
side_effect=always_conflict):
|
||
with self.assertRaises(sqlite3.IntegrityError):
|
||
self._merge(h)
|
||
self.assertEqual(calls["n"], 5)
|
||
self.assertEqual(self._status(), "completed")
|
||
self.assertEqual(self._merge_audits(), [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|