django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2025-07-31 17:29:44 +03:00
Code Issues Packages Projects Releases Wiki Activity

sort imports and add some commentary to cancellederror test

Browse Source
This commit is contained in:
Patrick Gingras 2020-11-02 22:34:09 -05:00
parent 49c00da715
commit af16649725
No known key found for this signature in database
GPG Key ID: F56B4E7C84CA0B49
1 changed files with 9 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
tests/test_websocket.py
View File

@ -6,9 +6,10 @@ from urllib import parse
import http_strategies
from http_base import DaphneTestCase, DaphneTestingInstance
from daphne.testing import BaseDaphneTestingInstance
from hypothesis import given, settings
from daphne.testing import BaseDaphneTestingInstance
class TestWebsocket(DaphneTestCase):
"""
@ -285,8 +286,10 @@ class TestWebsocket(DaphneTestCase):
async def cancelling_application(scope, receive, send):
import asyncio
from twisted.internet import reactor
# Stop the server after a short delay so that the teardown is run.
reactor.callLater(2, lambda: reactor.stop())
await send({"type": "websocket.accept"})
raise asyncio.CancelledError()
@ -303,6 +306,11 @@ class CancellingTestingInstance(BaseDaphneTestingInstance):
def process_teardown(self):
import multiprocessing
# Get a hold of the enclosing DaphneProcess (we're currently running in
# the same process as the application).
proc = multiprocessing.current_process()
# By now the (only) socket should have disconnected, and the
# application_checker should have run. If there are any connections
# still, it means that the application_checker did not clean them up.
if proc.server.connections:
raise ConnectionsNotEmpty()