"""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()