/usr/lib/python2.7/dist-packages/futurist/tests/test_waiters.py is in python-futurist 0.13.0-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | # -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import time
import eventlet
import testscenarios
import futurist
from futurist.tests import base
from futurist import waiters
# Module level functions need to be used since the process pool
# executor can not access instance or lambda level functions (since those
# are not pickleable).
def mini_delay(use_eventlet_sleep=False):
if use_eventlet_sleep:
eventlet.sleep(0.1)
else:
time.sleep(0.1)
return 1
class TestWaiters(testscenarios.TestWithScenarios, base.TestCase):
scenarios = [
('sync', {'executor_cls': futurist.SynchronousExecutor,
'executor_kwargs': {}, 'use_eventlet_sleep': False}),
('green_sync', {'executor_cls': futurist.SynchronousExecutor,
'executor_kwargs': {'green': True},
'use_eventlet_sleep': True}),
('green', {'executor_cls': futurist.GreenThreadPoolExecutor,
'executor_kwargs': {}, 'use_eventlet_sleep': True}),
('thread', {'executor_cls': futurist.ThreadPoolExecutor,
'executor_kwargs': {}, 'use_eventlet_sleep': False}),
('process', {'executor_cls': futurist.ProcessPoolExecutor,
'executor_kwargs': {}, 'use_eventlet_sleep': False}),
]
def setUp(self):
super(TestWaiters, self).setUp()
self.executor = self.executor_cls(**self.executor_kwargs)
def tearDown(self):
super(TestWaiters, self).tearDown()
self.executor.shutdown()
self.executor = None
def test_wait_for_any(self):
fs = []
for _i in range(0, 10):
fs.append(self.executor.submit(
mini_delay, use_eventlet_sleep=self.use_eventlet_sleep))
all_done_fs = []
total_fs = len(fs)
while len(all_done_fs) != total_fs:
done, not_done = waiters.wait_for_any(fs)
all_done_fs.extend(done)
fs = not_done
self.assertEqual(total_fs, sum(f.result() for f in all_done_fs))
def test_wait_for_all(self):
fs = []
for _i in range(0, 10):
fs.append(self.executor.submit(
mini_delay, use_eventlet_sleep=self.use_eventlet_sleep))
done_fs, not_done_fs = waiters.wait_for_all(fs)
self.assertEqual(len(fs), sum(f.result() for f in done_fs))
self.assertEqual(0, len(not_done_fs))
def test_no_mixed_wait_for_any(self):
fs = [futurist.GreenFuture(), futurist.Future()]
self.assertRaises(RuntimeError, waiters.wait_for_any, fs)
def test_no_mixed_wait_for_all(self):
fs = [futurist.GreenFuture(), futurist.Future()]
self.assertRaises(RuntimeError, waiters.wait_for_all, fs)
|