* 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()
# 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
class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase):
+ # pylint: disable=too-many-instance-attributes
"""vrouter testcase deployed with Cloudify Orchestrator."""
__logger = logging.getLogger(__name__)
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
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(
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)
+ return self.util.get_vnf_info_list(
+ self.cfy_manager_ip, self.deployment_name, target_vnf_name)
# ----------------------------------------------------------
# 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
# 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
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__)
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)
break
address = server.addresses[
- network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
+ network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
return address
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
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 : " +
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"]:
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"]
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)
# 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
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)
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 = {}
# 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
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')
"""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
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__)
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)
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
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
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
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
# 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
# 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 (
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)
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 = []
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 = []
# 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
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
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
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',
{'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.opnfv_tests.vnf.router.cloudify_vrouter.'
+ 'get_config',
+ return_value={
+ 'tenant_images': 'foo',
+ 'orchestrator': self.orchestrator,
+ 'vnf': self.vnf, 'vnf_test_suite': '',
+ 'version': 'whatever'}):
self.router_vnf = cloudify_vrouter.CloudifyVrouter()
#!/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()
functest.core
functest.energy
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.vnf.router
functest.tests.unit.utils.test_decorators
functest.utils.decorators
commands =
bash -c "\
pylint -f parseable --disable=locally-disabled functest | \
tee pylint.out | sed -ne '/Raw metrics/,//p'"
+ pylint --disable=locally-disabled --reports=n --errors-only functest
pylint --disable=locally-disabled --reports=n {[testenv:pylint]modules}
[testenv:yamllint]
Code Review / functest.git / commitdiff