ARG ODL_TAG=release/carbon
ARG OPENSTACK_TAG=stable/ocata
ARG VIMS_TAG=stable
-ARG VROUTER_TAG=stable
ARG REPOS_DIR=/home/opnfv/repos
ARG FUNCTEST_BASE_DIR=/home/opnfv/functest
ARG FUNCTEST_CONF_DIR=${FUNCTEST_BASE_DIR}/conf
# other repositories
RUN git clone --depth 1 -b $ODL_TAG https://git.opendaylight.org/gerrit/p/integration/test.git /src/odl_test
RUN git clone --depth 1 -b $VIMS_TAG https://github.com/boucherv-orange/clearwater-live-test /src/vims-test
-RUN git clone --depth 1 -b $VROUTER_TAG https://github.com/oolorg/opnfv-functest-vrouter.git ${REPOS_VNFS_DIR}/vrouter
# Install tempest venv and create symlink for running refstack-client
RUN ln -s /src/tempest /src/refstack-client/.tempest \
run:
module: 'functest.opnfv_tests.vnf.ims.orchestra_clearwaterims'
class: 'ClearwaterImsVnf'
+
+ -
+ case_name: vyos_vrouter
+ project_name: functest
+ criteria: 100
+ blocking: false
+ description: >-
+ This test case is vRouter testing.
+ dependencies:
+ installer: 'fuel'
+ scenario: 'nosdn-nofeature'
+ run:
+ module: 'functest.opnfv_tests.vnf.router.cloudify_vrouter'
+ class: 'CloudifyVrouter'
repo_odl_test: /src/odl_test
repo_fds: /src/fds
repo_securityscan: /home/opnfv/repos/securityscanning
- repo_vrouter: /home/opnfv/repos/vnfs/vrouter
functest: /home/opnfv/functest
results: /home/opnfv/functest/results
functest_conf: /home/opnfv/functest/conf
functest_data: /home/opnfv/functest/data
ims_data: /home/opnfv/functest/data/ims/
+ router_data: /home/opnfv/functest/data/router/
functest_images: /home/opnfv/functest/images
rally_inst: /root/.rally
tenant_name: orchestra_clearwaterims
tenant_description: Clearwater IMS deployed with Open Baton
config: orchestra.yaml
+ vyos_vrouter:
+ tenant_name: vrouter
+ tenant_description: vRouter
+ config: cloudify_vrouter.yaml
promise:
tenant_name: promise
http://artifacts.opnfv.org/functest/images/cirros-d161201-aarch64-kernel
https://cloud-images.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-cloudimg-arm64-uefi1.img
http://cloud.centos.org/altarch/7/images/aarch64/CentOS-7-aarch64-GenericCloud.qcow2.xz
+https://sourceforge.net/projects/ool-opnfv/files/vyos-1.1.7.img
EOF
xz --decompress --force --keep ${1:-/home/opnfv/functest/images}/CentOS-7-aarch64-GenericCloud.qcow2.xz
-
case_name: vyos_vrouter
- enabled: false
project_name: functest
criteria: 100
blocking: false
installer: 'fuel'
scenario: 'nosdn-nofeature'
run:
- module: 'functest.opnfv_tests.vnf.router.vyos_vrouter'
- class: 'VrouterVnf'
+ module: 'functest.opnfv_tests.vnf.router.cloudify_vrouter'
+ class: 'CloudifyVrouter'
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""vrouter testcase implementation."""
+
+import logging
+import os
+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.utils.constants import CONST
+import functest.utils.openstack_utils as os_utils
+
+from git import Repo
+
+from snaps.openstack.os_credentials import OSCreds
+from snaps.openstack.create_network import (NetworkSettings, SubnetSettings,
+ OpenStackNetwork)
+from snaps.openstack.create_security_group import (SecurityGroupSettings,
+ SecurityGroupRuleSettings,
+ Direction, Protocol,
+ OpenStackSecurityGroup)
+from snaps.openstack.create_router import RouterSettings, OpenStackRouter
+from snaps.openstack.create_instance import (VmInstanceSettings,
+ FloatingIpSettings,
+ OpenStackVmInstance)
+from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
+from snaps.openstack.create_image import ImageSettings, OpenStackImage
+from snaps.openstack.create_keypairs import KeypairSettings, OpenStackKeypair
+from snaps.openstack.create_network import PortSettings
+import snaps.openstack.utils.glance_utils as glance_utils
+
+from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
+
+__author__ = "Shuya Nakama <shuya.nakama@okinawaopenlabs.org>"
+
+
+class CloudifyVrouter(vrouter_base.VrouterOnBoardingBase):
+ """vrouter testcase deployed with Cloudify Orchestrator."""
+
+ __logger = logging.getLogger(__name__)
+ name = __name__
+
+ def __init__(self, **kwargs):
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = "vyos_vrouter"
+ super(CloudifyVrouter, self).__init__(**kwargs)
+
+ # Retrieve the configuration
+ try:
+ self.config = CONST.__getattribute__(
+ 'vnf_{}_config'.format(self.case_name))
+ except Exception:
+ raise Exception("VNF config file not found")
+
+ self.snaps_creds = ''
+ self.created_object = []
+
+ self.cfy_manager_ip = ''
+ self.util_info = {}
+ self.deployment_name = ''
+
+ config_file = os.path.join(self.case_dir, self.config)
+ self.orchestrator = dict(
+ requirements=get_config("orchestrator.requirements", config_file),
+ )
+ self.details['orchestrator'] = dict(
+ name=get_config("orchestrator.name", config_file),
+ version=get_config("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)
+ )
+ self.details['vnf'] = dict(
+ descriptor_version=self.vnf['descriptor']['version'],
+ name=get_config("vnf.name", config_file),
+ version=get_config("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)
+ )
+ self.images = get_config("tenant_images", config_file)
+ self.__logger.info("Images needed for vrouter: %s", self.images)
+
+ def prepare(self):
+ super(CloudifyVrouter, self).prepare()
+
+ self.__logger.info("Additional pre-configuration steps")
+
+ self.snaps_creds = OSCreds(
+ username=self.creds['username'],
+ password=self.creds['password'],
+ auth_url=self.creds['auth_url'],
+ project_name=self.creds['tenant'],
+ identity_api_version=int(os_utils.get_keystone_client_version()))
+
+ self.util.set_credentials(self.creds["username"],
+ self.creds["password"],
+ self.creds["auth_url"],
+ self.creds["tenant"])
+
+ # needs some images
+ self.__logger.info("Upload some OS images if it doesn't exist")
+ for image_name, image_file in self.images.iteritems():
+ self.__logger.info("image: %s, file: %s", image_name, image_file)
+ if image_file and image_name:
+ image_creator = OpenStackImage(
+ self.snaps_creds,
+ ImageSettings(name=image_name,
+ image_user='cloud',
+ img_format='qcow2',
+ image_file=image_file))
+ image_creator.create()
+ self.created_object.append(image_creator)
+
+ def deploy_orchestrator(self):
+ """
+ Deploy Cloudify Manager.
+ network, security group, fip, VM creation
+ """
+ # network creation
+
+ start_time = time.time()
+ self.__logger.info("Creating keypair ...")
+ kp_file = os.path.join(self.data_dir, "cloudify_vrouter.pem")
+ keypair_settings = KeypairSettings(name='cloudify_vrouter_kp',
+ private_filepath=kp_file)
+ keypair_creator = OpenStackKeypair(self.snaps_creds, keypair_settings)
+ keypair_creator.create()
+ self.created_object.append(keypair_creator)
+
+ self.__logger.info("Creating full network ...")
+ subnet_settings = SubnetSettings(name='cloudify_vrouter_subnet',
+ cidr='10.67.79.0/24')
+ network_settings = NetworkSettings(name='cloudify_vrouter_network',
+ subnet_settings=[subnet_settings])
+ network_creator = OpenStackNetwork(self.snaps_creds, network_settings)
+ network_creator.create()
+ self.created_object.append(network_creator)
+ ext_net_name = snaps_utils.get_ext_net_name(self.snaps_creds)
+ router_creator = OpenStackRouter(
+ self.snaps_creds,
+ RouterSettings(
+ name='cloudify_vrouter_router',
+ external_gateway=ext_net_name,
+ internal_subnets=[subnet_settings.name]))
+ router_creator.create()
+ self.created_object.append(router_creator)
+
+ # security group creation
+ self.__logger.info("Creating security group for cloudify manager vm")
+ sg_rules = list()
+ sg_rules.append(
+ SecurityGroupRuleSettings(sec_grp_name="sg-cloudify-manager",
+ direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=1,
+ port_range_max=65535))
+ sg_rules.append(
+ SecurityGroupRuleSettings(sec_grp_name="sg-cloudify-manager",
+ direction=Direction.ingress,
+ protocol=Protocol.udp, port_range_min=1,
+ port_range_max=65535))
+
+ security_group_creator = OpenStackSecurityGroup(
+ self.snaps_creds,
+ SecurityGroupSettings(
+ name="sg-cloudify-manager",
+ rule_settings=sg_rules))
+
+ security_group_creator.create()
+ self.created_object.append(security_group_creator)
+
+ # orchestrator VM flavor
+ self.__logger.info("Get or create flavor for cloudify manager vm ...")
+
+ flavor_settings = FlavorSettings(
+ name=self.orchestrator['requirements']['flavor']['name'],
+ ram=self.orchestrator['requirements']['flavor']['ram_min'],
+ disk=50,
+ vcpus=2)
+ flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
+ flavor_creator.create()
+ self.created_object.append(flavor_creator)
+ image_settings = ImageSettings(
+ name=self.orchestrator['requirements']['os_image'],
+ image_user='centos',
+ exists=True)
+
+ port_settings = PortSettings(name='cloudify_manager_port',
+ network_name=network_settings.name)
+
+ manager_settings = VmInstanceSettings(
+ name='cloudify_manager',
+ flavor=flavor_settings.name,
+ port_settings=[port_settings],
+ security_group_names=[
+ security_group_creator.sec_grp_settings.name],
+ floating_ip_settings=[FloatingIpSettings(
+ name='cloudify_manager_fip',
+ port_name=port_settings.name,
+ router_name=router_creator.router_settings.name)])
+
+ manager_creator = OpenStackVmInstance(self.snaps_creds,
+ manager_settings,
+ image_settings,
+ keypair_settings)
+
+ self.__logger.info("Creating cloudify manager VM")
+ manager_creator.create()
+ self.created_object.append(manager_creator)
+
+ public_auth_url = os_utils.get_endpoint('identity')
+
+ self.__logger.info("Set creds for cloudify manager")
+ cfy_creds = dict(keystone_username=self.tenant_name,
+ keystone_password=self.tenant_name,
+ keystone_tenant_name=self.tenant_name,
+ keystone_url=public_auth_url)
+
+ cfy_client = CloudifyClient(host=manager_creator.get_floating_ip().ip,
+ username='admin',
+ password='admin',
+ tenant='default_tenant')
+
+ self.orchestrator['object'] = cfy_client
+
+ self.cfy_manager_ip = manager_creator.get_floating_ip().ip
+
+ self.__logger.info("Attemps running status of the Manager")
+ cfy_status = None
+ retry = 10
+ while str(cfy_status) != 'running' and retry:
+ try:
+ cfy_status = cfy_client.manager.get_status()['status']
+ self.__logger.debug("The current manager status is %s",
+ cfy_status)
+ except Exception: # pylint: disable=broad-except
+ self.__logger.warning("Cloudify Manager isn't " +
+ "up and running. Retrying ...")
+ retry = retry - 1
+ time.sleep(30)
+
+ if str(cfy_status) == 'running':
+ self.__logger.info("Cloudify Manager is up and running")
+ else:
+ raise Exception("Cloudify Manager isn't up and running")
+
+ self.__logger.info("Put OpenStack creds in manager")
+ secrets_list = cfy_client.secrets.list()
+ for k, val in cfy_creds.iteritems():
+ if not any(d.get('key', None) == k for d in secrets_list):
+ cfy_client.secrets.create(k, val)
+ else:
+ cfy_client.secrets.update(k, val)
+
+ duration = time.time() - start_time
+
+ self.__logger.info("Put private keypair in manager")
+ if manager_creator.vm_ssh_active(block=True):
+ ssh = manager_creator.ssh_client()
+ scp = SCPClient(ssh.get_transport(), socket_timeout=15.0)
+ scp.put(kp_file, '~/')
+ cmd = "sudo cp ~/cloudify_vrouter.pem /etc/cloudify/"
+ run_blocking_ssh_command(ssh, cmd)
+ cmd = "sudo chmod 444 /etc/cloudify/cloudify_vrouter.pem"
+ run_blocking_ssh_command(ssh, cmd)
+ cmd = "sudo yum install -y gcc python-devel"
+ run_blocking_ssh_command(
+ ssh, cmd, "Unable to install packages on manager")
+
+ self.details['orchestrator'].update(status='PASS', duration=duration)
+
+ self.vnf['inputs'].update(dict(external_network_name=ext_net_name))
+
+ return True
+
+ def deploy_vnf(self):
+ start_time = time.time()
+
+ self.__logger.info("Upload VNFD")
+ cfy_client = self.orchestrator['object']
+ descriptor = self.vnf['descriptor']
+ self.deployment_name = descriptor.get('name')
+
+ vrouter_blueprint_dir = os.path.join(self.data_dir,
+ self.util.blueprint_dir)
+ if not os.path.exists(vrouter_blueprint_dir):
+ Repo.clone_from(descriptor.get('url'),
+ vrouter_blueprint_dir,
+ branch=descriptor.get('version'))
+
+ cfy_client.blueprints.upload(vrouter_blueprint_dir +
+ self.util.blueprint_file_name,
+ descriptor.get('name'))
+
+ self.__logger.info("Get or create flavor for vrouter")
+ flavor_settings = FlavorSettings(
+ name=self.vnf['requirements']['flavor']['name'],
+ ram=self.vnf['requirements']['flavor']['ram_min'],
+ disk=25,
+ vcpus=1)
+ flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
+ flavor = flavor_creator.create()
+ self.created_object.append(flavor_creator)
+
+ # set image name
+ glance = glance_utils.glance_client(self.snaps_creds)
+ image = glance_utils.get_image(glance,
+ "vyos1.1.7")
+ self.vnf['inputs'].update(dict(target_vnf_image_id=image.id))
+ self.vnf['inputs'].update(dict(reference_vnf_image_id=image.id))
+
+ # set flavor id
+ self.vnf['inputs'].update(dict(target_vnf_flavor_id=flavor.id))
+ self.vnf['inputs'].update(dict(reference_vnf_flavor_id=flavor.id))
+
+ self.vnf['inputs'].update(dict(keystone_username=self.tenant_name))
+ self.vnf['inputs'].update(dict(keystone_password=self.tenant_name))
+ self.vnf['inputs'].update(dict(keystone_tenant_name=self.tenant_name))
+ self.vnf['inputs'].update(
+ dict(keystone_url=os_utils.get_endpoint('identity')))
+
+ self.__logger.info("Create VNF Instance")
+ cfy_client.deployments.create(descriptor.get('name'),
+ descriptor.get('name'),
+ self.vnf.get('inputs'))
+
+ wait_for_execution(cfy_client,
+ get_execution_id(
+ cfy_client, descriptor.get('name')),
+ self.__logger,
+ timeout=7200)
+
+ self.__logger.info("Start the VNF Instance deployment")
+ execution = cfy_client.executions.start(descriptor.get('name'),
+ 'install')
+ # Show execution log
+ execution = wait_for_execution(cfy_client, execution, self.__logger)
+
+ duration = time.time() - start_time
+
+ self.__logger.info(execution)
+ if execution.status == 'terminated':
+ self.details['vnf'].update(status='PASS', duration=duration)
+ result = True
+ else:
+ self.details['vnf'].update(status='FAIL', duration=duration)
+ result = False
+ return result
+
+ def test_vnf(self):
+ cfy_client = self.orchestrator['object']
+ credentials = {"username": self.creds["username"],
+ "password": self.creds["password"],
+ "auth_url": self.creds["auth_url"],
+ "tenant_name": self.creds["tenant"],
+ "region_name": os.environ['OS_REGION_NAME']}
+
+ self.util_info = {"credentials": credentials,
+ "cfy": cfy_client,
+ "vnf_data_dir": self.util.vnf_data_dir}
+
+ start_time = time.time()
+
+ result, test_result_data = super(CloudifyVrouter, self).test_vnf()
+
+ duration = time.time() - start_time
+
+ if result:
+ self.details['test_vnf'].update(status='PASS',
+ result='OK',
+ full_result=test_result_data,
+ duration=duration)
+ else:
+ self.details['test_vnf'].update(status='FAIL',
+ result='NG',
+ full_result=test_result_data,
+ duration=duration)
+
+ return True
+
+ def clean(self):
+ try:
+ cfy_client = self.orchestrator['object']
+ dep_name = self.vnf['descriptor'].get('name')
+ # kill existing execution
+ self.__logger.info('Deleting the current deployment')
+ exec_list = cfy_client.executions.list(dep_name)
+ for execution in exec_list:
+ if execution['status'] == "started":
+ try:
+ cfy_client.executions.cancel(execution['id'],
+ force=True)
+ except: # pylint: disable=broad-except
+ self.__logger.warn("Can't cancel the current exec")
+
+ execution = cfy_client.executions.start(
+ dep_name,
+ 'uninstall',
+ parameters=dict(ignore_failure=True))
+
+ 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
+ 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
+
+
+def wait_for_execution(client, execution, logger, timeout=7200, ):
+ """Wait for a workflow execution on Cloudify Manager."""
+ # if execution already ended - return without waiting
+ if execution.status in Execution.END_STATES:
+ return execution
+
+ if timeout is not None:
+ deadline = time.time() + timeout
+
+ # Poll for execution status and execution logs, until execution ends
+ # and we receive an event of type in WORKFLOW_END_TYPES
+ offset = 0
+ batch_size = 50
+ event_list = []
+ execution_ended = False
+ while True:
+ event_list = client.events.list(
+ execution_id=execution.id,
+ _offset=offset,
+ _size=batch_size,
+ include_logs=False,
+ sort='@timestamp').items
+
+ offset = offset + len(event_list)
+ for event in event_list:
+ logger.debug(event.get('message'))
+
+ if timeout is not None:
+ if time.time() > deadline:
+ raise RuntimeError(
+ 'execution of operation {0} for deployment {1} '
+ 'timed out'.format(execution.workflow_id,
+ execution.deployment_id))
+ else:
+ # update the remaining timeout
+ timeout = deadline - time.time()
+
+ if not execution_ended:
+ execution = client.executions.get(execution.id)
+ execution_ended = execution.status in Execution.END_STATES
+
+ if execution_ended:
+ break
+
+ time.sleep(5)
+
+ return execution
+
+
+def get_execution_id(client, deployment_id):
+ """
+ Get the execution id of a env preparation.
+ network, security group, fip, VM creation
+ """
+ executions = client.executions.list(deployment_id=deployment_id)
+ for execution in executions:
+ if execution.workflow_id == 'create_deployment_environment':
+ return execution
+ raise RuntimeError('Failed to get create_deployment_environment '
+ 'workflow execution.'
+ 'Available executions: {0}'.format(executions))
+
+
+def run_blocking_ssh_command(ssh, cmd, error_msg="Unable to run this command"):
+ """Command to run ssh command with the exit status."""
+ (_, stdout, _) = ssh.exec_command(cmd)
+ if stdout.channel.recv_exit_status() != 0:
+ raise Exception(error_msg)
--- /dev/null
+tenant_images:
+ cloudify_manager_4.0: /home/opnfv/functest/images/cloudify-manager-premium-4.0.1.qcow2
+ vyos1.1.7: /home/opnfv/functest/images/vyos-1.1.7.img
+test_data:
+ url: 'https://github.com/oolorg/opnfv-vnf-data.git'
+ branch: 'master'
+orchestrator:
+ name: cloudify
+ version: '4.0'
+ requirements:
+ flavor:
+ name: m1.medium
+ ram_min: 4096
+ os_image: 'cloudify_manager_4.0'
+vnf:
+ name: vyos1.1.7
+ version: '1.1.7'
+ descriptor:
+ url: https://github.com/oolorg/opnfv-vnf-vyos-blueprint/
+ name: vrouter-opnfv
+ version: 'master'
+ requirements:
+ flavor:
+ name: m1.medium
+ ram_min: 2048
+ inputs:
+ external_network_name: admin_floating_net
+ region: RegionOne
+vnf_test_suite:
+ name: vyos-vrouter-test
+ version: "1.0"
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""vrouter function test execution module"""
+
+import logging
+import time
+import yaml
+
+from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
+from functest.opnfv_tests.vnf.router.vnf_controller.vnf_controller import (
+ VnfController)
+
+
+class FunctionTestExec(object):
+ """vrouter function test execution class"""
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self, util_info):
+ self.logger.debug("init test exec")
+ self.util = Utilvnf()
+ credentials = util_info["credentials"]
+ self.vnf_ctrl = VnfController(util_info)
+
+ test_cmd_map_file = open(self.util.vnf_data_dir +
+ self.util.opnfv_vnf_data_dir +
+ self.util.command_template_dir +
+ self.util.test_cmd_map_yaml_file,
+ 'r')
+ self.test_cmd_map_yaml = yaml.safe_load(test_cmd_map_file)
+ test_cmd_map_file.close()
+
+ self.util.set_credentials(credentials["username"],
+ credentials["password"],
+ credentials["auth_url"],
+ credentials["tenant_name"],
+ credentials["region_name"])
+
+ with open(self.util.test_env_config_yaml) as file_fd:
+ test_env_config_yaml = yaml.safe_load(file_fd)
+ file_fd.close()
+
+ self.protocol_stable_wait = test_env_config_yaml.get("general").get(
+ "protocol_stable_wait")
+
+ def config_target_vnf(self, target_vnf, reference_vnf, test_kind):
+ self.logger.debug("Configuration to target vnf")
+ test_info = self.test_cmd_map_yaml[target_vnf["os_type"]]
+ test_cmd_file_path = test_info[test_kind]["pre_command_target"]
+ target_parameter_file_path = test_info[test_kind]["parameter_target"]
+ prompt_file_path = test_info["prompt"]
+
+ return self.vnf_ctrl.config_vnf(target_vnf,
+ reference_vnf,
+ test_cmd_file_path,
+ target_parameter_file_path,
+ prompt_file_path)
+
+ def config_reference_vnf(self, target_vnf, reference_vnf, test_kind):
+ self.logger.debug("Configuration to reference vnf")
+ test_info = self.test_cmd_map_yaml[reference_vnf["os_type"]]
+ test_cmd_file_path = test_info[test_kind]["pre_command_reference"]
+ reference_parameter_file_path = test_info[test_kind][
+ "parameter_reference"]
+ prompt_file_path = test_info["prompt"]
+
+ return self.vnf_ctrl.config_vnf(reference_vnf,
+ target_vnf,
+ test_cmd_file_path,
+ reference_parameter_file_path,
+ prompt_file_path)
+
+ def result_check(self, target_vnf, reference_vnf, test_kind, test_list):
+ test_info = self.test_cmd_map_yaml[target_vnf["os_type"]]
+ target_parameter_file_path = test_info[test_kind]["parameter_target"]
+ prompt_file_path = test_info["prompt"]
+ check_rule_file_path_list = []
+
+ for test in test_list:
+ check_rule_file_path_list.append(test_info[test_kind][test])
+
+ return self.vnf_ctrl.result_check(target_vnf,
+ reference_vnf,
+ check_rule_file_path_list,
+ target_parameter_file_path,
+ prompt_file_path)
+
+ def run(self, target_vnf, reference_vnf_list, test_info, test_list):
+ test_result_data = {}
+ test_kind = test_info["protocol"]
+ for reference_vnf in reference_vnf_list:
+ self.logger.debug("Start config command " +
+ target_vnf["vnf_name"] + " and " +
+ reference_vnf["vnf_name"])
+
+ result = self.config_target_vnf(target_vnf,
+ reference_vnf,
+ test_kind)
+ if not result:
+ return False, test_result_data
+
+ result = self.config_reference_vnf(target_vnf,
+ reference_vnf,
+ test_kind)
+ if not result:
+ return False, test_result_data
+
+ self.logger.debug("Finish config command.")
+
+ self.logger.debug("Waiting for protocol stable.")
+ time.sleep(self.protocol_stable_wait)
+
+ self.logger.debug("Start check method")
+
+ (result, res_dict_data_list) = self.result_check(target_vnf,
+ reference_vnf,
+ test_kind,
+ test_list)
+
+ test_result_data = {"test_kind": test_info["test_kind"],
+ "protocol": test_info["protocol"],
+ "result": res_dict_data_list}
+
+ if not result:
+ self.logger.debug("Error check method.")
+ return False, test_result_data
+
+ self.logger.debug("Finish check method.")
+
+ return True, test_result_data
--- /dev/null
+test_scenario_list:
+ -
+ test_type: 'function_test'
+ vnf_list:
+ -
+ vnf_name: 'target_vnf'
+ os_type: 'vyos'
+ image_name: 'vyos1.1.7'
+ flavor_name: 'm1.medium'
+ -
+ vnf_name: 'reference_vnf'
+ os_type: 'vyos'
+ image_name: 'vyos1.1.7'
+ flavor_name: 'm1.medium'
+ function_test_list:
+ -
+ target_vnf_name: 'target_vnf'
+ test_list:
+ -
+ test_kind: 'Interoperability'
+ protocol: 'BGP'
+ BGP:
+ - 'Checking_the_peer_of_BGP'
+ - 'Checking_the_status_of_BGP'
+ - 'Checking_the_advertised_routes'
+ - 'Checking_the_received_routes'
+ - 'Checking_the_routing_table'
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+""" Utility module of vrouter testcase """
+
+import json
+import logging
+import os
+import pkg_resources
+import requests
+import yaml
+
+from functest.utils.constants import CONST
+from git import Repo
+from novaclient import client as novaclient
+from keystoneauth1 import session
+from keystoneauth1 import loading
+from requests.auth import HTTPBasicAuth
+
+RESULT_SPRIT_INDEX = {
+ "transfer": 8,
+ "bandwidth": 6,
+ "jitter": 4,
+ "los_total": 2,
+ "pkt_loss": 1
+}
+
+BIT_PER_BYTE = 8
+
+NOVA_CLIENT_API_VERSION = '2'
+NOVA_CILENT_NETWORK_INFO_INDEX = 0
+CFY_INFO_OUTPUT_FILE = "output.txt"
+
+CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
+PACKET_LOST_INFO_INDEX = 0
+PACKET_TOTAL_INFO_INDEX = 1
+
+NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
+NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
+NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
+NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
+
+
+class Utilvnf(object):
+ """ Utility class of vrouter testcase """
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self):
+ self.username = ""
+ self.password = ""
+ self.auth_url = ""
+ self.tenant_name = ""
+ self.region_name = ""
+
+ data_dir = data_dir = CONST.__getattribute__('dir_router_data')
+
+ self.vnf_data_dir = data_dir
+ self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
+ self.command_template_dir = "command_template/"
+ self.test_scenario_yaml = "test_scenario.yaml"
+ test_env_config_yaml_file = "test_env_config.yaml"
+ self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
+ self.test_env_config_yaml = os.path.join(
+ self.vnf_data_dir,
+ self.opnfv_vnf_data_dir,
+ test_env_config_yaml_file)
+
+ self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
+ self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
+
+ if not os.path.exists(self.vnf_data_dir):
+ os.makedirs(self.vnf_data_dir)
+
+ case_dir = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/vnf/router')
+
+ config_file_name = CONST.__getattribute__(
+ 'vnf_{}_config'.format("vyos_vrouter"))
+
+ config_file = os.path.join(case_dir, config_file_name)
+
+ with open(config_file) as file_fd:
+ vrouter_config_yaml = yaml.safe_load(file_fd)
+ file_fd.close()
+
+ test_data = vrouter_config_yaml.get("test_data")
+
+ self.logger.debug("Downloading the test data.")
+ vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
+
+ if not os.path.exists(vrouter_data_path):
+ Repo.clone_from(test_data['url'],
+ vrouter_data_path,
+ branch=test_data['branch'])
+
+ with open(self.test_env_config_yaml) as file_fd:
+ test_env_config_yaml = yaml.safe_load(file_fd)
+ file_fd.close()
+
+ self.image = test_env_config_yaml.get(
+ "general").get("images").get("vyos")
+ self.tester_image = test_env_config_yaml.get(
+ "general").get("images").get("tester_vm_os")
+
+ 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)
+
+ def get_nova_client(self):
+ creds = self.get_nova_credentials()
+ loader = loading.get_plugin_loader('password')
+ auth = loader.load_from_options(**creds)
+ sess = session.Session(auth=auth)
+ nova_client = novaclient.Client(NOVA_CLIENT_API_VERSION, session=sess)
+
+ return nova_client
+
+ def set_credentials(self, username, password, auth_url,
+ tenant_name, region_name="RegionOne"):
+ self.username = username
+ self.password = password
+ self.auth_url = auth_url
+ self.tenant_name = tenant_name
+ self.region_name = region_name
+
+ def get_nova_credentials(self):
+ creds = {}
+ creds['username'] = self.username
+ creds['password'] = self.password
+ creds['auth_url'] = self.auth_url
+ creds['tenant_name'] = self.tenant_name
+ return creds
+
+ def get_address(self, server_name, network_name):
+ nova_client = self.get_nova_client()
+ servers_list = nova_client.servers.list()
+ server = None
+
+ for server in servers_list:
+ if server.name == server_name:
+ break
+
+ address = server.addresses[
+ network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
+
+ return address
+
+ def get_mac_address(self, server_name, network_name):
+ nova_client = self.get_nova_client()
+ servers_list = nova_client.servers.list()
+ server = None
+
+ for server in servers_list:
+ if server.name == server_name:
+ break
+
+ mac_address = server.addresses[network_name][
+ NOVA_CILENT_NETWORK_INFO_INDEX][
+ "OS-EXT-IPS-MAC:mac_addr"]
+
+ return mac_address
+
+ def reboot_vm(self, server_name):
+ nova_client = self.get_nova_client()
+ servers_list = nova_client.servers.list()
+ server = None
+
+ for server in servers_list:
+ if server.name == server_name:
+ break
+
+ server.reboot()
+
+ return
+
+ def delete_vm(self, server_name):
+ nova_client = self.get_nova_client()
+ servers_list = nova_client.servers.list()
+ server = None
+
+ for server in servers_list:
+ if server.name == server_name:
+ nova_client.servers.delete(server)
+ break
+
+ return
+
+ def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
+ url = "http://%s/deployments/%s/outputs" % (
+ cfy_manager_ip, deployment_name)
+
+ response = requests.get(
+ url,
+ auth=HTTPBasicAuth('admin', 'admin'),
+ headers={'Tenant': 'default_tenant'})
+
+ resp_data = response.json()
+ self.logger.debug(resp_data)
+ data = resp_data["outputs"]
+ return data
+
+ def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
+ outputs = self.get_blueprint_outputs(cfy_manager_ip,
+ deployment_name)
+ vnfs = outputs["vnfs"]
+ vnf_list = []
+ for vnf_name in vnfs:
+ vnf_list.append(vnfs[vnf_name])
+ return vnf_list
+
+ def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
+ outputs = self.get_blueprint_outputs(cfy_manager_ip,
+ deployment_name)
+ networks = outputs["networks"]
+ network_list = []
+ for network_name in networks:
+ network_list.append(networks[network_name])
+ return network_list
+
+ def request_vnf_reboot(self, vnf_info_list):
+ for vnf in vnf_info_list:
+ self.logger.debug("reboot the " + vnf["vnf_name"])
+ self.reboot_vm(vnf["vnf_name"])
+
+ def request_vm_delete(self, vnf_info_list):
+ for vnf in vnf_info_list:
+ self.logger.debug("delete the " + vnf["vnf_name"])
+ self.delete_vm(vnf["vnf_name"])
+
+ def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
+ target_vnf_name):
+ network_list = self.get_blueprint_outputs_networks(
+ cfy_manager_ip,
+ topology_deploy_name)
+ vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
+ topology_deploy_name)
+ for vnf in vnf_info_list:
+ vnf_name = vnf["vnf_name"]
+ vnf["os_type"] = self.image["os_type"]
+ 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
+
+ self.logger.debug("vnf name : " + vnf_name)
+ self.logger.debug(vnf_name + " floating ip address : " +
+ vnf["floating_ip"])
+
+ for network in network_list:
+ network_name = network["network_name"]
+ ip_address = self.get_address(vnf["vnf_name"],
+ network["network_name"])
+ vnf[network_name + "_ip"] = ip_address
+ mac = self.get_mac_address(vnf["vnf_name"],
+ network["network_name"])
+ vnf[network_name + "_mac"] = mac
+
+ self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
+ " : " + vnf[network_name + "_ip"])
+ self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
+ " : " + vnf[network_name + "_mac"])
+
+ return vnf_info_list
+
+ def get_target_vnf(self, 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):
+ reference_vnf_list = []
+ for vnf in vnf_info_list:
+ if not vnf["target_vnf_flag"]:
+ reference_vnf_list.append(vnf)
+
+ return reference_vnf_list
+
+ def get_vnf_info(self, 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):
+ result = {}
+ for result_data in result_data_list:
+ test_kind = result_data["test_kind"]
+ protocol = result_data["protocol"]
+ test_result_data = result_data["result"]
+
+ if test_kind not in result:
+ result[test_kind] = []
+
+ result[test_kind].append({protocol: test_result_data})
+
+ return {"Functional_test": result}
+
+ def write_result_data(self, result_data):
+ test_result = []
+ if not os.path.isfile(self.test_result_json_file):
+ file_fd = open(self.test_result_json_file, "w")
+ file_fd.close()
+ else:
+ file_fd = open(self.test_result_json_file, "r")
+ test_result = json.load(file_fd)
+ file_fd.close()
+
+ test_result.append(result_data)
+
+ file_fd = open(self.test_result_json_file, "w")
+ json.dump(test_result, file_fd)
+ file_fd.close()
+
+ def output_test_result_json(self):
+ if os.path.isfile(self.test_result_json_file):
+ file_fd = open(self.test_result_json_file, "r")
+ test_result = json.load(file_fd)
+ file_fd.close()
+ output_json_data = json.dumps(test_result,
+ sort_keys=True,
+ indent=4)
+ self.logger.debug("test_result %s" % output_json_data)
+ else:
+ self.logger.debug("Not found %s" % self.test_result_json_file)
+
+ def get_test_scenario(self, file_path):
+ test_scenario_file = open(file_path,
+ 'r')
+ test_scenario_yaml = yaml.safe_load(test_scenario_file)
+ test_scenario_file.close()
+ return test_scenario_yaml["test_scenario_list"]
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""vrouter test result check module"""
+
+import json
+import logging
+import re
+
+from jinja2 import Environment, FileSystemLoader
+
+
+class Checker(object):
+ """vrouter test result check class"""
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self):
+ self.logger.debug("init checker")
+
+ def load_check_rule(self, rule_file_dir, rule_file_name, parameter):
+ loader = FileSystemLoader(rule_file_dir,
+ encoding='utf8')
+ env = Environment(loader=loader)
+ check_rule_template = env.get_template(rule_file_name)
+ check_rule = check_rule_template.render(parameter)
+ check_rule_data = json.loads(check_rule)
+ return check_rule_data
+
+ def regexp_information(self, response, rules):
+ status = False
+ result_data = {}
+
+ for rule in rules["rules"]:
+ result_data = {
+ "test_name": rule["description"],
+ "result": "NG"
+ }
+
+ match = re.search(rule["regexp"],
+ response)
+ rule["response"] = response
+ if match is None:
+ status = False
+ break
+
+ if not match.group(1) == rule["result"]:
+ status = False
+ else:
+ result_data["result"] = "OK"
+ status = True
+
+ return status, result_data
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""command generator module for vrouter testing"""
+
+import logging
+from jinja2 import Environment, FileSystemLoader
+
+
+class CommandGenerator(object):
+ """command generator class for vrouter testing"""
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self):
+ self.logger.debug("init command generator")
+
+ def load_template(self, template_dir, template):
+ loader = FileSystemLoader(template_dir,
+ encoding='utf8')
+ env = Environment(loader=loader)
+ return env.get_template(template)
+
+ def command_create(self, template, parameter):
+ commands = template.render(parameter)
+ return commands.split('\n')
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""ssh client module for vrouter testing"""
+
+import logging
+import paramiko
+import time
+import yaml
+
+from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
+
+RECEIVE_ROOP_WAIT = 1
+
+DEFAULT_CONNECT_TIMEOUT = 10
+DEFAULT_CONNECT_RETRY_COUNT = 10
+DEFAULT_SEND_TIMEOUT = 10
+
+
+class SshClient(object):
+ """ssh client class for vrouter testing"""
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self, ip_address, user, password=None, key_filename=None):
+ self.ip_address = ip_address
+ self.user = user
+ self.password = password
+ self.key_filename = key_filename
+ self.connected = False
+ self.shell = None
+
+ self.logger.setLevel(logging.INFO)
+
+ self.ssh = paramiko.SSHClient()
+ self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+ self.util = Utilvnf()
+ with open(self.util.test_env_config_yaml) as file_fd:
+ test_env_config_yaml = yaml.safe_load(file_fd)
+ file_fd.close()
+
+ self.ssh_revieve_buff = test_env_config_yaml.get("general").get(
+ "ssh_receive_buffer")
+
+ def connect(self, time_out=DEFAULT_CONNECT_TIMEOUT,
+ retrycount=DEFAULT_CONNECT_RETRY_COUNT):
+ while retrycount > 0:
+ try:
+ self.logger.info("SSH connect to %s.", self.ip_address)
+ self.ssh.connect(self.ip_address,
+ username=self.user,
+ password=self.password,
+ key_filename=self.key_filename,
+ timeout=time_out,
+ look_for_keys=False,
+ allow_agent=False)
+
+ self.logger.info("SSH connection established to %s.",
+ self.ip_address)
+
+ self.shell = self.ssh.invoke_shell()
+
+ while not self.shell.recv_ready():
+ time.sleep(RECEIVE_ROOP_WAIT)
+
+ self.shell.recv(self.ssh_revieve_buff)
+ break
+ except: # pylint: disable=broad-except
+ self.logger.info("SSH timeout for %s...", self.ip_address)
+ time.sleep(time_out)
+ retrycount -= 1
+
+ if retrycount == 0:
+ self.logger.error("Cannot establish connection to IP '%s'. " +
+ "Aborting",
+ self.ip_address)
+ self.connected = False
+ return self.connected
+
+ self.connected = True
+ return self.connected
+
+ def send(self, cmd, prompt, timeout=DEFAULT_SEND_TIMEOUT):
+ 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
+ self.logger.error("ssh send timeout : Command : '%s'", cmd)
+ return None
+
+ res_buff = ''
+ while not res_buff.endswith(prompt):
+ time.sleep(RECEIVE_ROOP_WAIT)
+ try:
+ res = self.shell.recv(self.ssh_revieve_buff)
+ except: # pylint: disable=broad-except
+ self.logger.error("ssh receive timeout : Command : '%s'",
+ cmd)
+ break
+
+ res_buff += res
+
+ self.logger.debug("Response : '%s'", res_buff)
+ return res_buff
+ else:
+ self.logger.error("Cannot connected to IP '%s'.", self.ip_address)
+ return None
+
+ def close(self):
+ if self.connected is True:
+ self.ssh.close()
+
+ def error_check(response, err_strs=["error",
+ "warn",
+ "unknown command",
+ "already exist"]):
+ for err in err_strs:
+ if err in response:
+ return False
+
+ return True
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""vm controll module"""
+
+import logging
+import os
+import time
+import yaml
+
+from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
+from functest.opnfv_tests.vnf.router.vnf_controller.command_generator import (
+ CommandGenerator)
+from functest.opnfv_tests.vnf.router.vnf_controller.ssh_client import (
+ SshClient)
+
+
+class VmController(object):
+ """vm controll class"""
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self, util_info):
+ self.logger.debug("initialize vm controller")
+ self.command_gen = CommandGenerator()
+ credentials = util_info["credentials"]
+
+ self.util = Utilvnf()
+ self.util.set_credentials(credentials["username"],
+ credentials["password"],
+ credentials["auth_url"],
+ credentials["tenant_name"],
+ credentials["region_name"])
+
+ with open(self.util.test_env_config_yaml) as file_fd:
+ test_env_config_yaml = yaml.safe_load(file_fd)
+ file_fd.close()
+
+ self.reboot_wait = test_env_config_yaml.get("general").get(
+ "reboot_wait")
+ self.command_wait = test_env_config_yaml.get("general").get(
+ "command_wait")
+ self.ssh_connect_timeout = test_env_config_yaml.get("general").get(
+ "ssh_connect_timeout")
+ self.ssh_connect_retry_count = test_env_config_yaml.get("general").get(
+ "ssh_connect_retry_count")
+
+ def command_gen_from_template(self, command_file_path, cmd_input_param):
+ (command_file_dir, command_file_name) = os.path.split(
+ command_file_path)
+ template = self.command_gen.load_template(command_file_dir,
+ command_file_name)
+ return self.command_gen.command_create(template,
+ cmd_input_param)
+
+ def config_vm(self, vm_info, test_cmd_file_path,
+ cmd_input_param, prompt_file_path):
+ ssh = self.connect_ssh_and_config_vm(vm_info,
+ test_cmd_file_path,
+ cmd_input_param,
+ prompt_file_path)
+ if ssh is None:
+ return False
+
+ ssh.close()
+
+ return True
+
+ def connect_ssh_and_config_vm(self, vm_info, test_cmd_file_path,
+ cmd_input_param, prompt_file_path):
+
+ key_filename = None
+ if "key_path" in vm_info:
+ key_filename = vm_info["key_path"]
+
+ ssh = SshClient(ip_address=vm_info["floating_ip"],
+ user=vm_info["user"],
+ password=vm_info["pass"],
+ key_filename=key_filename)
+
+ result = ssh.connect(self.ssh_connect_timeout,
+ self.ssh_connect_retry_count)
+ if not result:
+ self.logger.debug("try to vm reboot.")
+ self.util.reboot_vm(vm_info["vnf_name"])
+ time.sleep(self.reboot_wait)
+ result = ssh.connect(self.ssh_connect_timeout,
+ self.ssh_connect_retry_count)
+ if not result:
+ return None
+
+ (result, _) = self.command_create_and_execute(
+ ssh,
+ test_cmd_file_path,
+ cmd_input_param,
+ prompt_file_path)
+ if not result:
+ ssh.close()
+ return None
+
+ return ssh
+
+ def command_create_and_execute(self, ssh, test_cmd_file_path,
+ cmd_input_param, prompt_file_path):
+ prompt_file = open(prompt_file_path,
+ 'r')
+ prompt = yaml.safe_load(prompt_file)
+ prompt_file.close()
+ config_mode_prompt = prompt["config_mode"]
+
+ commands = self.command_gen_from_template(test_cmd_file_path,
+ cmd_input_param)
+ return self.command_list_execute(ssh,
+ commands,
+ config_mode_prompt)
+
+ def command_list_execute(self, ssh, command_list, prompt):
+ res_data_list = []
+ for command in command_list:
+ self.logger.debug("Command : " + command)
+ (res, res_data) = self.command_execute(ssh,
+ command,
+ prompt)
+ self.logger.debug("Response : " + res_data)
+ res_data_list.append(res_data)
+ if not res:
+ return res, res_data_list
+
+ time.sleep(self.command_wait)
+
+ return True, res_data_list
+
+ def command_execute(self, ssh, command, prompt):
+ res_data = ssh.send(command, prompt)
+ if res_data is None:
+ self.logger.info("retry send command : " + command)
+ res_data = ssh.send(command,
+ prompt)
+ if not ssh.error_check(res_data):
+ return False, res_data
+
+ return True, res_data
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""vrouter controll module"""
+
+import logging
+import os
+import prettytable
+import time
+import yaml
+
+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 (
+ SshClient)
+from functest.opnfv_tests.vnf.router.vnf_controller.vm_controller import (
+ VmController)
+
+
+class VnfController(object):
+ """vrouter controll class"""
+
+ logger = logging.getLogger(__name__)
+
+ def __init__(self, util_info):
+ self.logger.debug("init vnf controller")
+ self.util = Utilvnf()
+ self.vm_controller = VmController(util_info)
+
+ with open(self.util.test_env_config_yaml) as file_fd:
+ test_env_config_yaml = yaml.safe_load(file_fd)
+ file_fd.close()
+
+ self.cmd_wait = test_env_config_yaml.get("general").get("command_wait")
+ self.ssh_connect_timeout = test_env_config_yaml.get("general").get(
+ "ssh_connect_timeout")
+ self.ssh_connect_retry_count = test_env_config_yaml.get("general").get(
+ "ssh_connect_retry_count")
+
+ def config_vnf(self, source_vnf, destination_vnf, test_cmd_file_path,
+ parameter_file_path, prompt_file_path):
+ parameter_file = open(parameter_file_path,
+ 'r')
+ cmd_input_param = yaml.safe_load(parameter_file)
+ parameter_file.close()
+
+ cmd_input_param["macaddress"] = source_vnf["data_plane_network_mac"]
+ cmd_input_param["source_ip"] = source_vnf["data_plane_network_ip"]
+ cmd_input_param["destination_ip"] = destination_vnf[
+ "data_plane_network_ip"]
+
+ return self.vm_controller.config_vm(source_vnf,
+ test_cmd_file_path,
+ cmd_input_param,
+ prompt_file_path)
+
+ def result_check(self, target_vnf, reference_vnf,
+ check_rule_file_path_list, parameter_file_path,
+ prompt_file_path):
+
+ res_dict_data_list = []
+
+ parameter_file = open(parameter_file_path,
+ 'r')
+ cmd_input_param = yaml.safe_load(parameter_file)
+ parameter_file.close()
+
+ cmd_input_param["source_ip"] = target_vnf["data_plane_network_ip"]
+ cmd_input_param["destination_ip"] = reference_vnf[
+ "data_plane_network_ip"]
+
+ prompt_file = open(prompt_file_path,
+ 'r')
+ prompt = yaml.safe_load(prompt_file)
+ prompt_file.close()
+ terminal_mode_prompt = prompt["terminal_mode"]
+
+ ssh = SshClient(target_vnf["floating_ip"],
+ target_vnf["user"],
+ target_vnf["pass"])
+
+ result = ssh.connect(self.ssh_connect_timeout,
+ self.ssh_connect_retry_count)
+ if not result:
+ return False, res_dict_data_list
+
+ checker = Checker()
+
+ res_table = prettytable.PrettyTable(
+ header_style='upper', padding_width=5,
+ field_names=['test item', 'result'])
+
+ status = True
+ res_data_list = []
+ for check_rule_file_path in check_rule_file_path_list:
+ (check_rule_dir, check_rule_file) = os.path.split(
+ check_rule_file_path)
+ check_rules = checker.load_check_rule(check_rule_dir,
+ check_rule_file,
+ cmd_input_param)
+ (res, res_data) = self.vm_controller.command_execute(
+ ssh,
+ check_rules["command"],
+ terminal_mode_prompt)
+ res_data_list.append(res_data)
+ if not res:
+ status = False
+ break
+
+ (res, res_dict_data) = checker.regexp_information(res_data,
+ check_rules)
+ res_dict_data_list.append(res_dict_data)
+ res_table.add_row([res_dict_data["test_name"],
+ res_dict_data["result"]])
+ if not res:
+ status = False
+
+ time.sleep(self.cmd_wait)
+
+ ssh.close()
+
+ self.logger.info("Test result:\n\n%s\n", res_table.get_string())
+
+ self.output_check_result_detail_data(res_data_list)
+
+ return status, res_dict_data_list
+
+ def output_check_result_detail_data(self, res_data_list):
+ for res_data in res_data_list:
+ self.logger.debug(res_data)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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
+
+"""vrouter testing base class module"""
+
+import datetime
+import json
+import logging
+import os
+import pkg_resources
+import time
+
+import functest.core.vnf as vnf
+from functest.utils.constants import CONST
+from functest.opnfv_tests.vnf.router.test_controller import function_test_exec
+from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
+
+__author__ = "Shuya Nakama <shuya.nakama@okinawaopenlabs.org>"
+
+REBOOT_WAIT = 30
+
+
+class VrouterOnBoardingBase(vnf.VnfOnBoarding):
+ """vrouter testing base class"""
+
+ def __init__(self, **kwargs):
+ self.logger = logging.getLogger(__name__)
+ super(VrouterOnBoardingBase, self).__init__(**kwargs)
+ self.case_dir = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/vnf/router')
+ self.data_dir = CONST.__getattribute__('dir_router_data')
+ self.result_dir = os.path.join(CONST.__getattribute__('dir_results'),
+ self.case_name)
+ self.util = Utilvnf()
+ self.util_info = {}
+
+ self.vnf_list = []
+
+ if not os.path.exists(self.data_dir):
+ os.makedirs(self.data_dir)
+ if not os.path.exists(self.result_dir):
+ os.makedirs(self.result_dir)
+
+ def test_vnf(self):
+ """vrouter test execution"""
+ result = False
+ test_result_data_list = []
+ test_scenario_file_path = os.path.join(self.case_dir,
+ self.util.test_scenario_yaml)
+ test_scenario_list = self.util.get_test_scenario(
+ test_scenario_file_path)
+ for test_scenario in test_scenario_list:
+ if test_scenario["test_type"] == "function_test":
+ function_test_list = test_scenario["function_test_list"]
+ for function_test in function_test_list:
+ test_list = function_test["test_list"]
+ target_vnf_name = function_test["target_vnf_name"]
+ for test_info in test_list:
+ self.logger.info(test_info["protocol"] + " " +
+ test_info["test_kind"] +
+ " test.")
+ (result, result_data) = self.function_test_vrouter(
+ target_vnf_name,
+ test_info)
+ test_result_data_list.append(result_data)
+ if not result:
+ break
+
+ self.util.request_vm_delete(self.vnf_list)
+
+ test_result_data = json.dumps(test_result_data_list, indent=4)
+
+ return result, test_result_data
+
+ def function_test_vrouter(self, target_vnf_name, test_info):
+ """function test execution"""
+
+ test_protocol = test_info["protocol"]
+ test_list = test_info[test_protocol]
+
+ vnf_info_list = self.get_vnf_info_list(target_vnf_name)
+ self.vnf_list = vnf_info_list
+
+ self.logger.debug("request vnf's reboot.")
+ self.util.request_vnf_reboot(vnf_info_list)
+ time.sleep(REBOOT_WAIT)
+
+ target_vnf = self.util.get_target_vnf(vnf_info_list)
+
+ reference_vnf_list = self.util.get_reference_vnf_list(vnf_info_list)
+
+ test_exec = function_test_exec.FunctionTestExec(self.util_info)
+
+ # start test
+ start_time_ts = time.time()
+ self.logger.info("vRouter test Start Time:'%s'", (
+ datetime.datetime.fromtimestamp(start_time_ts).strftime(
+ '%Y-%m-%d %H:%M:%S')))
+
+ (result, test_result_data) = test_exec.run(target_vnf,
+ reference_vnf_list,
+ test_info,
+ test_list)
+
+ end_time_ts = time.time()
+ duration = round(end_time_ts - start_time_ts, 1)
+ self.logger.info("vRouter test duration :'%s'", duration)
+
+ return result, test_result_data
+
+ def get_vnf_info_list(self, target_vnf_name):
+ vnf_info_list = []
+ return vnf_info_list
+++ /dev/null
-#!/usr/bin/env python
-#
-# Copyright 2017 Okinawa Open Laboratory
-#
-# All rights reserved. This program and the accompanying materials
-# 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 functest.core.feature as base
-import json
-import os
-
-RESULT_DETAILS_FILE = "test_result.json"
-
-
-class VrouterVnf(base.Feature):
- def __init__(self, **kwargs):
- kwargs["repo"] = 'dir_repo_vrouter'
- if "case_name" not in kwargs:
- kwargs["case_name"] = "vyos_vrouter"
- super(VrouterVnf, self).__init__(**kwargs)
- self.cmd = 'cd %s && ./run.sh' % self.repo
-
- def set_result_details(self):
- filepath = os.path.join(self.repo, RESULT_DETAILS_FILE)
- if os.path.exists(filepath):
- f = open(filepath, 'r')
- self.details = json.load(f)
- f.close()
-
- def log_results(self):
- if self.result == 'PASS':
- self.set_result_details()
- super(VrouterVnf, self).log_results()
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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 logging
+import unittest
+
+import mock
+
+from functest.core import vnf
+from functest.opnfv_tests.vnf.router import cloudify_vrouter
+
+from snaps.openstack.os_credentials import OSCreds
+
+
+class CloudifyVrouterTesting(unittest.TestCase):
+
+ def setUp(self):
+
+ self.tenant = 'cloudify_vrouter'
+ self.creds = {'username': 'user',
+ 'password': 'pwd'}
+ self.orchestrator = {'name': 'cloudify',
+ 'version': '4.0',
+ 'object': 'foo',
+ 'requirements': {'flavor': {'name': 'm1.medium',
+ 'ram_min': 4096},
+ 'os_image': 'manager_4.0'}}
+
+ self.vnf = {'name': 'vrouter',
+ 'descriptor': {'version': '100',
+ 'file_name': 'function-test-' +
+ 'openstack-blueprint.yaml',
+ 'name': 'vrouter-opnfv',
+ 'url': 'https://foo',
+ 'requirements': {'flavor':
+ {'name': 'm1.medium',
+ 'ram_min': 2048}}}}
+
+ with mock.patch('functest.opnfv_tests.vnf.router.cloudify_vrouter.'
+ '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()
+
+ self.images = {'image1': 'url1',
+ 'image2': 'url2'}
+ self.details = {'orchestrator': {'status': 'PASS', 'duration': 120},
+ 'vnf': {},
+ 'test_vnf': {}}
+
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ return_value='test')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ return_value={'auth_url': 'test/v1'})
+ @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
+ def test_prepare_default(self, *args):
+ self.assertIsNone(self.router_vnf.prepare())
+ args[4].assert_called_once_with()
+
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ return_value='test')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ return_value={'auth_url': 'test/no_v'})
+ @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
+ def test_prepare_bad_auth_url(self, *args):
+ with self.assertRaises(Exception):
+ self.router_vnf.image_creator(
+ OSCreds(username='user', password='pass', auth_url='url',
+ project_name='project', identity_api_version=3),
+ mock.Mock())
+ args[0].assert_not_called()
+
+ def test_prepare_missing_param(self):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.router_vnf.prepare()
+
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ side_effect=Exception)
+ def test_prepare_keystone_exception(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.router_vnf.prepare()
+ args[0].assert_called_once_with()
+
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ return_value='test')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ side_effect=Exception)
+ def test_prepare_tenant_exception(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.router_vnf.prepare()
+ args[1].assert_called_once_with()
+
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ return_value='test')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
+ side_effect=Exception)
+ def test_prepare_user_exception(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.router_vnf.prepare()
+ args[2].assert_called_once_with()
+
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ return_value='test')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
+ return_value=True)
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ side_effect=Exception)
+ def test_prepare_credentials_exception(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.router_vnf.prepare()
+ args[0].assert_called_once_with()
+
+
+if __name__ == "__main__":
+ logging.disable(logging.CRITICAL)
+ unittest.main(verbosity=2)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Okinawa Open Laboratory and others.
+#
+# All rights reserved. This program and the accompanying materials
+# 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 logging
+import unittest
+
+import mock
+
+from functest.opnfv_tests.vnf.router import vrouter_base
+
+
+class VrouterOnBoardingBaseTesting(unittest.TestCase):
+
+ def setUp(self):
+ with mock.patch('functest.opnfv_tests.vnf.router.cloudify_vrouter.'
+ 'os.makedirs'):
+ self.vrouter_vnf = vrouter_base.VrouterOnBoardingBase()
+
+
+if __name__ == "__main__":
+ logging.disable(logging.CRITICAL)
+ unittest.main(verbosity=2)
Code Review / functest-xtesting.git / commitdiff