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, keystone_utils)
36 from snaps.openstack.utils.nova_utils import NovaException
38 __author__ = 'spisarski'
40 logger = logging.getLogger('nova_utils_tests')
43 class NovaSmokeTests(OSComponentTestCase):
45 Tests to ensure that the nova client can communicate with the cloud
48 def test_nova_connect_success(self):
50 Tests to ensure that the proper credentials can connect.
52 nova = nova_utils.nova_client(self.os_creds, self.os_session)
54 # This should not throw an exception
57 def test_nova_get_hypervisor_hosts(self):
59 Tests to ensure that get_hypervisors() function works.
61 nova = nova_utils.nova_client(self.os_creds, self.os_session)
63 hosts = nova_utils.get_hypervisor_hosts(nova)
64 # This should not throw an exception
65 self.assertGreaterEqual(len(hosts), 1)
67 def test_nova_connect_fail(self):
69 Tests to ensure that the improper credentials cannot connect.
71 from snaps.openstack.os_credentials import OSCreds
73 nova = nova_utils.nova_client(
74 OSCreds(username='user', password='pass',
75 auth_url=self.os_creds.auth_url,
76 project_name=self.os_creds.project_name,
77 proxy_settings=self.os_creds.proxy_settings))
79 # This should throw an exception
80 with self.assertRaises(Exception):
84 class NovaUtilsKeypairTests(OSComponentTestCase):
86 Test basic nova keypair functionality
91 Instantiates the CreateImage object that is responsible for downloading
92 and creating an OS image file within OpenStack
94 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
95 self.priv_key_file_path = 'tmp/' + guid
96 self.pub_key_file_path = self.priv_key_file_path + '.pub'
98 self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
99 self.keys = nova_utils.create_keys()
100 self.public_key = nova_utils.public_key_openssh(self.keys)
101 self.keypair_name = guid
106 Cleans the image and downloaded image file
110 nova_utils.delete_keypair(self.nova, self.keypair)
115 os.chmod(self.priv_key_file_path, 0o777)
116 os.remove(self.priv_key_file_path)
121 os.chmod(self.pub_key_file_path, 0o777)
122 os.remove(self.pub_key_file_path)
126 super(self.__class__, self).__clean__()
128 def test_create_keypair(self):
130 Tests the creation of an OpenStack keypair that does not exist.
132 self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name,
134 result = nova_utils.keypair_exists(self.nova, self.keypair)
135 self.assertEqual(self.keypair, result)
136 keypair = nova_utils.get_keypair_by_name(self.nova, self.keypair_name)
137 self.assertEqual(self.keypair, keypair)
139 def test_create_delete_keypair(self):
141 Tests the creation of an OpenStack keypair that does not exist.
143 self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name,
145 result = nova_utils.keypair_exists(self.nova, self.keypair)
146 self.assertEqual(self.keypair, result)
147 nova_utils.delete_keypair(self.nova, self.keypair)
148 result2 = nova_utils.keypair_exists(self.nova, self.keypair)
149 self.assertIsNone(result2)
151 def test_create_key_from_file(self):
153 Tests that the generated RSA keys are properly saved to files
156 file_utils.save_keys_to_files(self.keys, self.pub_key_file_path,
157 self.priv_key_file_path)
158 self.keypair = nova_utils.upload_keypair_file(self.nova,
160 self.pub_key_file_path)
161 pub_key_file = open(os.path.expanduser(self.pub_key_file_path))
162 pub_key = pub_key_file.read()
164 self.assertEqual(self.keypair.public_key, pub_key)
167 class NovaUtilsFlavorTests(OSComponentTestCase):
169 Test basic nova flavor functionality
174 Instantiates the CreateImage object that is responsible for downloading
175 and creating an OS image file within OpenStack
177 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
178 self.flavor_settings = FlavorConfig(
179 name=guid + '-name', flavor_id=guid + '-id', ram=1, disk=1,
180 vcpus=1, ephemeral=1, swap=2, rxtx_factor=3.0, is_public=False)
181 self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
186 Cleans the image and downloaded image file
190 nova_utils.delete_flavor(self.nova, self.flavor)
194 super(self.__class__, self).__clean__()
196 def test_create_flavor(self):
198 Tests the creation of an OpenStack keypair that does not exist.
200 self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings)
201 self.validate_flavor()
203 def test_create_delete_flavor(self):
205 Tests the creation of an OpenStack keypair that does not exist.
207 self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings)
208 self.validate_flavor()
209 nova_utils.delete_flavor(self.nova, self.flavor)
210 flavor = nova_utils.get_flavor_by_name(self.nova,
211 self.flavor_settings.name)
212 self.assertIsNone(flavor)
214 def validate_flavor(self):
216 Validates the flavor_settings against the OpenStack flavor object
218 self.assertIsNotNone(self.flavor)
219 self.assertEqual(self.flavor_settings.name, self.flavor.name)
220 self.assertEqual(self.flavor_settings.flavor_id, self.flavor.id)
221 self.assertEqual(self.flavor_settings.ram, self.flavor.ram)
222 self.assertEqual(self.flavor_settings.disk, self.flavor.disk)
223 self.assertEqual(self.flavor_settings.vcpus, self.flavor.vcpus)
224 self.assertEqual(self.flavor_settings.ephemeral, self.flavor.ephemeral)
226 if self.flavor_settings.swap == 0:
227 self.assertEqual('', self.flavor.swap)
229 self.assertEqual(self.flavor_settings.swap, self.flavor.swap)
231 self.assertEqual(self.flavor_settings.rxtx_factor,
232 self.flavor.rxtx_factor)
233 self.assertEqual(self.flavor_settings.is_public, self.flavor.is_public)
236 class NovaUtilsInstanceTests(OSComponentTestCase):
238 Tests the creation of VM instances via nova_utils.py
243 Setup objects required by VM instances
247 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
249 self.nova = nova_utils.nova_client(
250 self.os_creds, self.os_session)
251 self.keystone = keystone_utils.keystone_client(
252 self.os_creds, self.os_session)
253 self.neutron = neutron_utils.neutron_client(
254 self.os_creds, self.os_session)
255 self.glance = glance_utils.glance_client(
256 self.os_creds, self.os_session)
258 self.image_creator = None
259 self.network_creator = None
260 self.flavor_creator = None
265 image_settings = openstack_tests.cirros_image_settings(
266 name=guid + '-image', image_metadata=self.image_metadata)
267 self.image_creator = OpenStackImage(
268 self.os_creds, image_settings=image_settings)
269 self.image_creator.create()
271 network_settings = openstack_tests.get_priv_net_config(
272 project_name=self.os_creds.project_name,
273 net_name="{}-{}".format(guid, 'net'),
274 subnet_name="{}-{}".format(guid, 'subnet')).network_settings
275 self.network_creator = OpenStackNetwork(
276 self.os_creds, network_settings)
277 self.network_creator.create()
279 flavor_config = openstack_tests.get_flavor_config(
280 name="{}-{}".format(guid, 'flavor-name'), ram=256, disk=10,
281 vcpus=1, metadata=self.flavor_metadata)
282 self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_config)
283 self.flavor_creator.create()
285 port_settings = PortConfig(
286 name=guid + '-port', network_name=network_settings.name)
287 self.port = neutron_utils.create_port(
288 self.neutron, self.os_creds, port_settings)
290 self.instance_settings = VmInstanceConfig(
291 name=guid + '-vm_inst',
292 flavor=self.flavor_creator.flavor_settings.name,
293 port_settings=[port_settings])
300 Cleanup deployed resources
305 nova_utils.delete_vm_instance(self.nova, self.vm_inst)
310 neutron_utils.delete_port(self.neutron, self.port)
313 if self.flavor_creator:
315 self.flavor_creator.clean()
318 if self.network_creator:
320 self.network_creator.clean()
323 if self.image_creator:
325 self.image_creator.clean()
329 super(self.__class__, self).__clean__()
331 def test_create_instance(self):
333 Tests the nova_utils.create_server() method
337 self.vm_inst = nova_utils.create_server(
338 self.nova, self.keystone, self.neutron, self.glance,
339 self.instance_settings, self.image_creator.image_settings,
340 self.os_creds.project_name)
342 self.assertIsNotNone(self.vm_inst)
344 # Wait until instance is ACTIVE
349 status = nova_utils.get_server_status(self.nova, self.vm_inst)
350 if create_instance.STATUS_ACTIVE == status:
357 self.assertTrue(active, msg='VM {} status {} is not {}'.format(
358 self.vm_inst.name, status, create_instance.STATUS_ACTIVE))
359 vm_inst = nova_utils.get_latest_server_object(
360 self.nova, self.neutron, self.keystone, self.vm_inst,
361 self.os_creds.project_name)
363 self.assertEqual(self.vm_inst.name, vm_inst.name)
364 self.assertEqual(self.vm_inst.id, vm_inst.id)
367 class NovaUtilsInstanceVolumeTests(OSComponentTestCase):
369 Tests the creation of VM instances via nova_utils.py
374 Setup objects required by VM instances
378 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
380 self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
381 self.cinder = cinder_utils.cinder_client(
382 self.os_creds, self.os_session)
384 self.image_creator = None
385 self.network_creator = None
386 self.flavor_creator = None
387 self.volume_creator = None
388 self.instance_creator = None
391 image_settings = openstack_tests.cirros_image_settings(
392 name=guid + '-image', image_metadata=self.image_metadata)
393 self.image_creator = OpenStackImage(
394 self.os_creds, image_settings=image_settings)
395 self.image_creator.create()
397 network_settings = openstack_tests.get_priv_net_config(
398 project_name=self.os_creds.project_name,
399 net_name="{}-{}".format(guid, 'net'),
400 subnet_name="{}-{}".format(guid, 'subnet')).network_settings
402 self.network_creator = OpenStackNetwork(
403 self.os_creds, network_settings)
404 self.network_creator.create()
406 flavor_settings = openstack_tests.get_flavor_config(
407 name=guid + '-flavor', ram=256, disk=10, vcpus=1,
408 metadata=self.flavor_metadata)
409 self.flavor_creator = OpenStackFlavor(
410 self.os_creds, flavor_settings)
411 self.flavor_creator.create()
414 volume_settings = VolumeConfig(
415 name=self.__class__.__name__ + '-' + str(guid))
416 self.volume_creator = OpenStackVolume(
417 self.os_creds, volume_settings)
418 self.volume_creator.create(block=True)
420 port_settings = PortConfig(
421 name=guid + '-port', network_name=network_settings.name)
422 instance_settings = VmInstanceConfig(
423 name=guid + '-vm_inst',
424 flavor=self.flavor_creator.flavor_settings.name,
425 port_settings=[port_settings])
426 self.instance_creator = OpenStackVmInstance(
427 self.os_creds, instance_settings, image_settings)
428 self.instance_creator.create(block=True)
435 Cleanup deployed resources
438 if self.instance_creator:
440 self.instance_creator.clean()
443 if self.volume_creator:
445 self.volume_creator.clean()
448 if self.flavor_creator:
450 self.flavor_creator.clean()
453 if self.network_creator:
455 self.network_creator.clean()
458 if self.image_creator:
460 self.image_creator.clean()
464 super(self.__class__, self).__clean__()
466 def test_add_remove_volume(self):
468 Tests the nova_utils.attach_volume() and detach_volume functions with
473 self.assertIsNotNone(self.volume_creator.get_volume())
474 self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
476 # Attach volume to VM
477 neutron = neutron_utils.neutron_client(
478 self.os_creds, self.os_session)
479 keystone = keystone_utils.keystone_client(
480 self.os_creds, self.os_session)
481 self.assertIsNotNone(nova_utils.attach_volume(
482 self.nova, neutron, keystone, self.instance_creator.get_vm_inst(),
483 self.volume_creator.get_volume(), self.os_creds.project_name))
488 start_time = time.time()
489 while time.time() < start_time + 120:
490 vol_attach = cinder_utils.get_volume_by_id(
491 self.cinder, self.volume_creator.get_volume().id)
493 if len(vol_attach.attachments) > 0:
499 self.assertTrue(attached)
500 self.assertIsNotNone(vol_attach)
502 keystone = keystone_utils.keystone_client(
503 self.os_creds, self.os_session)
504 vm_attach = nova_utils.get_server_object_by_id(
505 self.nova, neutron, keystone,
506 self.instance_creator.get_vm_inst().id, self.os_creds.project_name)
508 # Validate Attachment
509 self.assertIsNotNone(vol_attach)
510 self.assertEqual(self.volume_creator.get_volume().id, vol_attach.id)
511 self.assertEqual(1, len(vol_attach.attachments))
512 self.assertEqual(vm_attach.volume_ids[0]['id'],
513 vol_attach.attachments[0]['volume_id'])
515 # Detach volume to VM
516 self.assertIsNotNone(nova_utils.detach_volume(
517 self.nova, neutron, keystone, self.instance_creator.get_vm_inst(),
518 self.volume_creator.get_volume(), self.os_creds.project_name))
520 start_time = time.time()
521 while time.time() < start_time + 120:
522 vol_detach = cinder_utils.get_volume_by_id(
523 self.cinder, self.volume_creator.get_volume().id)
524 if len(vol_detach.attachments) == 0:
530 self.assertFalse(attached)
531 self.assertIsNotNone(vol_detach)
533 vm_detach = nova_utils.get_server_object_by_id(
534 self.nova, neutron, keystone,
535 self.instance_creator.get_vm_inst().id, self.os_creds.project_name)
537 # Validate Detachment
538 self.assertIsNotNone(vol_detach)
539 self.assertEqual(self.volume_creator.get_volume().id, vol_detach.id)
541 self.assertEqual(0, len(vol_detach.attachments))
542 self.assertEqual(0, len(vm_detach.volume_ids))
544 def test_attach_volume_nowait(self):
546 Tests the nova_utils.attach_volume() with a timeout value that is too
547 small to have the volume attachment data to be included on the VmInst
548 object that was supposed to be returned
551 self.assertIsNotNone(self.volume_creator.get_volume())
552 self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
554 # Attach volume to VM
555 neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
556 keystone = keystone_utils.keystone_client(
557 self.os_creds, self.os_session)
558 with self.assertRaises(NovaException):
559 nova_utils.attach_volume(
560 self.nova, neutron, keystone,
561 self.instance_creator.get_vm_inst(),
562 self.volume_creator.get_volume(), self.os_creds.project_name,
565 def test_detach_volume_nowait(self):
567 Tests the nova_utils.detach_volume() with a timeout value that is too
568 small to have the volume attachment data to be included on the VmInst
569 object that was supposed to be returned
572 self.assertIsNotNone(self.volume_creator.get_volume())
573 self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
575 # Attach volume to VM
576 neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
577 keystone = keystone_utils.keystone_client(
578 self.os_creds, self.os_session)
579 nova_utils.attach_volume(
580 self.nova, neutron, keystone, self.instance_creator.get_vm_inst(),
581 self.volume_creator.get_volume(), self.os_creds.project_name)
583 # Check VmInst for attachment
584 keystone = keystone_utils.keystone_client(
585 self.os_creds, self.os_session)
586 latest_vm = nova_utils.get_server_object_by_id(
587 self.nova, neutron, keystone,
588 self.instance_creator.get_vm_inst().id, self.os_creds.project_name)
589 self.assertEqual(1, len(latest_vm.volume_ids))
591 # Check Volume for attachment
594 start_time = time.time()
595 while time.time() < start_time + 120:
596 vol_attach = cinder_utils.get_volume_by_id(
597 self.cinder, self.volume_creator.get_volume().id)
599 if len(vol_attach.attachments) > 0:
605 self.assertTrue(attached)
606 self.assertIsNotNone(vol_attach)
609 with self.assertRaises(NovaException):
610 nova_utils.detach_volume(
611 self.nova, neutron, keystone,
612 self.instance_creator.get_vm_inst(),
613 self.volume_creator.get_volume(), self.os_creds.project_name,