almalinux/pungi
14
1
Fork 0
Code Issues Pull Requests Releases Activity

[koji-wrapper] Get task id from failed runroot

Browse Source 
Even when runroot task fails, it can emit task_id.

Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
This commit is contained in:
Lubomír Sedlář 2016-02-12 12:38:45 +01:00
parent 93015d679c
commit e1895bff26
2 changed files with 23 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
pungi/wrappers/kojiwrapper.py
View File

@ -78,16 +78,23 @@ class KojiWrapper(object):
return cmd return cmd
def run_runroot_cmd(self, command, log_file=None): def run_runroot_cmd(self, command, log_file=None):
# runroot is blocking -> you probably want to run it in a thread """
Run koji runroot command and wait for results.
If the command specified --task-id, and the first line of output
contains the id, it will be captured and returned.
"""
task_id = None task_id = None
retcode, output = run(command, can_fail=True, logfile=log_file) retcode, output = run(command, can_fail=True, logfile=log_file)
if retcode == 0 and "--task-id" in command: if "--task-id" in command:
task_id = int(output.splitlines()[0]) first_line = output.splitlines()[0]
output_ends_with_eol = output.endswith("\n") if re.match(r'^\d+$', first_line):
output = "\n".join(output.splitlines()[1:]) task_id = int(first_line)
if output_ends_with_eol: # Remove first line from the output, preserving any trailing newlines.
output += "\n" output_ends_with_eol = output.endswith("\n")
output = "\n".join(output.splitlines()[1:])
if output_ends_with_eol:
output += "\n"
return { return {
"retcode": retcode, "retcode": retcode,

9
tests/test_koji_wrapper.py
View File

@ -382,6 +382,15 @@ class RunrootKojiWrapperTest(KojiWrapperBaseTestCase):
result = self.koji.run_runroot_cmd(cmd) result = self.koji.run_runroot_cmd(cmd)
self.assertDictEqual(result, {'retcode': 1, 'output': output, 'task_id': None}) self.assertDictEqual(result, {'retcode': 1, 'output': output, 'task_id': None})
@mock.patch('pungi.wrappers.kojiwrapper.run')
def test_run_runroot_cmd_with_task_id_and_fail_but_emit_id(self, run):
cmd = ['koji', 'runroot', '--task-id']
output = 'Nope, does not work.\n'
run.return_value = (1, '12345\n' + output)
result = self.koji.run_runroot_cmd(cmd)
self.assertDictEqual(result, {'retcode': 1, 'output': output, 'task_id': 12345})
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()