"""云端环境单测：选、建、管 + 任务绑定环境。

不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
SQLite；worker 侧注入假 git runner（clone/fetch/checkout 只
操作临时目录，不调真实 git）；console 侧只查 HTML 标记。

链路覆盖：建环境（creating）-> worker 准备（ready/failed）
-> 建任务选环境（执行器继承/冲突拒绝）-> 任务派发指到环境
检出 -> 归档（进行中任务在用时拒绝、幂等）。
"""

import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
import uuid
from pathlib import Path
from types import SimpleNamespace

try:
    from cloud.auth import AuthStore
    from cloud.sandbox import DockerExecutor
    from cloud.server import create_server, init_db
    from cloud.worker import Worker
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.sandbox import DockerExecutor
    from cloud.server import create_server, init_db
    from cloud.worker import Worker


def _req(port, method, path, body=None, key=None):
    conn = http.client.HTTPConnection("127.0.0.1", 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 _set_env_status(db_path, env_id, status):
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("UPDATE environments SET status = ? WHERE id = ?",
                     (status, env_id))
        conn.commit()
    finally:
        conn.close()


def _set_task_status(db_path, task_id, status):
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
                     (status, task_id))
        conn.commit()
    finally:
        conn.close()


class ServerEnvironmentsTestCase(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.other_id, self.other_key = self.auth.create_user("other")
        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 = _req(
            self.port, "POST", "/projects",
            {"name": "demo", "repo_url": "https://github.com/octocat/Hello-World.git"},
            self.owner_key)
        self.assertEqual(status, 201, self.project)
        status, self.plain_project = _req(
            self.port, "POST", "/projects", {"name": "plain"},
            self.owner_key)
        self.assertEqual(status, 201, self.plain_project)

    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 _create_env(self, project_id=None, name="dev", **extra):
        body = {"project_id": project_id or self.project["id"],
                "name": name}
        body.update(extra)
        return _req(self.port, "POST", "/environments", body,
                    self.owner_key)

    def _ready_env(self, name="dev", **extra):
        status, env = self._create_env(name=name, **extra)
        self.assertEqual(status, 201, env)
        _set_env_status(self.db_path, env["id"], "ready")
        return env

    def test_create_environment_roundtrip(self):
        status, env = self._create_env(
            name="dev", branch="main", executor="local")
        self.assertEqual(status, 201, env)
        self.assertEqual(env["status"], "creating")
        self.assertEqual(env["project_id"], self.project["id"])
        self.assertEqual(env["branch"], "main")
        self.assertEqual(env["executor"], "local")
        self.assertEqual(env["repo_url"],
                         "https://github.com/octocat/Hello-World.git")
        status, got = _req(
            self.port, "GET", "/environments/" + env["id"],
            key=self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(got["id"], env["id"])

    def test_create_environment_validation(self):
        # 项目没有 repo_url：环境无从检出。
        status, _ = self._create_env(
            project_id=self.plain_project["id"], name="x")
        self.assertEqual(status, 400)
        # 名字、分支、执行器非法。
        status, _ = self._create_env(name="  ")
        self.assertEqual(status, 400)
        status, _ = self._create_env(name="x", branch="../evil")
        self.assertEqual(status, 400)
        status, _ = self._create_env(name="x", branch="-bad")
        self.assertEqual(status, 400)
        status, _ = self._create_env(name="x", executor="cloud")
        self.assertEqual(status, 400)
        # 非成员、不存在的环境。
        status, _ = _req(
            self.port, "POST", "/environments",
            {"project_id": self.project["id"], "name": "x"},
            self.other_key)
        self.assertEqual(status, 403)
        status, _ = _req(self.port, "GET", "/environments/nope",
                         key=self.owner_key)
        self.assertEqual(status, 404)

    def test_environment_name_unique_per_project(self):
        status, _ = self._create_env(name="dev")
        self.assertEqual(status, 201)
        status, _ = self._create_env(name="dev")
        self.assertEqual(status, 409)
        # 另一项目同名可以；归档后原项目同名可重用。
        status, env2 = self._create_env(
            project_id=self.plain_project["id"], name="dev")
        # plain 项目无 repo_url，上面应 400——换一个带仓库的项目测同名。
        self.assertEqual(status, 400)
        status, other = _req(
            self.port, "POST", "/projects",
            {"name": "demo2",
             "repo_url": "https://github.com/octocat/Hello-World.git"},
            self.owner_key)
        self.assertEqual(status, 201)
        status, _ = self._create_env(project_id=other["id"], name="dev")
        self.assertEqual(status, 201)

    def test_list_environments_and_archive(self):
        env = self._ready_env(name="dev")
        status, listing = _req(
            self.port, "GET",
            "/environments?project_id=" + self.project["id"],
            key=self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(len(listing["environments"]), 1)
        # 缺 project_id 400；非成员 403。
        status, _ = _req(self.port, "GET", "/environments",
                         key=self.owner_key)
        self.assertEqual(status, 400)
        status, _ = _req(
            self.port, "GET",
            "/environments?project_id=" + self.project["id"],
            key=self.other_key)
        self.assertEqual(status, 403)
        # 归档后默认列表排除，include_archived=1 才出现；再归档幂等。
        status, archived = _req(
            self.port, "POST", "/environments/%s/archive" % env["id"],
            {}, self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(archived["status"], "archived")
        status, listing = _req(
            self.port, "GET",
            "/environments?project_id=" + self.project["id"],
            key=self.owner_key)
        self.assertEqual(listing["environments"], [])
        status, listing = _req(
            self.port, "GET",
            "/environments?project_id=" + self.project["id"]
            + "&include_archived=1",
            key=self.owner_key)
        self.assertEqual(len(listing["environments"]), 1)
        status, again = _req(
            self.port, "POST", "/environments/%s/archive" % env["id"],
            {}, self.owner_key)
        self.assertEqual(status, 200)

    def test_archive_blocked_when_task_using_it(self):
        env = self._ready_env(name="dev")
        status, task = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": env["id"]},
            self.owner_key)
        self.assertEqual(status, 201, task)
        status, _ = _req(
            self.port, "POST", "/environments/%s/archive" % env["id"],
            {}, self.owner_key)
        self.assertEqual(status, 409)
        _set_task_status(self.db_path, task["id"], "completed")
        status, archived = _req(
            self.port, "POST", "/environments/%s/archive" % env["id"],
            {}, self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(archived["status"], "archived")

    def test_task_with_environment_rules(self):
        # creating 的环境不能建任务。
        status, env = self._create_env(name="dev", executor="docker")
        self.assertEqual(status, 201)
        status, _ = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": env["id"]},
            self.owner_key)
        self.assertEqual(status, 400)
        # ready 后：执行器继承、冲突拒绝、cloud 拒绝、跨项目拒绝。
        _set_env_status(self.db_path, env["id"], "ready")
        status, task = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": env["id"]},
            self.owner_key)
        self.assertEqual(status, 201, task)
        self.assertEqual(task["executor"], "docker")
        self.assertEqual(task["environment_id"], env["id"])
        status, _ = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": env["id"], "executor": "local"},
            self.owner_key)
        self.assertEqual(status, 400)
        status, _ = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": env["id"], "executor": "cloud"},
            self.owner_key)
        self.assertEqual(status, 400)
        status, _ = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.plain_project["id"], "prompt": "p",
             "environment_id": env["id"]},
            self.owner_key)
        self.assertEqual(status, 400)
        status, _ = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": "nope"},
            self.owner_key)
        self.assertEqual(status, 404)
        # failed 的环境也不能建任务。
        _set_env_status(self.db_path, env["id"], "failed")
        status, _ = _req(
            self.port, "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "p",
             "environment_id": env["id"]},
            self.owner_key)
        self.assertEqual(status, 400)

    def test_console_has_environment_ui(self):
        path = Path(__file__).resolve().parents[1] / "console.html"
        html = path.read_text(encoding="utf-8")
        # React 版（2026-10-29 起）：旧单页的 loadEnvironments /
        # createEnvironment / newEnv* id 已不存在，环境 UI 改为
        # React 组件（EnvSection + useCatalog），断言意图不变，
        # 只换成构建产物里实际存在的标记：环境栏、建环境表单、
        # 刷新/新建/归档操作与环境列表接口。
        for marker in ("nav-env", "新环境名", "新建环境", "归档",
                       "refreshEnvs", "createEnv", "archiveEnv",
                       "/environments?project_id="):
            self.assertIn(marker, html)


class FakeGit:
    """假 git：clone 在临时目录建 .git 并记 origin；fetch 恒成功；
    checkout 只接受白名单分支，其余报分支不存在。"""

    def __init__(self, allowed_branches=("main", "feature/x")):
        self.allowed = set(allowed_branches)
        self.origins = {}
        self.commands = []

    def __call__(self, cmd):
        self.commands.append(list(cmd))
        if cmd[1] == "clone":
            target = cmd[3]
            os.makedirs(os.path.join(target, ".git"), exist_ok=True)
            self.origins[target] = cmd[2]
            return SimpleNamespace(returncode=0, stdout="", stderr="")
        if cmd[1] == "-C":
            target = cmd[2]
            if cmd[3] == "remote":
                return SimpleNamespace(
                    returncode=0, stdout=self.origins.get(target, ""),
                    stderr="")
            if cmd[3] == "fetch":
                return SimpleNamespace(returncode=0, stdout="", stderr="")
            if cmd[3] == "checkout":
                branch = cmd[4]
                if branch in self.allowed:
                    return SimpleNamespace(returncode=0, stdout="",
                                           stderr="")
                return SimpleNamespace(
                    returncode=1, stdout="",
                    stderr="pathspec '%s' did not match" % branch)
        return SimpleNamespace(returncode=1, stdout="",
                               stderr="unexpected git command")


def _insert_env(db_path, env_id="env1", project_id="p1",
                status="creating", branch=None, executor=None,
                name="dev",
                repo_url="https://github.com/octocat/Hello-World.git"):
    now = time.time()
    conn = sqlite3.connect(db_path)
    try:
        conn.execute(
            "INSERT INTO environments(id, project_id, name, branch,"
            " executor, status, error_text, repo_url, created_by,"
            " created_at, updated_at) VALUES(?, ?, ?, ?, ?, ?,"
            " NULL, ?, 'tester', ?, ?)",
            (env_id, project_id, name, branch, executor, status,
             repo_url, now, now))
        conn.commit()
    finally:
        conn.close()


def _env_row(db_path, env_id):
    conn = sqlite3.connect(db_path)
    try:
        row = conn.execute(
            "SELECT * FROM environments WHERE id = ?",
            (env_id,)).fetchone()
        cols = [d[0] for d in conn.execute(
            "SELECT * FROM environments LIMIT 0").description]
        return dict(zip(cols, row)) if row else None
    finally:
        conn.close()


class WorkerEnvironmentsTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        init_db(self.db_path)
        self.repos_dir = os.path.join(self.tmp.name, "repos")
        self.env = {"XCM_API_KEY": "test-placeholder"}

    def tearDown(self):
        self.tmp.cleanup()

    def _worker(self, git=None, **kwargs):
        kwargs.setdefault("env", self.env)
        kwargs.setdefault("repos_dir", self.repos_dir)
        kwargs.setdefault("repos_runner", git)
        return Worker(self.db_path, "/fake/host.py", **kwargs)

    def test_prepare_environment_ready(self):
        _insert_env(self.db_path, branch="feature/x")
        worker = self._worker(git=FakeGit())
        results = worker._prepare_pending_environments()
        self.assertEqual(results, [("env1", "ready", None)])
        row = _env_row(self.db_path, "env1")
        self.assertEqual(row["status"], "ready")
        target = os.path.join(self.repos_dir, "envs", "env1")
        self.assertTrue(os.path.isdir(os.path.join(target, ".git")))

    def test_prepare_environment_failed_branch(self):
        _insert_env(self.db_path, branch="no-such-branch")
        worker = self._worker(git=FakeGit())
        results = worker._prepare_pending_environments()
        self.assertEqual(results[0][1], "failed")
        self.assertIn("checkout", results[0][2])
        self.assertEqual(_env_row(self.db_path, "env1")["status"],
                         "failed")

    def test_prepare_environment_archived_during_prepare(self):
        # 准备期间被归档：CAS 改不到行，状态保持 archived。
        _insert_env(self.db_path)
        worker = self._worker(git=FakeGit())
        orig = worker._prepare_environment

        def _prepare_and_archive(env):
            err = orig(env)
            _set_env_status(self.db_path, env["id"], "archived")
            return err

        worker._prepare_environment = _prepare_and_archive
        results = worker._prepare_pending_environments()
        self.assertEqual(results[0][1], "archived")
        self.assertEqual(_env_row(self.db_path, "env1")["status"],
                         "archived")

    def test_apply_environment_repo_local(self):
        _insert_env(self.db_path, status="ready", executor="local")
        worker = self._worker()
        task = {"id": "T", "project_id": "p1", "environment_id": "env1",
                "executor": "local"}
        self.assertIsNone(worker._apply_project_repo(task))
        self.assertEqual(
            task["repo"],
            os.path.join(self.repos_dir, "envs", "env1"))

    def test_apply_environment_repo_tenant_docker(self):
        _insert_env(self.db_path, status="ready", executor="docker")
        worker = self._worker(
            executor=DockerExecutor(config_path="/fake/cfg.json"),
            tenant_isolation=True,
            tenants_root=os.path.join(self.tmp.name, "tenants"))
        task = {"id": "T", "project_id": "p1", "environment_id": "env1",
                "executor": "docker"}
        self.assertIsNone(worker._apply_project_repo(task))
        self.assertEqual(task["repo"],
                         "/var/lib/codex-app-server/envs/env1")

    def test_apply_environment_repo_rejections(self):
        worker = self._worker()
        # 环境不存在。
        task = {"id": "T", "project_id": "p1", "environment_id": "nope"}
        self.assertIn("not found", worker._apply_project_repo(task))
        # 未 ready、跨项目、执行器冲突（同项目多环境须不同名，
        # 部分唯一索引只拦同项目同名未归档行）。
        _insert_env(self.db_path, env_id="e2", name="e2",
                    status="creating")
        task = {"id": "T", "project_id": "p1", "environment_id": "e2"}
        self.assertIn("not ready", worker._apply_project_repo(task))
        _insert_env(self.db_path, env_id="e3", project_id="p2",
                    name="e3", status="ready")
        task = {"id": "T", "project_id": "p1", "environment_id": "e3"}
        self.assertIn("another project",
                      worker._apply_project_repo(task))
        _insert_env(self.db_path, env_id="e4", name="e4",
                    status="ready", executor="docker")
        task = {"id": "T", "project_id": "p1", "environment_id": "e4",
                "executor": "local"}
        self.assertIn("does not match",
                      worker._apply_project_repo(task))

    def test_tenant_mounts_include_envs(self):
        worker = self._worker(
            executor=DockerExecutor(config_path="/fake/cfg.json"),
            tenant_isolation=True,
            tenants_root=os.path.join(self.tmp.name, "tenants"))
        # 先播种租户目录（codex-home 模板缺失会报错，这里只测
        # envs 挂载条件：手动建好租户目录结构再调 _tenant_mounts
        # 的 envs 分支——目录不存在时不挂载，存在时挂载）。
        base = os.path.join(self.tmp.name, "tenants", "p1")
        os.makedirs(os.path.join(base, "work"), exist_ok=True)
        os.makedirs(os.path.join(base, "codex-home"), exist_ok=True)
        with open(os.path.join(base, "codex-home", "config.toml"),
                  "w") as f:
            f.write("")
        os.makedirs(os.path.join(base, "test-repo"), exist_ok=True)
        with open(os.path.join(base, "test-repo", "sample.py"),
                  "w") as f:
            f.write("")
        mounts = worker._tenant_mounts("p1")
        self.assertFalse(any("envs" in m for m in mounts))
        os.makedirs(os.path.join(base, "envs", "env1"), exist_ok=True)
        mounts = worker._tenant_mounts("p1")
        self.assertTrue(any(
            m.endswith(":/var/lib/codex-app-server/envs:rw")
            for m in mounts))


if __name__ == "__main__":
    unittest.main()