cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Fixing lack of failure when uploaded source is invalid (#37461)

Browse source 
Checking the response status for 400 and throwing exception.
Unit tests updated.

Fixes #37406
This commit is contained in:
Evgeny Fedoruk 2018-04-04 16:48:22 +03:00 committed by Chris Alfonso
parent a4e932ee9f
commit 5e990301bb
2 changed files with 72 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
lib/ansible/modules/network/radware/vdirect_file.py
View file

@ -165,6 +165,21 @@ meta_args = dict(
)
class FileException(Exception):
def __init__(self, reason, details):
self.reason = reason
self.details = details
def __str__(self):
return 'Reason: {0}. Details:{1}.'.format(self.reason, self.details)
class InvalidSourceException(FileException):
def __init__(self, message):
super(InvalidSourceException, self).__init__(
'Error parsing file', repr(message))
class VdirectFile(object):
def __init__(self, params):
self.client = rest_client.RestClient(params['vdirect_ip'],
@ -185,26 +200,31 @@ class VdirectFile(object):
runnable_file = open(fqn, 'r')
file_content = runnable_file.read()
result_to_return = CONFIGURATION_TEMPLATE_CREATED_SUCCESS
result = template.create_from_source(file_content, template_name, fail_if_invalid=True)
if result[rest_client.RESP_STATUS] == 409:
template.upload_source(file_content, template_name, fail_if_invalid=True)
result = CONFIGURATION_TEMPLATE_UPDATED_SUCCESS
else:
result = CONFIGURATION_TEMPLATE_CREATED_SUCCESS
result_to_return = CONFIGURATION_TEMPLATE_UPDATED_SUCCESS
result = template.upload_source(file_content, template_name, fail_if_invalid=True)
if result[rest_client.RESP_STATUS] == 400:
raise InvalidSourceException(str(result[rest_client.RESP_STR]))
elif fqn.endswith(WORKFLOW_EXTENSION):
workflow = rest_client.WorkflowTemplate(self.client)
runnable_file = open(fqn, 'rb')
file_content = runnable_file.read()
result_to_return = WORKFLOW_TEMPLATE_CREATED_SUCCESS
result = workflow.create_template_from_archive(file_content, fail_if_invalid=True)
if result[rest_client.RESP_STATUS] == 409:
workflow.update_archive(file_content, os.path.splitext(os.path.basename(fqn))[0])
result = WORKFLOW_TEMPLATE_UPDATED_SUCCESS
else:
result = WORKFLOW_TEMPLATE_CREATED_SUCCESS
result_to_return = WORKFLOW_TEMPLATE_UPDATED_SUCCESS
result = workflow.update_archive(file_content, os.path.splitext(os.path.basename(fqn))[0])
if result[rest_client.RESP_STATUS] == 400:
raise InvalidSourceException(str(result[rest_client.RESP_STR]))
else:
result = WRONG_EXTENSION_ERROR
return result
result_to_return = WRONG_EXTENSION_ERROR
return result_to_return
def main():

52
test/units/modules/network/radware/test_vdirect_file.py
View file

@ -138,7 +138,7 @@ class TestManager(unittest.TestCase):
except IOError:
assert True
def test_template_upload_create_success(self, *args):
def test_template_upload_create(self, *args):
module_mock = MagicMock()
with patch.dict('sys.modules', **{
'vdirect_client': module_mock,
@ -148,14 +148,22 @@ class TestManager(unittest.TestCase):
vdirect_file.rest_client.RESP_STATUS = 0
vdirect_file.rest_client.Template = Template
Template.set_create_from_source_result([400])
file = vdirect_file.VdirectFile(NONE_PARAMS)
path = os.path.dirname(os.path.abspath(__file__))
Template.set_create_from_source_result([201])
result = file.upload(os.path.join(path, "ct.vm"))
self.assertEqual(result, vdirect_file.CONFIGURATION_TEMPLATE_CREATED_SUCCESS,
'Unexpected result received:' + repr(result))
def test_template_upload_update_success(self, *args):
Template.set_create_from_source_result([400, "", "Parsing error", ""])
try:
result = file.upload(os.path.join(path, "ct.vm"))
self.fail("InvalidSourceException was not thrown")
except vdirect_file.InvalidSourceException:
assert True
def test_template_upload_update(self, *args):
module_mock = MagicMock()
with patch.dict('sys.modules', **{
'vdirect_client': module_mock,
@ -165,15 +173,23 @@ class TestManager(unittest.TestCase):
vdirect_file.rest_client.RESP_STATUS = 0
vdirect_file.rest_client.Template = Template
Template.set_create_from_source_result([409])
Template.set_upload_source_result([400])
file = vdirect_file.VdirectFile(NONE_PARAMS)
path = os.path.dirname(os.path.abspath(__file__))
Template.set_create_from_source_result([409])
Template.set_upload_source_result([201])
result = file.upload(os.path.join(path, "ct.vm"))
self.assertEqual(result, vdirect_file.CONFIGURATION_TEMPLATE_UPDATED_SUCCESS,
'Unexpected result received:' + repr(result))
def test_workflow_upload_create_success(self, *args):
Template.set_upload_source_result([400, "", "Parsing error", ""])
try:
result = file.upload(os.path.join(path, "ct.vm"))
self.fail("InvalidSourceException was not thrown")
except vdirect_file.InvalidSourceException:
assert True
def test_workflow_upload_create(self, *args):
module_mock = MagicMock()
with patch.dict('sys.modules', **{
'vdirect_client': module_mock,
@ -183,14 +199,22 @@ class TestManager(unittest.TestCase):
vdirect_file.rest_client.RESP_STATUS = 0
vdirect_file.rest_client.WorkflowTemplate = WorkflowTemplate
WorkflowTemplate.set_create_template_from_archive_result([400])
file = vdirect_file.VdirectFile(NONE_PARAMS)
path = os.path.dirname(os.path.abspath(__file__))
WorkflowTemplate.set_create_template_from_archive_result([201])
result = file.upload(os.path.join(path, "wt.zip"))
self.assertEqual(result, vdirect_file.WORKFLOW_TEMPLATE_CREATED_SUCCESS,
'Unexpected result received:' + repr(result))
def test_workflow_upload_update_success(self, *args):
WorkflowTemplate.set_create_template_from_archive_result([400, "", "Parsing error", ""])
try:
result = file.upload(os.path.join(path, "wt.zip"))
self.fail("InvalidSourceException was not thrown")
except vdirect_file.InvalidSourceException:
assert True
def test_workflow_upload_update(self, *args):
module_mock = MagicMock()
with patch.dict('sys.modules', **{
'vdirect_client': module_mock,
@ -200,10 +224,18 @@ class TestManager(unittest.TestCase):
vdirect_file.rest_client.RESP_STATUS = 0
vdirect_file.rest_client.WorkflowTemplate = WorkflowTemplate
WorkflowTemplate.set_create_template_from_archive_result([409])
WorkflowTemplate.set_update_archive_result([400])
file = vdirect_file.VdirectFile(NONE_PARAMS)
path = os.path.dirname(os.path.abspath(__file__))
WorkflowTemplate.set_create_template_from_archive_result([409])
WorkflowTemplate.set_update_archive_result([201])
result = file.upload(os.path.join(path, "wt.zip"))
self.assertEqual(result, vdirect_file.WORKFLOW_TEMPLATE_UPDATED_SUCCESS,
'Unexpected result received:' + repr(result))
WorkflowTemplate.set_update_archive_result([400, "", "Parsing error", ""])
try:
result = file.upload(os.path.join(path, "wt.zip"))
self.fail("InvalidSourceException was not thrown")
except vdirect_file.InvalidSourceException:
assert True