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

Remove worker group tests for now as they do not account for threading

Browse Source
This commit is contained in:
Andrew Godwin 2017-05-30 00:07:05 -07:00
parent 601056f712
commit e5c91a1299
1 changed files with 0 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
tests/test_worker.py
View File

@ -96,41 +96,3 @@ class WorkerTests(ChannelTestCase):
worker.run()
self.assertEqual(consumer.call_count, 1)
self.assertEqual(channel_layer.send.call_count, 0)
class WorkerGroupTests(ChannelTestCase):
"""
Test threaded workers.
"""
def setUp(self):
self.channel_layer = channel_layers[DEFAULT_CHANNEL_LAYER]
self.worker = WorkerGroup(self.channel_layer, n_threads=4)
self.subworkers = self.worker.workers
def test_subworkers_created(self):
self.assertEqual(len(self.subworkers), 3)
def test_subworkers_no_sigterm(self):
for wrk in self.subworkers:
self.assertFalse(wrk.signal_handlers)
def test_ready_signals_sent(self):
self.in_signal = 0
def handle_signal(sender, *args, **kwargs):
self.in_signal += 1
worker_ready.connect(handle_signal)
WorkerGroup(self.channel_layer, n_threads=4)
self.worker.ready()
self.assertEqual(self.in_signal, 4)
def test_sigterm_handler(self):
threads = []
for wkr in self.subworkers:
t = threading.Thread(target=wkr.run)
t.start()
threads.append(t)
with self.assertRaises(SystemExit):
self.worker.sigterm_handler(None, None)