websocket_server Brick：可复用的 WebSocket 传输层app/tests/test_lifecycle.py
"""使用真实 websockets 和本机回环端口验证生命周期；不需要开发板或业务协议。"""
import importlib.util
import inspect
import math
import socket
import sys
import threading
import time
import types
import unittest
from contextlib import ExitStack
from pathlib import Path
from unittest.mock import patch
from websockets.exceptions import ConnectionClosed
from websockets.sync.client import connect
from websockets.sync.server import Server, serve as real_serve
APP_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(APP_DIR / "bricks"))
# 只在测试环境缺少 Arduino 包时替代注册装饰器；绝不放入生产 Brick。
HAS_APP_UTILS = importlib.util.find_spec("arduino") is not None
if not HAS_APP_UTILS:
arduino_stub = types.ModuleType("arduino")
utils_stub = types.ModuleType("arduino.app_utils")
utils_stub.brick = lambda cls: cls
sys.modules["arduino"] = arduino_stub
sys.modules["arduino.app_utils"] = utils_stub
from websocket_server import WebSocketServer
import websocket_server.server as server_module
def unused_port():
"""
@description : 向操作系统申请本机空闲测试端口
@param : 无参数
@return : 释放临时占位 socket 后的端口号
"""
with socket.socket() as reservation:
reservation.bind(("127.0.0.1", 0))
return reservation.getsockname()[1]
def wait_until(predicate, timeout=3.0):
"""
@description : 在明确期限内等待测试条件成立，避免无限等待
@param predicate : 返回布尔值的检查函数
@param timeout : 最大等待秒数
@return : 无返回值，超时抛出断言错误
"""
deadline = time.monotonic() + timeout
while not predicate():
if time.monotonic() >= deadline:
raise AssertionError("condition wasn't met before deadline")
time.sleep(0.005)
class LifecycleTests(unittest.TestCase):
"""以同步屏障制造竞争，不依赖偶然的线程调度顺序。"""
def make_server(self, **options):
"""
@description : 创建独立端口的服务并登记测试后清理
@param options : 覆盖构造参数
@return : WebSocketServer 实例
"""
options.setdefault("host", "127.0.0.1")
options.setdefault("port", unused_port())
server = WebSocketServer(**options)
self.addCleanup(server._stop, 3)
return server
def url(self, server, path="/ws"):
"""
@description : 生成测试服务端点
@param server : 测试服务实例
@param path : 请求路径
@return : WebSocket URL
"""
return f"ws://127.0.0.1:{server.get_status()['port']}{path}"
def start_in_thread(self, server, outcomes):
"""
@description : 在调用线程记录 start 的结果或异常
@param server : 测试服务实例
@param outcomes : 接收结果的列表
@return : 已启动的调用线程
"""
def invoke():
"""
@description : 保存异步启动结果供主测试线程断言
@param : 无参数
@return : 无返回值
"""
try:
outcomes.append(server.start())
except Exception as error:
outcomes.append(error)
thread = threading.Thread(target=invoke, daemon=True)
thread.start()
self.addCleanup(thread.join, 3)
return thread
def test_ready_echo_idempotent_restart(self):
"""
@description : start 成功即可连接，重复启动不建新线程，清理后可以重启
@param : 无参数
@return : 无返回值
"""
server = self.make_server()
server.on_message(server.send)
for _ in range(3):
self.assertTrue(server.start())
worker = server._run.thread
self.assertTrue(server.start())
self.assertIs(server._run.thread, worker)
server.require_running()
with connect(self.url(server), open_timeout=2) as client:
for payload in ("hello 中文", b"\x00\xff"):
client.send(payload)
self.assertEqual(client.recv(timeout=2), payload)
self.assertTrue(server.stop())
self.assertFalse(worker.is_alive())
self.assertEqual(server.get_status()["state"], "stopped")
self.assertEqual(server.get_clients(), [])
self.assertTrue(server.stop())
def test_port_in_use_propagates_reason_and_can_retry(self):
"""
@description : 端口占用必须启动失败并保留原始原因，释放端口后可以重试
@param : 无参数
@return : 无返回值
"""
with socket.create_server(("127.0.0.1", 0)) as occupied:
server = self.make_server(port=occupied.getsockname()[1])
with self.assertRaises(RuntimeError) as raised:
server.start()
self.assertIsInstance(raised.exception.__cause__, OSError)
self.assertTrue(server.stop())
self.assertEqual(server.get_status()["state"], "failed")
self.assertIn("Error", server.get_status()["server_error"])
with self.assertRaises(RuntimeError):
server.require_running()
self.assertTrue(server.start())
self.assertIsNone(server.get_status()["server_error"])
def test_stop_before_start(self):
"""
@description : 从未启动的服务重复停止安全
@param : 无参数
@return : 无返回值
"""
server = self.make_server()
self.assertTrue(server.stop())
self.assertTrue(server.request_stop())
self.assertTrue(server.start())
def test_concurrent_start_waits_for_same_listener(self):
"""
@description : 两个并发 start 都等待同一次监听，不提前返回成功
@param : 无参数
@return : 无返回值
"""
entered, release = threading.Event(), threading.Event()
outcomes = []
server = self.make_server()
def delayed(*args, **kwargs):
"""
@description : 用屏障暂停绑定，精确制造并发启动
@param args : serve 位置参数
@param kwargs : serve 关键字参数
@return : 真实 websockets Server
"""
entered.set()
if not release.wait(3):
raise TimeoutError("test barrier")
return real_serve(*args, **kwargs)
with patch.object(server_module, "serve", side_effect=delayed) as factory:
first = self.start_in_thread(server, outcomes)
self.assertTrue(entered.wait(1))
second = self.start_in_thread(server, outcomes)
try:
self.assertEqual(server.get_status()["state"], "starting")
self.assertEqual(outcomes, [])
self.assertEqual(factory.call_count, 1)
finally:
release.set()
first.join(3)
second.join(3)
self.assertEqual(outcomes, [True, True])
def test_stop_cancels_delayed_start_and_blocks_new_generation(self):
"""
@description : 延迟启动遇到 stop 后不能正常监听，清理完成前禁止新一代
@param : 无参数
@return : 无返回值
"""
entered, release = threading.Event(), threading.Event()
outcomes = []
server = self.make_server(stop_timeout_s=0.05)
def delayed(*args, **kwargs):
"""
@description : 模拟尚未完成的绑定操作
@param args : serve 位置参数
@param kwargs : serve 关键字参数
@return : 真实服务对象
"""
entered.set()
release.wait(3)
return real_serve(*args, **kwargs)
with patch.object(server_module, "serve", side_effect=delayed):
starter = self.start_in_thread(server, outcomes)
self.assertTrue(entered.wait(1))
try:
started = time.monotonic()
self.assertFalse(server.stop())
self.assertLess(time.monotonic() - started, 0.5)
self.assertTrue(server.get_status()["cleanup_pending"])
self.assertFalse(server.get_status()["listening"])
with self.assertRaisesRegex(RuntimeError, "cleaning up"):
server.start()
finally:
release.set()
starter.join(3)
wait_until(lambda: not server.get_status()["running"])
self.assertTrue(server.stop())
self.assertIsInstance(outcomes[0], RuntimeError)
self.assertEqual(server.get_status()["state"], "stopped")
with socket.create_server(("127.0.0.1", server.get_status()["port"])):
pass
self.assertTrue(server.start())
def test_start_timeout_remains_cancelled_after_factory_returns(self):
"""
@description : 启动超时后保留错误，迟到的绑定结果必须清理且不能变为运行中
@param : 无参数
@return : 无返回值
"""
release = threading.Event()
server = self.make_server(start_timeout_s=0.05)
def delayed(*args, **kwargs):
"""
@description : 在启动超时之后才返回真实服务对象
@param args : serve 位置参数
@param kwargs : serve 关键字参数
@return : 真实服务对象
"""
release.wait(3)
return real_serve(*args, **kwargs)
with patch.object(server_module, "serve", side_effect=delayed):
try:
with self.assertRaises(TimeoutError):
server.start()
self.assertEqual(server.get_status()["state"], "failed")
self.assertTrue(server.get_status()["cleanup_pending"])
with self.assertRaises(RuntimeError):
server.start()
finally:
release.set()
self.assertTrue(server.stop())
self.assertFalse(server.get_status()["listening"])
self.assertIn("startup timed out", server.get_status()["server_error"])
with socket.create_server(("127.0.0.1", server.get_status()["port"])):
pass
def test_stop_budget_includes_blocked_shutdown_and_is_idempotent(self):
"""
@description : shutdown 阻塞不能突破 stop 总预算，并发 stop 只触发一次清理
@param : 无参数
@return : 无返回值
"""
entered, release = threading.Event(), threading.Event()
server = self.make_server(stop_timeout_s=0.05)
original = Server.shutdown
def delayed(raw_server, *args, **kwargs):
"""
@description : 暂停库的清理过程以检查公开 stop 的总预算
@param raw_server : 原始 websockets Server
@param args : shutdown 位置参数
@param kwargs : shutdown 关键字参数
@return : 原始 shutdown 的返回值
"""
entered.set()
release.wait(3)
return original(raw_server, *args, **kwargs)
with patch.object(Server, "shutdown", autospec=True, side_effect=delayed) as shutdown:
server.start()
try:
started = time.monotonic()
self.assertFalse(server.stop())
self.assertLess(time.monotonic() - started, 0.5)
self.assertTrue(entered.wait(1))
self.assertFalse(server.request_stop())
with self.assertRaises(RuntimeError):
server.start()
self.assertEqual(shutdown.call_count, 1)
finally:
release.set()
wait_until(lambda: not server.get_status()["running"])
self.assertTrue(server.stop())
self.assertTrue(server.start())
def test_blocked_callback_keeps_cleanup_pending_until_it_returns(self):
"""
@description : 阻塞回调不能使 stop 无限等待，也不能被误报为已清理
@param : 无参数
@return : 无返回值
"""
entered, release = threading.Event(), threading.Event()
server = self.make_server(stop_timeout_s=0.1)
def callback(client_id, payload):
"""
@description : 模拟用户回调暂时阻塞
@param client_id : 客户端标识
@param payload : 消息内容
@return : 无返回值
"""
entered.set()
release.wait(3)
server.on_message(callback)
server.start()
with connect(self.url(server), close_timeout=0.2) as client:
client.send("block")
self.assertTrue(entered.wait(1))
try:
started = time.monotonic()
self.assertFalse(server.stop())
self.assertLess(time.monotonic() - started, 0.6)
self.assertEqual(server.get_status()["state"], "stopping")
with self.assertRaises(RuntimeError):
server.start()
finally:
release.set()
wait_until(lambda: not server.get_status()["running"])
self.assertTrue(server.stop())
self.assertEqual(server.get_clients(), [])
self.assertTrue(server.start())
def test_callback_can_request_stop_without_joining_itself(self):
"""
@description : 消息回调内调用 stop 只请求停止，不等待自身造成死锁
@param : 无参数
@return : 无返回值
"""
results, returned = [], threading.Event()
server = self.make_server()
def callback(client_id, payload):
"""
@description : 在连接线程请求停止并保存返回值
@param client_id : 客户端标识
@param payload : 消息内容
@return : 无返回值
"""
results.append(server.stop())
returned.set()
server.on_message(callback)
server.start()
with connect(self.url(server), close_timeout=1) as client:
client.send("stop")
self.assertTrue(returned.wait(1))
self.assertEqual(results, [False])
self.assertTrue(server.stop())
def test_multiple_clients_paths_limits_and_callback_failure(self):
"""
@description : 保留多客户端、文本二进制透传、路径和连接数限制及回调异常隔离
@param : 无参数
@return : 无返回值
"""
server = self.make_server(max_clients=2)
server.on_message(server.send)
server.start()
with ExitStack() as stack:
clients = [stack.enter_context(connect(self.url(server), close_timeout=1)) for _ in range(2)]
for number, client in enumerate(clients):
client.send(str(number))
self.assertEqual(client.recv(timeout=2), str(number))
with connect(self.url(server), close_timeout=1) as extra:
with self.assertRaises(ConnectionClosed) as closed:
extra.recv(timeout=2)
self.assertEqual(closed.exception.rcvd.code, 1013)
wait_until(lambda: not server.get_clients())
with connect(self.url(server, "/wrong"), close_timeout=1) as wrong:
with self.assertRaises(ConnectionClosed) as closed:
wrong.recv(timeout=2)
self.assertEqual(closed.exception.rcvd.code, 1008)
def broken(client_id, payload):
"""
@description : 注入单个业务回调异常，验证不破坏监听
@param client_id : 客户端标识
@param payload : 消息内容
@return : 始终抛出 ValueError
"""
raise ValueError("callback test")
server.on_message(broken)
with connect(self.url(server), close_timeout=1) as client:
client.send("fail")
with self.assertRaises(ConnectionClosed) as closed:
client.recv(timeout=2)
self.assertEqual(closed.exception.rcvd.code, 1011)
server.require_running()
self.assertTrue(server.stop())
def test_thread_creation_failure_is_visible_and_retryable(self):
"""
@description : 线程创建失败保留原因且不会留下假运行状态
@param : 无参数
@return : 无返回值
"""
server = self.make_server()
with patch.object(threading.Thread, "start", side_effect=RuntimeError("thread test")):
with self.assertRaisesRegex(RuntimeError, "thread test"):
server.start()
self.assertTrue(server.stop())
self.assertIn("thread test", server.get_status()["server_error"])
self.assertTrue(server.start())
def test_timeout_validation(self):
"""
@description : 生命周期超时参数不能为非有限数、负数或布尔值
@param : 无参数
@return : 无返回值
"""
for value in (-1, math.nan, math.inf, True, "1"):
with self.subTest(value=value):
with self.assertRaises(ValueError):
self.make_server(start_timeout_s=value)
with self.assertRaises(ValueError):
self.make_server(stop_timeout_s=value)
with self.assertRaises(ValueError):
self.make_server(start_timeout_s=0)
def test_app_lab_lifecycle_signatures(self):
"""
@description : 即使本机没有 app_utils，也检查启停方法符合 App Lab 的签名要求
@param : 无参数
@return : 无返回值
"""
for method in (WebSocketServer.start, WebSocketServer.stop):
self.assertEqual(list(inspect.signature(method).parameters), ["self"])
def test_cleanup_failure_is_visible_and_prevents_restart(self):
"""
@description : 清理异常必须可见，不能当作正常停止后重新启动
@param : 无参数
@return : 无返回值
"""
server = self.make_server()
original = Server.shutdown
def failing(raw_server, *args, **kwargs):
"""
@description : 在真实资源清理完成后注入异常，避免测试本身泄漏端口
@param raw_server : 原始服务对象
@param args : 位置参数
@param kwargs : 关键字参数
@return : 始终抛出 OSError
"""
original(raw_server, *args, **kwargs)
raise OSError("cleanup test")
with patch.object(Server, "shutdown", autospec=True, side_effect=failing):
server.start()
self.assertFalse(server.stop())
self.assertEqual(server.get_status()["state"], "failed")
self.assertIn("cleanup test", server.get_status()["cleanup_error"])
with self.assertRaisesRegex(RuntimeError, "cleanup failed"):
server.start()
with self.assertRaisesRegex(RuntimeError, "cleanup test"):
server.require_running()
def test_unexpected_accept_exit_is_a_failure(self):
"""
@description : 监听循环意外退出必须传播失败，完成清理后才能重启
@param : 无参数
@return : 无返回值
"""
server = self.make_server()
server.start()
# 模拟底层监听意外结束，不设置 Brick 的停止事件。
server._run.server.shutdown()
wait_until(lambda: not server.get_status()["running"])
self.assertEqual(server.get_status()["state"], "failed")
with self.assertRaises(RuntimeError) as raised:
server.require_running()
self.assertIsNotNone(raised.exception.__cause__)
self.assertIsNotNone(server.get_status()["server_error"])
self.assertTrue(server.stop())
self.assertTrue(server.start())
def test_pending_handshakes_do_not_multiply_stop_budget(self):
"""
@description : 多个只建立 TCP 却不完成握手的客户端共享一次停机预算
@param : 无参数
@return : 无返回值
"""
server = self.make_server(stop_timeout_s=0.05)
server.start()
raw_server = server._run.server
with ExitStack() as stack:
for _ in range(3):
stack.enter_context(socket.create_connection(("127.0.0.1", server.get_status()["port"])))
wait_until(lambda: len(raw_server.handler_threads) >= 3)
started = time.monotonic()
self.assertFalse(server.stop())
self.assertLess(time.monotonic() - started, 0.5)
self.assertTrue(server.get_status()["cleanup_pending"])
# 等待库自身的握手超时，而不是靠测试端提前关闭连接。
wait_until(lambda: not server.get_status()["running"], timeout=3)
self.assertTrue(server.stop())
self.assertTrue(server.start())
if __name__ == "__main__":
unittest.main()