Resources to handle testcase related requests
"""
-import ConfigParser
import logging
import os
import re
from flask import jsonify
from flasgger.utils import swag_from
import pkg_resources
+from six.moves import configparser
from functest.api.base import ApiResource
from functest.api.common import api_utils, thread
def _update_logging_ini(self, task_id): # pylint: disable=no-self-use
""" Update the log file for each task"""
- rconfig = ConfigParser.RawConfigParser()
+ rconfig = configparser.RawConfigParser()
rconfig.read(
pkg_resources.resource_filename('xtesting', 'ci/logging.ini'))
log_path = os.path.join(getattr(config.CONF, 'dir_results'),
import inspect
import logging
import socket
-from urlparse import urljoin
import pkg_resources
from flask import Flask
from flask_restful import Api
from flasgger import Swagger
+import six
from functest.api.base import ApiResource
from functest.api.common import api_utils
def get_endpoint(url):
""" Obtain the endpoint of url """
address = socket.gethostbyname(socket.gethostname())
- return urljoin('http://{}:5000'.format(address), url)
+ return six.moves.urllib.parse.urljoin(
+ 'http://{}:5000'.format(address), url)
def api_add_resource():
import sys
import uuid
from copy import deepcopy
-from urlparse import urljoin
import pkg_resources
import yaml
-from functest.core import vnf
-from functest.opnfv_tests.openstack.snaps import snaps_utils
-from functest.utils import config
-from functest.utils import env
-
+import six
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.network import NetworkConfig, SubnetConfig
from snaps.openstack.utils import keystone_utils
from snaps.openstack.utils import nova_utils
+from functest.core import vnf
+from functest.opnfv_tests.openstack.snaps import snaps_utils
+from functest.utils import config
+from functest.utils import env
+
__author__ = "Amarendra Meher <amarendra@rebaca.com>"
__author__ = "Soumaya K Nayek <soumaya.nayek@rebaca.com>"
# it enforces a versioned public identity endpoint as juju simply
# adds /auth/tokens wich fails vs an unversioned endpoint.
if not self.public_auth_url.endswith(('v3', 'v3/', 'v2.0', 'v2.0/')):
- self.public_auth_url = urljoin(self.public_auth_url, 'v3')
+ self.public_auth_url = six.moves.urllib.parse.urljoin(
+ self.public_auth_url, 'v3')
self._register_cloud()
if self.snaps_creds.identity_api_version == 3:
self._register_credentials_v3()
self.__logger.info("Upload some OS images if it doesn't exist")
images = get_config("tenant_images", self.config_file)
self.__logger.info("Images needed for vEPC: %s", images)
- for image_name, image_file in images.iteritems():
+ for image_name, image_file in six.iteritems(images):
self.__logger.info("image: %s, file: %s", image_name, image_file)
if image_file and image_name:
image_creator = OpenStackImage(self.snaps_creds, ImageConfig(
from cloudify_rest_client import CloudifyClient
from cloudify_rest_client.executions import Execution
from scp import SCPClient
-import yaml
-
+import six
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
from snaps.openstack.create_user import OpenStackUser
from snaps.openstack.utils import keystone_utils
from xtesting.energy import energy
+import yaml
from functest.opnfv_tests.openstack.snaps import snaps_utils
-import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
+from functest.opnfv_tests.vnf.ims import clearwater_ims_base
from functest.utils import config
from functest.utils import env
# needs some images
self.__logger.info("Upload some OS images if it doesn't exist")
- for image_name, image_file in self.images.iteritems():
+ for image_name, image_file in six.iteritems(self.images):
self.__logger.info("image: %s, file: %s", image_name, image_file)
if image_file and image_name:
image_creator = OpenStackImage(
raise Exception("Cloudify Manager isn't up and running")
self.__logger.info("Put OpenStack creds in manager")
secrets_list = cfy_client.secrets.list()
- for k, val in cfy_creds.iteritems():
+ for k, val in six.iteritems(cfy_creds):
if not any(d.get('key', None) == k for d in secrets_list):
cfy_client.secrets.create(k, val)
else:
i = 0
for rdata in result:
i = i + 1
- print rdata.address
+ print(rdata.address)
iplistims += str(rdata.address)
if i != len(result):
iplistims += ';'
headers=urlHeadersJson)
logger.debug(response)
logger.debug(content)
- except Exception, e:
+ except Exception as e:
raise Exception('Got an error code: ', e)
return content
(response, content) = h.request(loginUrl, 'POST',
body=payload,
headers=urlHeadersJson)
- except Exception, e:
+ except Exception as e:
raise Exception('Got an error code: ', e)
return content
(response, content) = h.request(url, 'POST',
json.dumps(payload),
headers=urlHeadersJson)
- except Exception, e:
+ except Exception as e:
raise Exception('Got an error code: ', e)
return content
(response, content) = h.request(url, 'PATCH',
json.dumps(payload),
urlHeadersJson)
- except Exception, e:
+ except Exception as e:
- # print (response, content)
+ # print((response, content))
raise Exception('Got an error code: ', e)
return content
disable_ssl_certificate_validation=True)
(response, content) = h.request(url, 'DELETE', '', urlHeadersJson)
logger.debug('DELETE: ' + url)
- except Exception, e:
+ except Exception as e:
raise Exception('Got an error code: ', e)
if response.status not in okStates:
raise TestFailedError(json.loads(content)['error'])
disable_ssl_certificate_validation=True)
logger.debug('GET: ' + url)
(response, content) = h.request(url, 'GET', '', urlHeadersJson)
- except Exception, e:
+ except Exception as e:
raise Exception('Got an error code: ', e)
if response.status not in okStates:
raise TestFailedError(json.loads(content)['error'])
if actionStatusObj.status == kActionStatusError:
errorMsg += actionStatusObj.error
- print errorMsg
+ print(errorMsg)
sys.exit(1)
else:
with open(fileName, 'rb') as f:
resp = requests.post(url, data=f, params=params,
headers=headers)
- except requests.exceptions.ConnectionError, e:
+ except requests.exceptions.ConnectionError as 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:
+ except IOError as e:
raise Exception('Upload file failed. Received IO error: %s'
% str(e))
except Exception:
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
-from urlparse import urljoin
import requests
+import six
+from six.moves.urllib.parse import urljoin
+
+
def getConnection(server, port):
"""
Gets a Connection instance, that will be used to
"""
self.jsonOptions = {}
- for (key, value) in entries.iteritems():
+ for (key, value) in six.iteritems(entries):
webObj = _WebObject(value)
self.jsonOptions[key] = webObj
self.__dict__[key] = webObj
from snaps.openstack.utils import keystone_utils
+import six
+
+
__author__ = "Shuya Nakama <shuya.nakama@okinawaopenlabs.org>"
self.created_object.append(keypair_creator)
self.__logger.info("Upload some OS images if it doesn't exist")
- for image_name, image_file in self.images.iteritems():
+ for image_name, image_file in six.iteritems(self.images):
self.__logger.info("image: %s, file: %s", image_name, image_file)
if image_file and image_name:
image_creator = OpenStackImage(
dirs =
functest/tests/unit/ci
functest/tests/unit/cli
+ functest/tests/unit/core
functest/tests/unit/odl
- functest/tests/unit/openstack/rally
- functest/tests/unit/openstack/tempest
+ functest/tests/unit/openstack
+ functest/tests/unit/vnf/epc
functest/tests/unit/utils
commands = nosetests {[testenv:py35]dirs}
Code Review / functest.git / commitdiff