particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

CLI download-task more specific info for not CLOSED tasks.

Browse source 
Fixes: https://pagure.io/koji/issue/3487
This commit is contained in:
Jana Cupova 2022-09-07 11:09:36 +02:00 • committed by Tomas Kopecek
parent ffb7af26d8
commit d2c580a3f8
2 changed files with 112 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

14
cli/koji_cli/commands.py
View file

@ -6940,11 +6940,19 @@ def anon_handle_download_task(options, session, args):
required_tasks[task["id"]] = task
for task_id in required_tasks:
if required_tasks[task_id]["state"] != koji.TASK_STATES.get("CLOSED"):
task_state = koji.TASK_STATES.get(required_tasks[task_id]["state"])
if task_state != "CLOSED":
if task_id == base_task_id:
error("Task %d has not finished yet." % task_id)
start_error_msg = "Task"
else:
error("Child task %d has not finished yet." % task_id)
start_error_msg = "Child task"
if task_state == 'FAILED':
error("%s %d failed. You can use save-failed-tree plugin for FAILED tasks."
% (start_error_msg, task_id))
elif task_state == 'CANCELED':
error("%s %d was canceled." % (start_error_msg, task_id))
else:
error("%s %d has not finished yet." % (start_error_msg, task_id))
# get files for download
downloads = []

113
tests/test_cli/test_download_task.py
View file

@ -239,20 +239,112 @@ Default behavior without --all option downloads .rpm files only for build and bu
self.session, self.parent_task_id)
self.download_file.assert_not_called()
def test_handle_download_parent_not_finished(self):
def test_handle_download_parent_canceled_task(self):
args = [str(self.parent_task_id)]
self.session.getTaskInfo.return_value = {
'id': self.parent_task_id,
'method': 'buildArch',
'arch': 'taskarch',
'state': 3}
self.list_task_output_all_volumes.return_value = {
'somerpm.src.rpm': ['DEFAULT', 'vol1'],
'somerpm.x86_64.rpm': ['DEFAULT', 'vol2'],
'somerpm.noarch.rpm': ['vol3'],
'somelog.log': ['DEFAULT', 'vol1'],
'somezip.zip': ['DEFAULT']
}
# Run it and check immediate output
# args: task_id
# expected: failure
self.assert_system_exit(
anon_handle_download_task,
self.options, self.session, args,
stderr="Task 123 was canceled.\n",
stdout='',
activate_session=None,
exit_code=1)
# Finally, assert that things were called as we expected.
self.ensure_connection.assert_called_once_with(self.session, self.options)
self.session.getTaskInfo.assert_called_once_with(self.parent_task_id)
self.session.getTaskChildren.assert_called_once_with(self.parent_task_id)
self.list_task_output_all_volumes.assert_not_called()
self.download_file.assert_not_called()
def test_handle_download_child_canceled_task(self):
args = [str(self.parent_task_id)]
self.session.getTaskInfo.return_value = self.parent_task_info
self.session.getTaskChildren.return_value = [{
'id': 22222,
'method': 'buildArch',
'arch': 'noarch',
'state': 3}]
# Run it and check immediate output
# args: task_id
# expected: failure
self.assert_system_exit(
anon_handle_download_task,
self.options, self.session, args,
stderr="Child task 22222 was canceled.\n",
stdout='',
activate_session=None,
exit_code=1)
# Finally, assert that things were called as we expected.
self.ensure_connection.assert_called_once_with(self.session, self.options)
self.session.getTaskInfo.assert_called_once_with(self.parent_task_id)
self.session.getTaskChildren.assert_called_once_with(self.parent_task_id)
self.list_task_output_all_volumes.assert_not_called()
self.download_file.assert_not_called()
def test_handle_download_parent_failed_task(self):
args = [str(self.parent_task_id)]
self.session.getTaskInfo.return_value = {
'id': self.parent_task_id,
'method': 'buildArch',
'arch': 'taskarch',
'state': 5}
# Run it and check immediate output
# args: task_id
# expected: failure
self.assert_system_exit(
anon_handle_download_task,
self.options, self.session, args,
stderr="Task 123 failed. You can use save-failed-tree plugin for FAILED tasks.\n",
stdout='',
activate_session=None,
exit_code=1)
# Finally, assert that things were called as we expected.
self.ensure_connection.assert_called_once_with(self.session, self.options)
self.session.getTaskInfo.assert_called_once_with(self.parent_task_id)
self.session.getTaskChildren.assert_called_once_with(self.parent_task_id)
self.list_task_output_all_volumes.assert_not_called()
self.download_file.assert_not_called()
def test_handle_download_child_failed_task(self):
args = [str(self.parent_task_id)]
self.session.getTaskInfo.return_value = self.parent_task_info
self.session.getTaskChildren.return_value = [{
'id': 22222,
'method': 'buildArch',
'arch': 'noarch',
'state': 5}]
# Run it and check immediate output
# args: task_id
# expected: failure
self.assert_system_exit(
anon_handle_download_task,
self.options, self.session, args,
stderr="Child task 22222 failed. "
"You can use save-failed-tree plugin for FAILED tasks.\n",
stdout='',
activate_session=None,
exit_code=1)
# Finally, assert that things were called as we expected.
self.ensure_connection.assert_called_once_with(self.session, self.options)
self.session.getTaskInfo.assert_called_once_with(self.parent_task_id)
self.session.getTaskChildren.assert_called_once_with(self.parent_task_id)
self.list_task_output_all_volumes.assert_not_called()
self.download_file.assert_not_called()
def test_handle_download_parent_not_finished(self):
args = [str(self.parent_task_id)]
self.session.getTaskInfo.return_value = {
'id': self.parent_task_id,
'method': 'buildArch',
'arch': 'taskarch',
'state': 1}
# Run it and check immediate output
# args: task_id
# expected: failure
@ -277,10 +369,7 @@ Default behavior without --all option downloads .rpm files only for build and bu
'id': 22222,
'method': 'buildArch',
'arch': 'noarch',
'state': 3}]
self.list_task_output_all_volumes.side_effect = [
{'somerpm.src.rpm': ['DEFAULT', 'vol1']},
{'somenextrpm.src.rpm': ['DEFAULT', 'vol1']}]
'state': 1}]
# Run it and check immediate output
# args: task_id
# expected: failure