use fake client for watch_tasks testing
This commit is contained in:
Mike McLean
parent
cf19091d9c
commit
d97cf1f961
3 changed files with 277 additions and 181 deletions
|
|
@ -1,4 +1,5 @@
|
|||
from __future__ import absolute_import
|
||||
import json
|
||||
import mock
|
||||
import os
|
||||
import six
|
||||
|
|
@ -9,15 +10,35 @@ from mock import call
|
|||
from six.moves import range
|
||||
|
||||
from koji_cli.lib import watch_tasks
|
||||
from .fakeclient import FakeClientSession, RecordingClientSession
|
||||
|
||||
|
||||
class TestWatchTasks(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.options = mock.MagicMock()
|
||||
self.session = mock.MagicMock(name='sessionMock')
|
||||
self.session = FakeClientSession('SERVER', {})
|
||||
self.recording = False
|
||||
self.record_file = None
|
||||
self.args = mock.MagicMock()
|
||||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
if self.recording:
|
||||
# save recorded calls
|
||||
if self.record_file:
|
||||
with open(self.record_file, 'w') as fp:
|
||||
json.dump(self.session.get_calls(), fp, indent=4)
|
||||
else:
|
||||
json.dump(self.session.get_calls(), sys.stderr, indent=4)
|
||||
self.recording = False
|
||||
self.record_file = None
|
||||
|
||||
def setup_record(self, filename=None):
|
||||
self.session = RecordingClientSession('http://localhost/kojihub', {})
|
||||
self.recording = True
|
||||
self.record_file = filename
|
||||
|
||||
# Show long diffs in error output...
|
||||
maxDiff = None
|
||||
|
||||
|
|
@ -30,192 +51,30 @@ class TestWatchTasks(unittest.TestCase):
|
|||
self.assertEqual(actual, expected)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.lib.TaskWatcher')
|
||||
@mock.patch('koji_cli.lib.display_tasklist_status')
|
||||
@mock.patch('koji_cli.lib.display_task_results')
|
||||
def test_watch_tasks(self, dtrMock, dtsMock, twClzMock, stdout):
|
||||
self.options.poll_interval = 0
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(twClzMock, 'TaskWatcherMock')
|
||||
manager.attach_mock(dtrMock, 'display_task_results_mock')
|
||||
manager.attach_mock(dtsMock, 'display_tasklist_status_mock')
|
||||
tw1 = manager.tw1
|
||||
tw1.level = 0
|
||||
tw1.is_done.side_effect = [False, True, False, True, True]
|
||||
tw1.update.side_effect = [False, False, True, True, True]
|
||||
tw1.is_success.return_value = False
|
||||
tw2 = manager.tw2
|
||||
tw2.level = 0
|
||||
tw2.is_done.side_effect = [False, False, False, False, True]
|
||||
tw2.update.side_effect = [True, False, False, True, True]
|
||||
tw2.is_success.return_value = False
|
||||
self.session.getTaskChildren.side_effect = lambda p: [
|
||||
{'id': 11}, {'id': 12}] if (0 == p) else []
|
||||
manager.attach_mock(self.session, 'sessionMock')
|
||||
def test_watch_tasks(self, stdout):
|
||||
# self.setup_record('foo.json')
|
||||
cfile = os.path.dirname(__file__) + '/data/calls/watchtasks1.json'
|
||||
with open(cfile) as fp:
|
||||
cdata = json.load(fp)
|
||||
self.session.load_calls(cdata)
|
||||
rv = watch_tasks(self.session, [1188], quiet=False, poll_interval=0)
|
||||
self.assertEqual(rv, 0)
|
||||
expected = (
|
||||
'''Watching tasks (this may be safely interrupted)...
|
||||
1188 build (f24, /users/mikem/fake.git:adaf62586b4b4a23b24394da5586abd7cd9f679e): closed
|
||||
1189 buildSRPMFromSCM (/users/mikem/fake.git:adaf62586b4b4a23b24394da5586abd7cd9f679e): closed
|
||||
1190 buildArch (fake-1.1-21.src.rpm, noarch): closed
|
||||
|
||||
def side_effect(*args, **kwargs):
|
||||
rt = None
|
||||
if args[0] not in list(range(2)):
|
||||
rt = mock.MagicMock()
|
||||
rt.level = args[2]
|
||||
rt.is_done.return_value = True
|
||||
rt.update.return_value = True
|
||||
rt.is_success.return_value = True
|
||||
manager.attach_mock(rt, 'tw' + str(args[0]))
|
||||
else:
|
||||
rt = {0: tw1, 1: tw2}.get(args[0])
|
||||
return rt
|
||||
|
||||
twClzMock.side_effect = side_effect
|
||||
rv = watch_tasks(self.session, list(range(2)), quiet=False, poll_interval=0)
|
||||
actual = stdout.getvalue()
|
||||
self.assertMultiLineEqual(
|
||||
actual, "Watching tasks (this may be safely interrupted)...\n\n")
|
||||
self.assertEqual(rv, 1)
|
||||
self.assertEqual(manager.mock_calls,
|
||||
[call.TaskWatcherMock(0, self.session, quiet=False),
|
||||
call.TaskWatcherMock(1, self.session, quiet=False),
|
||||
call.tw1.update(),
|
||||
call.tw1.is_done(),
|
||||
call.sessionMock.getTaskChildren(0),
|
||||
call.TaskWatcherMock(11, self.session, 1, quiet=False),
|
||||
call.tw11.update(),
|
||||
call.TaskWatcherMock(12, self.session, 1, quiet=False),
|
||||
call.tw12.update(),
|
||||
call.tw2.update(),
|
||||
call.tw2.is_done(),
|
||||
call.sessionMock.getTaskChildren(1),
|
||||
call.tw1.update(),
|
||||
call.tw1.is_done(),
|
||||
call.tw1.is_success(),
|
||||
call.sessionMock.getTaskChildren(0),
|
||||
call.tw2.update(),
|
||||
call.tw2.is_done(),
|
||||
call.sessionMock.getTaskChildren(1),
|
||||
call.tw11.update(),
|
||||
call.tw11.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw11.is_success(),
|
||||
call.sessionMock.getTaskChildren(11),
|
||||
call.tw12.update(),
|
||||
call.tw12.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw12.is_success(),
|
||||
call.sessionMock.getTaskChildren(12),
|
||||
call.tw1.update(),
|
||||
call.tw1.is_done(),
|
||||
call.sessionMock.getTaskChildren(0),
|
||||
call.tw2.update(),
|
||||
call.tw2.is_done(),
|
||||
call.sessionMock.getTaskChildren(1),
|
||||
call.tw11.update(),
|
||||
call.tw11.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw11.is_success(),
|
||||
call.sessionMock.getTaskChildren(11),
|
||||
call.tw12.update(),
|
||||
call.tw12.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw12.is_success(),
|
||||
call.sessionMock.getTaskChildren(12),
|
||||
call.tw1.update(),
|
||||
call.tw1.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw1.is_success(),
|
||||
call.sessionMock.getTaskChildren(0),
|
||||
call.tw2.update(),
|
||||
call.tw2.is_done(),
|
||||
call.sessionMock.getTaskChildren(1),
|
||||
call.tw11.update(),
|
||||
call.tw11.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw11.is_success(),
|
||||
call.sessionMock.getTaskChildren(11),
|
||||
call.tw12.update(),
|
||||
call.tw12.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw12.is_success(),
|
||||
call.sessionMock.getTaskChildren(12),
|
||||
call.tw1.update(),
|
||||
call.tw1.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw1.is_success(),
|
||||
call.sessionMock.getTaskChildren(0),
|
||||
call.tw2.update(),
|
||||
call.tw2.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw2.is_success(),
|
||||
call.sessionMock.getTaskChildren(1),
|
||||
call.tw11.update(),
|
||||
call.tw11.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw11.is_success(),
|
||||
call.sessionMock.getTaskChildren(11),
|
||||
call.tw12.update(),
|
||||
call.tw12.is_done(),
|
||||
call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}),
|
||||
call.tw12.is_success(),
|
||||
call.sessionMock.getTaskChildren(12),
|
||||
call.display_task_results_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12})
|
||||
])
|
||||
1188 build (f24, /users/mikem/fake.git:adaf62586b4b4a23b24394da5586abd7cd9f679e) completed successfully
|
||||
''')
|
||||
self.assertMultiLineEqual(stdout.getvalue(), expected)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.lib.TaskWatcher')
|
||||
@mock.patch('koji_cli.lib.display_tasklist_status')
|
||||
@mock.patch('koji_cli.lib.display_task_results')
|
||||
def test_watch_tasks_with_keyboardinterrupt(
|
||||
self, dtrMock, dtsMock, twClzMock, stdout):
|
||||
def test_watch_tasks_with_keyboardinterrupt(self, stdout):
|
||||
"""Raise KeyboardInterrupt inner watch_tasks.
|
||||
Raising it by SIGNAL might be better"""
|
||||
self.options.poll_interval = 0
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(twClzMock, 'TaskWatcherMock')
|
||||
manager.attach_mock(dtrMock, 'display_task_results_mock')
|
||||
manager.attach_mock(dtsMock, 'display_tasklist_status_mock')
|
||||
tw1 = manager.tw1
|
||||
tw1.level = 0
|
||||
tw1.is_done.side_effect = [False, KeyboardInterrupt, False]
|
||||
tw1.update.side_effect = [False, False]
|
||||
tw1.is_success.return_value = False
|
||||
tw1.str.return_value = 'tw1'
|
||||
tw1.display_state.return_value = 'tw1.display_state'
|
||||
tw2 = manager.tw2
|
||||
tw2.level = 0
|
||||
tw2.is_done.side_effect = [False, False, False, False, True]
|
||||
tw2.update.side_effect = [True, False, False, True, True]
|
||||
tw2.is_success.return_value = False
|
||||
tw2.str.return_value = 'tw2'
|
||||
tw2.display_state.return_value = 'tw2.display_state'
|
||||
self.session.getTaskChildren.side_effect = lambda p: [
|
||||
{'id': 11}, {'id': 12}] if (0 == p) else []
|
||||
manager.attach_mock(self.session, 'sessionMock')
|
||||
pass # TODO
|
||||
|
||||
def side_effect(*args, **kwargs):
|
||||
rt = None
|
||||
if args[0] not in list(range(2)):
|
||||
rt = mock.MagicMock()
|
||||
rt.level = args[2]
|
||||
rt.is_done.return_value = True
|
||||
rt.update.return_value = True
|
||||
rt.is_success.return_value = True
|
||||
manager.attach_mock(rt, 'tw' + str(args[0]))
|
||||
else:
|
||||
rt = {0: tw1, 1: tw2}.get(args[0])
|
||||
return rt
|
||||
|
||||
twClzMock.side_effect = side_effect
|
||||
|
||||
with self.assertRaises(KeyboardInterrupt):
|
||||
watch_tasks(self.session, list(range(2)), quiet=False, poll_interval=0)
|
||||
|
||||
actual = stdout.getvalue()
|
||||
self.assertMultiLineEqual(
|
||||
actual, """Watching tasks (this may be safely interrupted)...
|
||||
Tasks still running. You can continue to watch with the '%s watch-task' command.
|
||||
Running Tasks:
|
||||
tw1: tw1.display_state
|
||||
tw2: tw2.display_state
|
||||
""" % (os.path.basename(sys.argv[0]) or 'koji'))
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue