565 lines
23 KiB
Python
565 lines
23 KiB
Python
# -*- 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()
|