unittest: enabling tests/test_lib for py2

This commit is contained in:
Yuming Zhu 2024-10-22 23:51:50 +08:00 committed by Tomas Kopecek
parent 51d8535eee
commit c8a27e525a
8 changed files with 45 additions and 14 deletions

View file

@ -320,9 +320,8 @@ class TasksTestCase(unittest.TestCase):
@patch('time.time')
@patch('time.sleep')
@patch('signal.sigtimedwait')
@patch('signal.pause')
def test_BaseTaskHandler_wait_timeout(self, pause, sigtimedwait, sleep, time):
def test_BaseTaskHandler_wait_timeout(self, pause, sleep, time):
"""Tests timeout behavior in the wait function"""
temp_path = self.get_tmp_dir_path('TaskTest')
obj = TaskTest(95, 'some_method', ['random_arg'], None, None, temp_path)
@ -330,11 +329,18 @@ class TasksTestCase(unittest.TestCase):
obj.session = MagicMock()
obj.session.host.taskWait.return_value = [[], [99, 100, 101]]
time.side_effect = list(range(0, 4000, 60))
if six.PY3:
sigtimedwait = patch('signal.sigtimedwait').start()
try:
obj.wait([99, 100, 101], timeout=3600)
raise Exception('A GenericError was not raised.')
except koji.GenericError as e:
self.assertEqual(e.args[0][:24], 'Subtasks timed out after')
if six.PY3:
sigtimedwait.stop()
obj.session.host.taskSetWait.assert_called_once_with(95, [99, 100, 101])
obj.session.cancelTaskChildren.assert_called_once_with(95)
obj.session.getTaskResult.assert_not_called()
@ -342,9 +348,8 @@ class TasksTestCase(unittest.TestCase):
@patch('time.time')
@patch('time.sleep')
@patch('signal.sigtimedwait')
@patch('signal.pause')
def test_BaseTaskHandler_wait_avoid_timeout(self, pause, sigtimedwait, sleep, time):
def test_BaseTaskHandler_wait_avoid_timeout(self, pause, sleep, time):
"""Tests that timeout does not happen if tasks finish in time"""
temp_path = self.get_tmp_dir_path('TaskTest')
obj = TaskTest(95, 'some_method', ['random_arg'], None, None, temp_path)
@ -359,8 +364,15 @@ class TasksTestCase(unittest.TestCase):
# and then report all done
taskWait_returns.append([[99, 100, 101], []])
obj.session.host.taskWait.side_effect = taskWait_returns
if six.PY3:
sigtimedwait = patch('signal.sigtimedwait').start()
obj.wait([99, 100, 101], timeout=3600)
if six.PY3:
sigtimedwait.stop()
obj.session.host.taskSetWait.assert_called_once_with(95, [99, 100, 101])
obj.session.cancelTaskChildren.assert_not_called()
pause.assert_not_called()