diff options
author | 2017-09-14 21:26:15 +0000 | |
---|---|---|
committer | 2017-09-14 21:26:16 +0000 | |
commit | f45b9587c5568c0b64cfedb94e4241e485ffce47 (patch) | |
tree | 09cfe848259df5bf3c6617a4c85a2bef6ccd1f5e /functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py | |
parent | 890076da4b4d80c392c7e952282e9c721508b0d6 (diff) | |
parent | 2f9219dfa44982779990e13c177a703f2239b488 (diff) |
Merge "New testcase creation named "cloudify_ims_perf""
Diffstat (limited to 'functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py')
-rw-r--r-- | functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py | 150 |
1 files changed, 150 insertions, 0 deletions
diff --git a/functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py b/functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py new file mode 100644 index 00000000..973e0264 --- /dev/null +++ b/functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python + +# Copyright (c) 2017 IXIA and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import httplib2 +import json +import logging + + +okStates = [200, 201, 202] +states = [ + 'Queued', + 'In Progress', + 'Manual Step Required', + 'Error', + 'Finished', + 'Aborted', + 'Retried', + 'IRebooting', + 'Force Continue', + 'Pending', + ] +notStartedState = 'Not_Started' +errorStates = ['Error', 'Aborted', 'Force Continue'] +finishedStates = ['Manual Step Required', 'Finished'] + +logger = logging.getLogger(__name__) + + +class TestFailedError(Exception): + pass + + +class ChassisRestAPI: + @staticmethod + def postWithPayload(loginUrl, payload=None): + urlHeadersJson = {'content-type': 'application/json'} + try: + h = httplib2.Http('.cache', + disable_ssl_certificate_validation=True) + if payload is None: + logger.debug('POST: ' + loginUrl) + (response, content) = h.request(loginUrl, 'POST', '', + urlHeadersJson) + logger.debug(content) + else: + logger.debug('POST: ' + loginUrl + ' <- Data: ' + str(payload)) + (response, content) = h.request(loginUrl, 'POST', + body=payload, + headers=urlHeadersJson) + logger.debug(response) + logger.debug(content) + except Exception, e: + raise Exception('Got an error code: ', e) + return content + + @staticmethod + def postWithPayloadAndHeaders(loginUrl, urlHeadersJson, + payload=None): + try: + h = httplib2.Http('.cache', + disable_ssl_certificate_validation=True) + if payload is None: + logger.debug('POST: ' + loginUrl) + (response, content) = h.request(loginUrl, 'POST', '', + urlHeadersJson) + else: + logger.debug('POST: ' + loginUrl + ' <- Data: ' + str(payload)) + (response, content) = h.request(loginUrl, 'POST', + body=payload, + headers=urlHeadersJson) + except Exception, e: + raise Exception('Got an error code: ', e) + return content + + @staticmethod + def postOperation(url, apiKey, payload=''): + urlHeadersJson = {'content-type': 'application/json', + 'X-Api-Key': '%s' % str(apiKey)} + try: + h = httplib2.Http('.cache', + disable_ssl_certificate_validation=True) + if payload is None: + logger.debug('POST: ' + url) + (response, content) = h.request(url, 'POST', + json.dumps(payload), + urlHeadersJson) + else: + logger.debug('POST: ' + url + ' <- Data: ' + str(payload)) + (response, content) = h.request(url, 'POST', + json.dumps(payload), + headers=urlHeadersJson) + except Exception, e: + raise Exception('Got an error code: ', e) + return content + + @staticmethod + def patch(url, payload, apiKey): + urlHeadersJson = {'content-type': 'application/json', + 'X-Api-Key': '%s' % str(apiKey)} + try: + h = httplib2.Http('.cache', + disable_ssl_certificate_validation=True) + logger.debug('PATCH: ' + url + ' <-- Attribute: ' + + str(payload)) + (response, content) = h.request(url, 'PATCH', + json.dumps(payload), + urlHeadersJson) + except Exception, e: + + # print (response, content) + + raise Exception('Got an error code: ', e) + return content + + @staticmethod + def delete(url, apiKey): + urlHeadersJson = {'content-type': 'application/json', + 'X-Api-Key': '%s' % str(apiKey)} + try: + h = httplib2.Http('.cache', + disable_ssl_certificate_validation=True) + (response, content) = h.request(url, 'DELETE', '', urlHeadersJson) + logger.debug('DELETE: ' + url) + except Exception, e: + raise Exception('Got an error code: ', e) + if response.status not in okStates: + raise TestFailedError(json.loads(content)['error']) + return json.loads(content) + + @staticmethod + def getWithHeaders(url, apiKey): + urlHeadersJson = {'content-type': 'application/json', + 'X-Api-Key': '%s' % str(apiKey)} + try: + h = httplib2.Http('.cache', + disable_ssl_certificate_validation=True) + logger.debug('GET: ' + url) + (response, content) = h.request(url, 'GET', '', urlHeadersJson) + except Exception, e: + raise Exception('Got an error code: ', e) + if response.status not in okStates: + raise TestFailedError(json.loads(content)['error']) + output = json.loads(content) + return output |