147 lines
6.2 KiB
Python
147 lines
6.2 KiB
Python
# -*- 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()
|