1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
2 # and others. All rights reserved.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at:
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
21 from snaps import file_utils
22 from snaps.config.flavor import FlavorConfig
23 from snaps.config.network import PortConfig
24 from snaps.config.vm_inst import VmInstanceConfig
25 from snaps.config.volume import VolumeConfig
26 from snaps.openstack import create_instance
27 from snaps.openstack.create_flavor import OpenStackFlavor
28 from snaps.openstack.create_image import OpenStackImage
29 from snaps.openstack.create_instance import OpenStackVmInstance
30 from snaps.openstack.create_network import OpenStackNetwork
31 from snaps.openstack.create_volume import OpenStackVolume
32 from snaps.openstack.tests import openstack_tests
33 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
34 from snaps.openstack.utils import (
35 nova_utils, neutron_utils, glance_utils, cinder_utils)
37 __author__ = 'spisarski'
39 logger = logging.getLogger('nova_utils_tests')
42 class NovaSmokeTests(OSComponentTestCase):
44 Tests to ensure that the nova client can communicate with the cloud
47 def test_nova_connect_success(self):
49 Tests to ensure that the proper credentials can connect.
51 nova = nova_utils.nova_client(self.os_creds)
53 # This should not throw an exception
56 def test_nova_get_hypervisor_hosts(self):
58 Tests to ensure that get_hypervisors() function works.
60 nova = nova_utils.nova_client(self.os_creds)
62 hosts = nova_utils.get_hypervisor_hosts(nova)
63 # This should not throw an exception
64 self.assertGreaterEqual(len(hosts), 1)
66 def test_nova_connect_fail(self):
68 Tests to ensure that the improper credentials cannot connect.
70 from snaps.openstack.os_credentials import OSCreds
72 nova = nova_utils.nova_client(
73 OSCreds(username='user', password='pass',
74 auth_url=self.os_creds.auth_url,
75 project_name=self.os_creds.project_name,
76 proxy_settings=self.os_creds.proxy_settings))
78 # This should throw an exception
79 with self.assertRaises(Exception):
83 class NovaUtilsKeypairTests(OSComponentTestCase):
85 Test basic nova keypair functionality
90 Instantiates the CreateImage object that is responsible for downloading
91 and creating an OS image file within OpenStack
93 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
94 self.priv_key_file_path = 'tmp/' + guid
95 self.pub_key_file_path = self.priv_key_file_path + '.pub'
97 self.nova = nova_utils.nova_client(self.os_creds)
98 self.keys = nova_utils.create_keys()
99 self.public_key = nova_utils.public_key_openssh(self.keys)
100 self.keypair_name = guid
105 Cleans the image and downloaded image file
109 nova_utils.delete_keypair(self.nova, self.keypair)
114 os.chmod(self.priv_key_file_path, 0o777)
115 os.remove(self.priv_key_file_path)
120 os.chmod(self.pub_key_file_path, 0o777)
121 os.remove(self.pub_key_file_path)
125 def test_create_keypair(self):
127 Tests the creation of an OpenStack keypair that does not exist.
129 self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name,
131 result = nova_utils.keypair_exists(self.nova, self.keypair)
132 self.assertEqual(self.keypair, result)
133 keypair = nova_utils.get_keypair_by_name(self.nova, self.keypair_name)
134 self.assertEqual(self.keypair, keypair)
136 def test_create_delete_keypair(self):
138 Tests the creation of an OpenStack keypair that does not exist.
140 self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name,
142 result = nova_utils.keypair_exists(self.nova, self.keypair)
143 self.assertEqual(self.keypair, result)
144 nova_utils.delete_keypair(self.nova, self.keypair)
145 result2 = nova_utils.keypair_exists(self.nova, self.keypair)
146 self.assertIsNone(result2)
148 def test_create_key_from_file(self):
150 Tests that the generated RSA keys are properly saved to files
153 file_utils.save_keys_to_files(self.keys, self.pub_key_file_path,
154 self.priv_key_file_path)
155 self.keypair = nova_utils.upload_keypair_file(self.nova,
157 self.pub_key_file_path)
158 pub_key_file = open(os.path.expanduser(self.pub_key_file_path))
159 pub_key = pub_key_file.read()
161 self.assertEqual(self.keypair.public_key, pub_key)
164 class NovaUtilsFlavorTests(OSComponentTestCase):
166 Test basic nova flavor functionality
171 Instantiates the CreateImage object that is responsible for downloading
172 and creating an OS image file within OpenStack
174 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
175 self.flavor_settings = FlavorConfig(
176 name=guid + '-name', flavor_id=guid + '-id', ram=1, disk=1,
177 vcpus=1, ephemeral=1, swap=2, rxtx_factor=3.0, is_public=False)
178 self.nova = nova_utils.nova_client(self.os_creds)
183 Cleans the image and downloaded image file
187 nova_utils.delete_flavor(self.nova, self.flavor)
191 def test_create_flavor(self):
193 Tests the creation of an OpenStack keypair that does not exist.
195 self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings)
196 self.validate_flavor()
198 def test_create_delete_flavor(self):
200 Tests the creation of an OpenStack keypair that does not exist.
202 self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings)
203 self.validate_flavor()
204 nova_utils.delete_flavor(self.nova, self.flavor)
205 flavor = nova_utils.get_flavor_by_name(self.nova,
206 self.flavor_settings.name)
207 self.assertIsNone(flavor)
209 def validate_flavor(self):
211 Validates the flavor_settings against the OpenStack flavor object
213 self.assertIsNotNone(self.flavor)
214 self.assertEqual(self.flavor_settings.name, self.flavor.name)
215 self.assertEqual(self.flavor_settings.flavor_id, self.flavor.id)
216 self.assertEqual(self.flavor_settings.ram, self.flavor.ram)
217 self.assertEqual(self.flavor_settings.disk, self.flavor.disk)
218 self.assertEqual(self.flavor_settings.vcpus, self.flavor.vcpus)
219 self.assertEqual(self.flavor_settings.ephemeral, self.flavor.ephemeral)
221 if self.flavor_settings.swap == 0:
222 self.assertEqual('', self.flavor.swap)
224 self.assertEqual(self.flavor_settings.swap, self.flavor.swap)
226 self.assertEqual(self.flavor_settings.rxtx_factor,
227 self.flavor.rxtx_factor)
228 self.assertEqual(self.flavor_settings.is_public, self.flavor.is_public)
231 class NovaUtilsInstanceTests(OSComponentTestCase):
233 Tests the creation of VM instances via nova_utils.py
238 Setup objects required by VM instances
242 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
244 self.nova = nova_utils.nova_client(self.os_creds)
245 self.neutron = neutron_utils.neutron_client(self.os_creds)
246 self.glance = glance_utils.glance_client(self.os_creds)
248 self.image_creator = None
249 self.network_creator = None
250 self.flavor_creator = None
255 image_settings = openstack_tests.cirros_image_settings(
256 name=guid + '-image', image_metadata=self.image_metadata)
257 self.image_creator = OpenStackImage(
258 self.os_creds, image_settings=image_settings)
259 self.image_creator.create()
261 network_settings = openstack_tests.get_priv_net_config(
262 guid + '-net', guid + '-subnet').network_settings
263 self.network_creator = OpenStackNetwork(
264 self.os_creds, network_settings)
265 self.network_creator.create()
267 self.flavor_creator = OpenStackFlavor(
270 name=guid + '-flavor-name', ram=256, disk=10, vcpus=1))
271 self.flavor_creator.create()
273 port_settings = PortConfig(
274 name=guid + '-port', network_name=network_settings.name)
275 self.port = neutron_utils.create_port(
276 self.neutron, self.os_creds, port_settings)
278 self.instance_settings = VmInstanceConfig(
279 name=guid + '-vm_inst',
280 flavor=self.flavor_creator.flavor_settings.name,
281 port_settings=[port_settings])
288 Cleanup deployed resources
293 nova_utils.delete_vm_instance(self.nova, self.vm_inst)
298 neutron_utils.delete_port(self.neutron, self.port)
301 if self.flavor_creator:
303 self.flavor_creator.clean()
306 if self.network_creator:
308 self.network_creator.clean()
311 if self.image_creator:
313 self.image_creator.clean()
317 def test_create_instance(self):
319 Tests the nova_utils.create_server() method
323 self.vm_inst = nova_utils.create_server(
324 self.nova, self.neutron, self.glance, self.instance_settings,
325 self.image_creator.image_settings)
327 self.assertIsNotNone(self.vm_inst)
329 # Wait until instance is ACTIVE
333 if create_instance.STATUS_ACTIVE == nova_utils.get_server_status(
334 self.nova, self.vm_inst):
341 self.assertTrue(active)
342 vm_inst = nova_utils.get_latest_server_object(self.nova, self.vm_inst)
344 self.assertEqual(self.vm_inst.name, vm_inst.name)
345 self.assertEqual(self.vm_inst.id, vm_inst.id)
348 class NovaUtilsInstanceVolumeTests(OSComponentTestCase):
350 Tests the creation of VM instances via nova_utils.py
355 Setup objects required by VM instances
359 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
361 self.nova = nova_utils.nova_client(self.os_creds)
362 self.cinder = cinder_utils.cinder_client(self.os_creds)
364 self.image_creator = None
365 self.network_creator = None
366 self.flavor_creator = None
367 self.volume_creator = None
368 self.instance_creator = None
371 image_settings = openstack_tests.cirros_image_settings(
372 name=guid + '-image', image_metadata=self.image_metadata)
373 self.image_creator = OpenStackImage(
374 self.os_creds, image_settings=image_settings)
375 self.image_creator.create()
377 network_settings = openstack_tests.get_priv_net_config(
378 guid + '-net', guid + '-subnet').network_settings
379 self.network_creator = OpenStackNetwork(
380 self.os_creds, network_settings)
381 self.network_creator.create()
383 self.flavor_creator = OpenStackFlavor(
386 name=guid + '-flavor-name', ram=256, disk=10, vcpus=1))
387 self.flavor_creator.create()
390 volume_settings = VolumeConfig(
391 name=self.__class__.__name__ + '-' + str(guid))
392 self.volume_creator = OpenStackVolume(
393 self.os_creds, volume_settings)
394 self.volume_creator.create(block=True)
396 port_settings = PortConfig(
397 name=guid + '-port', network_name=network_settings.name)
398 instance_settings = VmInstanceConfig(
399 name=guid + '-vm_inst',
400 flavor=self.flavor_creator.flavor_settings.name,
401 port_settings=[port_settings])
402 self.instance_creator = OpenStackVmInstance(
403 self.os_creds, instance_settings, image_settings)
404 self.instance_creator.create(block=True)
411 Cleanup deployed resources
414 if self.instance_creator:
416 self.instance_creator.clean()
419 if self.volume_creator:
421 self.volume_creator.clean()
424 if self.flavor_creator:
426 self.flavor_creator.clean()
429 if self.network_creator:
431 self.network_creator.clean()
434 if self.image_creator:
436 self.image_creator.clean()
440 def test_add_remove_volume(self):
442 Tests the nova_utils.create_server() method
446 self.assertIsNotNone(self.volume_creator.get_volume())
447 self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
449 # Attach volume to VM
450 nova_utils.attach_volume(
451 self.nova, self.instance_creator.get_vm_inst(),
452 self.volume_creator.get_volume())
456 vol_attach = cinder_utils.get_volume_by_id(
457 self.cinder, self.volume_creator.get_volume().id)
458 vm_attach = nova_utils.get_server_object_by_id(
459 self.nova, self.instance_creator.get_vm_inst().id)
461 # Detach volume to VM
462 nova_utils.detach_volume(
463 self.nova, self.instance_creator.get_vm_inst(),
464 self.volume_creator.get_volume())
468 vol_detach = cinder_utils.get_volume_by_id(
469 self.cinder, self.volume_creator.get_volume().id)
470 vm_detach = nova_utils.get_server_object_by_id(
471 self.nova, self.instance_creator.get_vm_inst().id)
473 # Validate Attachment
474 self.assertIsNotNone(vol_attach)
475 self.assertEqual(self.volume_creator.get_volume().id, vol_attach.id)
476 self.assertEqual(1, len(vol_attach.attachments))
477 self.assertEqual(vm_attach.volume_ids[0]['id'],
478 vol_attach.attachments[0]['volume_id'])
480 # Validate Detachment
481 self.assertIsNotNone(vol_detach)
482 self.assertEqual(self.volume_creator.get_volume().id, vol_detach.id)
483 self.assertEqual(0, len(vol_detach.attachments))
484 self.assertEqual(0, len(vm_detach.volume_ids))