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

"""运行时审批队列模块单测：cloud/approvals.py。

不跑真实模型、不联网：内存 SQLite。覆盖批准放行、拒绝拦截、
回执留痕、二次决定不覆盖、超时默认拒绝、非法输入、回包映射。
"""

import sqlite3
import sys
import unittest
from pathlib import Path

try:
    from cloud.approvals import (
        command_response,
        create_approval,
        decide_approval,
        elicitation_response,
        ensure_approvals_table,
        expire_overdue,
        get_approval,
        list_approvals,
    )
except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.approvals import (
        command_response,
        create_approval,
        decide_approval,
        elicitation_response,
        ensure_approvals_table,
        expire_overdue,
        get_approval,
        list_approvals,
    )


class ApprovalsModuleTestCase(unittest.TestCase):
    def setUp(self):
        self.conn = sqlite3.connect(":memory:")
        self.conn.row_factory = sqlite3.Row
        ensure_approvals_table(self.conn)
        self.conn.commit()

    def tearDown(self):
        self.conn.close()

    def _mk(self, **kw):
        kw.setdefault("task_id", "task-1")
        kw.setdefault("project_id", "proj-1")
        kw.setdefault("summary", "MCP 服务请求表单输入")
        return create_approval(self.conn, **kw)

    def test_create_defaults_to_pending_with_timeout(self):
        item = self._mk()
        self.assertEqual(item["status"], "pending")
        self.assertEqual(item["kind"], "elicitation")
        self.assertIsNone(item["decided_by"])
        self.assertIsNone(item["decision"])
        self.assertIsNotNone(item["expires_at"])
        self.assertGreater(item["expires_at"], item["created_at"])
        pending = list_approvals(self.conn, task_id="task-1",
                                 status="pending")
        self.assertEqual(len(pending), 1)

    def test_approve_releases_with_receipt(self):
        item = self._mk()
        status, decided = decide_approval(
            self.conn, item["id"], "approved", "user-9")
        self.assertEqual(status, "decided")
        self.assertEqual(decided["status"], "approved")
        self.assertEqual(decided["decision"], "approved")
        self.assertEqual(decided["decided_by"], "user-9")
        self.assertIsNotNone(decided["decided_at"])
        self.assertEqual(elicitation_response(decided["decision"]),
                         {"action": "accept", "content": {}})
        self.assertEqual(command_response(decided["decision"]),
                         {"decision": "accept"})

    def test_deny_blocks_with_receipt(self):
        item = self._mk()
        status, decided = decide_approval(
            self.conn, item["id"], "denied", "user-9")
        self.assertEqual(status, "decided")
        self.assertEqual(decided["status"], "denied")
        self.assertEqual(decided["decision"], "denied")
        self.assertEqual(decided["decided_by"], "user-9")
        self.assertEqual(elicitation_response(decided["decision"]),
                         {"action": "decline"})
        self.assertEqual(command_response(decided["decision"]),
                         {"decision": "decline"})

    def test_second_decision_never_overwrites_first(self):
        item = self._mk()
        decide_approval(self.conn, item["id"], "approved", "user-9")
        status, kept = decide_approval(
            self.conn, item["id"], "denied", "user-7")
        self.assertEqual(status, "already_decided")
        self.assertEqual(kept["status"], "approved")
        self.assertEqual(kept["decided_by"], "user-9")

    def test_expire_overdue_defaults_to_deny_not_allow(self):
        item = self._mk(now=1000.0, timeout_seconds=60)
        self.assertEqual(item["expires_at"], 1060.0)
        count = expire_overdue(self.conn, now=1061.0)
        self.assertEqual(count, 1)
        done = get_approval(self.conn, item["id"])
        self.assertEqual(done["status"], "expired")
        self.assertEqual(done["decision"], "denied")
        self.assertEqual(done["decided_by"], "system-timeout")
        self.assertEqual(elicitation_response(done["decision"]),
                         {"action": "decline"})
        status, kept = decide_approval(
            self.conn, item["id"], "approved", "user-9")
        self.assertEqual(status, "already_decided")
        self.assertEqual(kept["status"], "expired")

    def test_not_yet_due_is_not_expired(self):
        self._mk(now=1000.0, timeout_seconds=3600)
        self.assertEqual(expire_overdue(self.conn, now=2000.0), 0)
        self.assertEqual(len(list_approvals(self.conn, status="pending")),
                         1)

    def test_decide_missing_is_not_found(self):
        status, item = decide_approval(self.conn, "nope", "approved",
                                       "user-9")
        self.assertEqual(status, "not_found")
        self.assertIsNone(item)

    def test_invalid_inputs_raise_value_error(self):
        with self.assertRaises(ValueError):
            self._mk(kind="bogus")
        with self.assertRaises(ValueError):
            self._mk(summary="   ")
        with self.assertRaises(ValueError):
            self._mk(summary="x" * 2001)
        with self.assertRaises(ValueError):
            self._mk(timeout_seconds=-1)
        with self.assertRaises(ValueError):
            decide_approval(self.conn, "x", "maybe", "user-9")

    def test_ensure_is_idempotent_and_keeps_rows(self):
        item = self._mk()
        ensure_approvals_table(self.conn)
        self.conn.commit()
        self.assertEqual(get_approval(self.conn, item["id"])["id"],
                         item["id"])

    def test_response_mapping_never_defaults_to_allow(self):
        for weird in ("expired", "bogus", None, ""):
            self.assertEqual(elicitation_response(weird),
                             {"action": "decline"})
            self.assertEqual(command_response(weird),
                             {"decision": "decline"})


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