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_successful___result_is_successful(self):
response = self.app.get('/analysis_status/{}'.format('location'))
self.assertEqual(json.loads(response.data.decode('utf-8')), {
'id': -1,
'status': status.STATUS_SUCCESS,
'message': '',
'outputs_location': 'output_location',
})
fake_result('output_location', status.STATUS_SUCCESS),
]))
def test_celery_task_is_successful_with_location_on_seccond_attempt___result_is_successful(self):
response = self.app.get('/analysis_status/{}'.format('location'))
self.assertEqual(json.loads(response.data.decode('utf-8')), {
'id': -1,
'status': status.STATUS_SUCCESS,
'message': '',
'outputs_location': 'output_location',
})
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="",
items:
$ref: '#/definitions/AnalysisStatus'
404:
description: Resource not found.
parameters:
- name: location
in: path
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