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

Add basic test for DaphneProcess.

Browse Source
This commit is contained in:
Steven H Johnson 2019-10-29 12:32:48 -04:00
parent b7f19291d6
commit 657d89f80d
2 changed files with 23 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
daphne/testing.py
View File

@ -109,8 +109,8 @@ class DaphneProcess(multiprocessing.Process):
self.host = host self.host = host
self.application = application self.application = application
self.kwargs = kwargs or {} self.kwargs = kwargs or {}
self.setup = setup or (lambda: None) self.setup = setup
self.teardown = teardown or (lambda: None) self.teardown = teardown
self.port = multiprocessing.Value("i") self.port = multiprocessing.Value("i")
self.ready = multiprocessing.Event() self.ready = multiprocessing.Event()
self.errors = multiprocessing.Queue() self.errors = multiprocessing.Queue()
@ -139,11 +139,13 @@ class DaphneProcess(multiprocessing.Process):
# Set up a poller to look for the port # Set up a poller to look for the port
reactor.callLater(0.1, self.resolve_port) reactor.callLater(0.1, self.resolve_port)
# Run with setup/teardown # Run with setup/teardown
self.setup() if self.setup:
self.setup()
try: try:
self.server.run() self.server.run()
finally: finally:
self.teardown() if self.teardown:
self.teardown()
except Exception as e: except Exception as e:
# Put the error on our queue so the parent gets it # Put the error on our queue so the parent gets it
self.errors.put((e, traceback.format_exc())) self.errors.put((e, traceback.format_exc()))
@ -164,6 +166,7 @@ class TestApplication:
and then quits the server. For testing. and then quits the server. For testing.
""" """
__test__ = False # Prevent pytest from thinking this is a TestCase
setup_storage = os.path.join(tempfile.gettempdir(), "setup.testio") setup_storage = os.path.join(tempfile.gettempdir(), "setup.testio")
result_storage = os.path.join(tempfile.gettempdir(), "result.testio") result_storage = os.path.join(tempfile.gettempdir(), "result.testio")

16
tests/test_testing.py Normal file
View File

@ -0,0 +1,16 @@
from daphne.testing import DaphneProcess, TestApplication
def test_daphne_process():
"""
Minimal test of DaphneProcess.
"""
application = TestApplication
server_process = DaphneProcess("localhost", application)
server_process.start()
server_process.ready.wait()
port = server_process.port.value
server_process.terminate()
server_process.join()
assert port > 0, "Port was not set"