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

"""成员删除端点单测（残留批次 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))