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

from __future__ import annotations

import json
import socket
import tempfile
import threading
import unittest
from pathlib import Path

import admin_backend
import backend_client


class BackendIntegrationTest(unittest.TestCase):
    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
            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()
                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
                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:
            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)

            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)
                    synced = json.loads(
                        Path(ai_config._SETTINGS_FILE).read_text(encoding="utf-8")
                    )
                    self.assertEqual(synced["AI_MODEL"], "integration-test-model")
                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()