aboutsummaryrefslogtreecommitdiffstats
path: root/functest/opnfv_tests/vnf/ims/ixia/utils
diff options
context:
space:
mode:
Diffstat (limited to 'functest/opnfv_tests/vnf/ims/ixia/utils')
-rw-r--r--functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py150
-rw-r--r--functest/opnfv_tests/vnf/ims/ixia/utils/IxLoadUtils.py397
-rw-r--r--functest/opnfv_tests/vnf/ims/ixia/utils/IxRestUtils.py192
-rw-r--r--functest/opnfv_tests/vnf/ims/ixia/utils/__init__.py0
4 files changed, 739 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
diff --git a/functest/opnfv_tests/vnf/ims/ixia/utils/IxLoadUtils.py b/functest/opnfv_tests/vnf/ims/ixia/utils/IxLoadUtils.py
new file mode 100644
index 00000000..d8003e46
--- /dev/null
+++ b/functest/opnfv_tests/vnf/ims/ixia/utils/IxLoadUtils.py
@@ -0,0 +1,397 @@
+#!/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 requests
+import sys
+import time
+import logging
+from IxRestUtils import formatDictToJSONPayload
+
+kActionStateFinished = 'finished'
+kActionStatusSuccessful = 'Successful'
+kActionStatusError = 'Error'
+kTestStateUnconfigured = 'Unconfigured'
+
+logger = logging.getLogger(__name__)
+
+
+def stripApiAndVersionFromURL(url):
+
+ # remove the slash (if any) at the beginning of the url
+
+ if url[0] == '/':
+ url = url[1:]
+
+ urlElements = url.split('/')
+ if 'api' in url:
+
+ # strip the api/v0 part of the url
+
+ urlElements = urlElements[2:]
+
+ return '/'.join(urlElements)
+
+
+def waitForActionToFinish(connection, replyObj, actionUrl):
+ """
+ This method waits for an action to finish executing. after a POST request
+ is sent in order to start an action, The HTTP reply will contain,
+ in the header, a 'location' field, that contains an URL.
+ The action URL contains the status of the action. we perform a GET on that
+ URL every 0.5 seconds until the action finishes with a success.
+ If the action fails, we will throw an error and
+ print the action's error message.
+ """
+
+ actionResultURL = replyObj.headers.get('location')
+ if actionResultURL:
+ actionResultURL = stripApiAndVersionFromURL(actionResultURL)
+ actionFinished = False
+
+ while not actionFinished:
+ actionStatusObj = connection.httpGet(actionResultURL)
+
+ if actionStatusObj.state == kActionStateFinished:
+ if actionStatusObj.status == kActionStatusSuccessful:
+ actionFinished = True
+ else:
+ errorMsg = "Error while executing action '%s'." \
+ % actionUrl
+
+ if actionStatusObj.status == kActionStatusError:
+ errorMsg += actionStatusObj.error
+
+ print errorMsg
+
+ sys.exit(1)
+ else:
+ time.sleep(0.1)
+
+
+def performGenericOperation(connection, url, payloadDict):
+ """
+ This will perform a generic operation on the given url,
+ it will wait for it to finish.
+ """
+
+ data = formatDictToJSONPayload(payloadDict)
+ reply = connection.httpPost(url=url, data=data)
+
+ waitForActionToFinish(connection, reply, url)
+
+ return reply
+
+
+def performGenericPost(connection, listUrl, payloadDict):
+ """
+ This will perform a generic POST method on a given url
+ """
+
+ data = formatDictToJSONPayload(payloadDict)
+
+ reply = connection.httpPost(url=listUrl, data=data)
+ try:
+ newObjPath = reply.headers['location']
+ except:
+ raise Exception('Location header is not present. \
+ Please check if the action was created successfully.')
+
+ newObjID = newObjPath.split('/')[-1]
+ return newObjID
+
+
+def performGenericDelete(connection, listUrl, payloadDict):
+ """
+ This will perform a generic DELETE method on a given url
+ """
+
+ data = formatDictToJSONPayload(payloadDict)
+
+ reply = connection.httpDelete(url=listUrl, data=data)
+ return reply
+
+
+def performGenericPatch(connection, url, payloadDict):
+ """
+ This will perform a generic PATCH method on a given url
+ """
+
+ data = formatDictToJSONPayload(payloadDict)
+
+ reply = connection.httpPatch(url=url, data=data)
+ return reply
+
+
+def createSession(connection, ixLoadVersion):
+ """
+ This method is used to create a new session.
+ It will return the url of the newly created session
+ """
+
+ sessionsUrl = 'sessions'
+ data = {'ixLoadVersion': ixLoadVersion}
+
+ sessionId = performGenericPost(connection, sessionsUrl, data)
+
+ newSessionUrl = '%s/%s' % (sessionsUrl, sessionId)
+ startSessionUrl = '%s/operations/start' % newSessionUrl
+
+ # start the session
+
+ performGenericOperation(connection, startSessionUrl, {})
+
+ logger.debug('Created session no %s' % sessionId)
+
+ return newSessionUrl
+
+
+def deleteSession(connection, sessionUrl):
+ """
+ This method is used to delete an existing session.
+ """
+
+ deleteParams = {}
+ performGenericDelete(connection, sessionUrl, deleteParams)
+
+
+def uploadFile(connection, url, fileName, uploadPath, overwrite=True):
+ headers = {'Content-Type': 'multipart/form-data'}
+ params = {'overwrite': overwrite, 'uploadPath': uploadPath}
+
+ logger.debug('Uploading...')
+ try:
+ with open(fileName, 'rb') as f:
+ resp = requests.post(url, data=f, params=params,
+ headers=headers)
+ except requests.exceptions.ConnectionError, e:
+ raise Exception('Upload file failed. Received connection error. \
+ One common cause for this error is the size of the \
+ file to be uploaded.The web server sets a limit of 1GB\
+ for the uploaded file size. \
+ Received the following error: %s' % str(e))
+ except IOError, e:
+ raise Exception('Upload file failed. Received IO error: %s'
+ % str(e))
+ except Exception:
+ raise Exception('Upload file failed. Received the following error: %s'
+ % str(e))
+ else:
+ logger.debug('Upload file finished.')
+ logger.debug('Response status code %s' % resp.status_code)
+ logger.debug('Response text %s' % resp.text)
+
+
+def loadRepository(connection, sessionUrl, rxfFilePath):
+ """
+ This method will perform a POST request to load a repository.
+ """
+
+ loadTestUrl = '%s/ixload/test/operations/loadTest' % sessionUrl
+ data = {'fullPath': rxfFilePath}
+
+ performGenericOperation(connection, loadTestUrl, data)
+
+
+def saveRxf(connection, sessionUrl, rxfFilePath):
+ """
+ This method saves the current rxf to the disk of the machine on
+ which the IxLoad instance is running.
+ """
+
+ saveRxfUrl = '%s/ixload/test/operations/saveAs' % sessionUrl
+ rxfFilePath = rxfFilePath.replace('\\', '\\\\')
+ data = {'fullPath': rxfFilePath, 'overWrite': 1}
+
+ performGenericOperation(connection, saveRxfUrl, data)
+
+
+def runTest(connection, sessionUrl):
+ """
+ This method is used to start the currently loaded test.
+ After starting the 'Start Test' action, wait for the action to complete.
+ """
+
+ startRunUrl = '%s/ixload/test/operations/runTest' % sessionUrl
+ data = {}
+
+ performGenericOperation(connection, startRunUrl, data)
+
+
+def getTestCurrentState(connection, sessionUrl):
+ """
+ This method gets the test current state.
+ (for example - running, unconfigured, ..)
+ """
+
+ activeTestUrl = '%s/ixload/test/activeTest' % sessionUrl
+ testObj = connection.httpGet(activeTestUrl)
+
+ return testObj.currentState
+
+
+def getTestRunError(connection, sessionUrl):
+ """
+ This method gets the error that appeared during the last test run.
+ If no error appeared (the test ran successfully),
+ the return value will be 'None'.
+ """
+
+ activeTestUrl = '%s/ixload/test/activeTest' % sessionUrl
+ testObj = connection.httpGet(activeTestUrl)
+
+ return testObj.testRunError
+
+
+def waitForTestToReachUnconfiguredState(connection, sessionUrl):
+ """
+ This method waits for the current test to reach the 'Unconfigured' state.
+ """
+
+ while getTestCurrentState(connection, sessionUrl) \
+ != kTestStateUnconfigured:
+ time.sleep(0.1)
+
+
+def pollStats(connection, sessionUrl, watchedStatsDict, pollingInterval=4):
+ """
+ This method is used to poll the stats.
+ Polling stats is per request but this method does a continuous poll.
+ """
+
+ statSourceList = watchedStatsDict.keys()
+ statsDict = {}
+
+ collectedTimestamps = {}
+ testIsRunning = True
+
+ # check stat sources
+
+ for statSource in statSourceList[:]:
+ statSourceUrl = '%s/ixload/stats/%s/values' % (sessionUrl, statSource)
+ statSourceReply = connection.httpRequest('GET', statSourceUrl)
+ if statSourceReply.status_code != 200:
+ logger.debug("Warning - Stat source '%s' does not exist. \
+ Will ignore it." % statSource)
+ statSourceList.remove(statSource)
+
+ # check the test state, and poll stats while the test is still running
+
+ while testIsRunning:
+
+ # the polling interval is configurable.
+ # by default, it's set to 4 seconds
+
+ time.sleep(pollingInterval)
+
+ for statSource in statSourceList:
+ valuesUrl = '%s/ixload/stats/%s/values' % (sessionUrl, statSource)
+
+ valuesObj = connection.httpGet(valuesUrl)
+ valuesDict = valuesObj.getOptions()
+
+ # get just the new timestamps - that were not previously
+ # retrieved in another stats polling iteration
+
+ newTimestamps = [int(timestamp) for timestamp in
+ valuesDict.keys() if timestamp
+ not in collectedTimestamps.get(statSource,
+ [])]
+ newTimestamps.sort()
+
+ for timestamp in newTimestamps:
+ timeStampStr = str(timestamp)
+
+ collectedTimestamps.setdefault(statSource, []).append(
+ timeStampStr)
+
+ timestampDict = statsDict.setdefault(statSource,
+ {}).setdefault(
+ timestamp, {})
+
+ # save the values for the current timestamp,
+ # and later print them
+
+ logger.info(' -- ')
+ for (caption, value) in \
+ valuesDict[timeStampStr].getOptions().items():
+ if caption in watchedStatsDict[statSource]:
+ logger.info(' %s -> %s' % (caption, value))
+ timestampDict[caption] = value
+
+ testIsRunning = getTestCurrentState(connection, sessionUrl) \
+ == 'Running'
+
+ logger.debug('Stopped receiving stats.')
+ return timestampDict
+
+
+def clearChassisList(connection, sessionUrl):
+ """
+ This method is used to clear the chassis list.
+ After execution no chassis should be available in the chassisListself.
+ """
+
+ chassisListUrl = '%s/ixload/chassischain/chassisList' % sessionUrl
+ deleteParams = {}
+ performGenericDelete(connection, chassisListUrl, deleteParams)
+
+
+def configureLicenseServer(connection, sessionUrl, licenseServerIp):
+ """
+ This method is used to clear the chassis list.
+ After execution no chassis should be available in the chassisList.
+ """
+
+ chassisListUrl = '%s/ixload/preferences' % sessionUrl
+ patchParams = {'licenseServer': licenseServerIp}
+ performGenericPatch(connection, chassisListUrl, patchParams)
+
+
+def addChassisList(connection, sessionUrl, chassisList):
+ """
+ This method is used to add one or more chassis to the chassis list.
+ """
+
+ chassisListUrl = '%s/ixload/chassisChain/chassisList' % sessionUrl
+
+ for chassisName in chassisList:
+ data = {'name': chassisName}
+ chassisId = performGenericPost(connection, chassisListUrl, data)
+
+ # refresh the chassis
+
+ refreshConnectionUrl = '%s/%s/operations/refreshConnection' \
+ % (chassisListUrl, chassisId)
+ performGenericOperation(connection, refreshConnectionUrl, {})
+
+
+def assignPorts(connection, sessionUrl, portListPerCommunity):
+ """
+ This method is used to assign ports from a connected chassis
+ to the required NetTraffics.
+ """
+
+ communtiyListUrl = '%s/ixload/test/activeTest/communityList' \
+ % sessionUrl
+
+ communityList = connection.httpGet(url=communtiyListUrl)
+
+ for community in communityList:
+ portListForCommunity = portListPerCommunity.get(community.name)
+
+ portListUrl = '%s/%s/network/portList' % (communtiyListUrl,
+ community.objectID)
+
+ if portListForCommunity:
+ for portTuple in portListForCommunity:
+ (chassisId, cardId, portId) = portTuple
+ paramDict = {'chassisId': chassisId, 'cardId': cardId,
+ 'portId': portId}
+
+ performGenericPost(connection, portListUrl, paramDict)
diff --git a/functest/opnfv_tests/vnf/ims/ixia/utils/IxRestUtils.py b/functest/opnfv_tests/vnf/ims/ixia/utils/IxRestUtils.py
new file mode 100644
index 00000000..63db403b
--- /dev/null
+++ b/functest/opnfv_tests/vnf/ims/ixia/utils/IxRestUtils.py
@@ -0,0 +1,192 @@
+#!/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
+
+from urlparse import urljoin
+import requests
+
+
+def getConnection(server, port):
+ """
+ Gets a Connection instance, that will be used to
+ make the HTTP requests to the application
+ """
+ connectionUrl = 'http://%s:%s/' % (server, port)
+
+ conn = Connection(connectionUrl, 'v0')
+ return conn
+
+
+def formatDictToJSONPayload(dictionary):
+ """
+ Converts a given python dict instance to a string
+ JSON payload that can be sent to a REST API.
+ """
+ jsonPayload = '{'
+ optionsList = []
+ for (key, val) in dictionary.items():
+ valStr = str(val)
+ if type(val) is str:
+ valStr = '"%s"' % val
+ if type(val) is bool:
+ valStr = valStr.lower()
+ optionsList.append('"%s":%s' % (key, valStr))
+
+ jsonPayload += ','.join(optionsList)
+ jsonPayload += '}'
+
+ return jsonPayload
+
+
+class Connection(object):
+
+ """
+ Class that executes the HTTP requests to the application instance.
+ It handles creating the HTTP session and executing HTTP methods.
+ """
+
+ kHeaderContentType = 'content-type'
+ kContentJson = 'application/json'
+
+ def __init__(self, siteUrl, apiVersion):
+ self.httpSession = None
+
+ self.url = Connection.urljoin(siteUrl, 'api')
+ self.url = Connection.urljoin(self.url, apiVersion)
+
+ def _getHttpSession(self):
+ """
+ This is a lazy initializer for the HTTP session.
+ It does not need to be active until it is required.
+ """
+
+ if self.httpSession is None:
+ self.httpSession = requests.Session()
+ return self.httpSession
+
+ @classmethod
+ def urljoin(cls, base, end):
+ """ Join two URLs. If the second URL is absolute, the base is ignored.
+
+ Use this instead of urlparse.urljoin directly so that we can customize
+ its behavior if necessary.
+ Currently differs in that it
+ 1. appends a / to base if not present.
+ 2. casts end to a str as a convenience
+ """
+
+ if base and not base.endswith('/'):
+ base = base + '/'
+ return urljoin(base, str(end))
+
+ def httpRequest(self, method, url='', data='', params={}, headers={}):
+ """
+ Method for making a HTTP request.
+ The method type (GET, POST, PATCH, DELETE) will be sent as a parameter.
+ Along with the url and request data. The HTTP response is returned
+ """
+
+ headers[Connection.kHeaderContentType] = Connection.kContentJson
+
+ absUrl = Connection.urljoin(self.url, url)
+ result = self._getHttpSession().request(method,
+ absUrl,
+ data=str(data),
+ params=params,
+ headers=headers)
+ return result
+
+ def httpGet(self, url='', data='', params={}, headers={}):
+ """
+ Method for calling HTTP GET.
+ This will return a WebObject that has the fields returned
+ in JSON format by the GET operation.
+ """
+
+ reply = self.httpRequest('GET', url, data, params, headers)
+ return _WebObject(reply.json())
+
+ def httpPost(self, url='', data='', params={}, headers={}):
+ """
+ Method for calling HTTP POST. Will return the HTTP reply.
+ """
+
+ return self.httpRequest('POST', url, data, params, headers)
+
+ def httpPatch(self, url='', data='', params={}, headers={}):
+ """
+ Method for calling HTTP PATCH. Will return the HTTP reply.
+ """
+
+ return self.httpRequest('PATCH', url, data, params, headers)
+
+ def httpDelete(self, url='', data='', params={}, headers={}):
+ """
+ Method for calling HTTP DELETE. Will return the HTTP reply.
+ """
+
+ return self.httpRequest('DELETE', url, data, params, headers)
+
+
+def _WebObject(value):
+ """
+ Method used for creating a wrapper object corresponding to the JSON string
+ received on a GET request.
+ """
+
+ if isinstance(value, dict):
+ result = WebObject(**value)
+ elif isinstance(value, list):
+ result = WebList(entries=value)
+ else:
+ result = value
+ return result
+
+
+class WebList(list):
+
+ """
+ Using this class a JSON list will be transformed
+ in a list of WebObject instances.
+ """
+
+ def __init__(self, entries=[]):
+ """
+ Create a WebList from a list of items that
+ are processed by the _WebObject function
+ """
+
+ for item in entries:
+ self.append(_WebObject(item))
+
+
+class WebObject(object):
+
+ """
+ A WebObject instance will have its fields set to correspond to
+ the JSON format received on a GET request.
+ """
+
+ def __init__(self, **entries):
+ """
+ Create a WebObject instance by providing a dict having a
+ property - value structure.
+ """
+
+ self.jsonOptions = {}
+ for (key, value) in entries.iteritems():
+ webObj = _WebObject(value)
+ self.jsonOptions[key] = webObj
+ self.__dict__[key] = webObj
+
+ def getOptions(self):
+ '''
+ Get the JSON dictionary which represents the WebObject Instance
+ '''
+
+ return self.jsonOptions
diff --git a/functest/opnfv_tests/vnf/ims/ixia/utils/__init__.py b/functest/opnfv_tests/vnf/ims/ixia/utils/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/functest/opnfv_tests/vnf/ims/ixia/utils/__init__.py