cloudbridge/test/test_compute_service.py

import ipaddress

from test import helpers
from test.helpers import ProviderTestBase
from test.helpers import standard_interface_tests as sit

from cloudbridge.cloud.interfaces import InstanceState
from cloudbridge.cloud.interfaces import InvalidConfigurationException
from cloudbridge.cloud.interfaces import TestMockHelperMixin
from cloudbridge.cloud.interfaces.exceptions import WaitStateException
from cloudbridge.cloud.interfaces.resources import InstanceType
# from cloudbridge.cloud.interfaces.resources import SnapshotState

import six


class CloudComputeServiceTestCase(ProviderTestBase):

    @helpers.skipIfNoService(['compute.instances', 'networking.networks'])
    def test_crud_instance(self):
        name = "cb_instcrud-{0}".format(helpers.get_uuid())
        # Declare these variables and late binding will allow
        # the cleanup method access to the most current values
        inst = None
        net = None
        with helpers.cleanup_action(lambda: helpers.cleanup_test_resources(
                inst, net)):
            net, subnet = helpers.create_test_network(self.provider, name)
            inst = helpers.get_test_instance(self.provider, name,
                                             subnet=subnet)

            sit.check_standard_behaviour(
                self, self.provider.compute.instances, inst)
        deleted_inst = self.provider.compute.instances.get(
            inst.id)
        self.assertTrue(
            deleted_inst is None or deleted_inst.state in (
                InstanceState.TERMINATED,
                InstanceState.UNKNOWN),
            "Instance %s should have been deleted but still exists." %
            name)

    def _is_valid_ip(self, address):
        try:
            ipaddress.ip_address(address)
        except ValueError:
            return False
        return True

    @helpers.skipIfNoService(['compute.instances', 'networking.networks',
                              'security.security_groups',
                              'security.key_pairs'])
    def test_instance_properties(self):
        name = "cb_inst_props-{0}".format(helpers.get_uuid())

        # Declare these variables and late binding will allow
        # the cleanup method access to the most current values
        test_instance = None
        net = None
        sg = None
        kp = None
        with helpers.cleanup_action(lambda: helpers.cleanup_test_resources(
                test_instance, net, sg, kp)):
            net, subnet = helpers.create_test_network(self.provider, name)
            kp = self.provider.security.key_pairs.create(name=name)
            sg = self.provider.security.security_groups.create(
                name=name, description=name, network_id=net.id)
            test_instance = helpers.get_test_instance(self.provider,
                                                      name, key_pair=kp,
                                                      security_groups=[sg],
                                                      subnet=subnet)
            self.assertEqual(
                test_instance.name, name,
                "Instance name {0} is not equal to the expected name"
                " {1}".format(test_instance.name, name))
            image_id = helpers.get_provider_test_data(self.provider, "image")
            self.assertEqual(test_instance.image_id, image_id,
                             "Image id {0} is not equal to the expected id"
                             " {1}".format(test_instance.image_id, image_id))
            self.assertIsInstance(test_instance.zone_id,
                                  six.string_types)
            # FIXME: Moto is not returning the instance's placement zone
#             find_zone = [zone for zone in
#                          self.provider.compute.regions.current.zones
#                          if zone.id == test_instance.zone_id]
#             self.assertEqual(len(find_zone), 1,
#                              "Instance's placement zone could not be "
#                              " found in zones list")
            self.assertEqual(
                test_instance.image_id,
                helpers.get_provider_test_data(self.provider, "image"))
            self.assertIsInstance(test_instance.public_ips, list)
            self.assertIsInstance(test_instance.private_ips, list)
            self.assertEqual(
                test_instance.key_pair_name,
                kp.name)
            self.assertIsInstance(test_instance.security_groups, list)
            self.assertEqual(
                test_instance.security_groups[0],
                sg)
            self.assertIsInstance(test_instance.security_group_ids, list)
            self.assertEqual(
                test_instance.security_group_ids[0],
                sg.id)
            # Must have either a public or a private ip
            ip_private = test_instance.private_ips[0] \
                if test_instance.private_ips else None
            ip_address = test_instance.public_ips[0] \
                if test_instance.public_ips and test_instance.public_ips[0] \
                else ip_private
            self.assertIsNotNone(
                ip_address,
                "Instance must have either a public IP or a private IP")
            self.assertTrue(
                self._is_valid_ip(ip_address),
                "Instance must have a valid IP address")
            self.assertIsInstance(test_instance.instance_type_id,
                                  six.string_types)
            itype = self.provider.compute.instance_types.get(
                test_instance.instance_type_id)
            self.assertEqual(
                itype, test_instance.instance_type,
                "Instance type {0} does not match expected type {1}".format(
                    itype.name, test_instance.instance_type))
            self.assertIsInstance(itype, InstanceType)
            expected_type = helpers.get_provider_test_data(self.provider,
                                                           'instance_type')
            self.assertEqual(
                itype.name, expected_type,
                "Instance type {0} does not match expected type {1}".format(
                    itype.name, expected_type))

    @helpers.skipIfNoService(['compute.instances', 'compute.images',
                              'compute.instance_types'])
    def test_block_device_mapping_launch_config(self):
        lc = self.provider.compute.instances.create_launch_config()

        # specifying an invalid size should raise
        # an exception
        with self.assertRaises(InvalidConfigurationException):
            lc.add_volume_device(size=-1)

        # Attempting to add a blank volume without specifying a size
        # should raise an exception
        with self.assertRaises(InvalidConfigurationException):
            lc.add_volume_device(source=None)

        # block_devices should be empty so far
        self.assertListEqual(
            lc.block_devices, [], "No block devices should have been"
            " added to mappings list since the configuration was"
            " invalid")

        # Add a new volume
        lc.add_volume_device(size=1, delete_on_terminate=True)

        # Override root volume size
        image_id = helpers.get_provider_test_data(self.provider, "image")
        img = self.provider.compute.images.get(image_id)
        # The size should be greater then the ami size
        # and therefore, img.min_disk is used.
        lc.add_volume_device(
            is_root=True,
            source=img,
            size=img.min_disk if img and img.min_disk else 2,
            delete_on_terminate=True)

        # Attempting to add more than one root volume should raise an
        # exception.
        with self.assertRaises(InvalidConfigurationException):
            lc.add_volume_device(size=1, is_root=True)

        # Attempting to add an incorrect source should raise an exception
        with self.assertRaises(InvalidConfigurationException):
            lc.add_volume_device(
                source="invalid_source",
                delete_on_terminate=True)

        # Add all available ephemeral devices
        instance_type_name = helpers.get_provider_test_data(
            self.provider,
            "instance_type")
        inst_type = self.provider.compute.instance_types.find(
            name=instance_type_name)[0]
        for _ in range(inst_type.num_ephemeral_disks):
            lc.add_ephemeral_device()

        # block_devices should be populated
        self.assertTrue(
            len(lc.block_devices) == 2 + inst_type.num_ephemeral_disks,
            "Expected %d total block devices bit found %d" %
            (2 + inst_type.num_ephemeral_disks, len(lc.block_devices)))

    @helpers.skipIfNoService(['compute.instances', 'compute.images',
                              'compute.instance_types', 'block_store.volumes'])
    def test_block_device_mapping_attachments(self):
        name = "cb_blkattch-{0}".format(helpers.get_uuid())

        # Comment out BDM tests because OpenStack is not stable enough yet
        if True:
            if True:

                # test_vol = self.provider.block_store.volumes.create(
                #    name,
                #    1,
                #    helpers.get_provider_test_data(self.provider,
                #                                   "placement"))
                # with helpers.cleanup_action(lambda: test_vol.delete()):
                #    test_vol.wait_till_ready()
                #    test_snap = test_vol.create_snapshot(name=name,
                #                                         description=name)
                #
                #    def cleanup_snap(snap):
                #        snap.delete()
                #        snap.wait_for(
                #            [SnapshotState.UNKNOWN],
                #            terminal_states=[SnapshotState.ERROR])
                #
                #    with helpers.cleanup_action(lambda:
                #                                cleanup_snap(test_snap)):
                #         test_snap.wait_till_ready()

                lc = self.provider.compute.instances.create_launch_config()

#                 # Add a new blank volume
#                 lc.add_volume_device(size=1, delete_on_terminate=True)
#
#                 # Attach an existing volume
#                 lc.add_volume_device(size=1, source=test_vol,
#                                      delete_on_terminate=True)
#
#                 # Add a new volume based on a snapshot
#                 lc.add_volume_device(size=1, source=test_snap,
#                                      delete_on_terminate=True)

                # Override root volume size
                image_id = helpers.get_provider_test_data(
                    self.provider,
                    "image")
                img = self.provider.compute.images.get(image_id)
                # The size should be greater then the ami size
                # and therefore, img.min_disk is used.
                lc.add_volume_device(
                    is_root=True,
                    source=img,
                    size=img.min_disk if img and img.min_disk else 2,
                    delete_on_terminate=True)

                # Add all available ephemeral devices
                instance_type_name = helpers.get_provider_test_data(
                    self.provider,
                    "instance_type")
                inst_type = self.provider.compute.instance_types.find(
                    name=instance_type_name)[0]
                for _ in range(inst_type.num_ephemeral_disks):
                    lc.add_ephemeral_device()

                net, subnet = helpers.create_test_network(self.provider, name)

                with helpers.cleanup_action(lambda:
                                            helpers.delete_test_network(net)):

                    inst = helpers.create_test_instance(
                        self.provider,
                        name,
                        subnet=subnet,
                        launch_config=lc)

                    with helpers.cleanup_action(lambda:
                                                helpers.delete_test_instance(
                                                    inst)):
                        try:
                            inst.wait_till_ready()
                        except WaitStateException as e:
                            self.fail("The block device mapped launch did not "
                                      " complete successfully: %s" % e)
                        # TODO: Check instance attachments and make sure they
                        # correspond to requested mappings

    @helpers.skipIfNoService(['compute.instances', 'networking.networks',
                              'security.security_groups'])
    def test_instance_methods(self):
        name = "cb_instmethods-{0}".format(helpers.get_uuid())

        # Declare these variables and late binding will allow
        # the cleanup method access to the most current values
        test_inst = None
        net = None
        sg = None
        with helpers.cleanup_action(lambda: helpers.cleanup_test_resources(
                test_inst, net, sg)):
            net, subnet = helpers.create_test_network(self.provider, name)
            test_inst = helpers.get_test_instance(self.provider, name,
                                                  subnet=subnet)
            sg = self.provider.security.security_groups.create(
                name=name, description=name, network_id=net.id)

            # Check adding a security group to a running instance
            test_inst.add_security_group(sg)
            test_inst.refresh()
            self.assertTrue(
                sg in test_inst.security_groups, "Expected security group '%s'"
                " to be among instance security_groups: [%s]" %
                (sg, test_inst.security_groups))

            # Check removing a security group from a running instance
            test_inst.remove_security_group(sg)
            test_inst.refresh()
            self.assertTrue(
                sg not in test_inst.security_groups, "Expected security group"
                " '%s' to be removed from instance security_groups: [%s]" %
                (sg, test_inst.security_groups))

            # check floating ips
            router = self.provider.networking.routers.create(name, net)
            gateway = None

            def cleanup_router():
                with helpers.cleanup_action(lambda: router.delete()):
                    with helpers.cleanup_action(lambda: gateway.delete()):
                        router.detach_subnet(subnet)
                        router.detach_gateway(gateway)

            with helpers.cleanup_action(lambda: cleanup_router()):
                router.attach_subnet(subnet)
                gateway = (self.provider.networking.gateways
                           .get_or_create_inet_gateway(name))
                router.attach_gateway(gateway)
                # check whether adding an elastic ip works
                fip = (self.provider.networking.networks
                       .create_floating_ip())
                with helpers.cleanup_action(lambda: fip.delete()):
                    test_inst.add_floating_ip(fip.public_ip)
                    test_inst.refresh()
                    self.assertIn(fip.public_ip, test_inst.public_ips)
                    if isinstance(self.provider, TestMockHelperMixin):
                        # TODO: Moto bug does not refresh removed public ip
                        return
                    # check whether removing an elastic ip works
                    test_inst.remove_floating_ip(fip.public_ip)
                    test_inst.refresh()
                    self.assertNotIn(fip.public_ip, test_inst.public_ips)