"""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()