cloudbridge/tests/test_network_service.py

from cloudbridge.base import helpers as cb_helpers
from cloudbridge.base.resources import BaseNetwork
from cloudbridge.interfaces.resources import (FloatingIP, Network,
                                              NetworkState, RouterState,
                                              Subnet, SubnetState)

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


class CloudNetworkServiceTestCase(ProviderTestBase):

    _multiprocess_can_split_ = True

    @helpers.skipIfNoService(['networking.subnets',
                              'networking.networks',
                              'networking.routers'])
    def test_storage_services_event_pattern(self):
        # pylint:disable=protected-access
        self.assertEqual(
            self.provider.networking.networks._service_event_pattern,
            "provider.networking.networks",
            "Event pattern for {} service should be '{}', "
            "but found '{}'.".format("networks",
                                     "provider.networking.networks",
                                     self.provider.networking.networks.
                                     _service_event_pattern))
        # pylint:disable=protected-access
        self.assertEqual(
            self.provider.networking.subnets._service_event_pattern,
            "provider.networking.subnets",
            "Event pattern for {} service should be '{}', "
            "but found '{}'.".format("subnets",
                                     "provider.networking.subnets",
                                     self.provider.networking.subnets.
                                     _service_event_pattern))
        # pylint:disable=protected-access
        self.assertEqual(
            self.provider.networking.routers._service_event_pattern,
            "provider.networking.routers",
            "Event pattern for {} service should be '{}', "
            "but found '{}'.".format("routers",
                                     "provider.networking.routers",
                                     self.provider.networking.routers.
                                     _service_event_pattern))

    @helpers.skipIfNoService(['networking.networks'])
    def test_crud_network(self):

        def create_net(label):
            return self.provider.networking.networks.create(
                label=label, cidr_block=BaseNetwork.CB_DEFAULT_IPV4RANGE)

        def cleanup_net(net):
            if net:
                net.delete()
                net.wait_for([NetworkState.UNKNOWN],
                             terminal_states=[NetworkState.ERROR])
                self.assertTrue(
                    net.state == NetworkState.UNKNOWN,
                    "Network.state must be unknown after "
                    "a delete but got %s"
                    % net.state)

        sit.check_crud(self, self.provider.networking.networks, Network,
                       "cb-crudnetwork", create_net, cleanup_net)

    @helpers.skipIfNoService(['networking.networks'])
    def test_network_properties(self):
        label = 'cb-propnetwork-{0}'.format(helpers.get_uuid())
        subnet_label = 'cb-propsubnet-{0}'.format(helpers.get_uuid())
        net = self.provider.networking.networks.create(
            label=label, cidr_block=BaseNetwork.CB_DEFAULT_IPV4RANGE)
        with cb_helpers.cleanup_action(lambda: helpers.cleanup_network(net)):
            net.wait_till_ready()
            self.assertEqual(
                net.state, 'available',
                "Network in state '%s', yet should be 'available'" % net.state)

            sit.check_repr(self, net)

            self.assertIn(
                net.cidr_block, ['', BaseNetwork.CB_DEFAULT_IPV4RANGE],
                "Network CIDR %s does not contain the expected value %s."
                % (net.cidr_block, BaseNetwork.CB_DEFAULT_IPV4RANGE))

            cidr = '10.0.20.0/24'
            sn = net.subnets.create(
                label=subnet_label, cidr_block=cidr)
            with cb_helpers.cleanup_action(lambda: helpers.cleanup_subnet(sn)):
                self.assertTrue(
                    sn in net.subnets,
                    "Subnet ID %s should be listed in network subnets %s."
                    % (sn.id, net.subnets))

                self.assertTrue(
                    sn in self.provider.networking.subnets.list(network=net),
                    "Subnet ID %s should be included in the subnets list %s."
                    % (sn.id, self.provider.networking.subnets.list(net))
                )

                self.assertListEqual(
                    list(net.subnets), [sn],
                    "Network should have exactly one subnet: %s." % sn.id)

                self.assertEqual(
                    net.id, sn.network_id,
                    "Network ID %s and subnet's network id %s should be"
                    " equal." % (net.id, sn.network_id))

                self.assertEqual(
                    net, sn.network,
                    "Network obj %s and subnet's parent net obj %s"
                    " should be equal." % (net, sn.network))

                self.assertEqual(
                    cidr, sn.cidr_block,
                    "Should be exact cidr block that was requested")

                self.assertTrue(
                    BaseNetwork.cidr_blocks_overlap(cidr, sn.cidr_block),
                    "Subnet's CIDR %s should overlap the specified one %s." % (
                        sn.cidr_block, cidr))

    def test_crud_subnet(self):
        # Late binding will make sure that create_subnet gets the
        # correct value
        net = self.provider.networking.networks.create(
                  label="cb-crudsubnet",
                  cidr_block=BaseNetwork.CB_DEFAULT_IPV4RANGE)

        def create_subnet(label):
            return self.provider.networking.subnets.create(
                label=label, network=net, cidr_block="10.0.10.0/24")

        def cleanup_subnet(subnet):
            if subnet:
                net = subnet.network
                subnet.delete()
                subnet.wait_for([SubnetState.UNKNOWN],
                                terminal_states=[SubnetState.ERROR])
                self.assertTrue(
                    subnet.state == SubnetState.UNKNOWN,
                    "Subnet.state must be unknown after "
                    "a delete but got %s"
                    % subnet.state)
                net.delete()
                net.wait_for([NetworkState.UNKNOWN],
                             terminal_states=[NetworkState.ERROR])
                self.assertTrue(
                    net.state == NetworkState.UNKNOWN,
                    "Network.state must be unknown after "
                    "a delete but got %s"
                    % net.state)

        sit.check_crud(self, self.provider.networking.subnets, Subnet,
                       "cb-crudsubnet", create_subnet, cleanup_subnet)

    def test_crud_floating_ip(self):
        gw = helpers.get_test_gateway(
            self.provider)

        def create_fip(label):
            fip = gw.floating_ips.create()
            return fip

        def cleanup_fip(fip):
            if fip:
                gw.floating_ips.delete(fip.id)

        with cb_helpers.cleanup_action(
                lambda: helpers.cleanup_gateway(gw)):
            sit.check_crud(self, gw.floating_ips, FloatingIP,
                           "cb-crudfip", create_fip, cleanup_fip,
                           skip_name_check=True)

    def test_floating_ip_properties(self):
        # Check floating IP address
        gw = helpers.get_test_gateway(
            self.provider)
        fip = gw.floating_ips.create()
        with cb_helpers.cleanup_action(
                lambda: helpers.cleanup_gateway(gw)):
            with cb_helpers.cleanup_action(lambda: fip.delete()):
                fipl = list(gw.floating_ips)
                self.assertIn(fip, fipl)
                # 2016-08: address filtering not implemented in moto
                # empty_ipl = self.provider.network.floating_ips('dummy-net')
                # self.assertFalse(
                #     empty_ipl,
                #     "Bogus network should not have any floating IPs: {0}"
                #     .format(empty_ipl))
                self.assertFalse(
                    fip.private_ip,
                    "Floating IP should not have a private IP value ({0})."
                    .format(fip.private_ip))
                self.assertFalse(
                    fip.in_use,
                    "Newly created floating IP address should not be in use.")

    @helpers.skipIfNoService(['networking.routers'])
    def test_crud_router(self):

        def _cleanup(net, subnet, router, gateway):
            with cb_helpers.cleanup_action(
                    lambda: helpers.cleanup_network(net)):
                with cb_helpers.cleanup_action(
                        lambda: helpers.cleanup_subnet(subnet)):
                    with cb_helpers.cleanup_action(
                            lambda: router.delete()):
                        with cb_helpers.cleanup_action(
                                lambda: helpers.cleanup_gateway(gateway)):
                            router.detach_subnet(subnet)
                            router.detach_gateway(gateway)

        label = 'cb-crudrouter-{0}'.format(helpers.get_uuid())
        # Declare these variables and late binding will allow
        # the cleanup method access to the most current values
        net = None
        sn = None
        router = None
        gteway = None
        with cb_helpers.cleanup_action(
                lambda: _cleanup(net, sn, router, gteway)):
            net = self.provider.networking.networks.create(
                label=label, cidr_block=BaseNetwork.CB_DEFAULT_IPV4RANGE)
            router = self.provider.networking.routers.create(label=label,
                                                             network=net)
            cidr = '10.0.15.0/24'
            sn = net.subnets.create(label=label, cidr_block=cidr)

            # Check basic router properties
            sit.check_standard_behaviour(
                self, self.provider.networking.routers, router)
            if (self.provider.PROVIDER_ID != 'gcp'):
                if (self.provider.PROVIDER_ID != 'mock'):
                    self.assertEqual(
                        router.state, RouterState.DETACHED,
                        "Router {0} state {1} should be {2}.".format(
                            router.id, router.state, RouterState.DETACHED))

#                 self.assertEqual(
#                     router.network_id, net.id,  "Router {0} should be assoc."
#                     " with network {1}, but is associated with {2}"
#                     .format(router.id, net.id, router.network_id))

                self.assertTrue(
                    len(router.subnets) == 0,
                    "No subnet should be attached to router {0} but subnet {1}"
                    " is currently attached to it.".format(router, sn)
                )
                router.attach_subnet(sn)
                self.assertTrue(
                    len(router.subnets) == 1,
                    "Subnet {0} not attached to router {1}".format(sn, router)
                )
            gteway = net.gateways.get_or_create()
            router.attach_gateway(gteway)
            # TODO: add a check for routes after that's been implemented

        sit.check_delete(self, self.provider.networking.routers, router)
        # Also make sure that linked resources were properly cleaned up
        sit.check_delete(self, self.provider.networking.subnets, sn)
        sit.check_delete(self, self.provider.networking.networks, net)

    @helpers.skipIfNoService(['networking.networks'])
    def test_default_network(self):
        subnet = self.provider.networking.subnets.get_or_create_default()
        self.assertIsInstance(subnet, Subnet)