remove nova code related to Quantum v1 API

bug 1034692

Quantum team is removing all v1 support in Folsom, so this code
will not be used or supported with Quantum Folsom release.

Change-Id: I4c397d5238fcc3a15320e052bdc4735dfec1f176

3 files changed, 1 insertions, 891 deletions

diff --git a/nova/tests/fake_network.py b/nova/tests/fake_network.py
index 5c945fa75..476d7eab4 100644
--- a/nova/tests/fake_network.py
+++ b/nova/tests/fake_network.py
@@ -20,7 +20,7 @@ from nova import db
 from nova import exception
 from nova import flags
 from nova.network import manager as network_manager
-from nova.network.quantum import nova_ipam_lib
+from nova.network import nova_ipam_lib
 from nova import utils
 
 
diff --git a/nova/tests/network/test_quantum.py b/nova/tests/network/test_quantum.py
deleted file mode 100644
index b126994d0..000000000
--- a/nova/tests/network/test_quantum.py
+++ /dev/null
@@ -1,647 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2011,2012 Nicira, Inc.
-# 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 mox
-
-from nova import context
-from nova import db
-from nova.db.sqlalchemy import models
-from nova.db.sqlalchemy import session as sql_session
-from nova import exception
-from nova import flags
-from nova.network.quantum import client as quantum_client
-from nova.network.quantum import fake_client
-from nova.network.quantum import manager as quantum_manager
-from nova.network.quantum import melange_connection
-from nova.network.quantum import melange_ipam_lib
-from nova.network.quantum import quantum_connection
-from nova.openstack.common import log as logging
-
-from nova import test
-from nova import utils
-
-LOG = logging.getLogger(__name__)
-FLAGS = flags.FLAGS
-
-
-networks = [{'label': 'project1-net1',
-             'injected': False,
-             'multi_host': False,
-             'cidr': '100.168.0.0/24',
-             'cidr_v6': '100:1db8::/64',
-             'gateway_v6': '100:1db8::1',
-             'netmask_v6': '64',
-             'netmask': '255.255.255.0',
-             'bridge': None,
-             'bridge_interface': None,
-             'gateway': '100.168.0.1',
-             'broadcast': '100.168.0.255',
-             'dns1': '8.8.8.8',
-             'vlan': None,
-             'host': None,
-             'vpn_public_address': None,
-             'project_id': 'fake_project1',
-             'priority': 1},
-            {'label': 'project2-net1',
-             'injected': False,
-             'multi_host': False,
-             'cidr': '101.168.1.0/24',
-             'cidr_v6': '101:1db9::/64',
-             'gateway_v6': '101:1db9::1',
-             'netmask_v6': '64',
-             'netmask': '255.255.255.0',
-             'bridge': None,
-             'bridge_interface': None,
-             'gateway': '101.168.1.1',
-             'broadcast': '101.168.1.255',
-             'dns1': '8.8.8.8',
-             'vlan': None,
-             'host': None,
-             'project_id': 'fake_project2',
-             'priority': 1},
-             {'label': "public",
-             'injected': False,
-             'multi_host': False,
-             'cidr': '102.0.0.0/24',
-             'cidr_v6': '102:1dba::/64',
-             'gateway_v6': '102:1dba::1',
-             'netmask_v6': '64',
-             'netmask': '255.255.255.0',
-             'bridge': None,
-             'bridge_interface': None,
-             'gateway': '102.0.0.1',
-             'broadcast': '102.0.0.255',
-             'dns1': '8.8.8.8',
-             'vlan': None,
-             'host': None,
-             'project_id': None,
-             'priority': 0},
-             {'label': "project2-net2",
-             'injected': False,
-             'multi_host': False,
-             'cidr': '103.0.0.0/24',
-             'cidr_v6': '103:1dbb::/64',
-             'gateway_v6': '103:1dbb::1',
-             'netmask_v6': '64',
-             'netmask': '255.255.255.0',
-             'bridge': None,
-             'bridge_interface': None,
-             'gateway': '103.0.0.1',
-             'broadcast': '103.0.0.255',
-             'dns1': '8.8.8.8',
-             'vlan': None,
-             'host': None,
-             'project_id': "fake_project2",
-             'priority': 2}]
-
-
-class QuantumConnectionTestCase(test.TestCase):
-
-    def test_connection(self):
-        fc = fake_client.FakeClient(LOG)
-        qc = quantum_connection.QuantumClientConnection(client=fc)
-        t = "tenant1"
-        net1_name = "net1"
-        net1_uuid = qc.create_network(t, net1_name)
-        self.assertEquals(net1_name, qc.get_network_name(t, net1_uuid))
-        self.assertTrue(qc.network_exists(t, net1_uuid))
-        self.assertFalse(qc.network_exists(t, "fake-uuid"))
-        self.assertFalse(qc.network_exists("fake-tenant", net1_uuid))
-
-        nets = qc.get_networks(t)['networks']
-        self.assertEquals(len(nets), 1)
-        self.assertEquals(nets[0]['id'], net1_uuid)
-
-        num_ports = 10
-        for i in range(0, num_ports):
-            qc.create_and_attach_port(t, net1_uuid,
-                                'iface' + str(i), state='ACTIVE')
-
-        self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), num_ports)
-
-        for i in range(0, num_ports):
-            port_uuid = qc.get_port_by_attachment(t, net1_uuid,
-                                'iface' + str(i))
-            self.assertTrue(port_uuid)
-            qc.detach_and_delete_port(t, net1_uuid, port_uuid)
-
-        self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), 0)
-
-        # test port not found
-        qc.create_and_attach_port(t, net1_uuid, 'foo', state='ACTIVE')
-        port_uuid = qc.get_port_by_attachment(t, net1_uuid, 'foo')
-        qc.detach_and_delete_port(t, net1_uuid, port_uuid)
-        self.assertRaises(quantum_client.QuantumNotFoundException,
-                            qc.detach_and_delete_port, t,
-                            net1_uuid, port_uuid)
-
-        qc.delete_network(t, net1_uuid)
-        self.assertFalse(qc.network_exists(t, net1_uuid))
-        self.assertEquals(len(qc.get_networks(t)['networks']), 0)
-
-        self.assertRaises(quantum_client.QuantumNotFoundException,
-                            qc.get_network_name, t, net1_uuid)
-
-
-# this is a base class to be used by other QuantumManager Test classes
-class QuantumNovaTestCase(test.TestCase):
-
-    def setUp(self):
-        super(QuantumNovaTestCase, self).setUp()
-
-        self.flags(quantum_use_dhcp=True)
-        self.flags(l3_lib="nova.network.l3.LinuxNetL3")
-        linuxdrv = "nova.network.linux_net.LinuxOVSInterfaceDriver"
-        self.flags(linuxnet_interface_driver=linuxdrv)
-        fc = fake_client.FakeClient(LOG)
-        qc = quantum_connection.QuantumClientConnection(client=fc)
-
-        self.net_man = quantum_manager.QuantumManager(
-                ipam_lib="nova.network.quantum.nova_ipam_lib",
-                q_conn=qc)
-
-        def func(arg1, arg2):
-            pass
-
-        def func2(arg1, arg2, arg3):
-            pass
-
-        def func1(arg1):
-            pass
-
-        self.net_man.driver.update_dhcp_hostfile_with_text = func
-        self.net_man.driver.restart_dhcp = func2
-        self.net_man.driver.kill_dhcp = func1
-
-        # Tests seem to create some networks by default, which
-        # we don't want.  So we delete them.
-
-        ctx = context.RequestContext('user1', 'fake_project1').elevated()
-        for n in db.network_get_all(ctx):
-            db.network_delete_safe(ctx, n['id'])
-
-        # Other unit tests (e.g., test_compute.py) have a nasty
-        # habit of of creating fixed IPs and not cleaning up, which
-        # can confuse these tests, so we remove all existing fixed
-        # ips before starting.
-        session = sql_session.get_session()
-        result = session.query(models.FixedIp).all()
-        with session.begin():
-            for fip_ref in result:
-                session.delete(fip_ref)
-
-        self.net_man.init_host()
-
-    def _create_network(self, n):
-        ctx = context.RequestContext('user1', n['project_id'])
-        nwks = self.net_man.create_networks(
-            ctx,
-            label=n['label'], cidr=n['cidr'],
-            multi_host=n['multi_host'],
-            num_networks=1, network_size=256,
-            cidr_v6=n['cidr_v6'],
-            gateway=n['gateway'],
-            gateway_v6=n['gateway_v6'], bridge=None,
-            bridge_interface=None, dns1=n['dns1'],
-            project_id=n['project_id'],
-            priority=n['priority'])
-        n['uuid'] = nwks[0]['uuid']
-
-
-class QuantumAllocationTestCase(QuantumNovaTestCase):
-    def test_get_network_in_db(self):
-        context = self.mox.CreateMockAnything()
-        context.elevated().AndReturn('elevated')
-        self.mox.StubOutWithMock(db, 'network_get_by_uuid')
-        self.net_man.context = context
-        db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn(
-                                                                {'uuid': 1})
-
-        self.mox.ReplayAll()
-
-        network = self.net_man.get_network(context, ('quantum_net_id',
-                                                     'net_tenant_id'))
-        self.assertEquals(network['quantum_net_id'], 'quantum_net_id')
-        self.assertEquals(network['uuid'], 1)
-
-    def test_get_network_not_in_db(self):
-        context = self.mox.CreateMockAnything()
-        context.elevated().AndReturn('elevated')
-        self.mox.StubOutWithMock(db, 'network_get_by_uuid')
-        self.net_man.context = context
-        db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn(None)
-
-        self.mox.ReplayAll()
-
-        network = self.net_man.get_network(context, ('quantum_net_id',
-                                                     'net_tenant_id'))
-        self.assertEquals(network['quantum_net_id'], 'quantum_net_id')
-        self.assertEquals(network['uuid'], 'quantum_net_id')
-
-
-class QuantumDeallocationTestCase(QuantumNovaTestCase):
-    def test_deallocate_port(self):
-        quantum = self.mox.CreateMock(
-                                quantum_connection.QuantumClientConnection)
-        quantum.get_port_by_attachment('q_tenant_id', 'net_id',
-                                       'interface_id').AndReturn('port_id')
-        quantum.detach_and_delete_port('q_tenant_id', 'net_id', 'port_id')
-        self.net_man.q_conn = quantum
-
-        self.mox.ReplayAll()
-
-        self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id',
-                                     'instance_id')
-
-    def test_deallocate_port_logs_error(self):
-        quantum = self.mox.CreateMock(
-            quantum_connection.QuantumClientConnection)
-        quantum.get_port_by_attachment('q_tenant_id', 'net_id',
-                            'interface_id').AndRaise(Exception)
-        self.net_man.q_conn = quantum
-
-        self.mox.StubOutWithMock(quantum_manager.LOG, 'exception')
-        quantum_manager.LOG.exception(mox.Regex(r'port deallocation failed'))
-
-        self.mox.ReplayAll()
-
-        self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id',
-                                     'instance_id')
-
-    def test_deallocate_ip_address(self):
-        ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib)
-        ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1},
-                                     'project_id').AndReturn('ipam_tenant_id')
-        self.net_man.ipam = ipam
-        self.mox.ReplayAll()
-        self.net_man.deallocate_ip_address('context', 'net_id', 'project_id',
-                {'uuid': 1}, 'instance_id')
-
-    def test_deallocate_ip_address_2(self):
-        ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib)
-        ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1},
-                                     'project_id').AndRaise(Exception())
-        self.net_man.ipam = ipam
-
-        self.mox.StubOutWithMock(quantum_manager.LOG, 'exception')
-        quantum_manager.LOG.exception(mox.Regex(r'ipam deallocation failed'))
-
-        self.mox.ReplayAll()
-        self.net_man.deallocate_ip_address('context', 'net_id', 'project_id',
-                {'uuid': 1}, 'instance_id')
-
-
-class QuantumManagerTestCase(QuantumNovaTestCase):
-    def test_create_and_delete_nets(self):
-        self._create_nets()
-        self._delete_nets()
-
-    def _create_nets(self):
-        for n in networks:
-            self._create_network(n)
-
-    def _delete_nets(self):
-        for n in networks:
-            ctx = context.RequestContext('user1', n['project_id'])
-            self.net_man.delete_network(ctx, None, n['uuid'])
-        self.assertRaises(exception.NoNetworksFound,
-                          db.network_get_all, ctx.elevated())
-
-    def _validate_nw_info(self, nw_info, expected_net_labels):
-
-        self.assertEquals(len(nw_info), len(expected_net_labels))
-
-        ctx = context.RequestContext('user1', 'foo').elevated()
-        all_net_map = {}
-        for n in db.network_get_all(ctx):
-            all_net_map[n['label']] = n
-
-        for i in range(0, len(nw_info)):
-            vif = nw_info[i]
-            net = all_net_map[expected_net_labels[i]]
-
-            # simple test assumes that each starting prefix is unique
-            expected_v4_cidr_start = net['cidr'].split(".")[0].lower()
-            expected_v6_cidr_start = net['cidr_v6'].split(":")[0].lower()
-
-            for subnet in vif['network']['subnets']:
-                addr = subnet['ips'][0]['address']
-                if subnet['version'] == 4:
-                    address_start = addr.split(".")[0].lower()
-                    self.assertTrue(expected_v4_cidr_start, address_start)
-                else:
-                    address_start = addr.split(":")[0].lower()
-                    self.assertTrue(expected_v6_cidr_start, address_start)
-
-        # confirm that there is a DHCP device on corresponding net
-        for l in expected_net_labels:
-            n = all_net_map[l]
-            tenant_id = (n['project_id'] or
-                                FLAGS.quantum_default_tenant_id)
-            ports = self.net_man.q_conn.get_attached_ports(
-                                                    tenant_id, n['uuid'])
-            self.assertEquals(len(ports), 2)  # gw + instance VIF
-
-            # make sure we aren't allowed to delete network with
-            # active port
-            self.assertRaises(exception.NetworkBusy,
-                              self.net_man.delete_network,
-                              ctx, None, n['uuid'])
-
-    def _check_vifs(self, expect_num_vifs):
-        ctx = context.RequestContext('user1', "").elevated()
-        self.assertEqual(len(db.virtual_interface_get_all(ctx)),
-                        expect_num_vifs)
-
-    def _allocate_and_deallocate_instance(self, project_id, requested_networks,
-                                            expected_labels):
-
-        ctx = context.RequestContext('user1', project_id)
-        self._check_vifs(0)
-
-        instance_ref = db.instance_create(ctx,
-                                    {"project_id": project_id})
-
-        nw_info = self.net_man.allocate_for_instance(ctx.elevated(),
-                        instance_id=instance_ref['id'], host="",
-                        rxtx_factor=3,
-                        project_id=project_id,
-                        requested_networks=requested_networks)
-
-        self._check_vifs(len(nw_info))
-
-        self._validate_nw_info(nw_info, expected_labels)
-
-        nw_info = self.net_man.get_instance_nw_info(ctx, instance_ref['id'],
-                                instance_ref['uuid'],
-                                instance_ref['instance_type_id'], "",
-                                project_id=project_id)
-
-        self._check_vifs(len(nw_info))
-        self._validate_nw_info(nw_info, expected_labels)
-
-        port_net_pairs = []
-        for vif in nw_info:
-            nid = vif['network']['id']
-            pid = self.net_man.q_conn.get_port_by_attachment(
-                                project_id, nid, vif['id'])
-            if pid is None:
-                pid = self.net_man.q_conn.get_port_by_attachment(
-                                FLAGS.quantum_default_tenant_id,
-                                nid, vif['id'])
-            self.assertTrue(pid is not None)
-            port_net_pairs.append((pid, nid))
-
-        self.net_man.deallocate_for_instance(ctx,
-                    instance_id=instance_ref['id'],
-                    project_id=project_id)
-
-        for pid, nid in port_net_pairs:
-            self.assertRaises(quantum_client.QuantumNotFoundException,
-                            self.net_man.q_conn.detach_and_delete_port,
-                            project_id, nid, pid)
-            self.assertRaises(quantum_client.QuantumNotFoundException,
-                            self.net_man.q_conn.detach_and_delete_port,
-                            FLAGS.quantum_default_tenant_id, nid, pid)
-
-        self._check_vifs(0)
-
-    def test_allocate_and_deallocate_instance_static(self):
-        self._create_nets()
-        self._allocate_and_deallocate_instance("fake_project1", None,
-                                 ['public', 'project1-net1'])
-        self._delete_nets()
-
-    def test_allocate_and_deallocate_instance_dynamic(self):
-
-        self._create_nets()
-        project_id = "fake_project2"
-        ctx = context.RequestContext('user1', project_id)
-        all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
-                                                               ctx, project_id)
-        requested_networks = [(n[0], None) for n in all_valid_networks]
-
-        self.net_man.validate_networks(ctx, requested_networks)
-
-        label_map = {}
-        for n in db.network_get_all(ctx.elevated()):
-            label_map[n['uuid']] = n['label']
-        expected_labels = [label_map[uid] for uid, _i in requested_networks]
-
-        self._allocate_and_deallocate_instance(project_id, requested_networks,
-                                              expected_labels)
-        self._delete_nets()
-
-    def test_validate_bad_network(self):
-        ctx = context.RequestContext('user1', 'fake_project1')
-        self.assertRaises(exception.NetworkNotFound,
-                        self.net_man.validate_networks, ctx, [("", None)])
-
-    def test_create_net_external_uuid(self):
-        """Tests use case where network can be created directly via
-           Quantum API, then the UUID is passed in via nova-manage"""
-        project_id = "foo_project"
-        ctx = context.RequestContext('user1', project_id)
-        net_id = self.net_man.q_conn.create_network(project_id, 'net1')
-        self.net_man.create_networks(
-            ctx,
-            label='achtungbaby',
-            cidr="9.9.9.0/24",
-            multi_host=False,
-            num_networks=1,
-            network_size=256,
-            cidr_v6=None,
-            gateway="9.9.9.1",
-            gateway_v6=None,
-            bridge=None,
-            bridge_interface=None,
-            dns1="8.8.8.8",
-            project_id=project_id,
-            priority=9,
-            uuid=net_id)
-        net = db.network_get_by_uuid(ctx.elevated(), net_id)
-        self.assertTrue(net is not None)
-        self.assertEquals(net['uuid'], net_id)
-
-    def test_create_net_external_uuid_and_host_is_set(self):
-        """Make sure network['host'] is set when creating a network via the
-           network manager"""
-        project_id = "foo_project"
-        ctx = context.RequestContext('user1', project_id)
-        net_id = self.net_man.q_conn.create_network(project_id, 'net2')
-        self.net_man.create_networks(
-            ctx, label='achtungbaby2', cidr="9.9.8.0/24", multi_host=False,
-            num_networks=1, network_size=256, cidr_v6=None,
-            gateway="9.9.8.1", gateway_v6=None, bridge=None,
-            bridge_interface=None, dns1="8.8.8.8", project_id=project_id,
-            priority=8, uuid=net_id)
-        net = db.network_get_by_uuid(ctx.elevated(), net_id)
-        self.assertTrue(net is not None)
-        self.assertEquals(net['uuid'], net_id)
-        self.assertTrue(net['host'] != None)
-
-
-class QuantumNovaMACGenerationTestCase(QuantumNovaTestCase):
-    def test_local_mac_address_creation(self):
-        self.flags(use_melange_mac_generation=False)
-        fake_mac = "ab:cd:ef:ab:cd:ef"
-        self.stubs.Set(utils, "generate_mac_address",
-                       lambda: fake_mac)
-        project_id = "fake_project1"
-        ctx = context.RequestContext('user1', project_id)
-        self._create_network(networks[0])
-
-        all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
-                                                               ctx, project_id)
-        requested_networks = [(n[0], None) for n in all_valid_networks]
-
-        instance_ref = db.api.instance_create(ctx,
-                                    {"project_id": project_id})
-        nw_info = self.net_man.allocate_for_instance(ctx,
-                        instance_id=instance_ref['id'], host="",
-                        rxtx_factor=3,
-                        project_id=project_id,
-                        requested_networks=requested_networks)
-        self.assertEqual(nw_info[0]['address'], fake_mac)
-
-    def test_melange_mac_address_creation(self):
-        self.flags(use_melange_mac_generation=True)
-        fake_mac = "ab:cd:ef:ab:cd:ef"
-        self.stubs.Set(melange_connection.MelangeConnection, "create_vif",
-                       lambda w, x, y, z: fake_mac)
-        project_id = "fake_project1"
-        ctx = context.RequestContext('user1', project_id)
-        self._create_network(networks[0])
-
-        all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
-                                                               ctx, project_id)
-        requested_networks = [(n[0], None) for n in all_valid_networks]
-
-        instance_ref = db.api.instance_create(ctx,
-                                    {"project_id": project_id})
-        nw_info = self.net_man.allocate_for_instance(ctx,
-                        instance_id=instance_ref['id'], host="",
-                        rxtx_factor=3,
-                        project_id=project_id,
-                        requested_networks=requested_networks)
-        self.assertEqual(nw_info[0]['address'], fake_mac)
-
-
-class QuantumNovaPortSecurityTestCase(QuantumNovaTestCase):
-    def test_port_security(self):
-        self.flags(use_melange_mac_generation=True)
-        self.flags(quantum_use_port_security=True)
-        fake_mac = "ab:cd:ef:ab:cd:ef"
-        self.stubs.Set(melange_connection.MelangeConnection, "create_vif",
-                       lambda w, x, y, z: fake_mac)
-        project_id = "fake_project1"
-        ctx = context.RequestContext('user1', project_id)
-        self._create_network(networks[0])
-
-        all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
-                                                               ctx, project_id)
-        requested_networks = [(n[0], None) for n in all_valid_networks]
-
-        instance_ref = db.api.instance_create(ctx,
-                                    {"project_id": project_id})
-        oldfunc = self.net_man.q_conn.create_and_attach_port
-
-        # Make sure we get the appropriate mac set in allowed_address_pairs
-        # if port security is enabled.
-        def _instrumented_create_and_attach_port(tenant_id, net_id,
-                                                 interface_id, **kwargs):
-            self.assertTrue('allowed_address_pairs' in kwargs.keys())
-            pairs = kwargs['allowed_address_pairs']
-            self.assertTrue(pairs[0]['mac_address'] == fake_mac)
-            self.net_man.q_conn.create_and_attach_port = oldfunc
-            return oldfunc(tenant_id, net_id, interface_id, **kwargs)
-        _port_attach = _instrumented_create_and_attach_port
-        self.net_man.q_conn.create_and_attach_port = _port_attach
-        nw_info = self.net_man.allocate_for_instance(ctx,
-                        instance_id=instance_ref['id'], host="",
-                        rxtx_factor=3,
-                        project_id=project_id,
-                        requested_networks=requested_networks)
-        self.assertEqual(nw_info[0]['address'], fake_mac)
-
-    def test_port_security_negative(self):
-        self.flags(use_melange_mac_generation=True)
-        self.flags(quantum_use_port_security=False)
-        fake_mac = "ab:cd:ef:ab:cd:ef"
-        self.stubs.Set(melange_connection.MelangeConnection, "create_vif",
-                       lambda w, x, y, z: fake_mac)
-        project_id = "fake_project1"
-        ctx = context.RequestContext('user1', project_id)
-        self._create_network(networks[0])
-
-        all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
-                                                               ctx, project_id)
-        requested_networks = [(n[0], None) for n in all_valid_networks]
-
-        instance_ref = db.api.instance_create(ctx,
-                                              {"project_id": project_id})
-        oldfunc = self.net_man.q_conn.create_and_attach_port
-
-        # Make sure no pairs are passed in if port security is turned off
-        def _instrumented_create_and_attach_port(tenant_id, net_id,
-                                                 interface_id, **kwargs):
-            self.assertTrue('allowed_address_pairs' in kwargs.keys())
-            pairs = kwargs['allowed_address_pairs']
-            self.assertTrue(len(pairs) == 0)
-            self.net_man.q_conn.create_and_attach_port = oldfunc
-            return oldfunc(tenant_id, net_id, interface_id, **kwargs)
-        _port_attach = _instrumented_create_and_attach_port
-        self.net_man.q_conn.create_and_attach_port = _port_attach
-        nw_info = self.net_man.allocate_for_instance(ctx,
-                        instance_id=instance_ref['id'], host="",
-                        rxtx_factor=3,
-                        project_id=project_id,
-                        requested_networks=requested_networks)
-        self.assertEqual(nw_info[0]['address'], fake_mac)
-
-
-class QuantumMelangeTestCase(test.TestCase):
-    def setUp(self):
-        super(QuantumMelangeTestCase, self).setUp()
-
-        fc = fake_client.FakeClient(LOG)
-        qc = quantum_connection.QuantumClientConnection(client=fc)
-
-        self.net_man = quantum_manager.QuantumManager(
-            ipam_lib="nova.network.quantum.nova_ipam_lib",
-            q_conn=qc)
-
-    def test_get_instance_uuids_by_ip_filter(self):
-        fake_context = context.RequestContext('user', 'project')
-        address = '1.2.3.4'
-        filters = {'ip': address}
-
-        self.net_man.ipam = self.mox.CreateMockAnything()
-        self.net_man.ipam.get_instance_ids_by_ip_address(fake_context,
-                address).AndReturn(['instance_id'])
-
-        instance = self.mox.CreateMockAnything()
-        instance.uuid = 'instance_uuid'
-
-        self.mox.StubOutWithMock(db, 'instance_get')
-        db.instance_get(fake_context, 'instance_id').AndReturn(instance)
-
-        self.mox.ReplayAll()
-
-        uuids = self.net_man.get_instance_uuids_by_ip_filter(fake_context,
-                                                             filters)
-        self.assertEquals(uuids, [{'instance_uuid':'instance_uuid'}])
diff --git a/nova/tests/test_melange_ipam_lib.py b/nova/tests/test_melange_ipam_lib.py
deleted file mode 100644
index d752a5dee..000000000
--- a/nova/tests/test_melange_ipam_lib.py
+++ /dev/null
@@ -1,243 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2012 OpenStack LLC.
-# 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.
-
-from nova import db
-from nova import flags
-from nova.network.quantum import melange_connection
-from nova.network.quantum import melange_ipam_lib
-from nova import test
-
-FLAGS = flags.FLAGS
-
-
-class MelangeIpamLibTestCase(test.TestCase):
-    def setUp(self):
-        super(MelangeIpamLibTestCase, self).setUp()
-        self.m_conn = self.mox.CreateMock(melange_connection.MelangeConnection)
-        self.ipam = melange_ipam_lib.QuantumMelangeIPAMLib()
-        self.ipam.m_conn = self.m_conn
-
-    def _block_list(self, id='id', cidr='cidr', network_id='network_id'):
-        return {'ip_blocks': [{'id': id,
-                                'cidr': cidr,
-                               'network_id': network_id}]}
-
-    def test_allocate_fixed_ips_extracts_address(self):
-        self.m_conn.allocate_ip('quantum_net_id', 'network_tenant_id',
-                                'vif_ref_uuid', 'project_id',
-                                'vif_ref_address').AndReturn(
-                                                [{'address': 'ip_address'}])
-
-        self.mox.ReplayAll()
-
-        ips = self.ipam.allocate_fixed_ips('context',
-                                           'project_id',
-                                           'quantum_net_id',
-                                           'network_tenant_id',
-                                           {'uuid': 'vif_ref_uuid',
-                                            'address': 'vif_ref_address'})
-        self.assertEqual(ips[0], 'ip_address')
-
-    def test_delete_subnets_by_net_id_deletes_block(self):
-        context = self.mox.CreateMockAnything()
-        context.elevated().AndReturn('elevated')
-
-        self.m_conn.get_blocks('project_id').AndReturn(
-                                        self._block_list(id='block_id'))
-        self.m_conn.delete_block('block_id', 'project_id')
-
-        self.mox.StubOutWithMock(db, 'network_get_by_uuid')
-        db.network_get_by_uuid('elevated', 'network_id').AndReturn(
-                                                         {'id': 'network_id'})
-
-        self.mox.StubOutWithMock(db, 'network_delete_safe')
-        db.network_delete_safe(context, 'network_id')
-
-        self.mox.ReplayAll()
-        self.ipam.delete_subnets_by_net_id(context, 'network_id', 'project_id')
-
-    def test_get_networks_by_tenant_gets_all_networks(self):
-        block_list = self._block_list(network_id='net_1')
-        block_list['ip_blocks'] += self._block_list(
-                                        network_id='net_2')['ip_blocks']
-
-        self.m_conn.get_blocks('tenant_id').AndReturn(block_list)
-
-        self.mox.StubOutWithMock(db, 'network_get_by_uuid')
-        db.network_get_by_uuid('admin_context', 'net_1').AndReturn('network1')
-        db.network_get_by_uuid('admin_context', 'net_2').AndReturn('network2')
-
-        self.mox.ReplayAll()
-        values = self.ipam.get_networks_by_tenant('admin_context', 'tenant_id')
-        self.assertEquals(values, ['network1', 'network2'])
-
-    def test_get_global_networks(self):
-        FLAGS.quantum_default_tenant_id = 'quantum_default_tenant_id'
-        self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant')
-        self.ipam.get_networks_by_tenant('admin_context',
-                                    'quantum_default_tenant_id')
-
-        self.mox.ReplayAll()
-
-        self.ipam.get_global_networks('admin_context')
-
-    def test_get_project_networks(self):
-        context = self.mox.CreateMockAnything()
-        context.elevated().AndReturn('elevated')
-
-        networks = [{'project_id': 1}, {'project_id': None}]
-
-        self.mox.StubOutWithMock(db, 'network_get_all')
-        db.network_get_all('elevated').AndReturn(networks)
-
-        self.mox.ReplayAll()
-        values = self.ipam.get_project_networks(context)
-        self.assertEquals(values, [networks[0]])
-
-    def test_get_project_and_global_net_ids__by_priority(self):
-        context = self.mox.CreateMockAnything()
-        context.elevated().AndReturn('elevated')
-
-        FLAGS.quantum_default_tenant_id = 'default_tenant_id'
-
-        net1 = {'uuid': 'net1_uuid', 'priority': 'net1_priority'}
-        net2 = {'uuid': 'net2_uuid', 'priority': 'net2_priority'}
-
-        self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant')
-        self.ipam.get_networks_by_tenant('elevated',
-                                         'project_id').AndReturn([net1])
-        self.ipam.get_networks_by_tenant('elevated',
-                                         'default_tenant_id').AndReturn([net2])
-        self.mox.ReplayAll()
-        self.ipam.get_project_and_global_net_ids(context, 'project_id')
-
-    def test_get_tenant_id_by_net_id_returns_id(self):
-        FLAGS.quantum_default_tenant_id = 'qdti'
-
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                      'qdti').AndReturn({})
-        self.mox.ReplayAll()
-        value = self.ipam.get_tenant_id_by_net_id('context', 'net_id',
-                                          'vif_id', 'project_id')
-        self.assertEqual(value, 'qdti')
-
-    def test_get_tenant_id_by_net_id_returns_none_if_none_found(self):
-        FLAGS.quantum_default_tenant_id = 'qdti'
-
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                        'qdti').AndRaise(KeyError())
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                        'project_id').AndRaise(KeyError())
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                        None).AndRaise(KeyError())
-        self.mox.ReplayAll()
-        value = self.ipam.get_tenant_id_by_net_id('context', 'net_id',
-                                          'vif_id', 'project_id')
-        self.assertEqual(value, None)
-
-    def test_get_subnets_by_net_id(self):
-        ips = [{'ip_block': {'network_id': 'network_id',
-                            'id': 'id',
-                            'cidr': 'cidr',
-                            'gateway': 'gateway',
-                            'broadcast': 'broadcast',
-                            'netmask': 'netmask',
-                            'dns1': 'dns1',
-                            'dns2': 'dns2'},
-                'version': 4}]
-
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                      'tenant_id').AndReturn(ips)
-
-        self.mox.ReplayAll()
-        value = self.ipam.get_subnets_by_net_id('context', 'tenant_id',
-                                                'net_id', 'vif_id')
-        self.assertEquals(value[0]['cidr'], 'cidr')
-
-    def test_get_routes_by_ip_block(self):
-        self.m_conn.get_routes('block_id', 'project_id')
-        self.mox.ReplayAll()
-        self.ipam.get_routes_by_ip_block('context', 'block_id', 'project_id')
-
-    def test_get_v4_ips_by_interface(self):
-        self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface')
-        self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
-                                        'project_id', 4)
-        self.mox.ReplayAll()
-        self.ipam.get_v4_ips_by_interface('context', 'net_id', 'vif_id',
-                                          'project_id')
-
-    def test_get_v6_ips_by_interface(self):
-        self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface')
-        self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
-                                        'project_id', 6)
-        self.mox.ReplayAll()
-        self.ipam.get_v6_ips_by_interface('context', 'net_id', 'vif_id',
-                                          'project_id')
-
-    def test_get_ips_by_interface(self):
-        ips = [{'address': '10.10.10.10'}, {'address': '2001::CAFE'}]
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                      'tenant_id').AndReturn(ips)
-        self.m_conn.get_allocated_ips('net_id', 'vif_id',
-                                      'tenant_id').AndReturn(ips)
-        self.mox.ReplayAll()
-        values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
-                                        'tenant_id', 4)
-        self.assertEquals(values, ["10.10.10.10"])
-        values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
-                                        'tenant_id', 6)
-        self.assertEquals(values, ["2001::CAFE"])
-
-    def test_get_instance_ids_by_ip_address(self):
-        ips = [{'used_by_device': 'some_vif_uuid'}]
-        self.m_conn.get_allocated_ips_by_address('ip').AndReturn(ips)
-
-        self.mox.ReplayAll()
-        uuid = self.ipam.get_instance_ids_by_ip_address('context', 'ip')
-        self.assertEqual(uuid, ['some_vif_uuid'])
-
-    def test_verify_subnet_exists(self):
-        blocks = {'ip_blocks': [{'network_id': 'quantum_net_id'}]}
-        self.m_conn.get_blocks('tenant_id').AndReturn(blocks)
-        self.mox.ReplayAll()
-        value = self.ipam.verify_subnet_exists('context', 'tenant_id',
-                                       'quantum_net_id')
-        self.assertEquals(value, True)
-
-    def test_deallocate_ips_by_vif(self):
-        self.m_conn.deallocate_ips('net_id', 'uuid', 'tenant_id')
-        self.mox.ReplayAll()
-        self.ipam.deallocate_ips_by_vif('context', 'tenant_id', 'net_id',
-                {'uuid': 'uuid'})
-
-    def test_get_allocated_ips(self):
-        ips = [{'address': 'ip_address', 'interface_id': 'interface_id'}]
-        self.m_conn.get_allocated_ips_for_network('subnet_id',
-                                                  'project_id').AndReturn(ips)
-        self.mox.ReplayAll()
-        self.ipam.get_allocated_ips('context', 'subnet_id', 'project_id')
-
-    def test_create_vif(self):
-        self.m_conn.create_vif('vif_id', 'instance_id', 'project_id')
-        self.mox.ReplayAll()
-        self.ipam.create_vif('vif_id', 'instance_id', 'project_id')
-
-    def test_get_floating_ips_by_fixed_address(self):
-        value = self.ipam.get_floating_ips_by_fixed_address('context',
-                                                      'fixed_address')
-        self.assertEquals(value, [])