341 lines
14 KiB
Python
341 lines
14 KiB
Python
"""P2 切片 A（控制台管理面与筛选）单测。
|
||
|
||
只覆盖本切片新增或复用的服务端契约 + 控制台壳标记，不跑
|
||
浏览器 JS（与 test_console.py 同款：壳只验标记与路由）：
|
||
- 成员端点：非 owner 邀请 403、owner 邀请成功（user_id 与
|
||
username 两种入参）、重复邀请 409、非法角色 400、未知用户
|
||
404、成员列表读权限、改角色权限与唯一 owner 防降级、
|
||
非成员全局管理员可邀请/改角色。
|
||
- 管理区复用端点：非管理员 sweep 403、管理员配额读成功、
|
||
探针语义（非管理员 GET /admin/quotas 无参 403，管理员 400）。
|
||
- 链路全量：写 1200 条链路文件，GET /trace 全量返回（控制台
|
||
“下载完整链路”依赖的正是该端点，而非屏上截断的 1000 条）。
|
||
- Workers：心跳后 /workers 返回 epoch 秒 last_seen 与 alive。
|
||
- 控制台壳含本切片新增标记。
|
||
|
||
测试用的 key、用户 id 全是假值；不联网、不碰生产。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
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 P2ConsoleTestCase(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")
|
||
# 全局管理员：非项目成员的 admin，验“owner 或全局管理员”双通道。
|
||
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 _members(self, key=None):
|
||
return _req(self.port, "GET",
|
||
"/projects/%s/members" % self.project_id,
|
||
key=key)
|
||
|
||
# -- 成员邀请 / 列表 -------------------------------------------
|
||
def test_non_owner_invite_forbidden(self):
|
||
# maintainer 不是 owner、不是全局管理员：邀请必须 403，
|
||
# 且安全门不能因“项目成员”身份放行。
|
||
status, data = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "viewer"},
|
||
self.maint_key)
|
||
self.assertEqual(status, 403, data)
|
||
|
||
def test_outsider_invite_forbidden(self):
|
||
status, data = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "viewer"},
|
||
self.out_key)
|
||
self.assertEqual(status, 403, data)
|
||
|
||
def test_owner_invite_by_user_id_and_username(self):
|
||
status, data = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "contributor"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 201, data)
|
||
self.assertEqual(data["role"], "contributor")
|
||
# 第二个用户走 username 入参（控制台输入框两种都接受）。
|
||
status, data = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"username": "outsider", "role": "viewer"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 201, data)
|
||
self.assertEqual(data["user_id"], self.outsider_id)
|
||
|
||
def test_invite_validation(self):
|
||
status, _ = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "superuser"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 400)
|
||
status, _ = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": "no-such-user", "role": "viewer"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 404)
|
||
status, _ = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"role": "viewer"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 400)
|
||
|
||
def test_invite_duplicate_conflict(self):
|
||
status, _ = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "viewer"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 201)
|
||
# 已是成员：改角色走 PATCH，重复邀请 409，不静默覆盖。
|
||
status, data = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "maintainer"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 409, data)
|
||
|
||
def test_member_list_read_and_outsider_denied(self):
|
||
_req(self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "viewer"},
|
||
self.owner_key)
|
||
status, data = self._members(self.maint_key)
|
||
self.assertEqual(status, 200, data)
|
||
roles = {m["user_id"]: m["role"] for m in data["members"]}
|
||
self.assertEqual(roles[self.owner_id], "owner")
|
||
self.assertEqual(roles[self.maint_id], "maintainer")
|
||
self.assertEqual(roles[self.member_id], "viewer")
|
||
# 列表不泄露认证字段。
|
||
for m in data["members"]:
|
||
self.assertNotIn("api_key_hash", m)
|
||
self.assertNotIn("key_hash", m)
|
||
status, _ = self._members(self.out_key)
|
||
self.assertEqual(status, 403)
|
||
|
||
def test_global_admin_can_invite_without_membership(self):
|
||
# admin 不是项目成员，但凭 CODEX_GLOBAL_ADMINS 可邀请。
|
||
status, data = _req(
|
||
self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "contributor"},
|
||
self.admin_key)
|
||
self.assertEqual(status, 201, data)
|
||
|
||
# -- 改角色 / 唯一 owner 防降级 ---------------------------------
|
||
def test_patch_role_by_owner_and_non_owner_denied(self):
|
||
_req(self.port, "POST",
|
||
"/projects/%s/members" % self.project_id,
|
||
{"user_id": self.member_id, "role": "viewer"},
|
||
self.owner_key)
|
||
status, data = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, self.member_id),
|
||
{"role": "maintainer"}, self.maint_key)
|
||
self.assertEqual(status, 403, data)
|
||
status, data = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, self.member_id),
|
||
{"role": "maintainer"}, self.owner_key)
|
||
self.assertEqual(status, 200, data)
|
||
self.assertEqual(data["role"], "maintainer")
|
||
status, _ = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, "ghost"),
|
||
{"role": "viewer"}, self.owner_key)
|
||
self.assertEqual(status, 404)
|
||
|
||
def test_cannot_demote_only_owner(self):
|
||
status, data = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, self.owner_id),
|
||
{"role": "maintainer"}, self.owner_key)
|
||
self.assertEqual(status, 409, data)
|
||
# 先把另一位成员升为 owner，之后降原 owner 才放行。
|
||
status, _ = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, self.maint_id),
|
||
{"role": "owner"}, self.owner_key)
|
||
self.assertEqual(status, 200)
|
||
status, data = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, self.owner_id),
|
||
{"role": "maintainer"}, self.maint_key)
|
||
self.assertEqual(status, 200, data)
|
||
|
||
def test_global_admin_patch_blocked_by_sole_owner_guard(self):
|
||
# 全局管理员也不能把唯一 owner 降掉（防项目无 owner 可管）。
|
||
status, _ = _req(
|
||
self.port, "PATCH",
|
||
"/projects/%s/members/%s" % (self.project_id, self.owner_id),
|
||
{"role": "viewer"}, self.admin_key)
|
||
self.assertEqual(status, 409)
|
||
|
||
# -- 管理区复用端点 ---------------------------------------------
|
||
def test_admin_probe_and_quota_reuse(self):
|
||
# 探针语义：控制台凭此区分管理员与否，不新增端点。
|
||
status, _ = _req(self.port, "GET", "/admin/quotas",
|
||
key=self.maint_key)
|
||
self.assertEqual(status, 403)
|
||
status, _ = _req(self.port, "GET", "/admin/quotas",
|
||
key=self.admin_key)
|
||
self.assertEqual(status, 400)
|
||
status, data = _req(
|
||
self.port, "GET",
|
||
"/admin/quotas?user_id=%s" % self.owner_id,
|
||
key=self.admin_key)
|
||
self.assertEqual(status, 200, data)
|
||
self.assertIn("used_usd_micros", data)
|
||
status, _ = _req(
|
||
self.port, "GET",
|
||
"/admin/quotas?user_id=%s" % self.owner_id,
|
||
key=self.maint_key)
|
||
self.assertEqual(status, 403)
|
||
|
||
def test_non_admin_sweep_forbidden(self):
|
||
status, _ = _req(self.port, "POST", "/admin/sweep", {},
|
||
self.maint_key)
|
||
self.assertEqual(status, 403)
|
||
status, _ = _req(self.port, "POST", "/admin/backup", {},
|
||
self.maint_key)
|
||
self.assertEqual(status, 403)
|
||
|
||
# -- 链路全量 / Workers ------------------------------------------
|
||
def test_trace_endpoint_returns_full_entries(self):
|
||
status, task = _req(
|
||
self.port, "POST", "/tasks",
|
||
{"project_id": self.project_id, "prompt": "trace probe"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 201, task)
|
||
task_id = task["id"]
|
||
path = os.path.join(self.trace_dir, task_id + ".jsonl")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
for i in range(1200):
|
||
f.write(json.dumps(
|
||
{"id": "e%d" % i, "kind": "message",
|
||
"text": "line %d" % i}) + "\n")
|
||
status, data = _req(
|
||
self.port, "GET", "/tasks/%s/trace" % task_id,
|
||
key=self.owner_key)
|
||
self.assertEqual(status, 200, data)
|
||
# 服务端不截断：控制台下载按钮拿到的就是这 1200 条全量，
|
||
# 屏上 1000 条截断只是渲染层行为。
|
||
self.assertEqual(len(data["entries"]), 1200)
|
||
|
||
def test_workers_last_seen_is_epoch_and_alive(self):
|
||
status, data = _req(
|
||
self.port, "POST", "/workers/heartbeat",
|
||
{"worker_id": "w-p2", "executor": "docker",
|
||
"capacity": 2, "running": 0},
|
||
self.owner_key)
|
||
self.assertEqual(status, 200, data)
|
||
# 合并适配（P2 rebase 到 97f019d，P1 已合并）：P1 起完整
|
||
# worker 清单只给全局管理员，项目 owner 只能看到聚合
|
||
# （聚合行为由 P1 的 test_workers_admin_full_member_aggregate_only
|
||
# 覆盖）；本用例验的是完整清单里的 epoch last_seen 与
|
||
# alive，故用本夹具的全局管理员 admin_key 查，不降门。
|
||
status, data = _req(self.port, "GET", "/workers",
|
||
key=self.admin_key)
|
||
self.assertEqual(status, 200, data)
|
||
workers = data["workers"]
|
||
self.assertEqual(len(workers), 1)
|
||
# 控制台“x 秒/分钟前 + 120 秒在线阈值”的输入：epoch 秒。
|
||
self.assertIsInstance(workers[0]["last_seen"], (int, float))
|
||
self.assertLess(abs(workers[0]["last_seen"] - time.time()), 60)
|
||
self.assertTrue(workers[0]["alive"])
|
||
|
||
# -- 控制台壳标记 -------------------------------------------------
|
||
def test_console_contains_p2_markers(self):
|
||
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
||
conn.request("GET", "/console")
|
||
resp = conn.getresponse()
|
||
text = resp.read().decode("utf-8")
|
||
conn.close()
|
||
self.assertEqual(resp.status, 200)
|
||
for marker in ("/members", "inviteMember", "memberList",
|
||
"adminBody", "/admin/sweep", "/admin/backup",
|
||
"/admin/quotas", "taskStatusFilter",
|
||
"taskExecutorFilter", "taskSearch",
|
||
"btnDownloadTrace", "traceKindFilter",
|
||
"workerAgeText", "last_seen"):
|
||
self.assertIn(marker, text, marker)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|