-#!/usr/bin/python
+#!/usr/bin/env python
#
# Copyright (c) 2015 All rights reserved
# This program and the accompanying materials
#
# http://www.apache.org/licenses/LICENSE-2.0
-import os
-import re
-import sys
+
+"""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
-import argparse
-import paramiko
from scp import SCPClient
+import pkg_resources
+
+from functest.core.testcase import TestCase
+from functest.energy import energy
+from functest.opnfv_tests.openstack.snaps import snaps_utils
+from functest.opnfv_tests.openstack.vping import vping_base
+from functest.utils.constants import CONST
+from snaps.openstack.create_instance import FloatingIpSettings, \
+ VmInstanceSettings
-import functest.utils.functest_logger as ft_logger
-import functest.utils.openstack_utils as os_utils
-import vping_base
-import functest.core.testcase_base as testcase_base
+from snaps.openstack.create_keypairs import KeypairSettings
+from snaps.openstack.create_network import PortSettings
+from snaps.openstack.create_router import RouterSettings
+from snaps.openstack.create_security_group import Direction, Protocol, \
+ SecurityGroupSettings, SecurityGroupRuleSettings
+from snaps.openstack.utils import deploy_utils
class VPingSSH(vping_base.VPingBase):
+ """
+ VPingSSH testcase implementation.
- def __init__(self):
- super(VPingSSH, self).__init__()
- self.case_name = 'vping_ssh'
- self.logger = ft_logger.Logger(self.case_name).getLogger()
-
- def do_vping(self, vm, test_ip):
- floatip = self.add_float_ip(vm)
- if not floatip:
- return testcase_base.TestcaseBase.EX_RUN_ERROR
- ssh = self.establish_ssh(vm, floatip)
- if not ssh:
- return testcase_base.TestcaseBase.EX_RUN_ERROR
- if not self.transfer_ping_script(ssh, floatip):
- return testcase_base.TestcaseBase.EX_RUN_ERROR
- return self.do_vping_ssh(ssh, test_ip)
-
- def add_float_ip(self, vm):
- self.logger.info("Creating floating IP for VM '%s'..." % self.vm2_name)
- floatip_dic = os_utils.create_floating_ip(self.neutron_client)
- floatip = floatip_dic['fip_addr']
-
- if floatip is None:
- self.logger.error("Cannot create floating IP.")
- return None
- self.logger.info("Floating IP created: '%s'" % floatip)
-
- self.logger.info("Associating floating ip: '%s' to VM '%s' "
- % (floatip, self.vm2_name))
- if not os_utils.add_floating_ip(self.nova_client, vm.id, floatip):
- self.logger.error("Cannot associate floating IP to VM.")
- return None
-
- return floatip
-
- def establish_ssh(self, vm, floatip):
- self.logger.info("Trying to establish SSH connection to %s..."
- % floatip)
- username = 'cirros'
- password = 'cubswin:)'
- ssh = paramiko.SSHClient()
- ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
- timeout = 50
- nolease = False
- got_ip = False
- discover_count = 0
- cidr_first_octet = self.private_subnet_cidr.split('.')[0]
- while timeout > 0:
- try:
- ssh.connect(floatip, username=username,
- password=password, timeout=2)
- self.logger.debug("SSH connection established to %s."
- % floatip)
- break
- except:
- self.logger.debug("Waiting for %s..." % floatip)
- time.sleep(6)
- timeout -= 1
-
- console_log = vm.get_console_output()
-
- # print each "Sending discover" captured on the console log
- if (len(re.findall("Sending discover", console_log)) >
- discover_count and not got_ip):
- discover_count += 1
- self.logger.debug("Console-log '%s': Sending discover..."
- % self.vm2_name)
-
- # check if eth0 got an ip,the line looks like this:
- # "inet addr:192.168."....
- # if the dhcp agent fails to assing ip, this line will not appear
- if "inet addr:" + cidr_first_octet in console_log and not got_ip:
- got_ip = True
- self.logger.debug("The instance '%s' succeeded to get the IP "
- "from the dhcp agent." % self.vm2_name)
-
- # if dhcp not work,it shows "No lease, failing".The test will fail
- if "No lease, failing" in console_log \
- and not nolease \
- and not got_ip:
- nolease = True
- self.logger.debug("Console-log '%s': No lease, failing..."
- % self.vm2_name)
- self.logger.info("The instance failed to get an IP from DHCP "
- "agent. The test will probably timeout...")
-
- if timeout == 0: # 300 sec timeout (5 min)
- self.logger.error("Cannot establish connection to IP '%s'. "
- "Aborting" % floatip)
- return None
- return ssh
-
- def transfer_ping_script(self, ssh, floatip):
- self.logger.info("Trying to transfer ping.sh to %s..." % floatip)
+ Class to execute the vPing test using a Floating IP to connect to one VM
+ to issue the ping command to the second
+ """
+
+ def __init__(self, **kwargs):
+ """Initialize testcase."""
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = "vping_ssh"
+ super(VPingSSH, self).__init__(**kwargs)
+
+ self.kp_name = CONST.__getattribute__('vping_keypair_name') + self.guid
+ self.kp_priv_file = CONST.__getattribute__('vping_keypair_priv_file')
+ self.kp_pub_file = CONST.__getattribute__('vping_keypair_pub_file')
+ self.router_name = CONST.__getattribute__(
+ 'vping_router_name') + self.guid
+ self.sg_name = CONST.__getattribute__('vping_sg_name') + self.guid
+ self.sg_desc = CONST.__getattribute__('vping_sg_desc')
+
+ @energy.enable_recording
+ def run(self):
+ """
+ Excecute VPingSSH testcase.
+
+ Sets up the OpenStack keypair, router, security group, and VM instance
+ objects then validates the ping.
+ :return: the exit code from the super.execute() method
+ """
+ try:
+ super(VPingSSH, self).run()
+
+ log = "Creating keypair with name: '%s'" % self.kp_name
+ self.logger.info(log)
+ kp_creator = deploy_utils.create_keypair(
+ self.os_creds,
+ KeypairSettings(name=self.kp_name,
+ private_filepath=self.kp_priv_file,
+ public_filepath=self.kp_pub_file))
+ self.creators.append(kp_creator)
+
+ # Creating router to external network
+ log = "Creating router with name: '%s'" % self.router_name
+ self.logger.info(log)
+ net_set = self.network_creator.network_settings
+ sub_set = [net_set.subnet_settings[0].name]
+ ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
+ router_creator = deploy_utils.create_router(
+ self.os_creds,
+ RouterSettings(
+ name=self.router_name,
+ external_gateway=ext_net_name,
+ internal_subnets=sub_set))
+ self.creators.append(router_creator)
+
+ # Creating Instance 1
+ port1_settings = PortSettings(
+ name=self.vm1_name + '-vPingPort',
+ network_name=self.network_creator.network_settings.name)
+ instance1_settings = VmInstanceSettings(
+ name=self.vm1_name, flavor=self.flavor_name,
+ vm_boot_timeout=self.vm_boot_timeout,
+ vm_delete_timeout=self.vm_delete_timeout,
+ ssh_connect_timeout=self.vm_ssh_connect_timeout,
+ port_settings=[port1_settings])
+
+ log = ("Creating VM 1 instance with name: '%s'"
+ % instance1_settings.name)
+ self.logger.info(log)
+ self.vm1_creator = deploy_utils.create_vm_instance(
+ self.os_creds,
+ instance1_settings,
+ self.image_creator.image_settings,
+ keypair_creator=kp_creator)
+ self.creators.append(self.vm1_creator)
+
+ # Creating Instance 2
+ sg_creator = self.__create_security_group()
+ self.creators.append(sg_creator)
+
+ port2_settings = PortSettings(
+ name=self.vm2_name + '-vPingPort',
+ network_name=self.network_creator.network_settings.name)
+ instance2_settings = VmInstanceSettings(
+ name=self.vm2_name, flavor=self.flavor_name,
+ vm_boot_timeout=self.vm_boot_timeout,
+ vm_delete_timeout=self.vm_delete_timeout,
+ ssh_connect_timeout=self.vm_ssh_connect_timeout,
+ port_settings=[port2_settings],
+ security_group_names=[sg_creator.sec_grp_settings.name],
+ floating_ip_settings=[FloatingIpSettings(
+ name=self.vm2_name + '-FIPName',
+ port_name=port2_settings.name,
+ router_name=router_creator.router_settings.name)])
+
+ log = ("Creating VM 2 instance with name: '%s'"
+ % instance2_settings.name)
+ self.logger.info(log)
+ self.vm2_creator = deploy_utils.create_vm_instance(
+ self.os_creds,
+ instance2_settings,
+ self.image_creator.image_settings,
+ keypair_creator=kp_creator)
+ self.creators.append(self.vm2_creator)
+
+ return self._execute()
+ except Exception as exc: # pylint: disable=broad-except
+ self.logger.error('Unexpected error running test - ' + exc.message)
+ return TestCase.EX_RUN_ERROR
+ finally:
+ self._cleanup()
+
+ def _do_vping(self, vm_creator, test_ip):
+ """
+ Execute ping command.
+
+ Override from super
+ """
+ if vm_creator.vm_ssh_active(block=True):
+ ssh = vm_creator.ssh_client()
+ if not self._transfer_ping_script(ssh):
+ return TestCase.EX_RUN_ERROR
+ return self._do_vping_ssh(ssh, test_ip)
+ else:
+ return TestCase.EX_RUN_ERROR
+
+ def _transfer_ping_script(self, ssh):
+ """
+ Transfert vping script to VM.
+
+ Uses SCP to copy the ping script via the SSH client
+ :param ssh: the SSH client
+ :return:
+ """
+ self.logger.info("Trying to transfer ping.sh")
scp = SCPClient(ssh.get_transport())
- local_path = self.functest_repo + "/" + self.repo
- ping_script = os.path.join(local_path, "ping.sh")
+ ping_script = pkg_resources.resource_filename(
+ 'functest.opnfv_tests.openstack.vping', 'ping.sh')
try:
scp.put(ping_script, "~/")
- except:
- self.logger.error("Cannot SCP the file '%s' to VM '%s'"
- % (ping_script, floatip))
+ except Exception:
+ self.logger.error("Cannot SCP the file '%s'", ping_script)
return False
cmd = 'chmod 755 ~/ping.sh'
+ # pylint: disable=unused-variable
(stdin, stdout, stderr) = ssh.exec_command(cmd)
for line in stdout.readlines():
print line
return True
- def do_vping_ssh(self, ssh, test_ip):
- EXIT_CODE = -1
+ def _do_vping_ssh(self, ssh, test_ip):
+ """
+ Execute ping command via SSH.
+
+ Pings the test_ip via the SSH client
+ :param ssh: the SSH client used to issue the ping command
+ :param test_ip: the IP for the ping command to use
+ :return: exit_code (int)
+ """
+ exit_code = TestCase.EX_TESTCASE_FAILED
self.logger.info("Waiting for ping...")
sec = 0
while True:
time.sleep(1)
- (stdin, stdout, stderr) = ssh.exec_command(cmd)
+ (_, stdout, _) = ssh.exec_command(cmd)
output = stdout.readlines()
for line in output:
if "vPing OK" in line:
self.logger.info("vPing detected!")
- EXIT_CODE = 0
+ exit_code = TestCase.EX_OK
flag = True
break
break
if flag:
break
- self.logger.debug("Pinging %s. Waiting for response..." % test_ip)
+ log = "Pinging %s. Waiting for response..." % test_ip
+ self.logger.debug(log)
sec += 1
- return EXIT_CODE
+ return exit_code
+
+ def __create_security_group(self):
+ """
+ Configure OpenStack security groups.
+ Configures and deploys an OpenStack security group object
+ :return: the creator object
+ """
+ sg_rules = list()
+ sg_rules.append(
+ SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
+ direction=Direction.ingress,
+ protocol=Protocol.icmp))
+ sg_rules.append(
+ SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
+ direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=22,
+ port_range_max=22))
+ sg_rules.append(
+ SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
+ direction=Direction.egress,
+ protocol=Protocol.tcp, port_range_min=22,
+ port_range_max=22))
-if __name__ == '__main__':
- args_parser = argparse.ArgumentParser()
- args_parser.add_argument("-r", "--report",
- help="Create json result file",
- action="store_true")
- args = vars(args_parser.parse_args())
- sys.exit(vping_base.VPingMain(VPingSSH).main(**args))
+ log = "Security group with name: '%s'" % self.sg_name
+ self.logger.info(log)
+ return deploy_utils.create_security_group(self.os_creds,
+ SecurityGroupSettings(
+ name=self.sg_name,
+ description=self.sg_desc,
+ rule_settings=sg_rules))
Code Review / functest.git / blobdiff