876238a134
There is no testtools.main() at all (this was a
unittest.main() before). Let's remove this unused
and non-working code.
The following code can be used if someone needs
to start tests manually:
python -m unittest <test_file.py>
Change-Id: Id5162ac73825584df6c23dbe68786ff3355fc6ae
344 lines
14 KiB
Python
344 lines
14 KiB
Python
# Copyright (c) 2010-2013 OpenStack, LLC.
|
|
#
|
|
# 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 sys
|
|
import time
|
|
|
|
try:
|
|
from unittest import mock
|
|
except ImportError:
|
|
import mock
|
|
|
|
import testtools
|
|
import threading
|
|
import six
|
|
from six.moves.queue import Queue, Empty
|
|
|
|
from swiftclient import multithreading as mt
|
|
from swiftclient.exceptions import ClientException
|
|
|
|
|
|
class ThreadTestCase(testtools.TestCase):
|
|
def setUp(self):
|
|
super(ThreadTestCase, self).setUp()
|
|
self.got_args_kwargs = Queue()
|
|
self.starting_thread_count = threading.active_count()
|
|
|
|
def _func(self, q_item, *args, **kwargs):
|
|
self.got_items.put(q_item)
|
|
self.got_args_kwargs.put((args, kwargs))
|
|
|
|
if q_item == 'go boom':
|
|
raise Exception('I went boom!')
|
|
if q_item == 'c boom':
|
|
raise ClientException(
|
|
'Client Boom', http_scheme='http', http_host='192.168.22.1',
|
|
http_port=80, http_path='/booze', http_status=404,
|
|
http_reason='to much', http_response_content='no sir!')
|
|
|
|
return 'best result EVAR!'
|
|
|
|
def assertQueueContains(self, queue, expected_contents):
|
|
got_contents = []
|
|
try:
|
|
while True:
|
|
got_contents.append(queue.get(timeout=0.1))
|
|
except Empty:
|
|
pass
|
|
if isinstance(expected_contents, set):
|
|
got_contents = set(got_contents)
|
|
self.assertEqual(expected_contents, got_contents)
|
|
|
|
|
|
class TestQueueFunctionThread(ThreadTestCase):
|
|
def setUp(self):
|
|
super(TestQueueFunctionThread, self).setUp()
|
|
|
|
self.input_queue = Queue()
|
|
self.got_items = Queue()
|
|
self.stored_results = []
|
|
|
|
self.qft = mt.QueueFunctionThread(self.input_queue, self._func,
|
|
'one_arg', 'two_arg',
|
|
red_fish='blue_arg',
|
|
store_results=self.stored_results)
|
|
self.qft.start()
|
|
|
|
def tearDown(self):
|
|
if self.qft.is_alive():
|
|
self.finish_up_thread()
|
|
|
|
super(TestQueueFunctionThread, self).tearDown()
|
|
|
|
def finish_up_thread(self):
|
|
self.input_queue.put(mt.StopWorkerThreadSignal())
|
|
while self.qft.is_alive():
|
|
time.sleep(0.05)
|
|
|
|
def test_plumbing_and_store_results(self):
|
|
self.input_queue.put('abc')
|
|
self.input_queue.put(123)
|
|
self.finish_up_thread()
|
|
|
|
self.assertQueueContains(self.got_items, ['abc', 123])
|
|
self.assertQueueContains(self.got_args_kwargs, [
|
|
(('one_arg', 'two_arg'), {'red_fish': 'blue_arg'}),
|
|
(('one_arg', 'two_arg'), {'red_fish': 'blue_arg'})])
|
|
self.assertEqual(self.stored_results,
|
|
['best result EVAR!', 'best result EVAR!'])
|
|
|
|
def test_exception_handling(self):
|
|
self.input_queue.put('go boom')
|
|
self.input_queue.put('ok')
|
|
self.input_queue.put('go boom')
|
|
self.finish_up_thread()
|
|
|
|
self.assertQueueContains(self.got_items,
|
|
['go boom', 'ok', 'go boom'])
|
|
self.assertEqual(len(self.qft.exc_infos), 2)
|
|
self.assertEqual(Exception, self.qft.exc_infos[0][0])
|
|
self.assertEqual(Exception, self.qft.exc_infos[1][0])
|
|
self.assertEqual(('I went boom!',), self.qft.exc_infos[0][1].args)
|
|
self.assertEqual(('I went boom!',), self.qft.exc_infos[1][1].args)
|
|
|
|
|
|
class TestQueueFunctionManager(ThreadTestCase):
|
|
def setUp(self):
|
|
super(TestQueueFunctionManager, self).setUp()
|
|
self.thread_manager = mock.create_autospec(
|
|
mt.MultiThreadingManager, spec_set=True, instance=True)
|
|
self.thread_count = 4
|
|
self.error_counter = [0]
|
|
self.got_items = Queue()
|
|
self.stored_results = []
|
|
self.qfq = mt.QueueFunctionManager(
|
|
self._func, self.thread_count, self.thread_manager,
|
|
thread_args=('1arg', '2arg'),
|
|
thread_kwargs={'a': 'b', 'store_results': self.stored_results},
|
|
error_counter=self.error_counter,
|
|
connection_maker=self.connection_maker)
|
|
|
|
def connection_maker(self):
|
|
return 'yup, I made a connection'
|
|
|
|
def test_context_manager_without_error_counter(self):
|
|
self.qfq = mt.QueueFunctionManager(
|
|
self._func, self.thread_count, self.thread_manager,
|
|
thread_args=('1arg', '2arg'),
|
|
thread_kwargs={'a': 'b', 'store_results': self.stored_results},
|
|
connection_maker=self.connection_maker)
|
|
|
|
with self.qfq as input_queue:
|
|
self.assertEqual(self.starting_thread_count + self.thread_count,
|
|
threading.active_count())
|
|
input_queue.put('go boom')
|
|
|
|
self.assertEqual(self.starting_thread_count, threading.active_count())
|
|
error_strs = list(map(str, self.thread_manager.error.call_args_list))
|
|
self.assertEqual(1, len(error_strs))
|
|
self.assertTrue('Exception: I went boom!' in error_strs[0])
|
|
|
|
def test_context_manager_without_conn_maker_or_error_counter(self):
|
|
self.qfq = mt.QueueFunctionManager(
|
|
self._func, self.thread_count, self.thread_manager,
|
|
thread_args=('1arg', '2arg'), thread_kwargs={'a': 'b'})
|
|
|
|
with self.qfq as input_queue:
|
|
self.assertEqual(self.starting_thread_count + self.thread_count,
|
|
threading.active_count())
|
|
for i in range(20):
|
|
input_queue.put('slap%d' % i)
|
|
|
|
self.assertEqual(self.starting_thread_count, threading.active_count())
|
|
self.assertEqual([], self.thread_manager.error.call_args_list)
|
|
self.assertEqual(0, self.error_counter[0])
|
|
self.assertQueueContains(self.got_items,
|
|
set(['slap%d' % i for i in range(20)]))
|
|
self.assertQueueContains(
|
|
self.got_args_kwargs,
|
|
[(('1arg', '2arg'), {'a': 'b'})] * 20)
|
|
self.assertEqual(self.stored_results, [])
|
|
|
|
def test_context_manager_with_exceptions(self):
|
|
with self.qfq as input_queue:
|
|
self.assertEqual(self.starting_thread_count + self.thread_count,
|
|
threading.active_count())
|
|
for i in range(20):
|
|
input_queue.put('item%d' % i if i % 2 == 0 else 'go boom')
|
|
|
|
self.assertEqual(self.starting_thread_count, threading.active_count())
|
|
error_strs = list(map(str, self.thread_manager.error.call_args_list))
|
|
self.assertEqual(10, len(error_strs))
|
|
self.assertTrue(all(['Exception: I went boom!' in s for s in
|
|
error_strs]))
|
|
self.assertEqual(10, self.error_counter[0])
|
|
expected_items = set(['go boom'] +
|
|
['item%d' % i for i in range(20)
|
|
if i % 2 == 0])
|
|
self.assertQueueContains(self.got_items, expected_items)
|
|
self.assertQueueContains(
|
|
self.got_args_kwargs,
|
|
[(('yup, I made a connection', '1arg', '2arg'), {'a': 'b'})] * 20)
|
|
self.assertEqual(self.stored_results, ['best result EVAR!'] * 10)
|
|
|
|
def test_context_manager_with_client_exceptions(self):
|
|
with self.qfq as input_queue:
|
|
self.assertEqual(self.starting_thread_count + self.thread_count,
|
|
threading.active_count())
|
|
for i in range(20):
|
|
input_queue.put('item%d' % i if i % 2 == 0 else 'c boom')
|
|
|
|
self.assertEqual(self.starting_thread_count, threading.active_count())
|
|
error_strs = list(map(str, self.thread_manager.error.call_args_list))
|
|
self.assertEqual(10, len(error_strs))
|
|
stringification = 'Client Boom: ' \
|
|
'http://192.168.22.1:80/booze 404 to much no sir!'
|
|
self.assertTrue(all([stringification in s for s in error_strs]))
|
|
self.assertEqual(10, self.error_counter[0])
|
|
expected_items = set(['c boom'] +
|
|
['item%d' % i for i in range(20)
|
|
if i % 2 == 0])
|
|
self.assertQueueContains(self.got_items, expected_items)
|
|
self.assertQueueContains(
|
|
self.got_args_kwargs,
|
|
[(('yup, I made a connection', '1arg', '2arg'), {'a': 'b'})] * 20)
|
|
self.assertEqual(self.stored_results, ['best result EVAR!'] * 10)
|
|
|
|
def test_context_manager_with_connection_maker(self):
|
|
with self.qfq as input_queue:
|
|
self.assertEqual(self.starting_thread_count + self.thread_count,
|
|
threading.active_count())
|
|
for i in range(20):
|
|
input_queue.put('item%d' % i)
|
|
|
|
self.assertEqual(self.starting_thread_count, threading.active_count())
|
|
self.assertEqual([], self.thread_manager.error.call_args_list)
|
|
self.assertEqual(0, self.error_counter[0])
|
|
self.assertQueueContains(self.got_items,
|
|
set(['item%d' % i for i in range(20)]))
|
|
self.assertQueueContains(
|
|
self.got_args_kwargs,
|
|
[(('yup, I made a connection', '1arg', '2arg'), {'a': 'b'})] * 20)
|
|
self.assertEqual(self.stored_results, ['best result EVAR!'] * 20)
|
|
|
|
|
|
class TestMultiThreadingManager(ThreadTestCase):
|
|
|
|
@mock.patch('swiftclient.multithreading.QueueFunctionManager')
|
|
def test_instantiation(self, mock_qfq):
|
|
thread_manager = mt.MultiThreadingManager()
|
|
|
|
self.assertEqual([
|
|
mock.call(thread_manager._print, 1, thread_manager),
|
|
mock.call(thread_manager._print_error, 1, thread_manager),
|
|
], mock_qfq.call_args_list)
|
|
|
|
# These contexts don't get entered into until the
|
|
# MultiThreadingManager's context is entered.
|
|
self.assertEqual([], thread_manager.printer.__enter__.call_args_list)
|
|
self.assertEqual([],
|
|
thread_manager.error_printer.__enter__.call_args_list)
|
|
|
|
# Test default values for the streams.
|
|
self.assertEqual(sys.stdout, thread_manager.print_stream)
|
|
self.assertEqual(sys.stderr, thread_manager.error_stream)
|
|
|
|
@mock.patch('swiftclient.multithreading.QueueFunctionManager')
|
|
def test_queue_manager_no_args(self, mock_qfq):
|
|
thread_manager = mt.MultiThreadingManager()
|
|
|
|
mock_qfq.reset_mock()
|
|
mock_qfq.return_value = 'slap happy!'
|
|
|
|
self.assertEqual(
|
|
'slap happy!',
|
|
thread_manager.queue_manager(self._func, 88))
|
|
|
|
self.assertEqual([
|
|
mock.call(self._func, 88, thread_manager, thread_args=(),
|
|
thread_kwargs={}, connection_maker=None,
|
|
error_counter=None)
|
|
], mock_qfq.call_args_list)
|
|
|
|
@mock.patch('swiftclient.multithreading.QueueFunctionManager')
|
|
def test_queue_manager_with_args(self, mock_qfq):
|
|
thread_manager = mt.MultiThreadingManager()
|
|
|
|
mock_qfq.reset_mock()
|
|
mock_qfq.return_value = 'do run run'
|
|
|
|
self.assertEqual(
|
|
'do run run',
|
|
thread_manager.queue_manager(self._func, 88, 'fun', times='are',
|
|
connection_maker='abc', to='be had',
|
|
error_counter='def'))
|
|
|
|
self.assertEqual([
|
|
mock.call(self._func, 88, thread_manager, thread_args=('fun',),
|
|
thread_kwargs={'times': 'are', 'to': 'be had'},
|
|
connection_maker='abc', error_counter='def')
|
|
], mock_qfq.call_args_list)
|
|
|
|
def test_printers(self):
|
|
out_stream = six.StringIO()
|
|
err_stream = six.StringIO()
|
|
|
|
with mt.MultiThreadingManager(
|
|
print_stream=out_stream,
|
|
error_stream=err_stream) as thread_manager:
|
|
|
|
# Sanity-checking these gives power to the previous test which
|
|
# looked at the default values of thread_manager.print/error_stream
|
|
self.assertEqual(out_stream, thread_manager.print_stream)
|
|
self.assertEqual(err_stream, thread_manager.error_stream)
|
|
|
|
self.assertEqual(self.starting_thread_count + 2,
|
|
threading.active_count())
|
|
|
|
thread_manager.print_msg('one-argument')
|
|
thread_manager.print_msg('one %s, %d fish', 'fish', 88)
|
|
thread_manager.error('I have %d problems, but a %s is not one',
|
|
99, u'\u062A\u062A')
|
|
thread_manager.print_msg('some\n%s\nover the %r', 'where',
|
|
u'\u062A\u062A')
|
|
thread_manager.error('one-error-argument')
|
|
thread_manager.error('Sometimes\n%.1f%% just\ndoes not\nwork!',
|
|
3.14159)
|
|
|
|
self.assertEqual(self.starting_thread_count, threading.active_count())
|
|
|
|
out_stream.seek(0)
|
|
if six.PY3:
|
|
over_the = "over the '\u062a\u062a'\n"
|
|
else:
|
|
over_the = "over the u'\\u062a\\u062a'\n"
|
|
self.assertEqual([
|
|
'one-argument\n',
|
|
'one fish, 88 fish\n',
|
|
'some\n', 'where\n', over_the,
|
|
], list(out_stream.readlines()))
|
|
|
|
err_stream.seek(0)
|
|
first_item = u'I have 99 problems, but a \u062A\u062A is not one\n'
|
|
if six.PY2:
|
|
first_item = first_item.encode('utf8')
|
|
self.assertEqual([
|
|
first_item,
|
|
'one-error-argument\n',
|
|
'Sometimes\n', '3.1% just\n', 'does not\n', 'work!\n',
|
|
], list(err_stream.readlines()))
|
|
|
|
self.assertEqual(3, thread_manager.error_count)
|