ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2026-04-21 16:55:55 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

✨ Feature: WS 支持 ping interval/timeout 配置 (#3964)

Browse Source 
Co-authored-by: Ju4tCode <42488585+yanyongyu@users.noreply.github.com>
This commit is contained in:
StarHeart
2026-04-20 19:39:28 +08:00
committed by GitHub
parent 6b1c616860
commit 2b77b122af
4 changed files with 126 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
tests/test_driver.py
View File

@@ -878,6 +878,92 @@ async def test_websocket_client_timeout(driver: Driver, server_url: URL):
await anyio.sleep(1)
@pytest.mark.anyio
@pytest.mark.parametrize(
"driver",
[
pytest.param("nonebot.drivers.websockets:Driver", id="websockets"),
pytest.param("nonebot.drivers.aiohttp:Driver", id="aiohttp"),
],
indirect=True,
)
async def test_websocket_client_ping_timeout(driver: Driver, server_url: URL):
"""WebSocket connections work with different ping_timeout settings."""
assert isinstance(driver, WebSocketClientMixin)
ws_url = server_url.with_scheme("ws")
# ping timeout not set (UNSET), falls back to DEFAULT_TIMEOUT.ping
request = Request("GET", ws_url, timeout=Timeout())
async with driver.websocket(request) as ws:
await ws.send("quit")
with pytest.raises(WebSocketClosed):
await ws.receive()
await anyio.sleep(1)
# ping timeout explicitly set to None (disable ping timeout)
request = Request("GET", ws_url, timeout=Timeout(ping=None))
async with driver.websocket(request) as ws:
await ws.send("quit")
with pytest.raises(WebSocketClosed):
await ws.receive()
await anyio.sleep(1)
# ping timeout set to a float value
request = Request("GET", ws_url, timeout=Timeout(ping=20.0))
async with driver.websocket(request) as ws:
await ws.send("quit")
with pytest.raises(WebSocketClosed):
await ws.receive()
await anyio.sleep(1)
@pytest.mark.anyio
@pytest.mark.parametrize(
"driver",
[
pytest.param("nonebot.drivers.websockets:Driver", id="websockets"),
pytest.param("nonebot.drivers.aiohttp:Driver", id="aiohttp"),
],
indirect=True,
)
async def test_websocket_client_ping_interval(driver: Driver, server_url: URL):
"""WebSocket connections work with different ping_interval settings."""
assert isinstance(driver, WebSocketClientMixin)
ws_url = server_url.with_scheme("ws")
# ping_interval not set (UNSET), default behavior
request = Request("GET", ws_url)
async with driver.websocket(request) as ws:
await ws.send("quit")
with pytest.raises(WebSocketClosed):
await ws.receive()
await anyio.sleep(1)
# ping_interval explicitly set to None (disable ping)
request = Request("GET", ws_url, ping_interval=None)
async with driver.websocket(request) as ws:
await ws.send("quit")
with pytest.raises(WebSocketClosed):
await ws.receive()
await anyio.sleep(1)
# ping_interval set to a float value
request = Request("GET", ws_url, ping_interval=20.0)
async with driver.websocket(request) as ws:
await ws.send("quit")
with pytest.raises(WebSocketClosed):
await ws.receive()
await anyio.sleep(1)
@pytest.mark.parametrize(
("driver", "driver_type"),
[