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
8 # http://www.apache.org/licenses/LICENSE-2.0
11 """vPingSSH testcase."""
13 # This 1st import is here simply for pep8 as the 'os' package import appears
14 # to be required for mock and the unit tests will fail without it
15 import os # noqa # pylint: disable=unused-import
18 from scp import SCPClient
21 from functest.core.testcase import TestCase
22 from functest.energy import energy
23 from functest.opnfv_tests.openstack.snaps import snaps_utils
24 from functest.opnfv_tests.openstack.vping import vping_base
25 from functest.utils.constants import CONST
26 from snaps.openstack.create_instance import FloatingIpSettings, \
29 from snaps.openstack.create_keypairs import KeypairSettings
30 from snaps.openstack.create_network import PortSettings
31 from snaps.openstack.create_router import RouterSettings
32 from snaps.openstack.create_security_group import Direction, Protocol, \
33 SecurityGroupSettings, SecurityGroupRuleSettings
34 from snaps.openstack.utils import deploy_utils
37 class VPingSSH(vping_base.VPingBase):
39 VPingSSH testcase implementation.
41 Class to execute the vPing test using a Floating IP to connect to one VM
42 to issue the ping command to the second
45 def __init__(self, **kwargs):
46 """Initialize testcase."""
47 if "case_name" not in kwargs:
48 kwargs["case_name"] = "vping_ssh"
49 super(VPingSSH, self).__init__(**kwargs)
51 self.kp_name = CONST.__getattribute__('vping_keypair_name') + self.guid
52 self.kp_priv_file = CONST.__getattribute__('vping_keypair_priv_file')
53 self.kp_pub_file = CONST.__getattribute__('vping_keypair_pub_file')
54 self.router_name = CONST.__getattribute__(
55 'vping_router_name') + self.guid
56 self.sg_name = CONST.__getattribute__('vping_sg_name') + self.guid
57 self.sg_desc = CONST.__getattribute__('vping_sg_desc')
59 @energy.enable_recording
62 Excecute VPingSSH testcase.
64 Sets up the OpenStack keypair, router, security group, and VM instance
65 objects then validates the ping.
66 :return: the exit code from the super.execute() method
69 super(VPingSSH, self).run()
71 log = "Creating keypair with name: '%s'" % self.kp_name
73 kp_creator = deploy_utils.create_keypair(
75 KeypairSettings(name=self.kp_name,
76 private_filepath=self.kp_priv_file,
77 public_filepath=self.kp_pub_file))
78 self.creators.append(kp_creator)
80 # Creating router to external network
81 log = "Creating router with name: '%s'" % self.router_name
83 net_set = self.network_creator.network_settings
84 sub_set = [net_set.subnet_settings[0].name]
85 ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
86 router_creator = deploy_utils.create_router(
89 name=self.router_name,
90 external_gateway=ext_net_name,
91 internal_subnets=sub_set))
92 self.creators.append(router_creator)
95 port1_settings = PortSettings(
96 name=self.vm1_name + '-vPingPort',
97 network_name=self.network_creator.network_settings.name)
98 instance1_settings = VmInstanceSettings(
99 name=self.vm1_name, flavor=self.flavor_name,
100 vm_boot_timeout=self.vm_boot_timeout,
101 vm_delete_timeout=self.vm_delete_timeout,
102 ssh_connect_timeout=self.vm_ssh_connect_timeout,
103 port_settings=[port1_settings])
105 log = ("Creating VM 1 instance with name: '%s'"
106 % instance1_settings.name)
107 self.logger.info(log)
108 self.vm1_creator = deploy_utils.create_vm_instance(
111 self.image_creator.image_settings,
112 keypair_creator=kp_creator)
113 self.creators.append(self.vm1_creator)
115 # Creating Instance 2
116 sg_creator = self.__create_security_group()
117 self.creators.append(sg_creator)
119 port2_settings = PortSettings(
120 name=self.vm2_name + '-vPingPort',
121 network_name=self.network_creator.network_settings.name)
122 instance2_settings = VmInstanceSettings(
123 name=self.vm2_name, flavor=self.flavor_name,
124 vm_boot_timeout=self.vm_boot_timeout,
125 vm_delete_timeout=self.vm_delete_timeout,
126 ssh_connect_timeout=self.vm_ssh_connect_timeout,
127 port_settings=[port2_settings],
128 security_group_names=[sg_creator.sec_grp_settings.name],
129 floating_ip_settings=[FloatingIpSettings(
130 name=self.vm2_name + '-FIPName',
131 port_name=port2_settings.name,
132 router_name=router_creator.router_settings.name)])
134 log = ("Creating VM 2 instance with name: '%s'"
135 % instance2_settings.name)
136 self.logger.info(log)
137 self.vm2_creator = deploy_utils.create_vm_instance(
140 self.image_creator.image_settings,
141 keypair_creator=kp_creator)
142 self.creators.append(self.vm2_creator)
144 return self._execute()
145 except Exception as exc: # pylint: disable=broad-except
146 self.logger.error('Unexpected error running test - ' + exc.message)
147 return TestCase.EX_RUN_ERROR
151 def _do_vping(self, vm_creator, test_ip):
153 Execute ping command.
157 if vm_creator.vm_ssh_active(block=True):
158 ssh = vm_creator.ssh_client()
159 if not self._transfer_ping_script(ssh):
160 return TestCase.EX_RUN_ERROR
161 return self._do_vping_ssh(ssh, test_ip)
165 def _transfer_ping_script(self, ssh):
167 Transfert vping script to VM.
169 Uses SCP to copy the ping script via the SSH client
170 :param ssh: the SSH client
173 self.logger.info("Trying to transfer ping.sh")
174 scp = SCPClient(ssh.get_transport())
175 ping_script = pkg_resources.resource_filename(
176 'functest.opnfv_tests.openstack.vping', 'ping.sh')
178 scp.put(ping_script, "~/")
180 self.logger.error("Cannot SCP the file '%s'", ping_script)
183 cmd = 'chmod 755 ~/ping.sh'
184 # pylint: disable=unused-variable
185 (stdin, stdout, stderr) = ssh.exec_command(cmd)
186 for line in stdout.readlines():
191 def _do_vping_ssh(self, ssh, test_ip):
193 Execute ping command via SSH.
195 Pings the test_ip via the SSH client
196 :param ssh: the SSH client used to issue the ping command
197 :param test_ip: the IP for the ping command to use
198 :return: exit_code (int)
200 exit_code = TestCase.EX_TESTCASE_FAILED
201 self.logger.info("Waiting for ping...")
204 cmd = '~/ping.sh ' + test_ip
209 (_, stdout, _) = ssh.exec_command(cmd)
210 output = stdout.readlines()
213 if "vPing OK" in line:
214 self.logger.info("vPing detected!")
215 exit_code = TestCase.EX_OK
219 elif sec == self.ping_timeout:
220 self.logger.info("Timeout reached.")
225 log = "Pinging %s. Waiting for response..." % test_ip
226 self.logger.debug(log)
230 def __create_security_group(self):
232 Configure OpenStack security groups.
234 Configures and deploys an OpenStack security group object
235 :return: the creator object
239 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
240 direction=Direction.ingress,
241 protocol=Protocol.icmp))
243 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
244 direction=Direction.ingress,
245 protocol=Protocol.tcp, port_range_min=22,
248 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
249 direction=Direction.egress,
250 protocol=Protocol.tcp, port_range_min=22,
253 log = "Security group with name: '%s'" % self.sg_name
254 self.logger.info(log)
255 return deploy_utils.create_security_group(self.os_creds,
256 SecurityGroupSettings(
258 description=self.sg_desc,
259 rule_settings=sg_rules))