# -*- coding: utf-8 -*-
"""配置后台与桌面端同步的本地闭环测试。"""

from __future__ import annotations

import json
import hashlib
import socket
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from unittest import mock
from pathlib import Path

import admin_backend
import app_version
import backend_client


class BackendIntegrationTest(unittest.TestCase):
    def test_admin_page_contains_non_saving_model_test_button(self) -> None:
        config = admin_backend.load_initial_config()
        card = admin_backend.AdminHandler.config_card(
            {"role": "admin"},
            "csrf-token",
            {"version": 1, "updated_at": "now", "updated_by_name": "admin"},
            config,
        )
        self.assertIn("formaction='/admin/model/test'", card)
        self.assertIn("formtarget='_blank'", card)
        self.assertIn("测试模型连通性", card)
        self.assertIn("value='openai'", card)
        self.assertIn("value='dify'", card)
        self.assertIn("value='comfyui'", card)
        self.assertIn("name='AI_DEVELOPMENT_MODE'", card)
        self.assertIn("开启开发模式", card)
        self.assertIn("name='AI_UI_GUARD_ENABLED'", card)
        self.assertIn("启用 AI 页面守护", card)

    def test_development_diagnostics_are_cloud_controlled_and_redacted(self) -> None:
        config = {
            "AI_DEVELOPMENT_MODE": True,
            "AI_PROVIDER_TYPE": "openai",
            "AI_API_BASE": "https://api.example/v1",
            "AI_API_KEY": "top-secret-key",
            "AI_MODEL": "test-model",
            "AI_MCP_SERVERS": [
                {
                    "headers": {"Authorization": "Bearer hidden-auth"},
                    "env": {"ACCESS_TOKEN": "hidden-token"},
                    "args": [
                        "--token",
                        "hidden-argument",
                        "--password=hidden-inline",
                        "Bearer hidden-bearer",
                    ],
                }
            ],
        }
        response = {"version": 8, "updated_at": "now"}
        diagnostics = backend_client._config_diagnostics(
            config,
            response,
            "https://cloud.example/api/v1/desktop/config?api_key=query-secret",
        )
        rendered = "\n".join(diagnostics)
        self.assertIn("https://cloud.example/api/v1/desktop/config", rendered)
        self.assertIn("云端配置版本: v8", rendered)
        self.assertIn('"AI_MODEL": "test-model"', rendered)
        for secret in (
            "top-secret-key",
            "hidden-auth",
            "hidden-token",
            "hidden-argument",
            "hidden-inline",
            "hidden-bearer",
            "query-secret",
        ):
            self.assertNotIn(secret, rendered)
        self.assertEqual(
            backend_client._config_diagnostics(
                {**config, "AI_DEVELOPMENT_MODE": False}, response, "https://cloud.example/config"
            ),
            [],
        )

    def test_model_request_diagnostics_never_print_the_api_key(self) -> None:
        import ai_chat
        import ai_config

        previous = {
            "AI_DEVELOPMENT_MODE": getattr(ai_config, "AI_DEVELOPMENT_MODE", False),
            "AI_API_BASE": ai_config.AI_API_BASE,
            "AI_API_KEY": ai_config.AI_API_KEY,
            "AI_MODEL": ai_config.AI_MODEL,
        }
        try:
            ai_config.AI_DEVELOPMENT_MODE = True
            ai_config.AI_API_BASE = "https://api.example/v1"
            ai_config.AI_API_KEY = "never-print-this-key"
            ai_config.AI_MODEL = "diagnostic-model"
            with mock.patch("builtins.print") as printer:
                ai_chat._log_request_diagnostics(
                    "https://api.example/v1/chat/completions", "OpenAI 兼容"
                )
            rendered = "\n".join(
                " ".join(str(item) for item in call.args)
                for call in printer.call_args_list
            )
            self.assertIn("https://api.example/v1/chat/completions", rendered)
            self.assertIn("diagnostic-model", rendered)
            self.assertNotIn("never-print-this-key", rendered)
        finally:
            for key, value in previous.items():
                setattr(ai_config, key, value)

    def test_model_connection_uses_unsaved_values_and_saved_key(self) -> None:
        current = {
            "AI_API_BASE": "https://saved.example/v1",
            "AI_API_KEY": "saved-secret",
            "AI_MODEL": "saved-model",
            "AI_TIMEOUT": 120,
        }
        config = admin_backend.model_test_config(
            {
                "AI_API_BASE": "https://new.example/v1",
                "AI_API_KEY": "",
                "AI_MODEL": "new-model",
                "AI_TIMEOUT": "180",
            },
            current,
        )
        self.assertEqual(config["endpoint"], "https://new.example/v1/chat/completions")
        self.assertEqual(config["api_key"], "saved-secret")
        self.assertEqual(config["model"], "new-model")
        self.assertEqual(config["timeout"], 60)

    def test_model_connection_success_does_not_expose_key(self) -> None:
        config = admin_backend.model_test_config(
            {
                "AI_API_BASE": "https://api.example/v1",
                "AI_API_KEY": "top-secret-key",
                "AI_MODEL": "test-model",
                "AI_TIMEOUT": 10,
            },
            {},
        )
        response = json.dumps(
            {"choices": [{"message": {"content": "OK"}}]}
        ).encode("utf-8")
        with mock.patch(
            "admin_backend._perform_http_request", return_value=(200, response)
        ) as call:
            result = admin_backend.test_model_connection(config)
        self.assertTrue(result["ok"])
        self.assertEqual(result["http_status"], 200)
        self.assertNotIn("top-secret-key", json.dumps(result, ensure_ascii=False))
        self.assertEqual(call.call_args.args[0], config["endpoint"])
        self.assertEqual(
            call.call_args.kwargs["headers"]["Authorization"], "Bearer top-secret-key"
        )
        payload = call.call_args.kwargs["payload"]
        self.assertEqual(payload["model"], "test-model")

    def test_model_connection_error_redacts_key(self) -> None:
        config = admin_backend.model_test_config(
            {
                "AI_API_BASE": "https://api.example/v1",
                "AI_API_KEY": "top-secret-key",
                "AI_MODEL": "test-model",
                "AI_TIMEOUT": 10,
            },
            {},
        )
        response = json.dumps(
            {"error": {"message": "invalid top-secret-key"}}
        ).encode("utf-8")
        with mock.patch(
            "admin_backend._perform_http_request", return_value=(401, response)
        ):
            result = admin_backend.test_model_connection(config)
        self.assertFalse(result["ok"])
        self.assertEqual(result["http_status"], 401)
        self.assertIn("API Key 无效", result["message"])
        self.assertNotIn("top-secret-key", json.dumps(result, ensure_ascii=False))

    def test_dify_and_comfyui_use_provider_specific_endpoints(self) -> None:
        dify = admin_backend.model_test_config(
            {
                "AI_PROVIDER_TYPE": "dify",
                "AI_API_BASE": "https://dify.example/v1",
                "AI_API_KEY": "app-secret",
                "AI_MODEL": "",
                "AI_TIMEOUT": 10,
            },
            {},
        )
        self.assertEqual(dify["endpoint"], "https://dify.example/v1/chat-messages")
        self.assertEqual(dify["provider_type"], "dify")

        comfyui = admin_backend.model_test_config(
            {
                "AI_PROVIDER_TYPE": "comfyui",
                "AI_API_BASE": "http://127.0.0.1:8188",
                "AI_API_KEY": "",
                "AI_MODEL": "",
                "AI_TIMEOUT": 10,
            },
            {},
        )
        self.assertEqual(comfyui["endpoint"], "http://127.0.0.1:8188/system_stats")
        with mock.patch(
            "admin_backend._perform_http_request",
            return_value=(200, b'{"system": {"os": "windows"}, "devices": []}'),
        ) as call:
            result = admin_backend.test_model_connection(comfyui)
        self.assertTrue(result["ok"])
        self.assertEqual(call.call_args.kwargs["method"], "GET")
        self.assertIsNone(call.call_args.kwargs["payload"])

    def test_desktop_ai_respects_explicit_dify_provider(self) -> None:
        import ai_chat
        import ai_config

        old_provider = ai_config.AI_PROVIDER_TYPE
        old_base = ai_config.AI_API_BASE
        try:
            ai_config.AI_PROVIDER_TYPE = "dify"
            ai_config.AI_API_BASE = "https://dify.example/v1"
            self.assertTrue(ai_chat._is_dify_endpoint())
            self.assertEqual(
                ai_chat._completions_url(),
                "https://dify.example/v1/chat-messages",
            )
            ai_config.AI_PROVIDER_TYPE = "openai"
            self.assertFalse(ai_chat._is_dify_endpoint())
        finally:
            ai_config.AI_PROVIDER_TYPE = old_provider
            ai_config.AI_API_BASE = old_base

    def test_model_test_api_requires_edit_role(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            database = admin_backend.Database(Path(directory) / "test.db")
            database.initialize("InitialAdmin123")
            admin = database.authenticate("admin", "InitialAdmin123")
            database.change_password(
                admin["id"], "InitialAdmin123", "ChangedAdmin123", "127.0.0.1"
            )
            database.create_user(
                "readonly.user", "ViewerPassword123", "viewer", admin["id"], "127.0.0.1"
            )
            viewer = database.authenticate("readonly.user", "ViewerPassword123")
            database.change_password(
                viewer["id"], "ViewerPassword123", "ViewerChanged123", "127.0.0.1"
            )
            admin_token, _ = database.create_token(admin["id"], "api", "test", 3600)
            viewer_token, _ = database.create_token(viewer["id"], "api", "test", 3600)
            server = admin_backend.AdminServer(("127.0.0.1", 0), database)
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            url = f"http://127.0.0.1:{server.server_address[1]}/api/v1/model/test"
            payload = json.dumps(
                {
                    "AI_API_BASE": "https://api.example/v1",
                    "AI_MODEL": "test-model",
                    "AI_TIMEOUT": 10,
                }
            ).encode("utf-8")

            def request(token: str):
                return urllib.request.Request(
                    url,
                    data=payload,
                    headers={
                        "Authorization": f"Bearer {token}",
                        "Content-Type": "application/json",
                    },
                    method="POST",
                )

            fake_result = {
                "ok": True,
                "provider": "OpenAI 兼容",
                "model": "test-model",
                "endpoint": "https://api.example/v1/chat/completions",
                "http_status": 200,
                "latency_ms": 8,
                "message": "连接成功，模型回复：OK",
            }
            try:
                with mock.patch(
                    "admin_backend.test_model_connection", return_value=fake_result
                ):
                    with urllib.request.urlopen(request(admin_token), timeout=3) as response:
                        data = json.loads(response.read().decode("utf-8"))
                    self.assertTrue(data["ok"])
                    with self.assertRaises(urllib.error.HTTPError) as context:
                        urllib.request.urlopen(request(viewer_token), timeout=3)
                    self.assertEqual(context.exception.code, 403)
            finally:
                server.shutdown()
                server.server_close()
                thread.join(timeout=2)

    def test_release_status_detects_optional_and_forced_updates(self) -> None:
        optional = app_version.release_status(
            {"latest_version": "1.0.1", "force_upgrade": False}
        )
        self.assertTrue(optional["update_available"])
        self.assertFalse(optional["force_upgrade"])

        forced = app_version.release_status(
            {"latest_version": "1.0.1", "force_upgrade": True}
        )
        self.assertTrue(forced["force_upgrade"])

        current = app_version.release_status(
            {"latest_version": app_version.APP_VERSION, "force_upgrade": True}
        )
        self.assertFalse(current["update_available"])
        self.assertFalse(current["force_upgrade"])

    def test_release_form_validation(self) -> None:
        release = admin_backend.validate_release_form(
            {
                "latest_version": "v1.2.3",
                "download_url": "https://example.com/client.exe",
                "release_notes": "修复已知问题",
                "force_upgrade": "1",
            }
        )
        self.assertEqual(release["latest_version"], "1.2.3")
        self.assertTrue(release["force_upgrade"])
        with self.assertRaises(ValueError):
            admin_backend.validate_release_form(
                {"latest_version": "1.2", "force_upgrade": "1"}
            )
        with self.assertRaises(ValueError):
            admin_backend.validate_release_form(
                {"latest_version": "1.2.3", "force_upgrade": "1"}
            )

    def test_desktop_sync_key_matches_server(self) -> None:
        self.assertEqual(
            backend_client.DESKTOP_SYNC_KEY,
            admin_backend.DEFAULT_DESKTOP_SYNC_KEY,
        )

    def test_pbkdf2_fallback_matches_standard_library(self) -> None:
        password = b"FallbackPassword123"
        salt = b"0123456789abcdef"
        expected = hashlib.pbkdf2_hmac("sha256", password, salt, 1_000)
        native_pbkdf2 = admin_backend.hashlib.pbkdf2_hmac
        try:
            admin_backend.hashlib.pbkdf2_hmac = None
            actual = admin_backend.pbkdf2_sha256(password, salt, 1_000)
        finally:
            admin_backend.hashlib.pbkdf2_hmac = native_pbkdf2
        self.assertEqual(actual, expected)

    def test_stale_runtime_file_is_ignored(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            runtime_file = root / "backend_runtime.json"
            runtime_file.write_text(
                json.dumps(
                    {
                        "pid": 0,
                        "host": "127.0.0.1",
                        "port": 8765,
                        "server_url": "http://127.0.0.1:8765",
                        "local_sync_token": "stale-token",
                    }
                ),
                encoding="utf-8",
            )
            old_runtime_file = backend_client.RUNTIME_FILE
            try:
                backend_client.RUNTIME_FILE = runtime_file
                self.assertEqual(backend_client.discover_local_runtime(), {})
                self.assertEqual(backend_client.discover_local_sync_token(), "")
            finally:
                backend_client.RUNTIME_FILE = old_runtime_file

    def test_local_agent_name_overrides_cloud_and_blank_follows_cloud(self) -> None:
        import ai_config

        old_agent = ai_config.AI_AGENT_NAME
        old_cloud = ai_config.AI_CLOUD_AGENT_NAME
        old_local = ai_config.AI_LOCAL_AGENT_NAME
        old_prompt = ai_config.AI_SYSTEM_PROMPT
        try:
            ai_config.AI_CLOUD_AGENT_NAME = "旧云端客服"
            ai_config.AI_LOCAL_AGENT_NAME = ""
            ai_config.AI_AGENT_NAME = "旧云端客服"
            ai_config.set_local_agent_name_override("本机客服", persist=False)
            ai_config.apply_settings({"AI_AGENT_NAME": "新云端客服"}, persist=False)
            self.assertEqual(ai_config.AI_CLOUD_AGENT_NAME, "新云端客服")
            self.assertEqual(ai_config.AI_AGENT_NAME, "本机客服")
            ai_config.set_local_agent_name_override("", persist=False)
            self.assertEqual(ai_config.AI_AGENT_NAME, "新云端客服")
        finally:
            ai_config.AI_AGENT_NAME = old_agent
            ai_config.AI_CLOUD_AGENT_NAME = old_cloud
            ai_config.AI_LOCAL_AGENT_NAME = old_local
            ai_config.AI_SYSTEM_PROMPT = old_prompt

    def test_occupied_port_automatically_uses_next_port(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            database = admin_backend.Database(root / "test.db")
            database.initialize("InitialAdmin123")
            blocker = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            blocker.bind(("127.0.0.1", 0))
            blocker.listen(1)
            occupied_port = int(blocker.getsockname()[1])
            self.assertLess(occupied_port, 65535)
            server = None
            old_runtime_file = backend_client.RUNTIME_FILE
            old_connection_file = backend_client.CONNECTION_FILE
            old_default_server_url = backend_client.DEFAULT_SERVER_URL
            server_thread = None
            try:
                server, actual_port = admin_backend.create_server(
                    "127.0.0.1", occupied_port, database, max_attempts=5
                )
                self.assertGreater(actual_port, occupied_port)
                self.assertLessEqual(actual_port, occupied_port + 4)
                runtime_file = root / "backend_runtime.json"
                admin_backend.write_runtime_info(
                    runtime_file,
                    "127.0.0.1",
                    actual_port,
                    local_sync_token=server.local_sync_token,
                )
                backend_client.RUNTIME_FILE = runtime_file
                backend_client.CONNECTION_FILE = root / "connection.json"
                self.assertEqual(
                    backend_client.discover_local_server_url(),
                    f"http://127.0.0.1:{actual_port}",
                )
                self.assertTrue(backend_client.discover_local_sync_token())
                self.assertEqual(
                    backend_client.default_settings()["server_url"],
                    f"http://127.0.0.1:{actual_port}",
                )

                config = json.loads(database.config()["config_json"])
                config["AI_MODEL"] = "startup-detected-model"
                admin = database.authenticate("admin", "InitialAdmin123")
                database.save_config(config, admin["id"], "127.0.0.1")
                server_thread = threading.Thread(
                    target=server.serve_forever, daemon=True
                )
                server_thread.start()
                backend_client.DEFAULT_SERVER_URL = f"http://127.0.0.1:{actual_port}"
                import ai_config

                old_settings_file = ai_config._SETTINGS_FILE
                try:
                    ai_config._SETTINGS_FILE = str(root / "startup_ai_settings.json")
                    result = backend_client.startup_sync_config(timeout=3.0)
                    self.assertTrue(result["synced"])
                    self.assertEqual(ai_config.AI_MODEL, "startup-detected-model")
                finally:
                    ai_config._SETTINGS_FILE = old_settings_file
            finally:
                backend_client.RUNTIME_FILE = old_runtime_file
                backend_client.CONNECTION_FILE = old_connection_file
                backend_client.DEFAULT_SERVER_URL = old_default_server_url
                if server is not None:
                    if server_thread is not None:
                        server.shutdown()
                    server.server_close()
                if server_thread is not None:
                    server_thread.join(timeout=2)
                blocker.close()

    def test_login_roles_publish_and_desktop_sync(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            major, minor, patch = (
                int(part) for part in app_version.APP_VERSION.split(".")[:3]
            )
            newer_version = f"{major}.{minor}.{patch + 1}"
            root = Path(directory)
            database = admin_backend.Database(root / "test.db")
            self.assertTrue(database.initialize("InitialAdmin123"))
            admin = database.authenticate("admin", "InitialAdmin123")
            self.assertIsNotNone(admin)
            database.change_password(
                admin["id"], "InitialAdmin123", "ChangedAdmin123", "127.0.0.1"
            )
            database.create_user(
                "readonly.user", "ViewerPassword123", "viewer", admin["id"], "127.0.0.1"
            )
            viewer = database.authenticate("readonly.user", "ViewerPassword123")
            self.assertEqual(viewer["role"], "viewer")

            config_row = database.config()
            config = json.loads(config_row["config_json"])
            config["AI_MODEL"] = "integration-test-model"
            version = database.save_config(config, admin["id"], "127.0.0.1")
            self.assertEqual(version, 2)
            database.save_release(
                newer_version,
                "https://example.com/client.exe",
                "测试升级",
                True,
                admin["id"],
                "127.0.0.1",
            )
            release = database.release()
            self.assertEqual(release["latest_version"], newer_version)
            self.assertEqual(release["force_upgrade"], 1)

            server = admin_backend.AdminServer(("127.0.0.1", 0), database)
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            old_connection_file = backend_client.CONNECTION_FILE
            try:
                backend_client.CONNECTION_FILE = root / "connection.json"
                port = server.server_address[1]
                response = backend_client.login(
                    f"http://127.0.0.1:{port}",
                    "admin",
                    "ChangedAdmin123",
                )
                self.assertEqual(response["user"]["role"], "admin")

                import ai_config

                old_settings_file = ai_config._SETTINGS_FILE
                try:
                    ai_config._SETTINGS_FILE = str(root / "synced_ai_settings.json")
                    result = backend_client.sync_config(force=True)
                    self.assertTrue(result["synced"])
                    self.assertEqual(result["version"], 2)
                    self.assertTrue(result["update_available"])
                    self.assertTrue(result["force_upgrade"])
                    self.assertEqual(
                        result["release"]["latest_version"], newer_version
                    )
                    synced = json.loads(
                        Path(ai_config._SETTINGS_FILE).read_text(encoding="utf-8")
                    )
                    self.assertEqual(synced["AI_MODEL"], "integration-test-model")
                    self.assertTrue(
                        backend_client.cached_release_status()["force_upgrade"]
                    )
                finally:
                    ai_config._SETTINGS_FILE = old_settings_file
                backend_client.logout()
                self.assertFalse(backend_client.is_configured())
            finally:
                backend_client.CONNECTION_FILE = old_connection_file
                server.shutdown()
                server.server_close()
                thread.join(timeout=2)


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