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
12 from scp import SCPClient
15 from snaps.openstack.create_instance import FloatingIpSettings, \
17 from snaps.openstack.create_keypairs import KeypairSettings
18 from snaps.openstack.create_network import PortSettings
19 from snaps.openstack.create_router import RouterSettings
20 from snaps.openstack.create_security_group import Direction, Protocol, \
21 SecurityGroupSettings, SecurityGroupRuleSettings
22 from snaps.openstack.utils import deploy_utils
24 from functest.core.testcase import TestCase
25 from functest.opnfv_tests.openstack.snaps import snaps_utils
26 from functest.utils.constants import CONST
30 class VPingSSH(vping_base.VPingBase):
32 Class to execute the vPing test using a Floating IP to connect to one VM
33 to issue the ping command to the second
36 def __init__(self, **kwargs):
38 # This line is here simply for pep8 as the 'os' package import appears
39 # to be required for mock and the unit tests will fail without it
42 if "case_name" not in kwargs:
43 kwargs["case_name"] = "vping_ssh"
44 super(VPingSSH, self).__init__(**kwargs)
46 self.kp_name = CONST.__getattribute__('vping_keypair_name') + self.guid
47 self.kp_priv_file = CONST.__getattribute__('vping_keypair_priv_file')
48 self.kp_pub_file = CONST.__getattribute__('vping_keypair_pub_file')
49 self.router_name = CONST.__getattribute__(
50 'vping_router_name') + self.guid
51 self.sg_name = CONST.__getattribute__('vping_sg_name') + self.guid
52 self.sg_desc = CONST.__getattribute__('vping_sg_desc')
56 Sets up the OpenStack keypair, router, security group, and VM instance
57 objects then validates the ping.
58 :return: the exit code from the super.execute() method
61 super(VPingSSH, self).run()
63 self.logger.info("Creating keypair with name: '%s'" % self.kp_name)
64 kp_creator = deploy_utils.create_keypair(
66 KeypairSettings(name=self.kp_name,
67 private_filepath=self.kp_priv_file,
68 public_filepath=self.kp_pub_file))
69 self.creators.append(kp_creator)
71 # Creating router to external network
72 self.logger.info("Creating router with name: '%s'"
74 net_set = self.network_creator.network_settings
75 sub_set = [net_set.subnet_settings[0].name]
76 ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
77 router_creator = deploy_utils.create_router(
80 name=self.router_name,
81 external_gateway=ext_net_name,
82 internal_subnets=sub_set))
83 self.creators.append(router_creator)
86 port1_settings = PortSettings(
87 name=self.vm1_name + '-vPingPort',
88 network_name=self.network_creator.network_settings.name)
89 instance1_settings = VmInstanceSettings(
90 name=self.vm1_name, flavor=self.flavor_name,
91 vm_boot_timeout=self.vm_boot_timeout,
92 vm_delete_timeout=self.vm_delete_timeout,
93 ssh_connect_timeout=self.vm_ssh_connect_timeout,
94 port_settings=[port1_settings])
97 "Creating VM 1 instance with name: '%s'"
98 % instance1_settings.name)
99 self.vm1_creator = deploy_utils.create_vm_instance(
102 self.image_creator.image_settings,
103 keypair_creator=kp_creator)
104 self.creators.append(self.vm1_creator)
106 # Creating Instance 2
107 sg_creator = self.__create_security_group()
108 self.creators.append(sg_creator)
110 port2_settings = PortSettings(
111 name=self.vm2_name + '-vPingPort',
112 network_name=self.network_creator.network_settings.name)
113 instance2_settings = VmInstanceSettings(
114 name=self.vm2_name, flavor=self.flavor_name,
115 vm_boot_timeout=self.vm_boot_timeout,
116 vm_delete_timeout=self.vm_delete_timeout,
117 ssh_connect_timeout=self.vm_ssh_connect_timeout,
118 port_settings=[port2_settings],
119 security_group_names=[sg_creator.sec_grp_settings.name],
120 floating_ip_settings=[FloatingIpSettings(
121 name=self.vm2_name + '-FIPName',
122 port_name=port2_settings.name,
123 router_name=router_creator.router_settings.name)])
126 "Creating VM 2 instance with name: '%s'"
127 % instance2_settings.name)
128 self.vm2_creator = deploy_utils.create_vm_instance(
131 self.image_creator.image_settings,
132 keypair_creator=kp_creator)
133 self.creators.append(self.vm2_creator)
135 return self._execute()
136 except Exception as e:
137 self.logger.error('Unexpected error running test - ' + e.message)
138 return TestCase.EX_RUN_ERROR
142 def _do_vping(self, vm_creator, test_ip):
146 if vm_creator.vm_ssh_active(block=True):
147 ssh = vm_creator.ssh_client()
148 if not self._transfer_ping_script(ssh):
149 return TestCase.EX_RUN_ERROR
150 return self._do_vping_ssh(ssh, test_ip)
154 def _transfer_ping_script(self, ssh):
156 Uses SCP to copy the ping script via the SSH client
157 :param ssh: the SSH client
160 self.logger.info("Trying to transfer ping.sh")
161 scp = SCPClient(ssh.get_transport())
162 ping_script = pkg_resources.resource_filename(
163 'functest.opnfv_tests.openstack.vping', 'ping.sh')
165 scp.put(ping_script, "~/")
167 self.logger.error("Cannot SCP the file '%s'" % ping_script)
170 cmd = 'chmod 755 ~/ping.sh'
171 (stdin, stdout, stderr) = ssh.exec_command(cmd)
172 for line in stdout.readlines():
177 def _do_vping_ssh(self, ssh, test_ip):
179 Pings the test_ip via the SSH client
180 :param ssh: the SSH client used to issue the ping command
181 :param test_ip: the IP for the ping command to use
182 :return: exit_code (int)
184 exit_code = TestCase.EX_TESTCASE_FAILED
185 self.logger.info("Waiting for ping...")
188 cmd = '~/ping.sh ' + test_ip
193 (stdin, stdout, stderr) = ssh.exec_command(cmd)
194 output = stdout.readlines()
197 if "vPing OK" in line:
198 self.logger.info("vPing detected!")
199 exit_code = TestCase.EX_OK
203 elif sec == self.ping_timeout:
204 self.logger.info("Timeout reached.")
209 self.logger.debug("Pinging %s. Waiting for response..." % test_ip)
213 def __create_security_group(self):
215 Configures and deploys an OpenStack security group object
216 :return: the creator object
220 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
221 direction=Direction.ingress,
222 protocol=Protocol.icmp))
224 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
225 direction=Direction.ingress,
226 protocol=Protocol.tcp, port_range_min=22,
229 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
230 direction=Direction.egress,
231 protocol=Protocol.tcp, port_range_min=22,
234 self.logger.info("Security group with name: '%s'" % self.sg_name)
235 return deploy_utils.create_security_group(self.os_creds,
236 SecurityGroupSettings(
238 description=self.sg_desc,
239 rule_settings=sg_rules))