|
@@ -20,31 +20,31 @@ class test_TaskPool(AppCase):
|
|
|
|
|
|
with mask_modules('threadpool'):
|
|
|
with self.assertRaises(ImportError):
|
|
|
- TaskPool()
|
|
|
+ TaskPool(app=self.app)
|
|
|
|
|
|
def test_with_threadpool(self):
|
|
|
with mock_module('threadpool'):
|
|
|
- x = TaskPool()
|
|
|
+ x = TaskPool(app=self.app)
|
|
|
self.assertTrue(x.ThreadPool)
|
|
|
self.assertTrue(x.WorkRequest)
|
|
|
|
|
|
def test_on_start(self):
|
|
|
with mock_module('threadpool'):
|
|
|
- x = TaskPool()
|
|
|
+ x = TaskPool(app=self.app)
|
|
|
x.on_start()
|
|
|
self.assertTrue(x._pool)
|
|
|
self.assertIsInstance(x._pool.workRequests, NullDict)
|
|
|
|
|
|
def test_on_stop(self):
|
|
|
with mock_module('threadpool'):
|
|
|
- x = TaskPool()
|
|
|
+ x = TaskPool(app=self.app)
|
|
|
x.on_start()
|
|
|
x.on_stop()
|
|
|
x._pool.dismissWorkers.assert_called_with(x.limit, do_join=True)
|
|
|
|
|
|
def test_on_apply(self):
|
|
|
with mock_module('threadpool'):
|
|
|
- x = TaskPool()
|
|
|
+ x = TaskPool(app=self.app)
|
|
|
x.on_start()
|
|
|
callback = Mock()
|
|
|
accept_callback = Mock()
|