Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_celery_task_is_failure___result_is_failure(self):
response = self.app.get('/analysis_status/{}'.format('location'))
self.assertEqual(json.loads(response.data.decode('utf-8')), {
'id': -1,
'status': status.STATUS_FAILURE,
'message': "'message'",
'outputs_location': None,
})
@patch('src.server.app.CELERY.AsyncResult', Mock(return_value=fake_result('message', status.STATUS_FAILURE)))
def test_celery_task_is_failure___result_is_failure(self):
response = self.app.get('/analysis_status/{}'.format('location'))
self.assertEqual(json.loads(response.data.decode('utf-8')), {
'id': -1,
'status': status.STATUS_FAILURE,
'message': "'message'",
'outputs_location': None,
})
def _get_analysis_status(location):
"""
Get the status of an analysis
"""
result = CELERY.AsyncResult(location)
logging.debug("celery result:{}".format(result.result))
if result.state == status.STATUS_SUCCESS:
analysis_status = data.AnalysisStatus(
id=-1,
status=status.STATUS_SUCCESS,
message="",
outputs_location=result.result)
elif result.state == status.STATUS_FAILURE:
analysis_status = data.AnalysisStatus(
id=-1,
status=status.STATUS_FAILURE,
message=repr(result.result),
outputs_location=None)
else:
analysis_status = data.AnalysisStatus(
id=-1,
status=result.state,
message="",
outputs_location=None)
return analysis_status
description: The location of the outputs resource to download.
required: true
type: string
"""
analysis_status = _get_analysis_status(location)
# If there is no location for a successful analysis, retry once
# and then fail the analysis as something has gone wrong
if (analysis_status.status == status.STATUS_SUCCESS and analysis_status.outputs_location is None):
logging.info("Successful analysis has no location - retrying")
time.sleep(5)
analysis_status = _get_analysis_status(location)
if (analysis_status.status == status.STATUS_SUCCESS and analysis_status.outputs_location is None):
logging.info("Successful analysis still has no location - fail")
analysis_status.status = status.STATUS_FAILURE
response = jsonify(analysis_status)
logging.debug("Response: {}".format(response.data))
return response