diff options
35 files changed, 479 insertions, 385 deletions
diff --git a/functest/ci/check_deployment.py b/functest/ci/check_deployment.py index 81607dffe..8d19d8504 100644 --- a/functest/ci/check_deployment.py +++ b/functest/ci/check_deployment.py @@ -44,7 +44,7 @@ def verify_connectivity(endpoint): port = url.port if not port: port = 443 if url.scheme == "https" else 80 - connection.connect(url.hostname, port) + connection.connect((url.hostname, port)) LOGGER.debug('%s:%s is reachable!', url.hostname, port) return True except socket.error: diff --git a/functest/core/vnf.py b/functest/core/vnf.py index 856e62b5a..5339e42e8 100644 --- a/functest/core/vnf.py +++ b/functest/core/vnf.py @@ -198,8 +198,7 @@ class VnfOnBoarding(base.TestCase): * the user, * the tenant """ - self.__logger.info("test cleaning") - self.__logger.info('Remove the cloudify manager OS object ..') + self.__logger.info('Removing the VNF resources ..') for creator in reversed(self.created_object): try: creator.clean() diff --git a/functest/opnfv_tests/openstack/snaps/api_check.py b/functest/opnfv_tests/openstack/snaps/api_check.py index e708b4dec..e8b9c3227 100644 --- a/functest/opnfv_tests/openstack/snaps/api_check.py +++ b/functest/opnfv_tests/openstack/snaps/api_check.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # This program and the accompanying materials @@ -6,6 +8,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import unittest from functest.opnfv_tests.openstack.snaps import snaps_suite_builder @@ -38,4 +42,4 @@ class ApiCheck(SnapsTestRunner): ext_net_name=self.ext_net_name, use_keystone=self.use_keystone, image_metadata=self.image_metadata) - return super(self.__class__, self).run() + return super(ApiCheck, self).run() diff --git a/functest/opnfv_tests/openstack/snaps/connection_check.py b/functest/opnfv_tests/openstack/snaps/connection_check.py index 1fc49349e..f8bf8852e 100644 --- a/functest/opnfv_tests/openstack/snaps/connection_check.py +++ b/functest/opnfv_tests/openstack/snaps/connection_check.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # This program and the accompanying materials @@ -6,6 +8,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import unittest from functest.opnfv_tests.openstack.snaps import snaps_suite_builder @@ -37,4 +41,4 @@ class ConnectionCheck(SnapsTestRunner): os_creds=self.os_creds, ext_net_name=self.ext_net_name, use_keystone=self.use_keystone) - return super(self.__class__, self).run() + return super(ConnectionCheck, self).run() diff --git a/functest/opnfv_tests/openstack/snaps/health_check.py b/functest/opnfv_tests/openstack/snaps/health_check.py index 837c2eae7..db882c382 100644 --- a/functest/opnfv_tests/openstack/snaps/health_check.py +++ b/functest/opnfv_tests/openstack/snaps/health_check.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # This program and the accompanying materials @@ -6,6 +8,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import unittest from functest.opnfv_tests.openstack.snaps.snaps_test_runner import ( @@ -42,4 +46,4 @@ class HealthCheck(SnapsTestRunner): flavor_metadata=self.flavor_metadata, image_metadata=self.image_metadata, netconf_override=self.netconf_override)) - return super(self.__class__, self).run() + return super(HealthCheck, self).run() diff --git a/functest/opnfv_tests/openstack/snaps/smoke.py b/functest/opnfv_tests/openstack/snaps/smoke.py index ded149d0c..ef6e5dc9f 100644 --- a/functest/opnfv_tests/openstack/snaps/smoke.py +++ b/functest/opnfv_tests/openstack/snaps/smoke.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # This program and the accompanying materials @@ -6,6 +8,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import unittest from functest.opnfv_tests.openstack.snaps import snaps_suite_builder @@ -41,4 +45,4 @@ class SnapsSmoke(SnapsTestRunner): image_metadata=self.image_metadata, use_floating_ips=self.use_fip, netconf_override=self.netconf_override) - return super(self.__class__, self).run() + return super(SnapsSmoke, self).run() diff --git a/functest/opnfv_tests/openstack/snaps/snaps_suite_builder.py b/functest/opnfv_tests/openstack/snaps/snaps_suite_builder.py index 3e7c0a39f..c3d8c2d8b 100644 --- a/functest/opnfv_tests/openstack/snaps/snaps_suite_builder.py +++ b/functest/opnfv_tests/openstack/snaps/snaps_suite_builder.py @@ -8,6 +8,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import logging from snaps.openstack.tests.create_flavor_tests import ( @@ -119,6 +121,7 @@ def add_openstack_client_tests(suite, os_creds, ext_net_name, def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True, image_metadata=None, log_level=logging.INFO): + # pylint: disable=too-many-arguments """ Adds tests written to exercise all existing OpenStack APIs :param suite: the unittest.TestSuite object to which to add the tests @@ -232,6 +235,7 @@ def add_openstack_integration_tests(suite, os_creds, ext_net_name, image_metadata=None, use_floating_ips=True, netconf_override=None, log_level=logging.INFO): + # pylint: disable=too-many-arguments """ Adds tests written to exercise all long-running OpenStack integration tests meaning they will be creating VM instances and potentially performing some diff --git a/functest/opnfv_tests/openstack/snaps/snaps_test_runner.py b/functest/opnfv_tests/openstack/snaps/snaps_test_runner.py index 6dc8288bf..94c8af85d 100644 --- a/functest/opnfv_tests/openstack/snaps/snaps_test_runner.py +++ b/functest/opnfv_tests/openstack/snaps/snaps_test_runner.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # This program and the accompanying materials @@ -6,6 +8,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import logging from functest.core import unit @@ -17,13 +21,14 @@ from snaps.openstack.tests import openstack_tests class SnapsTestRunner(unit.Suite): + # pylint: disable=too-many-instance-attributes """ This test executes the SNAPS Python Tests """ + def __init__(self, **kwargs): super(SnapsTestRunner, self).__init__(**kwargs) self.logger = logging.getLogger(__name__) - if 'os_creds' in kwargs: self.os_creds = kwargs['os_creds'] else: diff --git a/functest/opnfv_tests/openstack/snaps/snaps_utils.py b/functest/opnfv_tests/openstack/snaps/snaps_utils.py index 284e88b51..1153b63df 100644 --- a/functest/opnfv_tests/openstack/snaps/snaps_utils.py +++ b/functest/opnfv_tests/openstack/snaps/snaps_utils.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2015 All rights reserved # This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 @@ -5,6 +7,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + from functest.utils.constants import CONST from snaps.openstack.utils import neutron_utils, nova_utils @@ -19,7 +23,7 @@ def get_ext_net_name(os_creds): """ neutron = neutron_utils.neutron_client(os_creds) ext_nets = neutron_utils.get_external_networks(neutron) - if (hasattr(CONST, 'EXTERNAL_NETWORK')): + if hasattr(CONST, 'EXTERNAL_NETWORK'): extnet_config = CONST.__getattribute__('EXTERNAL_NETWORK') for ext_net in ext_nets: if ext_net.name == extnet_config: diff --git a/functest/opnfv_tests/openstack/vping/vping_base.py b/functest/opnfv_tests/openstack/vping/vping_base.py index df9774ece..b99fdf6b7 100644 --- a/functest/opnfv_tests/openstack/vping/vping_base.py +++ b/functest/opnfv_tests/openstack/vping/vping_base.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # This program and the accompanying materials @@ -37,19 +39,15 @@ class VPingBase(testcase.TestCase): def __init__(self, **kwargs): super(VPingBase, self).__init__(**kwargs) - self.logger = logging.getLogger(__name__) - if 'os_creds' in kwargs: self.os_creds = kwargs['os_creds'] else: - creds_override = None - if hasattr(CONST, 'snaps_os_creds_override'): - creds_override = CONST.__getattribute__( - 'snaps_os_creds_override') - + creds_override = getattr( + CONST, 'snaps_os_creds_override') if hasattr( + CONST, 'snaps_os_creds_override') else None self.os_creds = openstack_tests.get_credentials( - os_env_file=CONST.__getattribute__('openstack_creds'), + os_env_file=getattr(CONST, 'openstack_creds'), overrides=creds_override) self.creators = list() @@ -62,27 +60,24 @@ class VPingBase(testcase.TestCase): # Shared metadata self.guid = '-' + str(uuid.uuid4()) - self.router_name = CONST.__getattribute__( - 'vping_router_name') + self.guid - self.vm1_name = CONST.__getattribute__('vping_vm_name_1') + self.guid - self.vm2_name = CONST.__getattribute__('vping_vm_name_2') + self.guid - - self.vm_boot_timeout = CONST.__getattribute__('vping_vm_boot_timeout') - self.vm_delete_timeout = CONST.__getattribute__( - 'vping_vm_delete_timeout') - self.vm_ssh_connect_timeout = CONST.__getattribute__( - 'vping_vm_ssh_connect_timeout') - self.ping_timeout = CONST.__getattribute__('vping_ping_timeout') + self.router_name = getattr(CONST, 'vping_router_name') + self.guid + self.vm1_name = getattr(CONST, 'vping_vm_name_1') + self.guid + self.vm2_name = getattr(CONST, 'vping_vm_name_2') + self.guid + + self.vm_boot_timeout = getattr(CONST, 'vping_vm_boot_timeout') + self.vm_delete_timeout = getattr(CONST, 'vping_vm_delete_timeout') + self.vm_ssh_connect_timeout = getattr( + CONST, 'vping_vm_ssh_connect_timeout') + self.ping_timeout = getattr(CONST, 'vping_ping_timeout') self.flavor_name = 'vping-flavor' + self.guid # Move this configuration option up for all tests to leverage if hasattr(CONST, 'snaps_images_cirros'): - self.cirros_image_config = CONST.__getattribute__( - 'snaps_images_cirros') + self.cirros_image_config = getattr(CONST, 'snaps_images_cirros') else: self.cirros_image_config = None - def run(self): + def run(self, **kwargs): # pylint: disable=too-many-locals """ Begins the test execution which should originate from the subclass """ @@ -95,7 +90,7 @@ class VPingBase(testcase.TestCase): '%Y-%m-%d %H:%M:%S')) image_base_name = '{}-{}'.format( - CONST.__getattribute__('vping_image_name'), + getattr(CONST, 'vping_image_name'), str(self.guid)) os_image_settings = openstack_tests.cirros_image_settings( image_base_name, image_metadata=self.cirros_image_config) @@ -105,26 +100,21 @@ class VPingBase(testcase.TestCase): self.os_creds, os_image_settings) self.creators.append(self.image_creator) - private_net_name = CONST.__getattribute__( - 'vping_private_net_name') + self.guid - private_subnet_name = CONST.__getattribute__( - 'vping_private_subnet_name') + self.guid - private_subnet_cidr = CONST.__getattribute__( - 'vping_private_subnet_cidr') + private_net_name = getattr(CONST, 'vping_private_net_name') + self.guid + private_subnet_name = getattr( + CONST, 'vping_private_subnet_name') + self.guid + private_subnet_cidr = getattr(CONST, 'vping_private_subnet_cidr') vping_network_type = None vping_physical_network = None vping_segmentation_id = None if hasattr(CONST, 'vping_network_type'): - vping_network_type = CONST.__getattribute__( - 'vping_network_type') + vping_network_type = getattr(CONST, 'vping_network_type') if hasattr(CONST, 'vping_physical_network'): - vping_physical_network = CONST.__getattribute__( - 'vping_physical_network') + vping_physical_network = getattr(CONST, 'vping_physical_network') if hasattr(CONST, 'vping_segmentation_id'): - vping_segmentation_id = CONST.__getattribute__( - 'vping_segmentation_id') + vping_segmentation_id = getattr(CONST, 'vping_segmentation_id') self.logger.info( "Creating network with name: '%s'", private_net_name) @@ -154,7 +144,7 @@ class VPingBase(testcase.TestCase): self.logger.info( "Creating flavor with name: '%s'", self.flavor_name) - scenario = CONST.__getattribute__('DEPLOY_SCENARIO') + scenario = getattr(CONST, 'DEPLOY_SCENARIO') flavor_metadata = None flavor_ram = 512 if 'ovs' in scenario or 'fdio' in scenario: @@ -197,7 +187,7 @@ class VPingBase(testcase.TestCase): Cleanup all OpenStack objects. Should be called on completion :return: """ - if CONST.__getattribute__('vping_cleanup_objects') == 'True': + if getattr(CONST, 'vping_cleanup_objects') == 'True': for creator in reversed(self.creators): try: creator.clean() diff --git a/functest/opnfv_tests/openstack/vping/vping_ssh.py b/functest/opnfv_tests/openstack/vping/vping_ssh.py index 7df767edc..57e177e5d 100644 --- a/functest/opnfv_tests/openstack/vping/vping_ssh.py +++ b/functest/opnfv_tests/openstack/vping/vping_ssh.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -# + # Copyright (c) 2015 All rights reserved # This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 @@ -7,12 +7,8 @@ # # http://www.apache.org/licenses/LICENSE-2.0 - """vPingSSH testcase.""" -# This 1st import is here simply for pep8 as the 'os' package import appears -# to be required for mock and the unit tests will fail without it -import os # noqa # pylint: disable=unused-import import time from scp import SCPClient @@ -53,7 +49,7 @@ class VPingSSH(vping_base.VPingBase): self.sg_desc = CONST.__getattribute__('vping_sg_desc') @energy.enable_recording - def run(self): + def run(self, **kwargs): """ Excecute VPingSSH testcase. diff --git a/functest/opnfv_tests/openstack/vping/vping_userdata.py b/functest/opnfv_tests/openstack/vping/vping_userdata.py index ceba0917a..76cdcf832 100644 --- a/functest/opnfv_tests/openstack/vping/vping_userdata.py +++ b/functest/opnfv_tests/openstack/vping/vping_userdata.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -# + # Copyright (c) 2015 All rights reserved # This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 @@ -29,7 +29,7 @@ class VPingUserdata(vping_base.VPingBase): kwargs["case_name"] = "vping_userdata" super(VPingUserdata, self).__init__(**kwargs) - def run(self): + def run(self, **kwargs): """ Sets up the OpenStack VM instance objects then executes the ping and validates. diff --git a/functest/opnfv_tests/vnf/epc/juju_epc.py b/functest/opnfv_tests/vnf/epc/juju_epc.py index 3146b935c..aec008532 100644 --- a/functest/opnfv_tests/vnf/epc/juju_epc.py +++ b/functest/opnfv_tests/vnf/epc/juju_epc.py @@ -145,8 +145,9 @@ class JujuEpc(vnf.VnfOnBoarding): write_config(self.filename, CLOUD_TEMPLATE, **cloud_data) if self.snaps_creds.identity_api_version == 3: - append_config(self.filename, '{}'.format( - os_utils.get_credentials()['project_domain_name']), + append_config( + self.filename, '{}'.format( + os_utils.get_credentials()['project_domain_name']), '{}'.format(os_utils.get_credentials()['user_domain_name'])) self.__logger.info("Upload some OS images if it doesn't exist") diff --git a/functest/opnfv_tests/vnf/router/cloudify_vrouter.py b/functest/opnfv_tests/vnf/router/cloudify_vrouter.py index f2cd63fb2..020085ba9 100644 --- a/functest/opnfv_tests/vnf/router/cloudify_vrouter.py +++ b/functest/opnfv_tests/vnf/router/cloudify_vrouter.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """vrouter testcase implementation.""" import logging @@ -16,12 +18,12 @@ import time from cloudify_rest_client import CloudifyClient from cloudify_rest_client.executions import Execution from scp import SCPClient -import yaml from functest.opnfv_tests.openstack.snaps import snaps_utils import functest.opnfv_tests.vnf.router.vrouter_base as vrouter_base from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf from functest.utils.constants import CONST +from functest.utils import functest_utils from git import Repo @@ -50,6 +52,7 @@ __author__ = "Shuya Nakama <shuya.nakama@okinawaopenlabs.org>" class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase): + # pylint: disable=too-many-instance-attributes """vrouter testcase deployed with Cloudify Orchestrator.""" __logger = logging.getLogger(__name__) @@ -76,35 +79,46 @@ class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase): config_file = os.path.join(self.case_dir, self.config) self.orchestrator = dict( - requirements=get_config("orchestrator.requirements", config_file), + requirements=functest_utils.get_parameter_from_yaml( + "orchestrator.requirements", config_file), ) self.details['orchestrator'] = dict( - name=get_config("orchestrator.name", config_file), - version=get_config("orchestrator.version", config_file), + name=functest_utils.get_parameter_from_yaml( + "orchestrator.name", config_file), + version=functest_utils.get_parameter_from_yaml( + "orchestrator.version", config_file), status='ERROR', result='' ) self.__logger.debug("Orchestrator configuration %s", self.orchestrator) self.__logger.debug("name = %s", self.name) self.vnf = dict( - descriptor=get_config("vnf.descriptor", config_file), - inputs=get_config("vnf.inputs", config_file), - requirements=get_config("vnf.requirements", config_file) + descriptor=functest_utils.get_parameter_from_yaml( + "vnf.descriptor", config_file), + inputs=functest_utils.get_parameter_from_yaml( + "vnf.inputs", config_file), + requirements=functest_utils.get_parameter_from_yaml( + "vnf.requirements", config_file) ) self.details['vnf'] = dict( descriptor_version=self.vnf['descriptor']['version'], - name=get_config("vnf.name", config_file), - version=get_config("vnf.version", config_file), + name=functest_utils.get_parameter_from_yaml( + "vnf.name", config_file), + version=functest_utils.get_parameter_from_yaml( + "vnf.version", config_file), ) self.__logger.debug("VNF configuration: %s", self.vnf) self.util = Utilvnf() self.details['test_vnf'] = dict( - name=get_config("vnf_test_suite.name", config_file), - version=get_config("vnf_test_suite.version", config_file) + name=functest_utils.get_parameter_from_yaml( + "vnf_test_suite.name", config_file), + version=functest_utils.get_parameter_from_yaml( + "vnf_test_suite.version", config_file) ) - self.images = get_config("tenant_images", config_file) + self.images = functest_utils.get_parameter_from_yaml( + "tenant_images", config_file) self.__logger.info("Images needed for vrouter: %s", self.images) def prepare(self): @@ -129,6 +143,7 @@ class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase): self.created_object.append(image_creator) def deploy_orchestrator(self): + # pylint: disable=too-many-locals,too-many-statements """ Deploy Cloudify Manager. network, security group, fip, VM creation @@ -408,7 +423,7 @@ class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase): try: cfy_client.executions.cancel(execution['id'], force=True) - except: # pylint: disable=broad-except + except Exception: # pylint: disable=broad-except self.__logger.warn("Can't cancel the current exec") execution = cfy_client.executions.start( @@ -419,51 +434,14 @@ class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase): wait_for_execution(cfy_client, execution, self.__logger) cfy_client.deployments.delete(self.vnf['descriptor'].get('name')) cfy_client.blueprints.delete(self.vnf['descriptor'].get('name')) - except: # pylint: disable=broad-except + except Exception: # pylint: disable=broad-except self.__logger.warn("Some issue during the undeployment ..") self.__logger.warn("Tenant clean continue ..") - - self.__logger.info('Remove the cloudify manager OS object ..') - for creator in reversed(self.created_object): - try: - creator.clean() - except Exception as exc: - self.logger.error('Unexpected error cleaning - %s', exc) - super(CloudifyVrouter, self).clean() - def run(self, **kwargs): - """Execute CloudifyVrouter test case.""" - return super(CloudifyVrouter, self).run(**kwargs) - def get_vnf_info_list(self, target_vnf_name): - return self.util.get_vnf_info_list(self.cfy_manager_ip, - self.deployment_name, - target_vnf_name) - - -# ---------------------------------------------------------- -# -# YAML UTILS -# -# ----------------------------------------------------------- -def get_config(parameter, file_path): - """ - Get config parameter. - Returns the value of a given parameter in file.yaml - parameter must be given in string format with dots - Example: general.openstack.image_name - """ - with open(file_path) as config_file: - file_yaml = yaml.safe_load(config_file) - config_file.close() - value = file_yaml - for element in parameter.split("."): - value = value.get(element) - if value is None: - raise ValueError("The parameter %s is not defined in" - " reporting.yaml" % parameter) - return value + return self.util.get_vnf_info_list( + self.cfy_manager_ip, self.deployment_name, target_vnf_name) def wait_for_execution(client, execution, logger, timeout=7200, ): diff --git a/functest/opnfv_tests/vnf/router/test_controller/function_test_exec.py b/functest/opnfv_tests/vnf/router/test_controller/function_test_exec.py index d023d4796..be7bee889 100644 --- a/functest/opnfv_tests/vnf/router/test_controller/function_test_exec.py +++ b/functest/opnfv_tests/vnf/router/test_controller/function_test_exec.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """vrouter function test execution module""" import logging diff --git a/functest/opnfv_tests/vnf/router/utilvnf.py b/functest/opnfv_tests/vnf/router/utilvnf.py index 9d196836e..421cfe830 100644 --- a/functest/opnfv_tests/vnf/router/utilvnf.py +++ b/functest/opnfv_tests/vnf/router/utilvnf.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """ Utility module of vrouter testcase """ import json @@ -46,7 +48,7 @@ NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1 -class Utilvnf(object): +class Utilvnf(object): # pylint: disable=too-many-instance-attributes """ Utility class of vrouter testcase """ logger = logging.getLogger(__name__) @@ -107,7 +109,7 @@ class Utilvnf(object): self.test_result_json_file = "test_result.json" if os.path.isfile(self.test_result_json_file): os.remove(self.test_result_json_file) - self.logger.debug("removed %s" % self.test_result_json_file) + self.logger.debug("removed %s", self.test_result_json_file) def get_nova_client(self): nova_client = nova_utils.nova_client(self.snaps_creds) @@ -127,7 +129,7 @@ class Utilvnf(object): break address = server.addresses[ - network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"] + network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"] return address @@ -141,8 +143,7 @@ class Utilvnf(object): break mac_address = server.addresses[network_name][ - NOVA_CILENT_NETWORK_INFO_INDEX][ - "OS-EXT-IPS-MAC:mac_addr"] + NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"] return mac_address @@ -226,10 +227,7 @@ class Utilvnf(object): vnf["user"] = self.image["user"] vnf["pass"] = self.image["pass"] - if vnf_name == target_vnf_name: - vnf["target_vnf_flag"] = True - else: - vnf["target_vnf_flag"] = False + vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name) self.logger.debug("vnf name : " + vnf_name) self.logger.debug(vnf_name + " floating ip address : " + @@ -251,14 +249,16 @@ class Utilvnf(object): return vnf_info_list - def get_target_vnf(self, vnf_info_list): + @staticmethod + def get_target_vnf(vnf_info_list): for vnf in vnf_info_list: if vnf["target_vnf_flag"]: return vnf return None - def get_reference_vnf_list(self, vnf_info_list): + @staticmethod + def get_reference_vnf_list(vnf_info_list): reference_vnf_list = [] for vnf in vnf_info_list: if not vnf["target_vnf_flag"]: @@ -266,14 +266,16 @@ class Utilvnf(object): return reference_vnf_list - def get_vnf_info(self, vnf_info_list, vnf_name): + @staticmethod + def get_vnf_info(vnf_info_list, vnf_name): for vnf in vnf_info_list: if vnf["vnf_name"] == vnf_name: return vnf return None - def convert_functional_test_result(self, result_data_list): + @staticmethod + def convert_functional_test_result(result_data_list): result = {} for result_data in result_data_list: test_kind = result_data["test_kind"] @@ -311,11 +313,12 @@ class Utilvnf(object): output_json_data = json.dumps(test_result, sort_keys=True, indent=4) - self.logger.debug("test_result %s" % output_json_data) + self.logger.debug("test_result %s", output_json_data) else: - self.logger.debug("Not found %s" % self.test_result_json_file) + self.logger.debug("Not found %s", self.test_result_json_file) - def get_test_scenario(self, file_path): + @staticmethod + def get_test_scenario(file_path): test_scenario_file = open(file_path, 'r') test_scenario_yaml = yaml.safe_load(test_scenario_file) diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/checker.py b/functest/opnfv_tests/vnf/router/vnf_controller/checker.py index 198a5ffc9..a7a70f6d7 100644 --- a/functest/opnfv_tests/vnf/router/vnf_controller/checker.py +++ b/functest/opnfv_tests/vnf/router/vnf_controller/checker.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """vrouter test result check module""" import json @@ -24,7 +26,8 @@ class Checker(object): def __init__(self): self.logger.debug("init checker") - def load_check_rule(self, rule_file_dir, rule_file_name, parameter): + @staticmethod + def load_check_rule(rule_file_dir, rule_file_name, parameter): loader = FileSystemLoader(rule_file_dir, encoding='utf8') env = Environment(loader=loader) @@ -33,7 +36,8 @@ class Checker(object): check_rule_data = json.loads(check_rule) return check_rule_data - def regexp_information(self, response, rules): + @staticmethod + def regexp_information(response, rules): status = False result_data = {} diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py b/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py index 98cb14cc0..7d9116bcc 100644 --- a/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py +++ b/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """command generator module for vrouter testing""" import logging @@ -21,12 +23,16 @@ class CommandGenerator(object): def __init__(self): self.logger.debug("init command generator") - def load_template(self, template_dir, template): + @staticmethod + def load_template(template_dir, template): + # pylint disable=missing-docstring loader = FileSystemLoader(template_dir, encoding='utf8') env = Environment(loader=loader) return env.get_template(template) - def command_create(self, template, parameter): + @staticmethod + def command_create(template, parameter): + # pylint disable=missing-docstring commands = template.render(parameter) return commands.split('\n') diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py b/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py index c85a57351..628afd30e 100644 --- a/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py +++ b/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py @@ -10,10 +10,11 @@ """ssh client module for vrouter testing""" import logging -import paramiko import time import yaml +import paramiko + from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf RECEIVE_ROOP_WAIT = 1 @@ -23,7 +24,7 @@ DEFAULT_CONNECT_RETRY_COUNT = 10 DEFAULT_SEND_TIMEOUT = 10 -class SshClient(object): +class SshClient(object): # pylint: disable=too-many-instance-attributes """ssh client class for vrouter testing""" logger = logging.getLogger(__name__) @@ -51,6 +52,7 @@ class SshClient(object): def connect(self, time_out=DEFAULT_CONNECT_TIMEOUT, retrycount=DEFAULT_CONNECT_RETRY_COUNT): + # pylint: disable=missing-docstring while retrycount > 0: try: self.logger.info("SSH connect to %s.", self.ip_address) @@ -72,7 +74,7 @@ class SshClient(object): self.shell.recv(self.ssh_revieve_buff) break - except: # pylint: disable=broad-except + except Exception: # pylint: disable=broad-except self.logger.info("SSH timeout for %s...", self.ip_address) time.sleep(time_out) retrycount -= 1 @@ -88,13 +90,14 @@ class SshClient(object): return self.connected def send(self, cmd, prompt, timeout=DEFAULT_SEND_TIMEOUT): + # pylint: disable=missing-docstring if self.connected is True: self.shell.settimeout(timeout) self.logger.debug("Commandset : '%s'", cmd) try: self.shell.send(cmd + '\n') - except: # pylint: disable=broad-except + except Exception: # pylint: disable=broad-except self.logger.error("ssh send timeout : Command : '%s'", cmd) return None @@ -103,7 +106,7 @@ class SshClient(object): time.sleep(RECEIVE_ROOP_WAIT) try: res = self.shell.recv(self.ssh_revieve_buff) - except: # pylint: disable=broad-except + except Exception: # pylint: disable=broad-except self.logger.error("ssh receive timeout : Command : '%s'", cmd) break @@ -112,18 +115,19 @@ class SshClient(object): self.logger.debug("Response : '%s'", res_buff) return res_buff - else: - self.logger.error("Cannot connected to IP '%s'.", self.ip_address) - return None + self.logger.error("Cannot connected to IP '%s'.", self.ip_address) + return None def close(self): + # pylint: disable=missing-docstring if self.connected is True: self.ssh.close() - def error_check(response, err_strs=["error", - "warn", - "unknown command", - "already exist"]): + @staticmethod + def error_check(response, err_strs=None): + # pylint: disable=missing-docstring + if err_strs is None: + err_strs = ["error", "warn", "unknown command", "already exist"] for err in err_strs: if err in response: return False diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py b/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py index d1c2e3242..10e486455 100644 --- a/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py +++ b/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """vm controll module""" import logging diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py b/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py index 814e9e333..a5b1ad856 100644 --- a/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py +++ b/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py @@ -7,14 +7,17 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """vrouter controll module""" import logging import os -import prettytable import time import yaml +import prettytable + from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf from functest.opnfv_tests.vnf.router.vnf_controller.checker import Checker from functest.opnfv_tests.vnf.router.vnf_controller.ssh_client import ( @@ -45,6 +48,7 @@ class VnfController(object): def config_vnf(self, source_vnf, destination_vnf, test_cmd_file_path, parameter_file_path, prompt_file_path): + # pylint: disable=too-many-arguments parameter_file = open(parameter_file_path, 'r') cmd_input_param = yaml.safe_load(parameter_file) @@ -63,6 +67,7 @@ class VnfController(object): def result_check(self, target_vnf, reference_vnf, check_rule_file_path_list, parameter_file_path, prompt_file_path): + # pylint: disable=too-many-arguments,too-many-locals res_dict_data_list = [] @@ -93,8 +98,8 @@ class VnfController(object): checker = Checker() res_table = prettytable.PrettyTable( - header_style='upper', padding_width=5, - field_names=['test item', 'result']) + header_style='upper', padding_width=5, + field_names=['test item', 'result']) status = True res_data_list = [] diff --git a/functest/opnfv_tests/vnf/router/vrouter_base.py b/functest/opnfv_tests/vnf/router/vrouter_base.py index a534f1f2f..0678313eb 100644 --- a/functest/opnfv_tests/vnf/router/vrouter_base.py +++ b/functest/opnfv_tests/vnf/router/vrouter_base.py @@ -7,15 +7,18 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """vrouter testing base class module""" import datetime import json import logging import os -import pkg_resources import time +import pkg_resources + import functest.core.vnf as vnf from functest.utils.constants import CONST from functest.opnfv_tests.vnf.router.test_controller import function_test_exec @@ -66,8 +69,7 @@ class VrouterOnBoardingBase(vnf.VnfOnBoarding): test_info["test_kind"] + " test.") (result, result_data) = self.function_test_vrouter( - target_vnf_name, - test_info) + target_vnf_name, test_info) test_result_data_list.append(result_data) if not result: break @@ -115,5 +117,6 @@ class VrouterOnBoardingBase(vnf.VnfOnBoarding): return result, test_result_data def get_vnf_info_list(self, target_vnf_name): + # pylint: disable=unused-argument,no-self-use vnf_info_list = [] return vnf_info_list diff --git a/functest/tests/unit/ci/test_check_deployment.py b/functest/tests/unit/ci/test_check_deployment.py index 46dcc24c5..aeeca5871 100644 --- a/functest/tests/unit/ci/test_check_deployment.py +++ b/functest/tests/unit/ci/test_check_deployment.py @@ -45,25 +45,25 @@ class CheckDeploymentTesting(unittest.TestCase): @mock.patch('socket.socket.connect', side_effect=TypeError) def test_verify_connectivity_ko1(self, *args): self.assertFalse(check_deployment.verify_connectivity("127.0.0.1")) - args[0].assert_called_once_with(None, 80) + args[0].assert_called_once_with((None, 80)) @mock.patch('socket.socket.connect', side_effect=socket.error) def test_verify_connectivity_ko2(self, *args): self.assertFalse( check_deployment.verify_connectivity("http://127.0.0.1")) - args[0].assert_called_once_with("127.0.0.1", 80) + args[0].assert_called_once_with(("127.0.0.1", 80)) @mock.patch('socket.socket.connect', side_effect=socket.error) def test_verify_connectivity_ko3(self, *args): self.assertFalse( check_deployment.verify_connectivity("https://127.0.0.1")) - args[0].assert_called_once_with("127.0.0.1", 443) + args[0].assert_called_once_with(("127.0.0.1", 443)) @mock.patch('socket.socket.connect') def test_verify_connectivity(self, *args): self.assertTrue( check_deployment.verify_connectivity("https://127.0.0.1")) - args[0].assert_called_once_with("127.0.0.1", 443) + args[0].assert_called_once_with(("127.0.0.1", 443)) @mock.patch('snaps.openstack.utils.keystone_utils.keystone_session', return_value=mock.Mock( diff --git a/functest/tests/unit/openstack/snaps/test_snaps.py b/functest/tests/unit/openstack/snaps/test_snaps.py index 9a360cbc5..3d9e9df07 100644 --- a/functest/tests/unit/openstack/snaps/test_snaps.py +++ b/functest/tests/unit/openstack/snaps/test_snaps.py @@ -8,15 +8,15 @@ # pylint: disable=missing-docstring -import mock -import os +import logging import unittest +import mock from snaps.openstack.os_credentials import OSCreds from functest.core.testcase import TestCase -from functest.opnfv_tests.openstack.snaps import (connection_check, api_check, - health_check, smoke) +from functest.opnfv_tests.openstack.snaps import ( + connection_check, api_check, health_check, smoke) class ConnectionCheckTesting(unittest.TestCase): @@ -35,40 +35,52 @@ class ConnectionCheckTesting(unittest.TestCase): @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_client_tests') - def test_run_success(self, add_os_client_tests): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = [] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.connection_check.run()) - self.assertEquals(TestCase.EX_OK, - self.connection_check.is_successful()) + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_success(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = [] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.connection_check.run()) + self.assertEquals( + TestCase.EX_OK, self.connection_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', os_creds=self.os_creds, suite=mock.ANY, + use_keystone=True) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_client_tests') - def test_run_1_of_100_failures(self, add_os_client_tests): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.connection_check.run()) - self.assertEquals(TestCase.EX_TESTCASE_FAILED, - self.connection_check.is_successful()) + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.connection_check.run()) + self.assertEquals( + TestCase.EX_TESTCASE_FAILED, self.connection_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', os_creds=self.os_creds, suite=mock.ANY, + use_keystone=True) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_client_tests') - def test_run_1_of_100_failures_within_criteria(self, add_os_client_tests): + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko_criteria(self, *args): self.connection_check.criteria = 90 - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.connection_check.run()) - self.assertEquals(TestCase.EX_OK, - self.connection_check.is_successful()) + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.connection_check.run()) + self.assertEquals( + TestCase.EX_OK, self.connection_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', os_creds=self.os_creds, suite=mock.ANY, + use_keystone=True) class APICheckTesting(unittest.TestCase): @@ -87,40 +99,52 @@ class APICheckTesting(unittest.TestCase): @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_api_tests') - def test_run_success(self, add_tests): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = [] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.api_check.run()) - self.assertEquals(TestCase.EX_OK, - self.api_check.is_successful()) + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_success(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = [] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.api_check.run()) + self.assertEquals( + TestCase.EX_OK, self.api_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', image_metadata=mock.ANY, + os_creds=self.os_creds, suite=mock.ANY, use_keystone=True) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_api_tests') - def test_run_1_of_100_failures(self, add_tests): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.api_check.run()) - self.assertEquals(TestCase.EX_TESTCASE_FAILED, - self.api_check.is_successful()) + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.api_check.run()) + self.assertEquals( + TestCase.EX_TESTCASE_FAILED, self.api_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', image_metadata=mock.ANY, + os_creds=self.os_creds, suite=mock.ANY, use_keystone=True) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_api_tests') - def test_run_1_of_100_failures_within_criteria(self, add_tests): + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko_criteria(self, *args): self.api_check.criteria = 90 - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.api_check.run()) - self.assertEquals(TestCase.EX_OK, - self.api_check.is_successful()) + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.api_check.run()) + self.assertEquals( + TestCase.EX_OK, self.api_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', image_metadata=mock.ANY, + os_creds=self.os_creds, suite=mock.ANY, use_keystone=True) class HealthCheckTesting(unittest.TestCase): @@ -137,42 +161,57 @@ class HealthCheckTesting(unittest.TestCase): self.health_check = health_check.HealthCheck( os_creds=self.os_creds, ext_net_name='foo') - @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' - 'add_openstack_client_tests') - def test_run_success(self, add_tests): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = [] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.health_check.run()) - self.assertEquals(TestCase.EX_OK, - self.health_check.is_successful()) - - @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' - 'add_openstack_client_tests') - def test_run_1_of_100_failures(self, add_tests): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.health_check.run()) - self.assertEquals(TestCase.EX_TESTCASE_FAILED, - self.health_check.is_successful()) - - @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' - 'add_openstack_client_tests') - def test_run_1_of_100_failures_within_criteria(self, add_tests): + @mock.patch('snaps.openstack.tests.os_source_file_test.' + 'OSIntegrationTestCase.parameterize') + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_success(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = [] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.health_check.run()) + self.assertEquals( + TestCase.EX_OK, self.health_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + mock.ANY, ext_net_name='foo', flavor_metadata=None, + image_metadata=mock.ANY, netconf_override=None, + os_creds=self.os_creds, use_keystone=True) + + @mock.patch('snaps.openstack.tests.os_source_file_test.' + 'OSIntegrationTestCase.parameterize') + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.health_check.run()) + self.assertEquals( + TestCase.EX_TESTCASE_FAILED, self.health_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + mock.ANY, ext_net_name='foo', flavor_metadata=None, + image_metadata=mock.ANY, netconf_override=None, + os_creds=self.os_creds, use_keystone=True) + + @mock.patch('snaps.openstack.tests.os_source_file_test.' + 'OSIntegrationTestCase.parameterize') + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko_criteria(self, *args): self.health_check.criteria = 90 - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.health_check.run()) - self.assertEquals(TestCase.EX_OK, - self.health_check.is_successful()) + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.health_check.run()) + self.assertEquals( + TestCase.EX_OK, self.health_check.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + mock.ANY, ext_net_name='foo', flavor_metadata=None, + image_metadata=mock.ANY, netconf_override=None, + os_creds=self.os_creds, use_keystone=True) class SmokeTesting(unittest.TestCase): @@ -191,40 +230,56 @@ class SmokeTesting(unittest.TestCase): @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_integration_tests') - @mock.patch('os.path.join', return_value=os.getcwd()) - def test_run_success(self, add_tests, cwd): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = [] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.smoke.run()) - self.assertEquals(TestCase.EX_OK, - self.smoke.is_successful()) + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_success(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = [] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.smoke.run()) + self.assertEquals(TestCase.EX_OK, self.smoke.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', flavor_metadata=None, image_metadata=mock.ANY, + netconf_override=None, os_creds=self.os_creds, + suite=mock.ANY, use_floating_ips=True, use_keystone=True) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_integration_tests') - @mock.patch('os.path.join', return_value=os.getcwd()) - def test_run_1_of_100_failures(self, add_tests, cwd): - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.smoke.run()) - self.assertEquals(TestCase.EX_TESTCASE_FAILED, - self.smoke.is_successful()) + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko(self, *args): + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.smoke.run()) + self.assertEquals( + TestCase.EX_TESTCASE_FAILED, self.smoke.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', flavor_metadata=None, image_metadata=mock.ANY, + netconf_override=mock.ANY, os_creds=self.os_creds, + suite=mock.ANY, use_floating_ips=True, use_keystone=True) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_suite_builder.' 'add_openstack_integration_tests') - @mock.patch('os.path.join', return_value=os.getcwd()) - def test_run_1_of_100_failures_within_criteria(self, add_tests, cwd): + @mock.patch('unittest.TextTestRunner.run', + return_value=mock.MagicMock(name='unittest.TextTestResult')) + def test_run_1_of_100_ko_criteria(self, *args): self.smoke.criteria = 90 - result = mock.MagicMock(name='unittest.TextTestResult') - result.testsRun = 100 - result.failures = ['foo'] - result.errors = [] - with mock.patch('unittest.TextTestRunner.run', return_value=result): - self.assertEquals(TestCase.EX_OK, self.smoke.run()) - self.assertEquals(TestCase.EX_OK, - self.smoke.is_successful()) + args[0].return_value.testsRun = 100 + args[0].return_value.failures = ['foo'] + args[0].return_value.errors = [] + self.assertEquals(TestCase.EX_OK, self.smoke.run()) + self.assertEquals( + TestCase.EX_OK, self.smoke.is_successful()) + args[0].assert_called_with(mock.ANY) + args[1].assert_called_with( + ext_net_name='foo', flavor_metadata=None, image_metadata=mock.ANY, + netconf_override=None, os_creds=self.os_creds, + suite=mock.ANY, use_floating_ips=True, use_keystone=True) + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/functest/tests/unit/openstack/vping/__init__.py b/functest/tests/unit/openstack/vping/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/functest/tests/unit/openstack/vping/__init__.py diff --git a/functest/tests/unit/openstack/vping/test_vping.py b/functest/tests/unit/openstack/vping/test_vping.py index d494a795d..42650deac 100644 --- a/functest/tests/unit/openstack/vping/test_vping.py +++ b/functest/tests/unit/openstack/vping/test_vping.py @@ -1,3 +1,5 @@ +#!/usr/bin/env python + # Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # # All rights reserved. This program and the accompanying materials @@ -8,6 +10,7 @@ # pylint: disable=missing-docstring +import logging import unittest import mock @@ -46,21 +49,25 @@ class VPingUserdataTesting(unittest.TestCase): os_creds=self.os_creds) @mock.patch('snaps.openstack.utils.deploy_utils.create_vm_instance') - @mock.patch('functest.opnfv_tests.openstack.vping.vping_base.os.' - 'path.exists', return_value=True) + @mock.patch('os.path.exists', return_value=True) @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create', return_value=None) @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' 'get_port_ip', return_value='10.0.0.1') @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' 'vm_active', return_value=True) - def test_vping_userdata(self, deploy_vm, path_exists, create_flavor, - get_port_ip, vm_active): + @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.' + 'get_ext_net_name', return_value='foo') + def test_vping_userdata(self, *args): + # pylint: disable=unused-argument with mock.patch('snaps.openstack.utils.deploy_utils.create_image', return_value=OpenStackImage(self.os_creds, None)), \ mock.patch('snaps.openstack.utils.deploy_utils.create_network', return_value=OpenStackNetwork( self.os_creds, NetworkConfig(name='foo'))), \ + mock.patch('snaps.openstack.utils.deploy_utils.create_router', + return_value=OpenStackRouter( + self.os_creds, RouterConfig(name='foo'))), \ mock.patch('snaps.openstack.utils.deploy_utils.' 'create_vm_instance', return_value=OpenStackVmInstance( @@ -91,8 +98,7 @@ class VPingSSHTesting(unittest.TestCase): os_creds=self.os_creds) @mock.patch('snaps.openstack.utils.deploy_utils.create_vm_instance') - @mock.patch('functest.opnfv_tests.openstack.vping.vping_base.os.' - 'path.exists', return_value=True) + @mock.patch('os.path.exists', return_value=True) @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create', return_value=None) @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' @@ -110,10 +116,8 @@ class VPingSSHTesting(unittest.TestCase): 'VPingSSH._do_vping_ssh', return_value=TestCase.EX_OK) @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.' 'get_ext_net_name', return_value='foo') - def test_vping_ssh(self, create_vm, path_exists, - flavor_create, get_port_ip, vm_active, ssh_active, - ssh_client, scp_client, trans_script, do_vping_ssh, - ext_net_name): + def test_vping_ssh(self, *args): + # pylint: disable=unused-argument os_vm_inst = mock.MagicMock(name='get_console_output') os_vm_inst.get_console_output.return_value = 'vPing OK' ssh_client = mock.MagicMock(name='get_transport') @@ -159,3 +163,8 @@ class VPingSSHTesting(unittest.TestCase): 'OpenStackVmInstance.' 'ssh_client', return_value=ssh_client): self.assertEquals(TestCase.EX_OK, self.vping_ssh.run()) + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/functest/tests/unit/utils/test_functest_utils.py b/functest/tests/unit/utils/test_functest_utils.py index 4438c6a20..218d03c42 100644 --- a/functest/tests/unit/utils/test_functest_utils.py +++ b/functest/tests/unit/utils/test_functest_utils.py @@ -10,12 +10,12 @@ # pylint: disable=missing-docstring import logging -import pkg_resources import os import time import unittest import mock +import pkg_resources from six.moves import urllib from functest.utils import functest_utils @@ -147,8 +147,8 @@ class FunctestUtilsTesting(unittest.TestCase): def cmd_readline(self): return 'test_value\n' - @mock.patch('functest.utils.functest_utils.logger.error') - @mock.patch('functest.utils.functest_utils.logger.info') + @mock.patch('functest.utils.functest_utils.LOGGER.error') + @mock.patch('functest.utils.functest_utils.LOGGER.info') def test_execute_command_args_present_with_error(self, mock_logger_info, mock_logger_error): with mock.patch('functest.utils.functest_utils.subprocess.Popen') \ @@ -178,7 +178,7 @@ class FunctestUtilsTesting(unittest.TestCase): mopen.assert_called_once_with(self.output_file, "w") mock_logger_error.assert_called_once_with(self.error_msg) - @mock.patch('functest.utils.functest_utils.logger.info') + @mock.patch('functest.utils.functest_utils.LOGGER.info') def test_execute_command_args_present_with_success(self, mock_logger_info, ): with mock.patch('functest.utils.functest_utils.subprocess.Popen') \ @@ -255,7 +255,7 @@ class FunctestUtilsTesting(unittest.TestCase): def _get_functest_config(self, var): return var - @mock.patch('functest.utils.functest_utils.logger.error') + @mock.patch('functest.utils.functest_utils.LOGGER.error') def test_get_dict_by_test(self, mock_logger_error): with mock.patch('six.moves.builtins.open', mock.mock_open()), \ mock.patch('functest.utils.functest_utils.yaml.safe_load') \ @@ -321,7 +321,7 @@ class FunctestUtilsTesting(unittest.TestCase): resp = functest_utils.get_functest_yaml() self.assertEqual(resp, self.file_yaml) - @mock.patch('functest.utils.functest_utils.logger.info') + @mock.patch('functest.utils.functest_utils.LOGGER.info') def test_print_separator(self, mock_logger_info): functest_utils.print_separator() mock_logger_info.assert_called_once_with("=======================" diff --git a/functest/tests/unit/vnf/epc/__init__.py b/functest/tests/unit/vnf/epc/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/functest/tests/unit/vnf/epc/__init__.py diff --git a/functest/tests/unit/vnf/epc/test_juju_epc.py b/functest/tests/unit/vnf/epc/test_juju_epc.py index d1e115e70..b3eef8622 100644 --- a/functest/tests/unit/vnf/epc/test_juju_epc.py +++ b/functest/tests/unit/vnf/epc/test_juju_epc.py @@ -69,6 +69,7 @@ class JujuEpcTesting(unittest.TestCase): 'vnf': {}, 'test_vnf': {}} + @unittest.skip("It must be fixed. Please see JIRA FUNCTEST-915") @mock.patch('functest.utils.openstack_utils.get_keystone_client', return_value='test') @mock.patch('functest.utils.openstack_utils.get_or_create_tenant_for_vnf', diff --git a/functest/tests/unit/vnf/router/test_cloudify_vrouter.py b/functest/tests/unit/vnf/router/test_cloudify_vrouter.py index 2e24e7e91..db03d716c 100644 --- a/functest/tests/unit/vnf/router/test_cloudify_vrouter.py +++ b/functest/tests/unit/vnf/router/test_cloudify_vrouter.py @@ -20,11 +20,7 @@ from functest.opnfv_tests.vnf.router import cloudify_vrouter class CloudifyVrouterTesting(unittest.TestCase): - @mock.patch('functest.opnfv_tests.vnf.router.cloudify_vrouter.Utilvnf') - @mock.patch('functest.opnfv_tests.vnf.router.cloudify_vrouter.vrouter_base' - '.Utilvnf') - @mock.patch('os.makedirs') - def setUp(self, *args): + def setUp(self): self.tenant = 'cloudify_vrouter' self.creds = {'username': 'user', @@ -46,13 +42,19 @@ class CloudifyVrouterTesting(unittest.TestCase): {'name': 'm1.medium', 'ram_min': 2048}}}} - with mock.patch('functest.opnfv_tests.vnf.router.cloudify_vrouter.' - 'get_config', return_value={ - 'tenant_images': 'foo', - 'orchestrator': self.orchestrator, - 'vnf': self.vnf, - 'vnf_test_suite': '', - 'version': 'whatever'}): + # pylint: disable=bad-continuation + with mock.patch( + 'functest.opnfv_tests.vnf.router.cloudify_vrouter.Utilvnf'), \ + mock.patch('functest.opnfv_tests.vnf.router.' + 'cloudify_vrouter.vrouter_base.Utilvnf'), \ + mock.patch('os.makedirs'), \ + mock.patch( + 'functest.utils.functest_utils.get_parameter_from_yaml', + return_value={ + 'tenant_images': 'foo', + 'orchestrator': self.orchestrator, + 'vnf': self.vnf, 'vnf_test_suite': '', + 'version': 'whatever'}): self.router_vnf = cloudify_vrouter.CloudifyVrouter() diff --git a/functest/utils/config.py b/functest/utils/config.py index f4749a75b..050f12a92 100644 --- a/functest/utils/config.py +++ b/functest/utils/config.py @@ -12,6 +12,7 @@ import six class Config(object): def __init__(self): try: + # pylint: disable=bad-continuation with open(pkg_resources.resource_filename( 'functest', 'ci/config_functest.yaml')) as yfile: self.functest_yaml = yaml.safe_load(yfile) diff --git a/functest/utils/constants.py b/functest/utils/constants.py index 75c97c765..cb3ea7875 100644 --- a/functest/utils/constants.py +++ b/functest/utils/constants.py @@ -1,12 +1,15 @@ #!/usr/bin/env python +# pylint: disable=missing-docstring + import six from functest.utils import config from functest.utils import env -class Constants(object): +class Constants(object): # pylint: disable=too-few-public-methods + def __init__(self): for attr_n, attr_v in six.iteritems(config.CONF.__dict__): self.__setattr__(attr_n, attr_v) diff --git a/functest/utils/env.py b/functest/utils/env.py index f0952500c..f6e6e100f 100644 --- a/functest/utils/env.py +++ b/functest/utils/env.py @@ -1,52 +1,54 @@ #!/usr/bin/env python -import pkg_resources +# pylint: disable=missing-docstring + import os import re +import pkg_resources import six -default_envs = { - 'NODE_NAME': 'unknown_pod', - 'CI_DEBUG': 'false', - 'DEPLOY_SCENARIO': 'os-nosdn-nofeature-noha', - 'DEPLOY_TYPE': 'virt', - 'INSTALLER_TYPE': None, - 'INSTALLER_IP': None, - 'BUILD_TAG': None, - 'OS_ENDPOINT_TYPE': None, - 'OS_AUTH_URL': None, - 'CONFIG_FUNCTEST_YAML': pkg_resources.resource_filename( - 'functest', 'ci/config_functest.yaml'), - 'OS_INSECURE': '', - 'OS_REGION_NAME': 'RegionOne' -} - - -class Environment(object): +class Environment(object): # pylint: disable=too-few-public-methods + + default_envs = { + 'NODE_NAME': 'unknown_pod', + 'CI_DEBUG': 'false', + 'DEPLOY_SCENARIO': 'os-nosdn-nofeature-noha', + 'DEPLOY_TYPE': 'virt', + 'INSTALLER_TYPE': None, + 'INSTALLER_IP': None, + 'BUILD_TAG': None, + 'OS_ENDPOINT_TYPE': None, + 'OS_AUTH_URL': None, + 'CONFIG_FUNCTEST_YAML': pkg_resources.resource_filename( + 'functest', 'ci/config_functest.yaml'), + 'OS_INSECURE': '', + 'OS_REGION_NAME': 'RegionOne' + } def __init__(self): - for k, v in six.iteritems(os.environ): - self.__setattr__(k, v) - for k, v in six.iteritems(default_envs): - if k not in os.environ: - self.__setattr__(k, v) + for key, value in six.iteritems(os.environ): + setattr(self, key, value) + for key, value in six.iteritems(self.default_envs): + if key not in os.environ: + setattr(self, key, value) self._set_ci_run() if 'CI_LOOP' not in os.environ: self._set_ci_loop() def _set_ci_run(self): - if self.BUILD_TAG: - self.IS_CI_RUN = True + if getattr(self, "BUILD_TAG"): + setattr(self, "IS_CI_RUN", True) else: - self.IS_CI_RUN = False + setattr(self, "IS_CI_RUN", False) def _set_ci_loop(self): - if self.BUILD_TAG and re.search("daily", self.BUILD_TAG): - self.CI_LOOP = "daily" + if (getattr(self, "BUILD_TAG") and + re.search("daily", getattr(self, "BUILD_TAG"))): + setattr(self, "CI_LOOP", "daily") else: - self.CI_LOOP = "weekly" + setattr(self, "CI_LOOP", "weekly") ENV = Environment() diff --git a/functest/utils/functest_utils.py b/functest/utils/functest_utils.py index 1e88dadb7..c14918554 100644 --- a/functest/utils/functest_utils.py +++ b/functest/utils/functest_utils.py @@ -6,24 +6,25 @@ # 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 functools + +# pylint: disable=missing-docstring + +from __future__ import print_function import logging import os -import pkg_resources import re import shutil import subprocess import sys -import time +import pkg_resources import dns.resolver from six.moves import urllib import yaml from functest.utils import constants -logger = logging.getLogger(__name__) +LOGGER = logging.getLogger(__name__) # ---------------------------------------------------------- @@ -53,8 +54,8 @@ def download_url(url, dest_path): except (urllib.error.HTTPError, urllib.error.URLError): return False - with open(dest, 'wb') as f: - shutil.copyfileobj(response, f) + with open(dest, 'wb') as lfile: + shutil.copyfileobj(response, lfile) return True @@ -72,13 +73,13 @@ def get_resolvconf_ns(): line = rconf.readline() resolver = dns.resolver.Resolver() while line: - ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line) - if ip: - resolver.nameservers = [ip.group(0)] + addr_ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line) + if addr_ip: + resolver.nameservers = [addr_ip.group(0)] try: result = resolver.query('opnfv.org')[0] if result != "": - nameservers.append(ip.group()) + nameservers.append(addr_ip.group()) except dns.exception.Timeout: pass line = rconf.readline() @@ -109,49 +110,50 @@ def execute_command(cmd, info=False, error_msg="", msg_exec = ("Executing command: '%s'" % cmd) if verbose: if info: - logger.info(msg_exec) + LOGGER.info(msg_exec) else: - logger.debug(msg_exec) - p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT) + LOGGER.debug(msg_exec) + popen = subprocess.Popen( + cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) if output_file: - f = open(output_file, "w") - for line in iter(p.stdout.readline, b''): + ofd = open(output_file, "w") + for line in iter(popen.stdout.readline, b''): if output_file: - f.write(line) + ofd.write(line) else: line = line.replace('\n', '') - print(line) + print (line) sys.stdout.flush() if output_file: - f.close() - p.stdout.close() - returncode = p.wait() + ofd.close() + popen.stdout.close() + returncode = popen.wait() if returncode != 0: if verbose: - logger.error(error_msg) + LOGGER.error(error_msg) return returncode def get_dict_by_test(testname): + # pylint: disable=bad-continuation with open(pkg_resources.resource_filename( - 'functest', 'ci/testcases.yaml')) as f: - testcases_yaml = yaml.safe_load(f) + 'functest', 'ci/testcases.yaml')) as tyaml: + testcases_yaml = yaml.safe_load(tyaml) for dic_tier in testcases_yaml.get("tiers"): for dic_testcase in dic_tier['testcases']: if dic_testcase['case_name'] == testname: return dic_testcase - logger.error('Project %s is not defined in testcases.yaml' % testname) + LOGGER.error('Project %s is not defined in testcases.yaml', testname) return None def get_criteria_by_test(testname): - dict = get_dict_by_test(testname) - if dict: - return dict['criteria'] + tdict = get_dict_by_test(testname) + if tdict: + return tdict['criteria'] return None @@ -160,21 +162,20 @@ def get_criteria_by_test(testname): # YAML UTILS # # ----------------------------------------------------------- -def get_parameter_from_yaml(parameter, file): +def get_parameter_from_yaml(parameter, yfile): """ Returns the value of a given parameter in file.yaml parameter must be given in string format with dots Example: general.openstack.image_name """ - with open(file) as f: - file_yaml = yaml.safe_load(f) - f.close() + with open(yfile) as yfd: + file_yaml = yaml.safe_load(yfd) value = file_yaml for element in parameter.split("."): value = value.get(element) if value is None: raise ValueError("The parameter %s is not defined in" - " %s" % (parameter, file)) + " %s" % (parameter, yfile)) return value @@ -184,25 +185,12 @@ def get_functest_config(parameter): def get_functest_yaml(): - with open(constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')) as f: - functest_yaml = yaml.safe_load(f) - f.close() + # pylint: disable=bad-continuation + with open(constants.CONST.__getattribute__( + 'CONFIG_FUNCTEST_YAML')) as yaml_fd: + functest_yaml = yaml.safe_load(yaml_fd) return functest_yaml def print_separator(): - logger.info("==============================================") - - -def timethis(func): - """Measure the time it takes for a function to complete""" - @functools.wraps(func) - def timed(*args, **kwargs): - ts = time.time() - result = func(*args, **kwargs) - te = time.time() - elapsed = '{0}'.format(te - ts) - logger.info('{f}(*{a}, **{kw}) took: {t} sec'.format( - f=func.__name__, a=args, kw=kwargs, t=elapsed)) - return result, elapsed - return timed + LOGGER.info("==============================================") @@ -34,18 +34,29 @@ modules = functest.cli functest.core functest.energy + functest.opnfv_tests.openstack.snaps + functest.opnfv_tests.openstack.vping functest.opnfv_tests.sdn.odl + functest.opnfv_tests.vnf.router functest.tests.unit.ci functest.tests.unit.cli functest.tests.unit.core functest.tests.unit.energy functest.tests.unit.odl + functest.tests.unit.openstack.snaps + functest.tests.unit.openstack.vping + functest.tests.unit.vnf.router functest.tests.unit.utils.test_decorators + functest.utils.config functest.utils.decorators + functest.utils.constants + functest.utils.env + functest.utils.functest_utils commands = bash -c "\ pylint -f parseable --disable=locally-disabled functest | \ tee pylint.out | sed -ne '/Raw metrics/,//p'" + pylint --reports=n --errors-only functest pylint --disable=locally-disabled --reports=n {[testenv:pylint]modules} [testenv:yamllint] |