functest/opnfv_tests/openstack/vping/vping_ssh.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2015 All rights reserved
   4 # This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 #
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 """vPingSSH testcase."""
  11
  12 import logging
  13 import os
  14 import tempfile
  15 import time
  16
  17 import paramiko
  18 from xtesting.core import testcase
  19 from xtesting.energy import energy
  20
  21 from functest.opnfv_tests.openstack.vping import vping_base
  22 from functest.utils import config
  23
  24
  25 class VPingSSH(vping_base.VPingBase):
  26     """
  27     VPingSSH testcase implementation.
  28
  29     Class to execute the vPing test using a Floating IP to connect to one VM
  30     to issue the ping command to the second
  31     """
  32
  33     def __init__(self, **kwargs):
  34         """Initialize testcase."""
  35         if "case_name" not in kwargs:
  36             kwargs["case_name"] = "vping_ssh"
  37         super(VPingSSH, self).__init__(**kwargs)
  38         self.logger = logging.getLogger(__name__)
  39         self.vm2 = None
  40         self.sec2 = None
  41         self.fip = None
  42         self.keypair = None
  43         self.ssh = paramiko.SSHClient()
  44         (_, self.key_filename) = tempfile.mkstemp()
  45
  46     @energy.enable_recording
  47     def run(self, **kwargs):
  48         """
  49         Excecute VPingSSH testcase.
  50
  51         Sets up the OpenStack keypair, router, security group, and VM instance
  52         objects then validates the ping.
  53         :return: the exit code from the super.execute() method
  54         """
  55         try:
  56             assert self.cloud
  57             super(VPingSSH, self).run()
  58
  59             kp_name = getattr(config.CONF, 'vping_keypair_name') + self.guid
  60             self.logger.info("Creating keypair with name: '%s'", kp_name)
  61             self.keypair = self.cloud.create_keypair(kp_name)
  62             self.logger.debug("keypair: %s", self.keypair)
  63             self.logger.debug("private_key: %s", self.keypair.private_key)
  64             with open(self.key_filename, 'w') as private_key_file:
  65                 private_key_file.write(self.keypair.private_key)
  66
  67             self.sec2 = self.cloud.create_security_group(
  68                 getattr(config.CONF, 'vping_sg_name') + self.guid,
  69                 getattr(config.CONF, 'vping_sg_desc'))
  70             self.cloud.create_security_group_rule(
  71                 self.sec2.id, port_range_min='22', port_range_max='22',
  72                 protocol='tcp', direction='ingress')
  73             self.cloud.create_security_group_rule(
  74                 self.sec2.id, protocol='icmp', direction='ingress')
  75
  76             vm2_name = "{}-{}-{}".format(
  77                 getattr(config.CONF, 'vping_vm_name_2'), "ssh", self.guid)
  78             self.logger.info(
  79                 "Creating VM 2 instance with name: '%s'", vm2_name)
  80             self.vm2 = self.cloud.create_server(
  81                 vm2_name, image=self.image.id, flavor=self.flavor.id,
  82                 key_name=self.keypair.id,
  83                 auto_ip=False, wait=True,
  84                 timeout=getattr(config.CONF, 'vping_vm_boot_timeout'),
  85                 network=self.network.id,
  86                 security_groups=[self.sec2.id])
  87             self.logger.debug("vm2: %s", self.vm2)
  88             self.fip = self.cloud.create_floating_ip(
  89                 network=self.ext_net.id, server=self.vm2)
  90             self.logger.debug("floating_ip2: %s", self.fip)
  91             self.vm2 = self.cloud.wait_for_server(self.vm2, auto_ip=False)
  92             p_console = self.cloud.get_server_console(self.vm2.id)
  93             self.logger.debug("vm2 console: \n%s", p_console)
  94             return self._execute()
  95         except Exception:  # pylint: disable=broad-except
  96             self.logger.exception('Unexpected error running vping_ssh')
  97             return testcase.TestCase.EX_RUN_ERROR
  98
  99     def _do_vping(self):
 100         time.sleep(10)
 101         self.ssh.set_missing_host_key_policy(paramiko.client.AutoAddPolicy())
 102         self.ssh.connect(
 103             self.vm2.public_v4,
 104             username=getattr(config.CONF, 'openstack_image_user'),
 105             key_filename=self.key_filename,
 106             timeout=getattr(config.CONF, 'vping_vm_ssh_connect_timeout'))
 107         self.logger.debug("ssh: %s", self.ssh)
 108         (_, stdout, _) = self.ssh.exec_command(
 109             'ping -c 1 ' + self.vm1.private_v4)
 110         self.logger.debug("ping output: %s", stdout)
 111         return stdout.channel.recv_exit_status()
 112
 113     def clean(self):
 114         assert self.cloud
 115         os.remove(self.key_filename)
 116         self.cloud.delete_server(
 117             self.vm2, wait=True,
 118             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 119         self.cloud.delete_security_group(self.sec2.id)
 120         self.cloud.delete_keypair(self.keypair.id)
 121         self.cloud.delete_floating_ip(self.fip.id)
 122         super(VPingSSH, self).clean()