External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-09 23:12:08 -03:30
Code Issues Packages Projects Releases Wiki Activity

Make workflow_job.assert_successful() give specifics

Browse Source
This commit is contained in:
Alan Rominger 2021-01-05 10:15:32 -05:00
parent 7e74f823f4
commit d79b96b6cc
No known key found for this signature in database
GPG Key ID: C2D7EAAA12B63559
2 changed files with 38 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
awxkit/awxkit/api/mixins/has_status.py
View File

@ -47,6 +47,13 @@ class HasStatus(object):
def wait_until_started(self, interval=1, timeout=60):
return self.wait_until_status(self.started_statuses, interval=interval, timeout=timeout)
def failure_output_details(self):
if getattr(self, 'result_stdout', ''):
output = bytes_to_str(self.result_stdout)
if output:
return '\nstdout:\n{}'.format(output)
return ''
def assert_status(self, status_list, msg=None):
if isinstance(status_list, str):
status_list = [status_list]
@ -65,10 +72,9 @@ class HasStatus(object):
msg += '\njob_explanation: {}'.format(bytes_to_str(self.job_explanation))
if getattr(self, 'result_traceback', ''):
msg += '\nresult_traceback:\n{}'.format(bytes_to_str(self.result_traceback))
if getattr(self, 'result_stdout', ''):
output = bytes_to_str(self.result_stdout)
if output:
msg = msg + '\nstdout:\n{}'.format(output)
msg += self.failure_output_details()
if getattr(self, 'job_explanation', '').startswith('Previous Task Failed'):
try:
data = json.loads(self.job_explanation.replace('Previous Task Failed: ', ''))

30
awxkit/awxkit/api/pages/workflow_jobs.py
View File

@ -13,11 +13,37 @@ class WorkflowJob(UnifiedJob):
result = self.related.relaunch.post(payload)
return self.walk(result.url)
def failure_output_details(self):
"""Special implementation of this part of assert_status so that
workflow_job.assert_successful() will give a breakdown of failure
"""
msg = '\nNode summary:'
node_list = self.related.workflow_nodes.get().results
for node in node_list:
msg += '\n{}:'.format(node.id)
if node.job:
msg += ' {}'.format(node.summary_fields.job)
else:
msg += ' None'
for rel in ('failure_nodes', 'always_nodes', 'success_nodes'):
val = getattr(node, rel, [])
if val:
msg += ' {} {}'.format(rel, val)
msg += '\n\nUnhandled individual job failures:\n'
for node in node_list:
if node.job and not (node.failure_nodes or node.always_nodes):
job = node.related.job.get()
try:
job.assert_successful()
except Exception as e:
msg += str(e)
return msg
@property
def result_stdout(self):
# workflow jobs do not have result_stdout
# which is problematic for the UnifiedJob.is_successful reliance on
# related stdout endpoint.
if 'result_stdout' not in self.json:
return 'Unprovided AWX field.'
else: