"""成员删除端点单测（残留批次 A 任务 1）。 DELETE /projects/{id}/members/{user_id}： - 限项目 owner 或全局管理员，其余 403； - 目标非成员 404，项目不存在 404； - 删唯一 owner 时 409（owner 与全局管理员均受此限）； - 成功后成员列表无此人，且记 remove_member 审计。 测试用的 key、用户 id 全是假值；不联网、不碰生产。 """ import http.client import json import os import sys import tempfile import threading import unittest from pathlib import Path try: 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.auth import AuthStore from cloud.server import create_server def _req(port, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) headers = {"Content-Type": "application/json"} if key: headers["X-API-Key"] = key conn.request(method, path, json.dumps(body) if body is not None else None, headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) except ValueError: data = None return resp.status, data class MemberDeleteTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.trace_dir = os.path.join(self.tmp.name, "host-traces") os.makedirs(self.trace_dir, exist_ok=True) self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.maint_id, self.maint_key = self.auth.create_user("maint") self.member_id, self.member_key = self.auth.create_user("member") self.admin_id, self.admin_key = self.auth.create_user("admin") self.outsider_id, self.out_key = self.auth.create_user("outsider") self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") os.environ["CODEX_GLOBAL_ADMINS"] = self.admin_id self.server = create_server(self.db_path, "127.0.0.1", 0) self.server.trace_dir = self.trace_dir self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, proj = _req(self.port, "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201, proj) self.project_id = proj["id"] self.auth.set_member(self.project_id, self.owner_id, self.maint_id, "maintainer") 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() if self._old_admins is None: os.environ.pop("CODEX_GLOBAL_ADMINS", None) else: os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins def _delete(self, target, key): return _req(self.port, "DELETE", "/projects/%s/members/%s" % (self.project_id, target), key=key) def _invite(self, target, role="contributor", key=None): return _req(self.port, "POST", "/projects/%s/members" % self.project_id, {"user_id": target, "role": role}, key or self.owner_key) def _member_ids(self, key=None): status, data = _req(self.port, "GET", "/projects/%s/members" % self.project_id, key=key or self.owner_key) self.assertEqual(status, 200, data) return {m["user_id"] for m in data["members"]} def _audit_actions(self): return [(a["action"], a["detail"]) for a in self.auth.list_audit(self.project_id)] def test_owner_delete_member_and_audit(self): self.assertEqual(self._invite(self.member_id)[0], 201) status, data = self._delete(self.member_id, self.owner_key) self.assertEqual(status, 200, data) self.assertEqual(data["user_id"], self.member_id) self.assertNotIn(self.member_id, self._member_ids()) self.assertIn(("remove_member", self.member_id), self._audit_actions()) def test_non_owner_and_outsider_denied(self): self.assertEqual(self._invite(self.member_id)[0], 201) status, _ = self._delete(self.member_id, self.maint_key) self.assertEqual(status, 403) status, _ = self._delete(self.member_id, self.out_key) self.assertEqual(status, 403) # 拒绝后成员仍在。 self.assertIn(self.member_id, self._member_ids()) def test_global_admin_delete_without_membership(self): self.assertEqual(self._invite(self.member_id)[0], 201) status, data = self._delete(self.member_id, self.admin_key) self.assertEqual(status, 200, data) self.assertNotIn(self.member_id, self._member_ids()) self.assertIn(("remove_member", self.member_id), self._audit_actions()) def test_delete_non_member_and_missing_project(self): status, _ = self._delete("ghost", self.owner_key) self.assertEqual(status, 404) status, _ = _req(self.port, "DELETE", "/projects/no-such/members/%s" % self.member_id, key=self.owner_key) self.assertEqual(status, 404) def test_cannot_remove_only_owner(self): status, data = self._delete(self.owner_id, self.owner_key) self.assertEqual(status, 409, data) status, _ = self._delete(self.owner_id, self.admin_key) self.assertEqual(status, 409) # 升第二位 owner 后，删原 owner 放行。 self.assertEqual(self._invite(self.member_id)[0], 201) status, _ = _req(self.port, "PATCH", "/projects/%s/members/%s" % (self.project_id, self.member_id), {"role": "owner"}, self.owner_key) self.assertEqual(status, 200) status, _ = self._delete(self.owner_id, self.maint_key) # maint 不是 owner 也不是全局管理员：仍 403（守卫先行）。 self.assertEqual(status, 403) status, data = self._delete(self.owner_id, self.member_key) self.assertEqual(status, 200, data) self.assertNotIn(self.owner_id, self._member_ids(self.member_key))