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
16 from snaps.openstack.create_instance import FloatingIpSettings, \
18 from snaps.openstack.create_keypairs import KeypairSettings
19 from snaps.openstack.create_network import PortSettings
20 from snaps.openstack.create_router import RouterSettings
21 from snaps.openstack.create_security_group import Direction, Protocol, \
22 SecurityGroupSettings, SecurityGroupRuleSettings
23 from snaps.openstack.utils import deploy_utils
25 from functest.core.testcase import TestCase
26 from functest.opnfv_tests.openstack.snaps import snaps_utils
27 from functest.utils.constants import CONST
31 class VPingSSH(vping_base.VPingBase):
33 Class to execute the vPing test using a Floating IP to connect to one VM
34 to issue the ping command to the second
37 def __init__(self, **kwargs):
38 if "case_name" not in kwargs:
39 kwargs["case_name"] = "vping_ssh"
40 super(VPingSSH, self).__init__(**kwargs)
42 self.kp_name = CONST.__getattribute__('vping_keypair_name') + self.guid
43 self.kp_priv_file = CONST.__getattribute__('vping_keypair_priv_file')
44 self.kp_pub_file = CONST.__getattribute__('vping_keypair_pub_file')
46 CONST.__getattribute__('vping_router_name') + self.guid
47 self.sg_name = CONST.__getattribute__('vping_sg_name') + self.guid
48 self.sg_desc = CONST.__getattribute__('vping_sg_desc')
50 self.ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
54 Sets up the OpenStack keypair, router, security group, and VM instance
55 objects then validates the ping.
56 :return: the exit code from the super.execute() method
59 super(VPingSSH, self).run()
61 self.logger.info("Creating keypair with name: '%s'" % self.kp_name)
62 kp_creator = deploy_utils.create_keypair(
64 KeypairSettings(name=self.kp_name,
65 private_filepath=self.kp_priv_file,
66 public_filepath=self.kp_pub_file))
67 self.creators.append(kp_creator)
69 # Creating router to external network
70 self.logger.info("Creating router with name: '%s'"
72 net_set = self.network_creator.network_settings
73 sub_set = [net_set.subnet_settings[0].name]
74 router_creator = deploy_utils.create_router(
77 name=self.router_name,
78 external_gateway=self.ext_net_name,
79 internal_subnets=sub_set))
80 self.creators.append(router_creator)
83 port1_settings = PortSettings(
84 name=self.vm1_name + '-vPingPort',
85 network_name=self.network_creator.network_settings.name)
86 instance1_settings = VmInstanceSettings(
87 name=self.vm1_name, flavor=self.flavor_name,
88 vm_boot_timeout=self.vm_boot_timeout,
89 vm_delete_timeout=self.vm_delete_timeout,
90 ssh_connect_timeout=self.vm_ssh_connect_timeout,
91 port_settings=[port1_settings])
94 "Creating VM 1 instance with name: '%s'"
95 % instance1_settings.name)
96 self.vm1_creator = deploy_utils.create_vm_instance(
99 self.image_creator.image_settings,
100 keypair_creator=kp_creator)
101 self.creators.append(self.vm1_creator)
103 # Creating Instance 2
104 sg_creator = self.__create_security_group()
105 self.creators.append(sg_creator)
107 port2_settings = PortSettings(
108 name=self.vm2_name + '-vPingPort',
109 network_name=self.network_creator.network_settings.name)
110 instance2_settings = VmInstanceSettings(
111 name=self.vm2_name, flavor=self.flavor_name,
112 vm_boot_timeout=self.vm_boot_timeout,
113 vm_delete_timeout=self.vm_delete_timeout,
114 ssh_connect_timeout=self.vm_ssh_connect_timeout,
115 port_settings=[port2_settings],
116 security_group_names=[sg_creator.sec_grp_settings.name],
117 floating_ip_settings=[FloatingIpSettings(
118 name=self.vm2_name + '-FIPName',
119 port_name=port2_settings.name,
120 router_name=router_creator.router_settings.name)])
123 "Creating VM 2 instance with name: '%s'"
124 % instance2_settings.name)
125 self.vm2_creator = deploy_utils.create_vm_instance(
128 self.image_creator.image_settings,
129 keypair_creator=kp_creator)
130 self.creators.append(self.vm2_creator)
132 return self._execute()
133 except Exception as e:
134 self.logger.error('Unexpected error running test - ' + e.message)
135 return TestCase.EX_RUN_ERROR
139 def _do_vping(self, vm_creator, test_ip):
143 if vm_creator.vm_ssh_active(block=True):
144 ssh = vm_creator.ssh_client()
145 if not self.__transfer_ping_script(ssh):
146 return TestCase.EX_RUN_ERROR
147 return self.__do_vping_ssh(ssh, test_ip)
151 def __transfer_ping_script(self, ssh):
153 Uses SCP to copy the ping script via the SSH client
154 :param ssh: the SSH client
157 self.logger.info("Trying to transfer ping.sh")
158 scp = SCPClient(ssh.get_transport())
159 local_path = self.functest_repo + "/" + self.repo
160 ping_script = os.path.join(local_path, "ping.sh")
162 scp.put(ping_script, "~/")
164 self.logger.error("Cannot SCP the file '%s'" % ping_script)
167 cmd = 'chmod 755 ~/ping.sh'
168 (stdin, stdout, stderr) = ssh.exec_command(cmd)
169 for line in stdout.readlines():
174 def __do_vping_ssh(self, ssh, test_ip):
176 Pings the test_ip via the SSH client
177 :param ssh: the SSH client used to issue the ping command
178 :param test_ip: the IP for the ping command to use
179 :return: exit_code (int)
181 exit_code = TestCase.EX_TESTCASE_FAILED
182 self.logger.info("Waiting for ping...")
185 cmd = '~/ping.sh ' + test_ip
190 (stdin, stdout, stderr) = ssh.exec_command(cmd)
191 output = stdout.readlines()
194 if "vPing OK" in line:
195 self.logger.info("vPing detected!")
196 exit_code = TestCase.EX_OK
200 elif sec == self.ping_timeout:
201 self.logger.info("Timeout reached.")
206 self.logger.debug("Pinging %s. Waiting for response..." % test_ip)
210 def __create_security_group(self):
212 Configures and deploys an OpenStack security group object
213 :return: the creator object
217 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
218 direction=Direction.ingress,
219 protocol=Protocol.icmp))
221 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
222 direction=Direction.ingress,
223 protocol=Protocol.tcp, port_range_min=22,
226 SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
227 direction=Direction.egress,
228 protocol=Protocol.tcp, port_range_min=22,
231 self.logger.info("Security group with name: '%s'" % self.sg_name)
232 return deploy_utils.create_security_group(self.os_creds,
233 SecurityGroupSettings(
235 description=self.sg_desc,
236 rule_settings=sg_rules))
239 if __name__ == '__main__':
240 args_parser = argparse.ArgumentParser()
241 args_parser.add_argument("-r", "--report",
242 help="Create json result file",
244 args = vars(args_parser.parse_args())
245 sys.exit(vping_base.VPingMain(VPingSSH).main(**args))