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
10 from datetime import datetime
16 from functest.core.testcase import TestCase
17 from functest.utils import functest_utils
18 from functest.utils.constants import CONST
20 from snaps.openstack import create_flavor
21 from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
22 from snaps.openstack.create_network import NetworkSettings, SubnetSettings
23 from snaps.openstack.tests import openstack_tests
24 from snaps.openstack.utils import deploy_utils, nova_utils
27 class VPingBase(TestCase):
30 Base class for vPing tests that check connectivity between two VMs shared
32 This class is responsible for creating the image, internal network.
35 def __init__(self, **kwargs):
36 super(VPingBase, self).__init__(**kwargs)
38 self.logger = logging.getLogger(self.__class__.__name__)
40 self.functest_repo = CONST.__getattribute__('dir_repo_functest')
42 if CONST.__getattribute__('vping_unique_names'):
43 self.guid = '-' + str(uuid.uuid4())
45 self.os_creds = openstack_tests.get_credentials(
46 os_env_file=CONST.__getattribute__('openstack_creds'))
48 self.repo = CONST.__getattribute__('dir_vping')
50 self.creators = list()
51 self.image_creator = None
52 self.network_creator = None
53 self.vm1_creator = None
54 self.vm2_creator = None
56 self.self_cleanup = CONST.__getattribute__('vping_cleanup_objects')
60 CONST.__getattribute__('vping_image_name') + self.guid
63 self.vm1_name = CONST.__getattribute__('vping_vm_name_1') + self.guid
64 self.vm2_name = CONST.__getattribute__('vping_vm_name_2') + self.guid
65 self.vm_boot_timeout = CONST.__getattribute__('vping_vm_boot_timeout')
66 self.vm_delete_timeout =\
67 CONST.__getattribute__('vping_vm_delete_timeout')
68 self.vm_ssh_connect_timeout = CONST.vping_vm_ssh_connect_timeout
69 self.ping_timeout = CONST.__getattribute__('vping_ping_timeout')
70 self.flavor_name = 'vping-flavor' + self.guid
72 # NEUTRON Private Network parameters
73 self.private_net_name =\
74 CONST.__getattribute__('vping_private_net_name') + self.guid
75 self.private_subnet_name =\
76 CONST.__getattribute__('vping_private_subnet_name') + self.guid
77 self.private_subnet_cidr =\
78 CONST.__getattribute__('vping_private_subnet_cidr')
80 scenario = functest_utils.get_scenario()
82 self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_ANY
83 if 'ovs' in scenario or 'fdio' in scenario:
84 self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
86 self.cirros_image_config = None
88 # Move this configuration option up for all tests to leverage
89 if hasattr(CONST, 'snaps_images_cirros'):
90 self.cirros_image_config = CONST.__getattribute__(
91 'snaps_images_cirros')
95 Begins the test execution which should originate from the subclass
98 if not os.path.exists(self.functest_repo):
100 "Functest repository not found '%s'" % self.functest_repo)
102 self.logger.info('Begin virtual environment setup')
104 self.start_time = time.time()
105 self.logger.info("vPing Start Time:'%s'" % (
106 datetime.fromtimestamp(self.start_time).strftime(
107 '%Y-%m-%d %H:%M:%S')))
109 self.__delete_exist_vms()
111 image_base_name = self.image_name + '-' + str(self.guid)
112 os_image_settings = openstack_tests.cirros_image_settings(
113 image_base_name, image_metadata=self.cirros_image_config)
114 self.logger.info("Creating image with name: '%s'" % self.image_name)
116 self.image_creator = deploy_utils.create_image(
117 self.os_creds, os_image_settings)
118 self.creators.append(self.image_creator)
121 "Creating network with name: '%s'" % self.private_net_name)
122 self.network_creator = deploy_utils.create_network(
124 NetworkSettings(name=self.private_net_name,
125 subnet_settings=[SubnetSettings(
126 name=self.private_subnet_name,
127 cidr=self.private_subnet_cidr)]))
128 self.creators.append(self.network_creator)
131 "Creating flavor with name: '%s'" % self.flavor_name)
132 flavor_creator = OpenStackFlavor(
134 FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1,
135 metadata=self.flavor_metadata))
136 flavor_creator.create()
137 self.creators.append(flavor_creator)
141 Method called by subclasses after environment has been setup
142 :return: the exit code
144 self.logger.info('Begin test execution')
146 test_ip = self.vm1_creator.get_port_ip(
147 self.vm1_creator.instance_settings.port_settings[0].name)
149 if self.vm1_creator.vm_active(
150 block=True) and self.vm2_creator.vm_active(block=True):
151 result = self._do_vping(self.vm2_creator, test_ip)
153 raise Exception('VMs never became active')
155 if result == TestCase.EX_RUN_ERROR:
156 return TestCase.EX_RUN_ERROR
158 self.stop_time = time.time()
160 return TestCase.EX_OK
164 Cleanup all OpenStack objects. Should be called on completion
167 if self.self_cleanup:
168 for creator in reversed(self.creators):
171 except Exception as e:
172 self.logger.error('Unexpected error cleaning - %s', e)
174 def _do_vping(self, vm_creator, test_ip):
176 Method to be implemented by subclasses
177 Begins the real test after the OpenStack environment has been setup
178 :param vm_creator: the SNAPS VM instance creator object
179 :param test_ip: the IP to which the VM needs to issue the ping
182 raise NotImplementedError('vping execution is not implemented')
184 def __delete_exist_vms(self):
186 Cleans any existing VMs using the same name.
188 nova_client = nova_utils.nova_client(self.os_creds)
189 servers = nova_client.servers.list()
190 for server in servers:
191 if server.name == self.vm1_name or server.name == self.vm2_name:
192 self.logger.info("Deleting instance %s..." % server.name)
196 class VPingMain(object):
198 def __init__(self, vping_cls):
199 self.vping = vping_cls()
201 def main(self, **kwargs):
203 result = self.vping.run(**kwargs)
204 if result != VPingBase.EX_OK:
207 return self.vping.publish_report()
209 return VPingBase.EX_RUN_ERROR