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
  14 from functest.core import singlevm
  15 from functest.utils import config
  16
  17
  18 class VPingSSH(singlevm.SingleVm2):
  19     """
  20     VPingSSH testcase implementation.
  21
  22     Class to execute the vPing test using a Floating IP to connect to one VM
  23     to issue the ping command to the second
  24     """
  25
  26     __logger = logging.getLogger(__name__)
  27
  28     def __init__(self, **kwargs):
  29         """Initialize testcase."""
  30         if "case_name" not in kwargs:
  31             kwargs["case_name"] = "vping_ssh"
  32         super().__init__(**kwargs)
  33         self.vm2 = None
  34
  35     def prepare(self):
  36         super().prepare()
  37         self.vm2 = self.boot_vm(
  38             '{}-vm2_{}'.format(self.case_name, self.guid),
  39             security_groups=[self.sec.id])
  40
  41     def execute(self):
  42         """Ping the second VM
  43
  44         Returns: ping exit codes
  45         """
  46         assert self.ssh
  47         if not self.check_regex_in_console(self.vm2.name):
  48             return 1
  49         (_, stdout, stderr) = self.ssh.exec_command(
  50             'ping -c 1 {}'.format(
  51                 self.vm2.private_v4 or self.vm2.addresses[
  52                     self.network.name][0].addr))
  53         self.__logger.info("output:\n%s", stdout.read().decode("utf-8"))
  54         self.__logger.info("error:\n%s", stderr.read().decode("utf-8"))
  55         return stdout.channel.recv_exit_status()
  56
  57     def clean(self):
  58         assert self.cloud
  59         if self.vm2:
  60             self.cloud.delete_server(
  61                 self.vm2, wait=True,
  62                 timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
  63         super().clean()