summary |
shortlog |
log |
commit | commitdiff |
review |
tree
raw |
patch |
inline | side by side (from parent 1:
09d6e6f)
It has to be noted that cloudify_rest_client doesn't support py3 then
next unit tests vs cloudify-based testcases would be excluded.
Change-Id: I0b0e64929d1238665c778ce51fd6abd5ccf06b6c
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>
Resources to handle testcase related requests
"""
Resources to handle testcase related requests
"""
import logging
import os
import re
import logging
import os
import re
from flask import jsonify
from flasgger.utils import swag_from
import pkg_resources
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
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"""
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'),
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
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 pkg_resources
from flask import Flask
from flask_restful import Api
from flasgger import Swagger
from functest.api.base import ApiResource
from functest.api.common import api_utils
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())
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)
import sys
import uuid
from copy import deepcopy
import sys
import uuid
from copy import deepcopy
-from urlparse import urljoin
import pkg_resources
import yaml
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
-
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.network import NetworkConfig, SubnetConfig
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 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>"
__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/')):
# 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._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)
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(
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
from cloudify_rest_client import CloudifyClient
from cloudify_rest_client.executions import Execution
from scp import SCPClient
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
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
from snaps.openstack.create_user import OpenStackUser
from snaps.openstack.utils import keystone_utils
from xtesting.energy import energy
from functest.opnfv_tests.openstack.snaps import snaps_utils
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
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")
# 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(
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()
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:
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
i = 0
for rdata in result:
i = i + 1
iplistims += str(rdata.address)
if i != len(result):
iplistims += ';'
iplistims += str(rdata.address)
if i != len(result):
iplistims += ';'
headers=urlHeadersJson)
logger.debug(response)
logger.debug(content)
headers=urlHeadersJson)
logger.debug(response)
logger.debug(content)
raise Exception('Got an error code: ', e)
return content
raise Exception('Got an error code: ', e)
return content
(response, content) = h.request(loginUrl, 'POST',
body=payload,
headers=urlHeadersJson)
(response, content) = h.request(loginUrl, 'POST',
body=payload,
headers=urlHeadersJson)
raise Exception('Got an error code: ', e)
return content
raise Exception('Got an error code: ', e)
return content
(response, content) = h.request(url, 'POST',
json.dumps(payload),
headers=urlHeadersJson)
(response, content) = h.request(url, 'POST',
json.dumps(payload),
headers=urlHeadersJson)
raise Exception('Got an error code: ', e)
return content
raise Exception('Got an error code: ', e)
return content
(response, content) = h.request(url, 'PATCH',
json.dumps(payload),
urlHeadersJson)
(response, content) = h.request(url, 'PATCH',
json.dumps(payload),
urlHeadersJson)
- # print (response, content)
+ # print((response, content))
raise Exception('Got an error code: ', e)
return 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)
disable_ssl_certificate_validation=True)
(response, content) = h.request(url, 'DELETE', '', urlHeadersJson)
logger.debug('DELETE: ' + url)
raise Exception('Got an error code: ', e)
if response.status not in okStates:
raise TestFailedError(json.loads(content)['error'])
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)
disable_ssl_certificate_validation=True)
logger.debug('GET: ' + url)
(response, content) = h.request(url, 'GET', '', urlHeadersJson)
raise Exception('Got an error code: ', e)
if response.status not in okStates:
raise TestFailedError(json.loads(content)['error'])
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
if actionStatusObj.status == kActionStatusError:
errorMsg += actionStatusObj.error
with open(fileName, 'rb') as f:
resp = requests.post(url, data=f, params=params,
headers=headers)
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))
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))
raise Exception('Upload file failed. Received IO error: %s'
% str(e))
except Exception:
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
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
-from urlparse import urljoin
+import six
+from six.moves.urllib.parse import urljoin
+
+
def getConnection(server, port):
"""
Gets a Connection instance, that will be used to
def getConnection(server, port):
"""
Gets a Connection instance, that will be used to
"""
self.jsonOptions = {}
"""
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
webObj = _WebObject(value)
self.jsonOptions[key] = webObj
self.__dict__[key] = webObj
from snaps.openstack.utils import keystone_utils
from snaps.openstack.utils import keystone_utils
__author__ = "Shuya Nakama <shuya.nakama@okinawaopenlabs.org>"
__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")
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(
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
dirs =
functest/tests/unit/ci
functest/tests/unit/cli
+ functest/tests/unit/core
- 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}
functest/tests/unit/utils
commands = nosetests {[testenv:py35]dirs}