gapfill G7: make _merge_task status commit and success audit atomic (same connection, audit_on_conn, one commit; audit failure rolls back to completed instead of merged-plus-500; ledger-head IntegrityError retries whole segment 5x mirroring AuthStore.audit); success semantics unchanged
This commit is contained in:
host committed
1 parent
5e1782be69
commit
a1a50f1e81
2 files changed
+223
-8
No files matched your search
@@ -0,0 +1,184 @@
|
||||
"""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()
|
||||
Reference in new issue
Block a user