Skip to content

Commit

Permalink
Add test to simulate lost connection on server
Browse files Browse the repository at this point in the history
  • Loading branch information
stefanw committed Aug 17, 2020
1 parent a9a46e8 commit 6aff2d0
Showing 1 changed file with 27 additions and 0 deletions.
27 changes: 27 additions & 0 deletions tests/protocols/test_websocket.py
Expand Up @@ -466,3 +466,30 @@ async def get_subprotocol(url):
accepted_subprotocol = loop.run_until_complete(get_subprotocol(url))
assert accepted_subprotocol == subprotocol
loop.close()


@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_server_lost_connection(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
await self.send({"type": "websocket.accept"})
await self.receive()

# Simulate a lost connection
# without receiving a close frame
self.send.__self__.connection_lost(None)

await self.send({"type": "websocket.send", "text": "123"})

async def websocket_session(url):
async with websockets.connect(url) as websocket:
await websocket.ping()
await websocket.send("abc")
# Delay exiting context manager
# to avoid sending a close frame before server attempts send
await asyncio.sleep(1)

with run_server(App, protocol_cls=protocol_cls) as url:
loop = asyncio.new_event_loop()
loop.run_until_complete(websocket_session(url))
loop.close()

0 comments on commit 6aff2d0

Please sign in to comment.