tests/test_server.py
Rama master · réplica de lectura del repositorio original.
"""无桌面入口的进程退出与现有数据保护。"""
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from peach import server, settings_file
from starlette.requests import Request
class ServerTests(unittest.TestCase):
def test_reader_initialization_does_not_launch_a_ledger_migration(self):
with tempfile.TemporaryDirectory() as directory:
config = settings_file.load_config(environ={'PEACH_DATA_ROOT': str(Path(directory).resolve())})
settings_file.write(config)
config.directory('database').mkdir()
database = config.directory('database') / 'ledger.db'
database.write_bytes(b'unchanged reader ledger')
config = settings_file.load_config(environ={'PEACH_DATA_ROOT': str(config.data_root)})
with patch.object(server.settings_file, 'load_config', return_value=config), patch(
'peach.runtime_prepare.configured_ledger_writer', return_value=(False, '读取端')), \
patch.object(server, 'launch') as launch:
server.initialize()
launch.assert_not_called()
self.assertEqual(database.read_bytes(), b'unchanged reader ledger')
def test_headless_configuration_accepts_private_peers_and_rejects_forwarding(self):
from peach.routes_configuration import local_client, managed_configuration
with patch.dict("os.environ", {"PEACH_HEADLESS_MANAGED": "1"}):
self.assertTrue(managed_configuration())
for peer, headers, allowed in (
("192.0.2.1", [], True), ("198.51.100.10", [], True),
("8.8.8.8", [], False), ("169.254.1.1", [], False),
("0.0.0.0", [], False), ("240.0.0.1", [], False),
("127.0.0.1", [], True), ("fd00::1", [], True),
("192.0.2.1", [(b"x-forwarded-for", b"8.8.8.8")], False),
("192.0.2.1", [(b"x-forwarded-host", b"example.com")], False),
):
request = Request({"type": "http", "client": (peer, 1234),
"server": ("0.0.0.0", 8900), "headers": headers,
"app": SimpleNamespace(state=SimpleNamespace(tunnel=None))})
with self.subTest(peer=peer, headers=headers):
self.assertEqual(local_client(request), allowed)
def test_cli_without_mdns_or_tunnel_needs_no_network_route(self):
from peach import cli
with patch.object(cli, "_require_readable_settings"), \
patch.object(cli, "_serve_token", return_value="test"), \
patch.object(cli, "_build_sync", return_value=None), \
patch.object(cli, "create_app"), patch.object(cli, "TUNNEL_ENABLED", False), \
patch("peach.mdns.lan_ipv4", side_effect=OSError("no route")) as route, \
patch.object(cli, "_run_uvicorn"):
self.assertEqual(cli.main(["serve", "--host", "0.0.0.0", "--no-mdns"]), 0)
route.assert_not_called()
def test_every_served_listener_gets_the_handler_that_can_listen_again(self):
"""客户端在 accept 完成前断开、监听被标准库关掉时，处理器要拿着 uvicorn 的 Server 才能重新监听。"""
import asyncio
from peach import cli, error_reporting
handlers = []
for arguments in (["serve", "--host", "0.0.0.0", "--no-mdns"],
["serve", "--host", "0.0.0.0", "--port", "80", "--redirect-origin", "https://peach.test"]):
handlers.clear()
with self.subTest(arguments=arguments), \
patch.object(cli, "_require_readable_settings"), \
patch.object(cli, "_serve_token", return_value="test"), \
patch.object(cli, "_build_sync", return_value=None), \
patch.object(cli, "create_app"), patch.object(cli, "TUNNEL_ENABLED", False), \
patch("uvicorn.Config") as config, patch("uvicorn.Server") as uvicorn_server:
config.return_value.get_loop_factory.return_value = None
uvicorn_server.return_value.serve = AsyncMock(
side_effect=lambda: handlers.append(asyncio.get_running_loop().get_exception_handler()))
self.assertEqual(cli.main(arguments), 0)
handler, = handlers
self.assertIs(handler.func, error_reporting.asyncio_handler)
self.assertEqual(handler.keywords, {"uvicorn_server": uvicorn_server.return_value})
def test_initialization_preserves_existing_configuration(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(root)})
settings_file.write(config)
config.directory('database').mkdir()
config.directory('database').joinpath('ledger.db').touch()
existing = config.path.read_bytes()
with patch.object(server.settings_file, "load_config", return_value=settings_file.load_config(
environ={"PEACH_DATA_ROOT": str(root)})), patch.object(server, "launch") as launch:
launch.return_value.wait.return_value = 0
server.initialize()
launch.assert_called_once_with('migrate', 'upgrade', '--yes', '--db', str(config.directory('database') / 'ledger.db'))
self.assertEqual(config.path.read_bytes(), existing)
def test_initialization_rejects_database_without_configuration(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(root)})
config.directory("database").mkdir()
config.directory("database").joinpath("ledger.db").write_bytes(b"existing")
with patch.object(server.settings_file, "load_config", return_value=config), \
patch.object(server, "launch") as launch:
with self.assertRaisesRegex(RuntimeError, "已有数据库"):
server.initialize()
launch.assert_not_called()
def test_stop_waits_for_child_exit(self):
process = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"],
stdin=subprocess.DEVNULL)
try:
server.stop_process(process)
self.assertIsNotNone(process.poll())
finally:
server.stop_process(process)
def test_relocation_restarts_the_service_on_the_new_data_root(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(root)})
process = subprocess.Popen([sys.executable, "-c", "raise SystemExit(17)"], stdin=subprocess.DEVNULL)
process.wait(timeout=15)
calls = []
def relocation(stop, restart):
if calls:
return False
calls.append(True)
stop()
restart({"PEACH_DATA_ROOT": str(root / "moved")})
return True
with patch.dict(server.os.environ, {}), \
patch.object(server.settings_file, "load_config", return_value=config), \
patch("peach.data_storage.poll_headless", side_effect=relocation), \
patch.object(server, "launch", return_value=process) as launch:
self.assertEqual(server.supervise("127.0.0.1", 8900, threading.Event()), 17)
# 迁移后看守进程自己也按新位置读设置，拉起的服务继承这个位置。
self.assertEqual(server.os.environ["PEACH_DATA_ROOT"], str(root / "moved"))
self.assertEqual(launch.call_count, 1)
def test_service_failure_exits_supervisor(self):
with tempfile.TemporaryDirectory() as directory:
config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(Path(directory).resolve())})
process = subprocess.Popen([sys.executable, "-c", "raise SystemExit(17)"],
stdin=subprocess.DEVNULL)
process.wait(timeout=15)
with patch.object(server.settings_file, "load_config", return_value=config), \
patch.object(server, "launch", return_value=process):
self.assertEqual(server.supervise("127.0.0.1", 8900, threading.Event()), 17)
def test_a_fresh_docker_volume_starts_unconfigured_without_init(self):
with tempfile.TemporaryDirectory() as directory:
config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(Path(directory).resolve())})
with patch.dict(server.os.environ, {"PEACH_DISTRIBUTION": "docker"}), \
patch.object(server.settings_file, "load_config", return_value=config), \
patch.object(server, "launch") as launch:
server.initialize()
launch.assert_not_called()
self.assertFalse(config.path.exists())
config.directory("database").mkdir(parents=True)
config.directory("database").joinpath("ledger.db").write_bytes(b"existing")
with self.assertRaisesRegex(RuntimeError, "已有数据库"):
server.initialize()
launch.assert_not_called()
def test_the_setup_service_is_replaced_once_the_settings_file_appears(self):
from dataclasses import replace
from unittest.mock import Mock
with tempfile.TemporaryDirectory() as directory:
absent = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(Path(directory).resolve())})
present = replace(absent, present=True)
state = {"configured": False}
stopping = threading.Event()
launched = []
def start(*arguments):
launched.append(arguments)
if len(launched) == 1:
state["configured"] = True # 首次设置提交，设置文件写好了
else:
stopping.set()
return Mock(poll=Mock(return_value=None))
with patch.object(server.settings_file, "load_config",
side_effect=lambda: present if state["configured"] else absent), \
patch("peach.data_storage.poll_headless", return_value=False), \
patch.object(server, "launch", side_effect=start):
self.assertEqual(server.supervise("0.0.0.0", 8900, stopping), 0)
self.assertEqual([arguments[:2] for arguments in launched], [("serve", "--setup"), ("serve", "--host")])
def test_access_password_from_the_environment_is_written_once_and_never_echoed(self):
from peach import access
with tempfile.TemporaryDirectory() as directory:
config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(Path(directory).resolve())})
path = config.directory("secrets") / "access.json"
with patch.dict(server.os.environ, {server.ACCESS_PASSWORD_ENV: "short"}):
message = server.seed_access_password(config)
self.assertNotIn(server.ACCESS_PASSWORD_ENV, server.os.environ)
self.assertIn("8–256", message)
self.assertFalse(path.exists())
with patch.dict(server.os.environ, {server.ACCESS_PASSWORD_ENV: "container-secret"}):
message = server.seed_access_password(config)
self.assertNotIn(server.ACCESS_PASSWORD_ENV, server.os.environ, "不传给服务子进程")
self.assertNotIn("container-secret", message)
policy = access.load(path)
self.assertEqual(policy["mode"], "password")
self.assertTrue(access.verify(policy, "container-secret"))
self.assertNotIn("container-secret", path.read_text(encoding="utf-8"))
with patch.dict(server.os.environ, {server.ACCESS_PASSWORD_ENV: "another-secret"}):
message = server.seed_access_password(config)
self.assertIn("忽略", message)
self.assertNotIn("another-secret", message)
self.assertEqual(access.load(path)["revision"], policy["revision"])
with patch.dict(server.os.environ, {}, clear=False):
server.os.environ.pop(server.ACCESS_PASSWORD_ENV, None)
self.assertEqual(server.seed_access_password(config), "")