"""web /v1 路由测试(无 DB 面)—— 拆分后首批路由级回归。

覆盖三类不碰 DB 的行为(handler 里带 DB 查询的端点只测 401 门 —— 依赖先于
handler 执行,永远到不了 DB):
- 公开端点:/healthz、/v1/changelog、/ 302
- 鉴权门:11 个 router 的代表端点无 token 一律 401;坏 platform_key 403;
  platform_key 对但 user_id 非法 400(在 ensure_user_row 落库之前就拦下)
- FS 类业务:kb 建/详/删、skills 列/详/删门、memory 只读、files 子目录
  列/下载/改名/删/拷(顶层目录与 upload 走 DB,留给 ZCBOT_TEST_DB_URL 套件)

隔离纪律:token 用测试自设的 PLATFORM_KEY/JWT_SECRET 直签(先于 import web.app
写进 env,litellm 的 dotenv 副作用不会覆盖已存在值);user_id 随机,产生的
workspace/users/<uid>/ 子树 teardown 整树删除;TestClient 不进 with(不跑
lifespan → 不碰 DB/后台协程)。
"""
from __future__ import annotations

import os
import shutil
import sys
import unittest
import uuid
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

# 必须先于 import web.app 设好(AuthConfig.from_env fail-fast)。注意全量跑时更早
# 导入的模块可能已触发 litellm 的 dotenv 副作用把 .env 真值灌进 env —— setdefault
# 不覆盖,所以下面一律用 AuthConfig.from_env()(与 app 同源),不假设具体 key 值。
os.environ.setdefault("PLATFORM_KEY", "test-platform-key-nodb")
os.environ.setdefault("JWT_SECRET", "test-jwt-secret-nodb")

from starlette.testclient import TestClient  # noqa: E402

from web.app import create_app  # noqa: E402
from web.auth import AuthConfig, mint_token  # noqa: E402
from web.schemas import TaskCreateRequest  # noqa: E402

_app = create_app()
_client = TestClient(_app)   # 刻意不进 with:不跑 lifespan(reaper/调度都不启动)

_CFG = AuthConfig.from_env()
_UID = uuid.uuid4()
_TOKEN, _ = mint_token(_CFG, _UID)
_AUTH = {"Authorization": f"Bearer {_TOKEN}"}


def _user_root() -> Path:
    from core.agent_builder import resolve_workspace, user_root
    return user_root(resolve_workspace(None), _UID)


def tearDownModule() -> None:
    d = _user_root()
    if d.is_dir():
        shutil.rmtree(d, ignore_errors=True)


class PublicEndpointTests(unittest.TestCase):
    def test_task_create_schema_accepts_working_dir_without_name(self):
        body = TaskCreateRequest(working_dir="低碳水泥项目")
        self.assertEqual(body.name, "")
        self.assertEqual(body.working_dir, "低碳水泥项目")
        self.assertFalse(body.auto_title)  # 自动命名由路由据空 name 推断

    def test_healthz(self):
        r = _client.get("/healthz")
        self.assertEqual(r.status_code, 200)
        body = r.json()
        self.assertEqual(body["status"], "ok")
        self.assertIn("version", body)
        self.assertIn("brand", body)

    def test_changelog_public(self):
        r = _client.get("/v1/changelog?limit=2")
        self.assertEqual(r.status_code, 200)
        self.assertIn("entries", r.json())
        # limit 越界 clamp,不 500
        self.assertEqual(_client.get("/v1/changelog?limit=9999").status_code, 200)

    def test_root_redirects_to_dev_spa(self):
        r = _client.get("/", follow_redirects=False)
        self.assertEqual(r.status_code, 302)
        self.assertEqual(r.headers["location"], "/static/dev.html")


class AuthGateTests(unittest.TestCase):
    # 每个 router 至少一个代表端点:无 token 必须 401(鉴权门接线回归 ——
    # 拆分 router 后若哪个模块忘了 Depends(require_user),这里当场红)
    _PROTECTED = [
        ("GET", "/v1/me"),
        ("GET", "/v1/tasks"),
        ("GET", "/v1/channel_tasks"),
        ("GET", "/v1/folders"),
        ("GET", "/v1/models"),
        ("GET", "/v1/image_models"),
        ("GET", "/v1/skills"),
        ("GET", "/v1/memory"),
        ("GET", "/v1/kb"),
        ("GET", "/v1/schedules"),
        ("GET", "/v1/files"),
        ("GET", "/v1/user/storage"),
        ("GET", "/v1/procs"),
        ("GET", "/v1/wechat/bind"),
        ("GET", "/v1/wecom/bind"),
        ("GET", "/v1/external-system-providers"),
        ("GET", "/v1/external-systems"),
        ("GET", "/v1/tasks/00000000-0000-0000-0000-000000000000/messages"),
        ("POST", "/v1/tasks"),
        ("POST", "/v1/asr/transcribe"),
        ("GET", "/v1/admin/overview"),
        ("GET", "/v1/admin/software-nodes"),
        ("GET", "/v1/software-jobs"),
        ("POST", "/v1/software-jobs/00000000-0000-0000-0000-000000000000/cancel"),
        ("DELETE", "/v1/admin/software-nodes/00000000-0000-0000-0000-000000000000"),
        ("GET", "/v1/admin/tool-wire-health"),
        ("GET", "/v1/admin/external-system-definitions"),
        ("GET", "/v1/admin/external-system-users"),
    ]

    def test_protected_endpoints_401_without_token(self):
        for method, path in self._PROTECTED:
            with self.subTest(path=path):
                r = _client.request(method, path)
                self.assertEqual(r.status_code, 401, f"{method} {path} -> {r.status_code}")

    def test_garbage_token_401(self):
        r = _client.get("/v1/me", headers={"Authorization": "Bearer not-a-jwt"})
        self.assertEqual(r.status_code, 401)

    def test_login_bad_platform_key_403(self):
        r = _client.post("/v1/auth/login", json={"user_id": str(uuid.uuid4()), "platform_key": "wrong"})
        self.assertEqual(r.status_code, 403)

    def test_login_bad_user_id_400(self):
        # platform_key 对(与 app 同源读 env)、user_id 非 UUID:在 ensure_user_row
        # 落库之前就 400(仅进程内请求,不外发)
        r = _client.post("/v1/auth/login", json={"user_id": "not-a-uuid", "platform_key": _CFG.platform_key})
        self.assertEqual(r.status_code, 400)


class ExternalSystemRoutesTests(unittest.TestCase):
    def test_provider_catalog_and_create_are_user_scoped(self):
        provider = {
            "provider": "generic_openapi",
            "title": "通用 OpenAPI 系统",
            "configured": True,
        }
        with patch("web.routers.external_systems.provider_catalog", return_value=[provider]) as catalog:
            r = _client.get("/v1/external-system-providers", headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()["providers"], [provider])
        catalog.assert_called_once_with(_UID)

        created = {"external_system_id": str(uuid.uuid4()), "username_masked": "me***r"}
        definition_id = uuid.uuid4()
        with patch("web.routers.external_systems.create_external_system", return_value=created) as create:
            r = _client.post(
                "/v1/external-systems",
                headers=_AUTH,
                json={
                    "definition_id": str(definition_id),
                    "name": "材料数据平台",
                    "credentials": {
                        "username": "mes-user",
                        "password": "secret",
                    },
                },
            )
        self.assertEqual(r.status_code, 201)
        self.assertEqual(r.json(), created)
        self.assertEqual(create.call_args.args[0], _UID)
        self.assertEqual(create.call_args.kwargs["definition_id"], definition_id)
        self.assertEqual(
            create.call_args.kwargs["credentials"],
            {"username": "mes-user", "password": "secret"},
        )

    def test_create_accepts_dynamic_credentials(self):
        created = {"external_system_id": str(uuid.uuid4()), "username_masked": "***"}
        definition_id = uuid.uuid4()
        with patch("web.routers.external_systems.create_external_system", return_value=created) as create:
            response = _client.post(
                "/v1/external-systems",
                headers=_AUTH,
                json={
                    "definition_id": str(definition_id),
                    "name": "LIMS",
                    "credentials": {"api_key": "secret-key"},
                },
            )
        self.assertEqual(response.status_code, 201)
        self.assertEqual(create.call_args.kwargs["credentials"], {"api_key": "secret-key"})

    def test_invalid_connection_id_is_not_forwarded_to_service(self):
        with patch("web.routers.external_systems.test_external_system") as test_connection:
            r = _client.post("/v1/external-systems/not-a-uuid/test", headers=_AUTH)
        self.assertEqual(r.status_code, 404)
        test_connection.assert_not_called()


class KbRoutesTests(unittest.TestCase):
    def test_kb_crud_fs_only(self):
        self.assertEqual(_client.get("/v1/kb", headers=_AUTH).json(), {"results": []})
        # 建库
        r = _client.post("/v1/kb", json={"name": "测试库"}, headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()["name"], "测试库")
        # 非法名(路径字符)→ 400
        self.assertEqual(_client.post("/v1/kb", json={"name": "../逃逸"}, headers=_AUTH).status_code, 400)
        # 详情(空库,ingest 进度为内存态)
        r = _client.get("/v1/kb/测试库", headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertIn("ingest", r.json())
        # 不存在的库 / 单篇 → 404
        self.assertEqual(_client.get("/v1/kb/不存在", headers=_AUTH).status_code, 404)
        self.assertEqual(_client.get("/v1/kb/测试库/docs/nope.md", headers=_AUTH).status_code, 404)
        # 删库
        self.assertEqual(_client.delete("/v1/kb/测试库", headers=_AUTH).json(), {"deleted": "测试库"})
        self.assertEqual(_client.delete("/v1/kb/测试库", headers=_AUTH).status_code, 404)


class SkillMemoryRoutesTests(unittest.TestCase):
    def test_skills_list_and_detail(self):
        r = _client.get("/v1/skills", headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        names = {s["name"] for s in r.json()["skills"]}
        self.assertIn("ppt", names)      # 内置 skill 该在
        detail = _client.get("/v1/skills/ppt", headers=_AUTH)
        self.assertEqual(detail.status_code, 200)
        self.assertTrue(detail.json()["content"].strip())
        self.assertEqual(_client.get("/v1/skills/不存在的", headers=_AUTH).status_code, 404)

    def test_builtin_skill_not_deletable(self):
        # 只能删 user 源;对内置返回 404(等同"用户那里没有这个可删的")
        self.assertEqual(_client.delete("/v1/skills/ppt", headers=_AUTH).status_code, 404)

    def test_memory_readonly_view(self):
        r = _client.get("/v1/memory", headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertEqual(_client.get("/v1/memory/extended/nope.md", headers=_AUTH).status_code, 404)
        # 路径穿越文件名 → 404(校验收口 core/memory.py)
        self.assertEqual(_client.get("/v1/memory/extended/..%2Fcore.md", headers=_AUTH).status_code, 404)


class FilesRoutesTests(unittest.TestCase):
    """files 的纯 FS 面:子目录列表 / 下载 / 越界 / 非顶层改名 / 删 / 拷。
    顶层目录 rename/delete(DB-aware)与 upload(磁盘配额)留给 DB 套件。"""

    @classmethod
    def setUpClass(cls):
        cls.wd = _user_root() / "route-test-wd"
        (cls.wd / "sub").mkdir(parents=True, exist_ok=True)
        (cls.wd / "figures" / ".meta").mkdir(parents=True, exist_ok=True)
        (cls.wd / "figures" / ".private").mkdir(parents=True, exist_ok=True)
        (cls.wd / "a.txt").write_text("hello", encoding="utf-8")
        (cls.wd / "sub" / "b.txt").write_text("world", encoding="utf-8")
        (cls.wd / "figures" / ".meta" / "image.png.json").write_text(
            "{}", encoding="utf-8"
        )
        (cls.wd / "figures" / ".env").write_text("SECRET=x", encoding="utf-8")
        (_user_root() / ".root-private").mkdir(exist_ok=True)

    def test_list_subdir(self):
        r = _client.get("/v1/files", params={"path": "route-test-wd"}, headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        body = r.json()
        self.assertTrue(body["exists"])
        names = {e["name"] for e in body["entries"]}
        self.assertEqual(names, {"sub", "figures", "a.txt"})
        self.assertEqual(body["current"], "route-test-wd")

    def test_dot_directories_are_hidden_by_default(self):
        r = _client.get(
            "/v1/files",
            params={"path": "route-test-wd/figures"},
            headers=_AUTH,
        )
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()["entries"], [])

    def test_enumerator_can_include_dot_directories_but_never_dotfiles(self):
        from web.userfiles import enumerate_files

        entries, _crumbs, _exists = enumerate_files(
            _user_root(),
            self.wd / "figures",
            include_hidden_dirs=True,
        )
        self.assertEqual(
            {entry["name"] for entry in entries},
            {".meta", ".private"},
        )

    def test_api_hidden_toggle_is_limited_to_selected_task_tree(self):
        task_id = uuid.uuid4()
        with patch(
            "web.routers.files._task_working_dir",
            return_value=(task_id, self.wd),
        ):
            shown = _client.get(
                "/v1/files",
                params={
                    "path": "route-test-wd/figures",
                    "include_hidden": "true",
                    "task_id": str(task_id),
                },
                headers=_AUTH,
            ).json()
            self.assertEqual(
                {entry["name"] for entry in shown["entries"]},
                {".meta", ".private"},
            )
            self.assertTrue(shown["hidden_dirs_included"])

            with patch("web.routers.files.system_wd_names", return_value=set()):
                root = _client.get(
                    "/v1/files",
                    params={"include_hidden": "true", "task_id": str(task_id)},
                    headers=_AUTH,
                ).json()
        self.assertNotIn(
            ".root-private",
            {entry["name"] for entry in root["entries"]},
        )
        self.assertFalse(root["hidden_dirs_included"])

    def test_create_directory_at_root_and_in_current_directory(self):
        root_name = "mkdir-api-root"
        root_dir = _user_root() / root_name
        try:
            created = _client.post(
                "/v1/files/mkdir",
                json={"path": "", "name": f"  {root_name}  "},
                headers=_AUTH,
            )
            self.assertEqual(created.status_code, 200)
            self.assertEqual(created.json()["path"], root_name)
            self.assertTrue(root_dir.is_dir())

            child = _client.post(
                "/v1/files/mkdir",
                json={"path": root_name, "name": "资料"},
                headers=_AUTH,
            )
            self.assertEqual(child.status_code, 200)
            self.assertEqual(child.json()["path"], f"{root_name}/资料")
            self.assertTrue((root_dir / "资料").is_dir())

            duplicate = _client.post(
                "/v1/files/mkdir",
                json={"path": root_name, "name": "资料"},
                headers=_AUTH,
            )
            self.assertEqual(duplicate.status_code, 409)

            for name in (".hidden", "../escape", "a/b"):
                with self.subTest(name=name):
                    invalid = _client.post(
                        "/v1/files/mkdir",
                        json={"path": root_name, "name": name},
                        headers=_AUTH,
                    )
                    self.assertEqual(invalid.status_code, 400)
        finally:
            shutil.rmtree(root_dir, ignore_errors=True)

    def test_download_and_errors(self):
        r = _client.get("/v1/files/download", params={"path": "route-test-wd/a.txt"}, headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.content, b"hello")
        self.assertEqual(
            _client.get("/v1/files/download", params={"path": "route-test-wd/nope.txt"}, headers=_AUTH).status_code,
            404,
        )
        self.assertEqual(
            _client.get("/v1/files/download", params={"path": "route-test-wd"}, headers=_AUTH).status_code,
            400,  # 目录不是文件
        )

    def test_path_escape_rejected(self):
        for bad in ("../x", "/etc/passwd", "..\\x"):
            with self.subTest(path=bad):
                r = _client.get("/v1/files/download", params={"path": bad}, headers=_AUTH)
                self.assertEqual(r.status_code, 400)

    def test_rename_delete_copy_nontop(self):
        (self.wd / "sub" / "c.txt").write_text("c", encoding="utf-8")
        # 非顶层改名:纯 FS,tasks_updated=0
        with patch("core.artifact_lifecycle.rename_active_artifacts", return_value=0):
            r = _client.post("/v1/files/rename",
                             json={"path": "route-test-wd/sub/c.txt", "new_name": "c2.txt"}, headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()["tasks_updated"], 0)
        self.assertTrue((self.wd / "sub" / "c2.txt").is_file())
        # 拷贝到子目录内(非顶层,无 DB 闸)
        (self.wd / "dest").mkdir(exist_ok=True)
        with patch("core.artifact_lifecycle.copy_active_artifacts", return_value=0):
            r = _client.post("/v1/files/copy",
                             json={"paths": ["route-test-wd/sub/c2.txt"], "dest_dir": "route-test-wd/dest"},
                             headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertTrue((self.wd / "dest" / "c2.txt").is_file())
        # 目标已存在 → 409(预检整批 abort)
        r = _client.post("/v1/files/copy",
                         json={"paths": ["route-test-wd/sub/c2.txt"], "dest_dir": "route-test-wd/dest"},
                         headers=_AUTH)
        self.assertEqual(r.status_code, 409)
        # 删文件
        with patch("core.artifact_lifecycle.trash_active_artifacts", return_value=0):
            r = _client.post("/v1/files/delete", json={"path": "route-test-wd/dest/c2.txt"}, headers=_AUTH)
        self.assertEqual(r.status_code, 200)
        self.assertFalse((self.wd / "dest" / "c2.txt").exists())
        # 删非空目录不带 recursive → 400
        with patch("core.artifact_lifecycle.trash_active_artifacts", return_value=0):
            r = _client.post("/v1/files/delete", json={"path": "route-test-wd/sub"}, headers=_AUTH)
        self.assertEqual(r.status_code, 400)


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