path: root/nova/tests/api/openstack/compute/contrib/test_networks.py

# Copyright 2011 Grid Dynamics
# Copyright 2011 OpenStack Foundation
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import copy
import math
import netaddr
import uuid

from oslo.config import cfg
import webob

from nova.api.openstack.compute.contrib import networks_associate
from nova.api.openstack.compute.contrib import os_networks as networks
from nova import exception
from nova import test
from nova.tests.api.openstack import fakes

CONF = cfg.CONF

FAKE_NETWORKS = [
    {
        'bridge': 'br100', 'vpn_public_port': 1000,
        'dhcp_start': '10.0.0.3', 'bridge_interface': 'eth0',
        'updated_at': '2011-08-16 09:26:13.048257',
        'id': 1, 'uuid': '20c8acc0-f747-4d71-a389-46d078ebf047',
        'cidr_v6': None, 'deleted_at': None,
        'gateway': '10.0.0.1', 'label': 'mynet_0',
        'project_id': '1234', 'rxtx_base': None,
        'vpn_private_address': '10.0.0.2', 'deleted': False,
        'vlan': 100, 'broadcast': '10.0.0.7',
        'netmask': '255.255.255.248', 'injected': False,
        'cidr': '10.0.0.0/29',
        'vpn_public_address': '127.0.0.1', 'multi_host': False,
        'dns1': None, 'dns2': None, 'host': 'nsokolov-desktop',
        'gateway_v6': None, 'netmask_v6': None, 'priority': None,
        'created_at': '2011-08-15 06:19:19.387525',
    },
    {
        'bridge': 'br101', 'vpn_public_port': 1001,
        'dhcp_start': '10.0.0.11', 'bridge_interface': 'eth0',
        'updated_at': None, 'id': 2, 'cidr_v6': None,
        'uuid': '20c8acc0-f747-4d71-a389-46d078ebf000',
        'deleted_at': None, 'gateway': '10.0.0.9',
        'label': 'mynet_1', 'project_id': None,
        'vpn_private_address': '10.0.0.10', 'deleted': False,
        'vlan': 101, 'broadcast': '10.0.0.15', 'rxtx_base': None,
        'netmask': '255.255.255.248', 'injected': False,
        'cidr': '10.0.0.10/29', 'vpn_public_address': None,
        'multi_host': False, 'dns1': None, 'dns2': None, 'host': None,
        'gateway_v6': None, 'netmask_v6': None, 'priority': None,
        'created_at': '2011-08-15 06:19:19.885495',
    },
]


FAKE_USER_NETWORKS = [
    {
        'id': 1, 'cidr': '10.0.0.0/29', 'netmask': '255.255.255.248',
        'gateway': '10.0.0.1', 'broadcast': '10.0.0.7', 'dns1': None,
        'dns2': None, 'cidr_v6': None, 'gateway_v6': None, 'label': 'mynet_0',
        'netmask_v6': None, 'uuid': '20c8acc0-f747-4d71-a389-46d078ebf047',
    },
    {
        'id': 2, 'cidr': '10.0.0.10/29', 'netmask': '255.255.255.248',
        'gateway': '10.0.0.9', 'broadcast': '10.0.0.15', 'dns1': None,
        'dns2': None, 'cidr_v6': None, 'gateway_v6': None, 'label': 'mynet_1',
        'netmask_v6': None, 'uuid': '20c8acc0-f747-4d71-a389-46d078ebf000',
    },
]

NEW_NETWORK = {
    "network": {
        "bridge_interface": "eth0",
        "cidr": "10.20.105.0/24",
        "label": "new net 111",
        "vlan_start": 111,
    }
}


class FakeNetworkAPI(object):

    _sentinel = object()
    _vlan_is_disabled = False

    def __init__(self):
        self.networks = copy.deepcopy(FAKE_NETWORKS)

    def disable_vlan(self):
        self._vlan_is_disabled = True

    def delete(self, context, network_id):
        for i, network in enumerate(self.networks):
            if network['id'] == network_id:
                del self.networks[0]
                return True
        raise exception.NetworkNotFoundForUUID(uuid=network_id)

    def disassociate(self, context, network_uuid):
        for network in self.networks:
            if network.get('uuid') == network_uuid:
                network['project_id'] = None
                return True
        raise exception.NetworkNotFound(network_id=network_uuid)

    def associate(self, context, network_uuid, host=_sentinel,
                  project=_sentinel):
        for network in self.networks:
            if network.get('uuid') == network_uuid:
                if host is not FakeNetworkAPI._sentinel:
                    network['host'] = host
                if project is not FakeNetworkAPI._sentinel:
                    network['project_id'] = project
                return True
        raise exception.NetworkNotFound(network_id=network_uuid)

    def add_network_to_project(self, context,
                               project_id, network_uuid=None):
        if self._vlan_is_disabled:
            raise NotImplementedError()
        if network_uuid:
            for network in self.networks:
                if network.get('project_id', None) is None:
                    network['project_id'] = project_id
                    return
            return
        for network in self.networks:
            if network.get('uuid') == network_uuid:
                network['project_id'] = project_id
                return

    def get_all(self, context):
        return self.networks

    def get(self, context, network_id):
        for network in self.networks:
            if network.get('uuid') == network_id:
                return network
        raise exception.NetworkNotFound(network_id=network_id)

    def create(self, context, **kwargs):
        subnet_bits = int(math.ceil(math.log(kwargs.get(
                        'network_size', CONF.network_size), 2)))
        fixed_net_v4 = netaddr.IPNetwork(kwargs['cidr'])
        prefixlen_v4 = 32 - subnet_bits
        subnets_v4 = list(fixed_net_v4.subnet(
                prefixlen_v4,
                count=kwargs.get('num_networks', CONF.num_networks)))
        new_networks = []
        new_id = max((net['id'] for net in self.networks))
        for index, subnet_v4 in enumerate(subnets_v4):
            new_id += 1
            net = {'id': new_id, 'uuid': str(uuid.uuid4())}

            net['cidr'] = str(subnet_v4)
            net['netmask'] = str(subnet_v4.netmask)
            net['gateway'] = kwargs.get('gateway') or str(subnet_v4[1])
            net['broadcast'] = str(subnet_v4.broadcast)
            net['dhcp_start'] = str(subnet_v4[2])

            for key in FAKE_NETWORKS[0].iterkeys():
                net.setdefault(key, kwargs.get(key))
            new_networks.append(net)
        self.networks += new_networks
        return new_networks


class NetworksTest(test.TestCase):

    def setUp(self):
        super(NetworksTest, self).setUp()
        self.fake_network_api = FakeNetworkAPI()
        self.controller = networks.NetworkController(
                                                self.fake_network_api)
        self.associate_controller = networks_associate\
            .NetworkAssociateActionController(self.fake_network_api)
        fakes.stub_out_networking(self.stubs)
        fakes.stub_out_rate_limiting(self.stubs)

    @staticmethod
    def network_uuid_to_id(network):
        network['id'] = network['uuid']
        del network['uuid']

    def test_network_list_all_as_user(self):
        self.maxDiff = None
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks')
        res_dict = self.controller.index(req)
        expected = copy.deepcopy(FAKE_USER_NETWORKS)
        for network in expected:
            self.network_uuid_to_id(network)
        self.assertEquals(res_dict, {'networks': expected})

    def test_network_list_all_as_admin(self):
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks')
        req.environ["nova.context"].is_admin = True
        res_dict = self.controller.index(req)
        expected = copy.deepcopy(FAKE_NETWORKS)
        for network in expected:
            self.network_uuid_to_id(network)
        self.assertEquals(res_dict, {'networks': expected})

    def test_network_disassociate(self):
        uuid = FAKE_NETWORKS[0]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s/action' % uuid)
        res = self.controller._disassociate_host_and_project(
            req, uuid, {'disassociate': None})
        self.assertEqual(res.status_int, 202)
        self.assertEqual(self.fake_network_api.networks[0]['project_id'], None)
        self.assertEqual(self.fake_network_api.networks[0]['host'], None)

    def test_network_disassociate_host_only(self):
        uuid = FAKE_NETWORKS[0]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s/action' % uuid)
        res = self.associate_controller._disassociate_host_only(
            req, uuid, {'disassociate_host': None})
        self.assertEqual(res.status_int, 202)
        self.assertNotEqual(self.fake_network_api.networks[0]['project_id'],
                            None)
        self.assertEqual(self.fake_network_api.networks[0]['host'], None)

    def test_network_disassociate_project_only(self):
        uuid = FAKE_NETWORKS[0]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s/action' % uuid)
        res = self.associate_controller._disassociate_project_only(
            req, uuid, {'disassociate_project': None})
        self.assertEqual(res.status_int, 202)
        self.assertEqual(self.fake_network_api.networks[0]['project_id'], None)
        self.assertNotEqual(self.fake_network_api.networks[0]['host'], None)

    def test_network_disassociate_not_found(self):
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/100/action')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.controller._disassociate_host_and_project,
                          req, 100, {'disassociate': None})

    def test_network_get_as_user(self):
        uuid = FAKE_USER_NETWORKS[0]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s' % uuid)
        res_dict = self.controller.show(req, uuid)
        expected = {'network': copy.deepcopy(FAKE_USER_NETWORKS[0])}
        self.network_uuid_to_id(expected['network'])
        self.assertEqual(res_dict, expected)

    def test_network_get_as_admin(self):
        uuid = FAKE_NETWORKS[0]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s' % uuid)
        req.environ["nova.context"].is_admin = True
        res_dict = self.controller.show(req, uuid)
        expected = {'network': copy.deepcopy(FAKE_NETWORKS[0])}
        self.network_uuid_to_id(expected['network'])
        self.assertEqual(res_dict, expected)

    def test_network_get_not_found(self):
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/100')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.controller.show, req, 100)

    def test_network_delete(self):
        uuid = FAKE_NETWORKS[0]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s' % uuid)
        res = self.controller.delete(req, 1)
        self.assertEqual(res.status_int, 202)

    def test_network_delete_not_found(self):
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/100')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.controller.delete, req, 100)

    def test_network_add_vlan_disabled(self):
        self.fake_network_api.disable_vlan()
        uuid = FAKE_NETWORKS[1]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/add')
        self.assertRaises(webob.exc.HTTPNotImplemented,
                          self.controller.add, req, {'id': uuid})

    def test_network_add(self):
        uuid = FAKE_NETWORKS[1]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/add')
        res = self.controller.add(req, {'id': uuid})
        self.assertEqual(res.status_int, 202)
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s' % uuid)
        req.environ["nova.context"].is_admin = True
        res_dict = self.controller.show(req, uuid)
        self.assertEqual(res_dict['network']['project_id'], 'fake')

    def test_network_associate_with_host(self):
        uuid = FAKE_NETWORKS[1]['uuid']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s/action' % uuid)
        res = self.associate_controller._associate_host(
            req, uuid, {'associate_host': "TestHost"})
        self.assertEqual(res.status_int, 202)
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s' % uuid)
        req.environ["nova.context"].is_admin = True
        res_dict = self.controller.show(req, uuid)
        self.assertEqual(res_dict['network']['host'], 'TestHost')

    def test_network_create(self):
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks')
        res_dict = self.controller.create(req, NEW_NETWORK)
        self.assertTrue('network' in res_dict)
        uuid = res_dict['network']['id']
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks/%s' % uuid)
        res_dict = self.controller.show(req, uuid)
        self.assertTrue(res_dict['network']['label'].
                        startswith(NEW_NETWORK['network']['label']))

    def test_network_create_large(self):
        req = fakes.HTTPRequest.blank('/v2/1234/os-networks')
        large_network = copy.deepcopy(NEW_NETWORK)
        large_network['network']['cidr'] = '128.0.0.0/4'
        res_dict = self.controller.create(req, large_network)
        self.assertEqual(res_dict['network']['cidr'],
                         large_network['network']['cidr'])