summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--functest/ci/check_deployment.py24
-rw-r--r--functest/ci/run_tests.py121
-rw-r--r--functest/ci/tier_builder.py7
-rw-r--r--functest/ci/tier_handler.py14
-rw-r--r--functest/core/testcase.py35
-rw-r--r--functest/opnfv_tests/vnf/epc/juju_epc.py33
-rw-r--r--functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py30
-rw-r--r--functest/opnfv_tests/vnf/ims/orchestra_openims.py26
-rw-r--r--functest/tests/unit/ci/test_check_deployment.py7
-rw-r--r--functest/tests/unit/ci/test_run_tests.py24
-rw-r--r--functest/tests/unit/ci/test_tier_builder.py2
-rw-r--r--functest/tests/unit/ci/test_tier_handler.py8
-rw-r--r--functest/tests/unit/core/test_testcase.py46
-rw-r--r--functest/utils/openstack_utils.py1
-rw-r--r--tox.ini3
15 files changed, 171 insertions, 210 deletions
diff --git a/functest/ci/check_deployment.py b/functest/ci/check_deployment.py
index 8cc522f69..ffea6beb4 100644
--- a/functest/ci/check_deployment.py
+++ b/functest/ci/check_deployment.py
@@ -19,7 +19,7 @@ import logging
import logging.config
import os
import pkg_resources
-from six.moves.urllib.parse import urlparse
+from six.moves import urllib
import socket
from functest.opnfv_tests.openstack.snaps import snaps_utils
@@ -36,19 +36,19 @@ LOGGER = logging.getLogger(__name__)
def verify_connectivity(endpoint):
- """ Returns true if an ip/port is reachable"""
+ """ Returns true if an hostname/port is reachable"""
connection = socket.socket()
connection.settimeout(10)
- ip = urlparse(endpoint).hostname
- port = urlparse(endpoint).port
+ hostname = urllib.parse.urlparse(endpoint).hostname
+ port = urllib.parse.urlparse(endpoint).port
if not port:
- port = 443 if urlparse(endpoint).scheme == "https" else 80
+ port = 443 if urllib.parse.urlparse(endpoint).scheme == "https" else 80
try:
- connection.connect((ip, port))
- LOGGER.debug('%s:%s is reachable!', ip, port)
+ connection.connect((hostname, port))
+ LOGGER.debug('%s:%s is reachable!', hostname, port)
return True
except socket.error:
- LOGGER.exception('%s:%s is not reachable.', ip, port)
+ LOGGER.exception('%s:%s is not reachable.', hostname, port)
return False
@@ -71,7 +71,7 @@ class CheckDeployment(object):
def check_auth_endpoint(self):
""" Verifies connectivity to the OS_AUTH_URL given in the RC file """
rc_endpoint = self.os_creds.auth_url
- if not (verify_connectivity(rc_endpoint)):
+ if not verify_connectivity(rc_endpoint):
raise Exception("OS_AUTH_URL {} is not reachable.".
format(rc_endpoint))
LOGGER.info("Connectivity to OS_AUTH_URL %s ...OK", rc_endpoint)
@@ -81,7 +81,7 @@ class CheckDeployment(object):
public_endpoint = keystone_utils.get_endpoint(self.os_creds,
'identity',
interface='public')
- if not (verify_connectivity(public_endpoint)):
+ if not verify_connectivity(public_endpoint):
raise Exception("Public endpoint {} is not reachable.".
format(public_endpoint))
LOGGER.info("Connectivity to the public endpoint %s ...OK",
@@ -92,7 +92,7 @@ class CheckDeployment(object):
endpoint = keystone_utils.get_endpoint(self.os_creds,
service,
interface='public')
- if not (verify_connectivity(endpoint)):
+ if not verify_connectivity(endpoint):
raise Exception("{} endpoint {} is not reachable.".
format(service, endpoint))
LOGGER.info("Connectivity to endpoint '%s' %s ...OK",
@@ -132,7 +132,7 @@ class CheckDeployment(object):
""" checks if external network exists """
ext_net = snaps_utils.get_ext_net_name(self.os_creds)
if ext_net:
- LOGGER.info("External network found: %s" % ext_net)
+ LOGGER.info("External network found: %s", ext_net)
else:
raise Exception("ERROR: No external networks in the deployment.")
diff --git a/functest/ci/run_tests.py b/functest/ci/run_tests.py
index b8da3be62..166dbfc9e 100644
--- a/functest/ci/run_tests.py
+++ b/functest/ci/run_tests.py
@@ -7,19 +7,24 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
+""" The entry of running tests:
+1) Parses functest/ci/testcases.yaml to check which testcase(s) to be run
+2) Execute the common operations on every testcase (run, push results to db...)
+3) Return the right status code
+"""
+
import argparse
import enum
import importlib
import logging
import logging.config
import os
-import pkg_resources
import re
import sys
import textwrap
+import pkg_resources
import prettytable
-import six
import functest.ci.tier_builder as tb
import functest.core.testcase as testcase
@@ -28,26 +33,32 @@ import functest.utils.openstack_utils as os_utils
from functest.utils.constants import CONST
# __name__ cannot be used here
-logger = logging.getLogger('functest.ci.run_tests')
+LOGGER = logging.getLogger('functest.ci.run_tests')
CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
'functest', 'ci/config_functest.yaml')
class Result(enum.Enum):
+ """The overall result in enumerated type"""
+ # pylint: disable=too-few-public-methods
EX_OK = os.EX_OK
EX_ERROR = -1
class BlockingTestFailed(Exception):
+ """Exception when the blocking test fails"""
pass
class TestNotEnabled(Exception):
+ """Exception when the test is not enabled"""
pass
class RunTestsParser(object):
+ """Parser to run tests"""
+ # pylint: disable=too-few-public-methods
def __init__(self):
self.parser = argparse.ArgumentParser()
@@ -63,11 +74,19 @@ class RunTestsParser(object):
"database (default=false).",
action="store_true")
- def parse_args(self, argv=[]):
+ def parse_args(self, argv=None):
+ """Parse arguments.
+
+ It can call sys.exit if arguments are incorrect.
+
+ Returns:
+ the arguments from cmdline
+ """
return vars(self.parser.parse_args(argv))
class Runner(object):
+ """Runner class"""
def __init__(self):
self.executed_test_cases = {}
@@ -81,42 +100,34 @@ class Runner(object):
@staticmethod
def source_rc_file():
+ """Set the environmental vars from openstack.creds"""
+
rc_file = CONST.__getattribute__('openstack_creds')
if not os.path.isfile(rc_file):
raise Exception("RC file %s does not exist..." % rc_file)
- logger.debug("Sourcing the OpenStack RC file...")
+ LOGGER.debug("Sourcing the OpenStack RC file...")
os_utils.source_credentials(rc_file)
- for key, value in six.iteritems(os.environ):
- if re.search("OS_", key):
- if key == 'OS_AUTH_URL':
- CONST.__setattr__('OS_AUTH_URL', value)
- elif key == 'OS_USERNAME':
- CONST.__setattr__('OS_USERNAME', value)
- elif key == 'OS_TENANT_NAME':
- CONST.__setattr__('OS_TENANT_NAME', value)
- elif key == 'OS_PASSWORD':
- CONST.__setattr__('OS_PASSWORD', value)
- elif key == "OS_PROJECT_DOMAIN_NAME":
- CONST.__setattr__('OS_PROJECT_DOMAIN_NAME', value)
@staticmethod
def get_run_dict(testname):
+ """Obtain the the 'run' block of the testcase from testcases.yaml"""
try:
- dict = ft_utils.get_dict_by_test(testname)
- if not dict:
- logger.error("Cannot get {}'s config options".format(testname))
- elif 'run' in dict:
- return dict['run']
+ dic_testcase = ft_utils.get_dict_by_test(testname)
+ if not dic_testcase:
+ LOGGER.error("Cannot get %s's config options", testname)
+ elif 'run' in dic_testcase:
+ return dic_testcase['run']
return None
- except Exception:
- logger.exception("Cannot get {}'s config options".format(testname))
+ except Exception: # pylint: disable=broad-except
+ LOGGER.exception("Cannot get %s's config options", testname)
return None
def run_test(self, test):
+ """Run one test case"""
if not test.is_enabled():
raise TestNotEnabled(
"The test case {} is not enabled".format(test.get_name()))
- logger.info("Running test case '%s'...", test.get_name())
+ LOGGER.info("Running test case '%s'...", test.get_name())
result = testcase.TestCase.EX_RUN_ERROR
run_dict = self.get_run_dict(test.get_name())
if run_dict:
@@ -140,33 +151,32 @@ class Runner(object):
result = test_case.is_successful()
else:
result = testcase.TestCase.EX_OK
- logger.info("Test result:\n\n%s\n", test_case)
+ LOGGER.info("Test result:\n\n%s\n", test_case)
if self.clean_flag:
test_case.clean()
except ImportError:
- logger.exception("Cannot import module {}".format(
- run_dict['module']))
+ LOGGER.exception("Cannot import module %s", run_dict['module'])
except AttributeError:
- logger.exception("Cannot get class {}".format(
- run_dict['class']))
+ LOGGER.exception("Cannot get class %s", run_dict['class'])
else:
raise Exception("Cannot import the class for the test case.")
return result
def run_tier(self, tier):
+ """Run one tier"""
tier_name = tier.get_name()
tests = tier.get_tests()
- if tests is None or len(tests) == 0:
- logger.info("There are no supported test cases in this tier "
+ if not tests:
+ LOGGER.info("There are no supported test cases in this tier "
"for the given scenario")
self.overall_result = Result.EX_ERROR
else:
- logger.info("Running tier '%s'" % tier_name)
+ LOGGER.info("Running tier '%s'", tier_name)
for test in tests:
self.run_test(test)
test_case = self.executed_test_cases[test.get_name()]
if test_case.is_successful() != testcase.TestCase.EX_OK:
- logger.error("The test case '%s' failed.", test.get_name())
+ LOGGER.error("The test case '%s' failed.", test.get_name())
if test.get_project() == "functest":
self.overall_result = Result.EX_ERROR
if test.is_blocking():
@@ -176,13 +186,14 @@ class Runner(object):
return self.overall_result
def run_all(self):
+ """Run all available testcases"""
tiers_to_run = []
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
field_names=['tiers', 'order', 'CI Loop', 'description',
'testcases'])
for tier in self._tiers.get_tiers():
- if (len(tier.get_tests()) != 0 and
+ if (tier.get_tests() and
re.search(CONST.__getattribute__('CI_LOOP'),
tier.get_ci_loop()) is not None):
tiers_to_run.append(tier)
@@ -191,11 +202,12 @@ class Runner(object):
textwrap.fill(tier.description, width=40),
textwrap.fill(' '.join([str(x.get_name(
)) for x in tier.get_tests()]), width=40)])
- logger.info("TESTS TO BE EXECUTED:\n\n%s\n", msg)
+ LOGGER.info("TESTS TO BE EXECUTED:\n\n%s\n", msg)
for tier in tiers_to_run:
self.run_tier(tier)
def main(self, **kwargs):
+ """Entry point of class Runner"""
if 'noclean' in kwargs:
self.clean_flag = not kwargs['noclean']
if 'report' in kwargs:
@@ -203,59 +215,59 @@ class Runner(object):
try:
if 'test' in kwargs:
self.source_rc_file()
- logger.debug("Test args: %s", kwargs['test'])
+ LOGGER.debug("Test args: %s", kwargs['test'])
if self._tiers.get_tier(kwargs['test']):
self.run_tier(self._tiers.get_tier(kwargs['test']))
elif self._tiers.get_test(kwargs['test']):
result = self.run_test(
self._tiers.get_test(kwargs['test']))
if result != testcase.TestCase.EX_OK:
- logger.error("The test case '%s' failed.",
+ LOGGER.error("The test case '%s' failed.",
kwargs['test'])
self.overall_result = Result.EX_ERROR
elif kwargs['test'] == "all":
self.run_all()
else:
- logger.error("Unknown test case or tier '%s', "
- "or not supported by "
- "the given scenario '%s'."
- % (kwargs['test'],
- CONST.__getattribute__('DEPLOY_SCENARIO')))
- logger.debug("Available tiers are:\n\n%s",
+ LOGGER.error("Unknown test case or tier '%s', or not "
+ "supported by the given scenario '%s'.",
+ kwargs['test'],
+ CONST.__getattribute__('DEPLOY_SCENARIO'))
+ LOGGER.debug("Available tiers are:\n\n%s",
self._tiers)
return Result.EX_ERROR
else:
self.run_all()
except BlockingTestFailed:
pass
- except Exception:
- logger.exception("Failures when running testcase(s)")
+ except Exception: # pylint: disable=broad-except
+ LOGGER.exception("Failures when running testcase(s)")
self.overall_result = Result.EX_ERROR
if not self._tiers.get_test(kwargs['test']):
self.summary(self._tiers.get_tier(kwargs['test']))
- logger.info("Execution exit value: %s" % self.overall_result)
+ LOGGER.info("Execution exit value: %s", self.overall_result)
return self.overall_result
def summary(self, tier=None):
+ """To generate functest report showing the overall results"""
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
field_names=['env var', 'value'])
for env_var in ['INSTALLER_TYPE', 'DEPLOY_SCENARIO', 'BUILD_TAG',
'CI_LOOP']:
msg.add_row([env_var, CONST.__getattribute__(env_var)])
- logger.info("Deployment description:\n\n%s\n", msg)
+ LOGGER.info("Deployment description:\n\n%s\n", msg)
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
field_names=['test case', 'project', 'tier',
'duration', 'result'])
tiers = [tier] if tier else self._tiers.get_tiers()
- for tier in tiers:
- for test in tier.get_tests():
+ for each_tier in tiers:
+ for test in each_tier.get_tests():
try:
test_case = self.executed_test_cases[test.get_name()]
except KeyError:
msg.add_row([test.get_name(), test.get_project(),
- tier.get_name(), "00:00", "SKIP"])
+ each_tier.get_name(), "00:00", "SKIP"])
else:
result = 'PASS' if(test_case.is_successful(
) == test_case.EX_OK) else 'FAIL'
@@ -263,13 +275,14 @@ class Runner(object):
[test_case.case_name, test_case.project_name,
self._tiers.get_tier_name(test_case.case_name),
test_case.get_duration(), result])
- for test in tier.get_skipped_test():
+ for test in each_tier.get_skipped_test():
msg.add_row([test.get_name(), test.get_project(),
- tier.get_name(), "00:00", "SKIP"])
- logger.info("FUNCTEST REPORT:\n\n%s\n", msg)
+ each_tier.get_name(), "00:00", "SKIP"])
+ LOGGER.info("FUNCTEST REPORT:\n\n%s\n", msg)
def main():
+ """Entry point"""
logging.config.fileConfig(pkg_resources.resource_filename(
'functest', 'ci/logging.ini'))
logging.captureWarnings(True)
diff --git a/functest/ci/tier_builder.py b/functest/ci/tier_builder.py
index c9969b612..9e92599d8 100644
--- a/functest/ci/tier_builder.py
+++ b/functest/ci/tier_builder.py
@@ -7,12 +7,15 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
+"""TierBuilder class to parse testcases config file"""
+
import yaml
import functest.ci.tier_handler as th
class TierBuilder(object):
+ # pylint: disable=missing-docstring
def __init__(self, ci_installer, ci_scenario, testcases_file):
self.ci_installer = ci_installer
@@ -24,8 +27,8 @@ class TierBuilder(object):
self.generate_tiers()
def read_test_yaml(self):
- with open(self.testcases_file) as f:
- self.testcases_yaml = yaml.safe_load(f)
+ with open(self.testcases_file) as tc_file:
+ self.testcases_yaml = yaml.safe_load(tc_file)
self.dic_tier_array = []
for tier in self.testcases_yaml.get("tiers"):
diff --git a/functest/ci/tier_handler.py b/functest/ci/tier_handler.py
index dd3e77ce3..9fc3f24d8 100644
--- a/functest/ci/tier_handler.py
+++ b/functest/ci/tier_handler.py
@@ -7,6 +7,8 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
+"""Tier and TestCase classes to wrap the testcases config file"""
+# pylint: disable=missing-docstring
import re
import textwrap
@@ -100,13 +102,9 @@ class Tier(object):
class TestCase(object):
- def __init__(self, name,
- enabled,
- dependency,
- criteria,
- blocking,
- description="",
- project=""):
+ def __init__(self, name, enabled, dependency, criteria, blocking,
+ description="", project=""):
+ # pylint: disable=too-many-arguments
self.name = name
self.enabled = enabled
self.dependency = dependency
@@ -117,7 +115,7 @@ class TestCase(object):
@staticmethod
def is_none(item):
- return item is None or item is ""
+ return item is None or item == ""
def is_compatible(self, ci_installer, ci_scenario):
try:
diff --git a/functest/core/testcase.py b/functest/core/testcase.py
index a7dc47c4e..ae0da9d7e 100644
--- a/functest/core/testcase.py
+++ b/functest/core/testcase.py
@@ -12,11 +12,10 @@
import logging
import os
+import functest.utils.functest_utils as ft_utils
+
import prettytable
-import functest.utils.functest_utils as ft_utils
-import functest.utils.openstack_clean as os_clean
-import functest.utils.openstack_snapshot as os_snapshot
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
@@ -196,33 +195,3 @@ class TestCase(object):
It can be overriden if resources must be deleted after
running the test case.
"""
-
-
-class OSGCTestCase(TestCase):
- """Model for single test case which requires an OpenStack Garbage
- Collector."""
-
- __logger = logging.getLogger(__name__)
-
- def create_snapshot(self):
- """Create a snapshot listing the OpenStack resources.
-
- Returns:
- TestCase.EX_OK if os_snapshot.main() returns 0.
- TestCase.EX_RUN_ERROR otherwise.
- """
- try:
- assert os_snapshot.main() == 0
- self.__logger.info("OpenStack resources snapshot created")
- return TestCase.EX_OK
- except Exception: # pylint: disable=broad-except
- self.__logger.exception("Cannot create the snapshot")
- return TestCase.EX_RUN_ERROR
-
- def clean(self):
- """Clean the OpenStack resources."""
- try:
- assert os_clean.main() == 0
- self.__logger.info("OpenStack resources cleaned")
- except Exception: # pylint: disable=broad-except
- self.__logger.exception("Cannot clean the OpenStack resources")
diff --git a/functest/opnfv_tests/vnf/epc/juju_epc.py b/functest/opnfv_tests/vnf/epc/juju_epc.py
index d9c14219e..be0f3ab21 100644
--- a/functest/opnfv_tests/vnf/epc/juju_epc.py
+++ b/functest/opnfv_tests/vnf/epc/juju_epc.py
@@ -15,14 +15,14 @@ import time
import json
import sys
from copy import deepcopy
-import yaml
-import functest.utils.openstack_utils as os_utils
-import functest.core.vnf as vnf
-import pkg_resources
+from urlparse import urljoin
+import functest.core.vnf as vnf
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.utils.constants import CONST
+import functest.utils.openstack_utils as os_utils
+import pkg_resources
from snaps.openstack.os_credentials import OSCreds
from snaps.openstack.create_network import (NetworkSettings,
SubnetSettings, OpenStackNetwork)
@@ -31,12 +31,14 @@ from snaps.openstack.create_flavor import (FlavorSettings, OpenStackFlavor)
from snaps.openstack.create_image import (ImageSettings, OpenStackImage)
from snaps.openstack.tests import openstack_tests
from snaps.openstack.utils import keystone_utils
+import yaml
__author__ = "Amarendra Meher <amarendra@rebaca.com>"
__author__ = "Soumaya K Nayek <soumaya.nayek@rebaca.com>"
class JujuEpc(vnf.VnfOnBoarding):
+ # pylint:disable=too-many-instance-attributes
"""Abot EPC deployed with JUJU Orchestrator Case"""
__logger = logging.getLogger(__name__)
@@ -94,6 +96,10 @@ class JujuEpc(vnf.VnfOnBoarding):
self.glance_client = os_utils.get_glance_client()
self.neutron_client = os_utils.get_neutron_client()
self.nova_client = os_utils.get_nova_client()
+ self.sec_group_id = None
+ self.public_auth_url = None
+ self.creds = None
+ self.filename = None
def prepare(self):
"""Prepare testcase (Additional pre-configuration steps)."""
@@ -104,6 +110,10 @@ class JujuEpc(vnf.VnfOnBoarding):
self.__logger.info("Additional pre-configuration steps")
self.public_auth_url = keystone_utils.get_endpoint(
self.snaps_creds, 'identity')
+ # 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.creds = {
"tenant": self.tenant_name,
@@ -127,7 +137,7 @@ class JujuEpc(vnf.VnfOnBoarding):
}
self.__logger.info("Cloud DATA: %s", cloud_data)
self.filename = os.path.join(self.case_dir, 'abot-epc.yaml')
- self.__logger.info("Cretae %s to add cloud info", self.filename)
+ self.__logger.info("Create %s to add cloud info", self.filename)
write_config(self.filename, CLOUD_TEMPLATE, **cloud_data)
if self.snaps_creds.identity_api_version == 3:
@@ -149,7 +159,7 @@ class JujuEpc(vnf.VnfOnBoarding):
image_creator.create()
self.created_object.append(image_creator)
- def deploy_orchestrator(self):
+ def deploy_orchestrator(self): # pylint: disable=too-many-locals
"""
Create network, subnet, router
@@ -288,8 +298,7 @@ class JujuEpc(vnf.VnfOnBoarding):
count = count + 1
os.system('juju-wait')
return True
- else:
- return False
+ return False
def test_vnf(self):
"""Run test on ABoT."""
@@ -455,8 +464,8 @@ def update_data(obj):
for tag in element['tags']:
element[tag['name']] = 1
- except:
- logging.error("Error in updating data, %s" % (sys.exc_info()[0]))
+ except Exception: # pylint: disable=broad-except
+ logging.error("Error in updating data, %s", sys.exc_info()[0])
raise
return obj
@@ -467,8 +476,8 @@ def get_instance_metadata(nova_client, instance):
try:
instance = nova_client.servers.get(instance.id)
return instance.metadata
- except Exception as e:
- logging.error("Error [get_instance_status(nova_client)]: %s" % e)
+ except Exception as exc: # pylint: disable=broad-except
+ logging.error("Error [get_instance_status(nova_client)]: %s", exc)
return None
diff --git a/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py b/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py
index 4593a1b8b..3db53e35e 100644
--- a/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py
+++ b/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py
@@ -17,6 +17,13 @@ import time
import pkg_resources
import yaml
+import functest.core.vnf as vnf
+import functest.utils.openstack_utils as os_utils
+from functest.opnfv_tests.openstack.snaps import snaps_utils
+from functest.utils.constants import CONST
+
+from org.openbaton.cli.errors.errors import NfvoException
+from org.openbaton.cli.agents.agents import MainAgent
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
@@ -24,7 +31,7 @@ from snaps.config.router import RouterConfig
from snaps.config.security_group import (
Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
from snaps.config.vm_inst import VmInstanceConfig
-
+from snaps.openstack.utils import keystone_utils
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_instance import OpenStackVmInstance
@@ -32,15 +39,6 @@ from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.create_router import OpenStackRouter
from snaps.openstack.create_security_group import OpenStackSecurityGroup
-from functest.opnfv_tests.openstack.snaps import snaps_utils
-
-import functest.core.vnf as vnf
-import functest.utils.openstack_utils as os_utils
-from functest.utils.constants import CONST
-
-from org.openbaton.cli.errors.errors import NfvoException
-from org.openbaton.cli.agents.agents import MainAgent
-
__author__ = "Pauls, Michael <michael.pauls@fokus.fraunhofer.de>"
# ----------------------------------------------------------
@@ -209,11 +207,14 @@ class ClearwaterImsVnf(vnf.VnfOnBoarding):
self.logger.info("Additional pre-configuration steps")
+ public_auth_url = keystone_utils.get_endpoint(
+ self.snaps_creds, 'identity')
+
self.creds = {
"tenant": self.tenant_name,
"username": self.tenant_name,
"password": self.tenant_name,
- "auth_url": os_utils.get_credentials()['auth_url']
+ "auth_url": public_auth_url
}
self.prepare_images()
self.prepare_flavor()
@@ -356,7 +357,6 @@ class ClearwaterImsVnf(vnf.VnfOnBoarding):
my_floating_ips = []
# Filter Floating IPs with tenant id
for floating_ip in floating_ips:
- # self.logger.info("Floating IP: %s", floating_ip)
if floating_ip.get('tenant_id') == tenant_id:
my_floating_ips.append(floating_ip.get('floating_ip_address'))
# Select if Floating IP exist else create new one
@@ -649,19 +649,19 @@ class ClearwaterImsVnf(vnf.VnfOnBoarding):
port_name='%s_port' % self.case_name)
snaps_utils.neutron_utils.delete_port(neutron_client, port)
time.sleep(10)
- except Exception as exc:
+ except Exception as exc: # pylint: disable=broad-except
self.logger.error('Unexpected error cleaning - %s', exc)
try:
self.logger.info("Deleting Open Baton Floating IP...")
snaps_utils.neutron_utils.delete_floating_ip(
neutron_client, self.mano['details']['fip'])
- except Exception as exc:
+ except Exception as exc: # pylint: disable=broad-except
self.logger.error('Unexpected error cleaning - %s', exc)
for resource in reversed(self.created_resources):
try:
self.logger.info("Cleaning %s", str(resource))
resource.clean()
- except Exception as exc:
+ except Exception as exc: # pylint: disable=broad-except
self.logger.error('Unexpected error cleaning - %s', exc)
super(ClearwaterImsVnf, self).clean()
diff --git a/functest/opnfv_tests/vnf/ims/orchestra_openims.py b/functest/opnfv_tests/vnf/ims/orchestra_openims.py
index fcb65e734..d609a0fba 100644
--- a/functest/opnfv_tests/vnf/ims/orchestra_openims.py
+++ b/functest/opnfv_tests/vnf/ims/orchestra_openims.py
@@ -17,6 +17,12 @@ import time
import pkg_resources
import yaml
+import functest.core.vnf as vnf
+import functest.utils.openstack_utils as os_utils
+from functest.utils.constants import CONST
+
+from org.openbaton.cli.errors.errors import NfvoException
+from org.openbaton.cli.agents.agents import MainAgent
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
@@ -24,7 +30,7 @@ from snaps.config.router import RouterConfig
from snaps.config.security_group import (
Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
from snaps.config.vm_inst import VmInstanceConfig
-
+from snaps.openstack.utils import keystone_utils
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_security_group import OpenStackSecurityGroup
@@ -34,13 +40,6 @@ from snaps.openstack.create_instance import OpenStackVmInstance
from functest.opnfv_tests.openstack.snaps import snaps_utils
-import functest.core.vnf as vnf
-import functest.utils.openstack_utils as os_utils
-from functest.utils.constants import CONST
-
-from org.openbaton.cli.errors.errors import NfvoException
-from org.openbaton.cli.agents.agents import MainAgent
-
__author__ = "Pauls, Michael <michael.pauls@fokus.fraunhofer.de>"
# ----------------------------------------------------------
@@ -202,12 +201,15 @@ class OpenImsVnf(vnf.VnfOnBoarding):
"""Prepare testscase (Additional pre-configuration steps)."""
super(OpenImsVnf, self).prepare()
+ public_auth_url = keystone_utils.get_endpoint(
+ self.snaps_creds, 'identity')
+
self.logger.info("Additional pre-configuration steps")
self.creds = {
"tenant": self.tenant_name,
"username": self.tenant_name,
"password": self.tenant_name,
- "auth_url": os_utils.get_credentials()['auth_url']
+ "auth_url": public_auth_url
}
self.prepare_images()
self.prepare_flavor()
@@ -659,8 +661,6 @@ class OpenImsVnf(vnf.VnfOnBoarding):
time.sleep(60)
else:
self.logger.info("No need to terminate the VNF...")
- # os_utils.delete_instance(nova_client=os_utils.get_nova_client(),
- # instance_id=self.mano_instance_id)
except (NfvoException, KeyError) as exc:
self.logger.error('Unexpected error cleaning - %s', exc)
@@ -672,13 +672,13 @@ class OpenImsVnf(vnf.VnfOnBoarding):
port_name='%s_port' % self.case_name)
snaps_utils.neutron_utils.delete_port(neutron_client, port)
time.sleep(10)
- except Exception as exc:
+ except Exception as exc: # pylint: disable=broad-except
self.logger.error('Unexpected error cleaning - %s', exc)
try:
self.logger.info("Deleting Open Baton Floating IP...")
snaps_utils.neutron_utils.delete_floating_ip(
neutron_client, self.mano['details']['fip'])
- except Exception as exc:
+ except Exception as exc: # pylint: disable=broad-except
self.logger.error('Unexpected error cleaning - %s', exc)
for resource in reversed(self.created_resources):
diff --git a/functest/tests/unit/ci/test_check_deployment.py b/functest/tests/unit/ci/test_check_deployment.py
index 66d1b7afd..fa2f21306 100644
--- a/functest/tests/unit/ci/test_check_deployment.py
+++ b/functest/tests/unit/ci/test_check_deployment.py
@@ -172,11 +172,14 @@ class CheckDeploymentTesting(unittest.TestCase):
@mock.patch('functest.ci.check_deployment.LOGGER.info')
@mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
- 'get_ext_net_name', return_value='ext-net')
+ 'get_ext_net_name')
def test_check_extnet(self, mock_getext, mock_loginfo):
+ test_network = 'ext-net'
+ mock_getext.return_value = test_network
self.deployment.check_ext_net()
self.assertTrue(mock_getext.called)
- mock_loginfo.assert_called_once_with("External network found: ext-net")
+ mock_loginfo.assert_called_once_with(
+ "External network found: %s", test_network)
@mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
'get_ext_net_name', return_value='')
diff --git a/functest/tests/unit/ci/test_run_tests.py b/functest/tests/unit/ci/test_run_tests.py
index bc95f8f3d..93cbfccdf 100644
--- a/functest/tests/unit/ci/test_run_tests.py
+++ b/functest/tests/unit/ci/test_run_tests.py
@@ -58,14 +58,14 @@ class RunTestsTesting(unittest.TestCase):
self.run_tests_parser = run_tests.RunTestsParser()
- @mock.patch('functest.ci.run_tests.logger.error')
+ @mock.patch('functest.ci.run_tests.LOGGER.error')
def test_source_rc_file_missing_file(self, mock_logger_error):
with mock.patch('functest.ci.run_tests.os.path.isfile',
return_value=False), \
self.assertRaises(Exception):
self.runner.source_rc_file()
- @mock.patch('functest.ci.run_tests.logger.debug')
+ @mock.patch('functest.ci.run_tests.LOGGER.debug')
@mock.patch('functest.ci.run_tests.os.path.isfile',
return_value=True)
def test_source_rc_file_default(self, *args):
@@ -81,7 +81,7 @@ class RunTestsTesting(unittest.TestCase):
self.assertEqual(self.runner.get_run_dict('test_name'),
mock_obj)
- @mock.patch('functest.ci.run_tests.logger.error')
+ @mock.patch('functest.ci.run_tests.LOGGER.error')
def test_get_run_dict_if_defined_missing_config_option(self,
mock_logger_error):
with mock.patch('functest.ci.run_tests.'
@@ -90,9 +90,8 @@ class RunTestsTesting(unittest.TestCase):
testname = 'test_name'
self.assertEqual(self.runner.get_run_dict(testname),
None)
- mock_logger_error.assert_called_once_with("Cannot get {}'s config "
- "options"
- .format(testname))
+ mock_logger_error.assert_called_once_with(
+ "Cannot get %s's config options", testname)
with mock.patch('functest.ci.run_tests.'
'ft_utils.get_dict_by_test',
@@ -101,7 +100,7 @@ class RunTestsTesting(unittest.TestCase):
self.assertEqual(self.runner.get_run_dict(testname),
None)
- @mock.patch('functest.ci.run_tests.logger.exception')
+ @mock.patch('functest.ci.run_tests.LOGGER.exception')
def test_get_run_dict_if_defined_exception(self,
mock_logger_except):
with mock.patch('functest.ci.run_tests.'
@@ -110,9 +109,8 @@ class RunTestsTesting(unittest.TestCase):
testname = 'test_name'
self.assertEqual(self.runner.get_run_dict(testname),
None)
- mock_logger_except.assert_called_once_with("Cannot get {}'s config"
- " options"
- .format(testname))
+ mock_logger_except.assert_called_once_with(
+ "Cannot get %s's config options", testname)
def test_run_tests_import_test_class_exception(self):
mock_test = mock.Mock()
@@ -153,14 +151,14 @@ class RunTestsTesting(unittest.TestCase):
run_tests.Result.EX_OK)
mock_methods[0].assert_called_with(mock.ANY)
- @mock.patch('functest.ci.run_tests.logger.info')
+ @mock.patch('functest.ci.run_tests.LOGGER.info')
def test_run_tier_missing_test(self, mock_logger_info):
self.tier.get_tests.return_value = None
self.assertEqual(self.runner.run_tier(self.tier),
run_tests.Result.EX_ERROR)
self.assertTrue(mock_logger_info.called)
- @mock.patch('functest.ci.run_tests.logger.info')
+ @mock.patch('functest.ci.run_tests.LOGGER.info')
@mock.patch('functest.ci.run_tests.Runner.run_tier')
@mock.patch('functest.ci.run_tests.Runner.summary')
def test_run_all_default(self, *mock_methods):
@@ -169,7 +167,7 @@ class RunTestsTesting(unittest.TestCase):
mock_methods[1].assert_not_called()
self.assertTrue(mock_methods[2].called)
- @mock.patch('functest.ci.run_tests.logger.info')
+ @mock.patch('functest.ci.run_tests.LOGGER.info')
@mock.patch('functest.ci.run_tests.Runner.summary')
def test_run_all_missing_tier(self, *mock_methods):
CONST.__setattr__('CI_LOOP', 'loop_re_not_available')
diff --git a/functest/tests/unit/ci/test_tier_builder.py b/functest/tests/unit/ci/test_tier_builder.py
index 1dec9aed6..d832ca3f0 100644
--- a/functest/tests/unit/ci/test_tier_builder.py
+++ b/functest/tests/unit/ci/test_tier_builder.py
@@ -5,6 +5,8 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
+# pylint: disable=missing-docstring
+
import logging
import unittest
diff --git a/functest/tests/unit/ci/test_tier_handler.py b/functest/tests/unit/ci/test_tier_handler.py
index 1909ac222..871220db3 100644
--- a/functest/tests/unit/ci/test_tier_handler.py
+++ b/functest/tests/unit/ci/test_tier_handler.py
@@ -5,6 +5,8 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
+# pylint: disable=missing-docstring
+
import logging
import unittest
@@ -89,11 +91,11 @@ class TierHandlerTesting(unittest.TestCase):
self.assertEqual(self.tier.get_ci_loop(),
'test_ci_loop')
- def test_testcase_is_none_present_item(self):
+ def test_testcase_is_none_in_item(self):
self.assertEqual(tier_handler.TestCase.is_none("item"),
False)
- def test_testcase_is_none_missing_item(self):
+ def test_testcase_is_none_no_item(self):
self.assertEqual(tier_handler.TestCase.is_none(None),
True)
@@ -102,7 +104,7 @@ class TierHandlerTesting(unittest.TestCase):
'test_scenario'),
True)
- def test_testcase_is_compatible_missing_installer_scenario(self):
+ def test_testcase_is_compatible_2(self):
self.assertEqual(self.testcase.is_compatible('missing_installer',
'test_scenario'),
False)
diff --git a/functest/tests/unit/core/test_testcase.py b/functest/tests/unit/core/test_testcase.py
index 0a6f0c9f4..6fbf62e22 100644
--- a/functest/tests/unit/core/test_testcase.py
+++ b/functest/tests/unit/core/test_testcase.py
@@ -12,15 +12,17 @@
import logging
import unittest
+from functest.core import testcase
+
import mock
-from functest.core import testcase
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
class TestCaseTesting(unittest.TestCase):
"""The class testing TestCase."""
+
# pylint: disable=missing-docstring,too-many-public-methods
_case_name = "base"
@@ -229,48 +231,6 @@ class TestCaseTesting(unittest.TestCase):
self.assertEqual(self.test.clean(), None)
-class OSGCTestCaseTesting(unittest.TestCase):
- """The class testing OSGCTestCase."""
- # pylint: disable=missing-docstring
-
- def setUp(self):
- self.test = testcase.OSGCTestCase()
-
- @mock.patch('functest.utils.openstack_snapshot.main',
- side_effect=Exception)
- def test_create_snapshot_exc(self, mock_method=None):
- self.assertEqual(self.test.create_snapshot(),
- testcase.TestCase.EX_RUN_ERROR)
- mock_method.assert_called_once_with()
-
- @mock.patch('functest.utils.openstack_snapshot.main', return_value=-1)
- def test_create_snapshot_ko(self, mock_method=None):
- self.assertEqual(self.test.create_snapshot(),
- testcase.TestCase.EX_RUN_ERROR)
- mock_method.assert_called_once_with()
-
- @mock.patch('functest.utils.openstack_snapshot.main', return_value=0)
- def test_create_snapshot_env(self, mock_method=None):
- self.assertEqual(self.test.create_snapshot(),
- testcase.TestCase.EX_OK)
- mock_method.assert_called_once_with()
-
- @mock.patch('functest.utils.openstack_clean.main', side_effect=Exception)
- def test_clean_exc(self, mock_method=None):
- self.assertEqual(self.test.clean(), None)
- mock_method.assert_called_once_with()
-
- @mock.patch('functest.utils.openstack_clean.main', return_value=-1)
- def test_clean_ko(self, mock_method=None):
- self.assertEqual(self.test.clean(), None)
- mock_method.assert_called_once_with()
-
- @mock.patch('functest.utils.openstack_clean.main', return_value=0)
- def test_clean(self, mock_method=None):
- self.assertEqual(self.test.clean(), None)
- mock_method.assert_called_once_with()
-
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
unittest.main(verbosity=2)
diff --git a/functest/utils/openstack_utils.py b/functest/utils/openstack_utils.py
index b479198d2..1d7cdafb9 100644
--- a/functest/utils/openstack_utils.py
+++ b/functest/utils/openstack_utils.py
@@ -127,6 +127,7 @@ def source_credentials(rc_file):
key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
os.environ[key] = value
+ CONST.__setattr__(key, value)
def get_session_auth(other_creds={}):
diff --git a/tox.ini b/tox.ini
index 8207dcb73..7d92b17c1 100644
--- a/tox.ini
+++ b/tox.ini
@@ -30,9 +30,12 @@ basepython = python2.7
whitelist_externals = bash
modules =
functest.api
+ functest.ci
functest.core
functest.energy
functest.opnfv_tests.sdn.odl
+ functest.tests.unit.ci.test_tier_handler
+ functest.tests.unit.ci.test_tier_builder
functest.tests.unit.core
functest.tests.unit.energy
functest.tests.unit.odl