223 lines
8.7 KiB
Python
223 lines
8.7 KiB
Python
"""运行时审批队列服务端单测：审批端点。
|
|
|
|
不跑真实模型、不联网：本地临时端口 + 临时 SQLite，与
|
|
test_peer_messages_server.py 同款骨架。覆盖创建、列表、批准、
|
|
拒绝、回执、二次决定 409、过期扫荡、权限、校验、跨任务隔离。
|
|
"""
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.approvals import ensure_approvals_table
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server
|
|
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.approvals import ensure_approvals_table
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server
|
|
|
|
|
|
class ApprovalsServerTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.auth = AuthStore(self.db_path)
|
|
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
|
self.outsider_id, self.outsider_key = self.auth.create_user(
|
|
"outsider")
|
|
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
|
self.port = self.server.server_address[1]
|
|
self.thread = threading.Thread(
|
|
target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
status, self.project = self._req(
|
|
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
|
self.assertEqual(status, 201)
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=5)
|
|
try:
|
|
self.auth.conn.close()
|
|
except Exception:
|
|
pass
|
|
self.tmp.cleanup()
|
|
|
|
def _req(self, method, path, body=None, key=None):
|
|
conn = http.client.HTTPConnection(
|
|
"127.0.0.1", self.port, timeout=10)
|
|
hdrs = {}
|
|
payload = None
|
|
if body is not None:
|
|
payload = json.dumps(body).encode("utf-8")
|
|
hdrs["Content-Type"] = "application/json"
|
|
if key is not None:
|
|
hdrs["X-API-Key"] = key
|
|
try:
|
|
conn.request(method, path, body=payload, headers=hdrs)
|
|
resp = conn.getresponse()
|
|
data = resp.read()
|
|
finally:
|
|
conn.close()
|
|
try:
|
|
parsed = json.loads(data.decode("utf-8")) if data else None
|
|
except ValueError:
|
|
parsed = None
|
|
return resp.status, parsed
|
|
|
|
def _create_task(self, project_id):
|
|
status, task = self._req(
|
|
"POST", "/tasks",
|
|
{"project_id": project_id, "prompt": "做事"}, self.owner_key)
|
|
self.assertEqual(status, 201, task)
|
|
return task
|
|
|
|
def _create_approval(self, task_id, **extra):
|
|
body = {"summary": "MCP 服务请求表单输入"}
|
|
body.update(extra)
|
|
return self._req(
|
|
"POST", "/tasks/%s/approvals" % task_id, body, self.owner_key)
|
|
|
|
def _insert_expired_row(self, task_id, project_id):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
ensure_approvals_table(conn)
|
|
conn.execute(
|
|
"INSERT INTO task_approvals(id, task_id, project_id,"
|
|
" kind, summary, detail, status, requested_by,"
|
|
" decided_by, decided_at, decision,"
|
|
" created_at, updated_at, expires_at)"
|
|
" VALUES(?, ?, ?, 'elicitation', '过期项', NULL,"
|
|
" 'pending', NULL, NULL, NULL, NULL, ?, ?, ?)",
|
|
("expired-1", task_id, project_id,
|
|
time.time() - 7200, time.time() - 7200,
|
|
time.time() - 3600))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def test_create_list_approve_receipt(self):
|
|
task = self._create_task(self.project["id"])
|
|
status, body = self._create_approval(task["id"])
|
|
self.assertEqual(status, 201, body)
|
|
item = body["approval"]
|
|
self.assertEqual(item["status"], "pending")
|
|
self.assertEqual(item["kind"], "elicitation")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/approvals" % task["id"],
|
|
None, self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(len(body["approvals"]), 1)
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/approvals/%s/approve" % (
|
|
task["id"], item["id"]), {}, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
decided = body["approval"]
|
|
self.assertEqual(decided["status"], "approved")
|
|
self.assertEqual(decided["decision"], "approved")
|
|
self.assertEqual(decided["decided_by"], self.owner_id)
|
|
self.assertIsNotNone(decided["decided_at"])
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/approvals" % task["id"],
|
|
None, self.owner_key)
|
|
self.assertEqual(body["approvals"][0]["status"], "approved")
|
|
|
|
def test_deny_and_second_decision_conflicts(self):
|
|
task = self._create_task(self.project["id"])
|
|
_, body = self._create_approval(task["id"])
|
|
item = body["approval"]
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/approvals/%s/deny" % (
|
|
task["id"], item["id"]), {}, self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body["approval"]["status"], "denied")
|
|
self.assertEqual(body["approval"]["decided_by"], self.owner_id)
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/approvals/%s/approve" % (
|
|
task["id"], item["id"]), {}, self.owner_key)
|
|
self.assertEqual(status, 409, body)
|
|
self.assertEqual(body["approval"]["status"], "denied")
|
|
|
|
def test_list_sweeps_overdue_to_expired_deny(self):
|
|
task = self._create_task(self.project["id"])
|
|
self._insert_expired_row(task["id"], self.project["id"])
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/approvals" % task["id"],
|
|
None, self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(len(body["approvals"]), 1)
|
|
swept = body["approvals"][0]
|
|
self.assertEqual(swept["status"], "expired")
|
|
self.assertEqual(swept["decision"], "denied")
|
|
self.assertEqual(swept["decided_by"], "system-timeout")
|
|
|
|
def test_validation_and_not_found(self):
|
|
task = self._create_task(self.project["id"])
|
|
base = "/tasks/%s/approvals" % task["id"]
|
|
status, _ = self._req("POST", base, {}, self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
status, _ = self._req(
|
|
"POST", base, {"summary": "x", "kind": "bogus"},
|
|
self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
status, _ = self._req(
|
|
"POST", base, {"summary": "x", "timeout_seconds": 10},
|
|
self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
status, _ = self._req(
|
|
"POST", base, {"summary": "x", "timeout_seconds": 90000},
|
|
self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
status, _ = self._req(
|
|
"GET", "/tasks/nope/approvals", None, self.owner_key)
|
|
self.assertEqual(status, 404)
|
|
status, _ = self._req(
|
|
"POST", base, {"summary": "x"}, None)
|
|
self.assertEqual(status, 401)
|
|
|
|
def test_forbidden_for_outsider(self):
|
|
task = self._create_task(self.project["id"])
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/approvals" % task["id"],
|
|
{"summary": "x"}, self.outsider_key)
|
|
self.assertEqual(status, 403)
|
|
status, _ = self._req(
|
|
"GET", "/tasks/%s/approvals" % task["id"],
|
|
None, self.outsider_key)
|
|
self.assertEqual(status, 403)
|
|
_, body = self._create_approval(task["id"])
|
|
item = body["approval"]
|
|
status, _ = self._req(
|
|
"POST", "/tasks/%s/approvals/%s/approve" % (
|
|
task["id"], item["id"]), {}, self.outsider_key)
|
|
self.assertEqual(status, 403)
|
|
|
|
def test_cross_task_approval_id_is_not_found(self):
|
|
task_a = self._create_task(self.project["id"])
|
|
task_b = self._create_task(self.project["id"])
|
|
_, body = self._create_approval(task_a["id"])
|
|
item = body["approval"]
|
|
status, _ = self._req(
|
|
"POST", "/tasks/%s/approvals/%s/approve" % (
|
|
task_b["id"], item["id"]), {}, self.owner_key)
|
|
self.assertEqual(status, 404)
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/approvals" % task_b["id"],
|
|
None, self.owner_key)
|
|
self.assertEqual(body["approvals"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|