From 38373bf8f60dd068dec69933d1456a8deb75bf8e Mon Sep 17 00:00:00 2001 From: Dan Wendlandt Date: Thu, 1 Sep 2011 15:02:09 -0700 Subject: move content of quantum/fake.py to test_quantum.py in unit testing class (most original content has been removed anyway) --- nova/network/quantum/fake.py | 92 ----------------------------------------- nova/network/quantum/manager.py | 9 ++-- nova/tests/test_quantum.py | 76 +++++++++++++++++++++++++++++++++- 3 files changed, 78 insertions(+), 99 deletions(-) delete mode 100644 nova/network/quantum/fake.py (limited to 'nova') diff --git a/nova/network/quantum/fake.py b/nova/network/quantum/fake.py deleted file mode 100644 index 4ecddd2ae..000000000 --- a/nova/network/quantum/fake.py +++ /dev/null @@ -1,92 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2011 Nicira Networks, 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. - -from nova import exception -from nova import ipv6 -from nova import log as logging -from nova import utils - - -LOG = logging.getLogger("network.quantum.fake") - - -# this class can be used for unit functional/testing on nova, -# as it does not actually make remote calls to the Quantum service -class FakeQuantumClientConnection(object): - - def __init__(self): - self.nets = {} - - def get_networks_for_tenant(self, tenant_id): - net_ids = [] - for net_id, n in self.nets.items(): - if n['tenant-id'] == tenant_id: - net_ids.append(net_id) - return net_ids - - def create_network(self, tenant_id, network_name): - - uuid = str(utils.gen_uuid()) - self.nets[uuid] = {'net-name': network_name, - 'tenant-id': tenant_id, - 'ports': {}} - return uuid - - def delete_network(self, tenant_id, net_id): - if self.nets[net_id]['tenant-id'] == tenant_id: - del self.nets[net_id] - - def network_exists(self, tenant_id, net_id): - try: - return self.nets[net_id]['tenant-id'] == tenant_id - except KeyError: - return False - - def _confirm_not_attached(self, interface_id): - for n in self.nets.values(): - for p in n['ports'].values(): - if p['attachment-id'] == interface_id: - raise Exception(_("interface '%s' is already attached" % - interface_id)) - - def create_and_attach_port(self, tenant_id, net_id, interface_id): - if not self.network_exists(tenant_id, net_id): - raise Exception( - _("network %(net_id)s does not exist for tenant %(tenant_id)" - % locals())) - - self._confirm_not_attached(interface_id) - uuid = str(utils.gen_uuid()) - self.nets[net_id]['ports'][uuid] = \ - {"port-state": "ACTIVE", - "attachment-id": interface_id} - - def detach_and_delete_port(self, tenant_id, net_id, port_id): - if not self.network_exists(tenant_id, net_id): - raise exception.NotFound( - _("network %(net_id)s does not exist " - "for tenant %(tenant_id)s" % locals())) - del self.nets[net_id]['ports'][port_id] - - def get_port_by_attachment(self, tenant_id, attachment_id): - for net_id, n in self.nets.items(): - if n['tenant-id'] == tenant_id: - for port_id, p in n['ports'].items(): - if p['attachment-id'] == attachment_id: - return (net_id, port_id) - - return (None, None) diff --git a/nova/network/quantum/manager.py b/nova/network/quantum/manager.py index fa16475ac..db2b000de 100644 --- a/nova/network/quantum/manager.py +++ b/nova/network/quantum/manager.py @@ -50,17 +50,16 @@ class QuantumManager(manager.FlatManager): Support for these capabilities are targted for future releases. """ - def __init__(self, ipam_lib=None, *args, **kwargs): + def __init__(self, q_conn=None, ipam_lib=None, *args, **kwargs): """ Initialize two key libraries, the connection to a Quantum service, and the library for implementing IPAM. Calls inherited FlatManager constructor. """ - if FLAGS.fake_network: - self.q_conn = fake.FakeQuantumClientConnection() - else: - self.q_conn = quantum_connection.QuantumClientConnection() + if not q_conn: + q_conn = quantum_connection.QuantumClientConnection() + self.q_conn = q_conn if not ipam_lib: ipam_lib = FLAGS.quantum_ipam_lib diff --git a/nova/tests/test_quantum.py b/nova/tests/test_quantum.py index 42acf03e7..0b1a1f204 100644 --- a/nova/tests/test_quantum.py +++ b/nova/tests/test_quantum.py @@ -20,12 +20,82 @@ from nova import db from nova.db.sqlalchemy import models from nova.db.sqlalchemy.session import get_session from nova import exception +from nova import ipv6 from nova import log as logging from nova.network.quantum import manager as quantum_manager from nova import test +from nova import utils LOG = logging.getLogger('nova.tests.quantum_network') + +# this class can be used for unit functional/testing on nova, +# as it does not actually make remote calls to the Quantum service +class FakeQuantumClientConnection(object): + + def __init__(self): + self.nets = {} + + def get_networks_for_tenant(self, tenant_id): + net_ids = [] + for net_id, n in self.nets.items(): + if n['tenant-id'] == tenant_id: + net_ids.append(net_id) + return net_ids + + def create_network(self, tenant_id, network_name): + + uuid = str(utils.gen_uuid()) + self.nets[uuid] = {'net-name': network_name, + 'tenant-id': tenant_id, + 'ports': {}} + return uuid + + def delete_network(self, tenant_id, net_id): + if self.nets[net_id]['tenant-id'] == tenant_id: + del self.nets[net_id] + + def network_exists(self, tenant_id, net_id): + try: + return self.nets[net_id]['tenant-id'] == tenant_id + except KeyError: + return False + + def _confirm_not_attached(self, interface_id): + for n in self.nets.values(): + for p in n['ports'].values(): + if p['attachment-id'] == interface_id: + raise Exception(_("interface '%s' is already attached" % + interface_id)) + + def create_and_attach_port(self, tenant_id, net_id, interface_id): + if not self.network_exists(tenant_id, net_id): + raise Exception( + _("network %(net_id)s does not exist for tenant %(tenant_id)" + % locals())) + + self._confirm_not_attached(interface_id) + uuid = str(utils.gen_uuid()) + self.nets[net_id]['ports'][uuid] = \ + {"port-state": "ACTIVE", + "attachment-id": interface_id} + + def detach_and_delete_port(self, tenant_id, net_id, port_id): + if not self.network_exists(tenant_id, net_id): + raise exception.NotFound( + _("network %(net_id)s does not exist " + "for tenant %(tenant_id)s" % locals())) + del self.nets[net_id]['ports'][port_id] + + def get_port_by_attachment(self, tenant_id, attachment_id): + for net_id, n in self.nets.items(): + if n['tenant-id'] == tenant_id: + for port_id, p in n['ports'].items(): + if p['attachment-id'] == attachment_id: + return (net_id, port_id) + + return (None, None) + networks = [{'label': 'project1-net1', 'injected': False, 'multi_host': False, @@ -230,8 +300,10 @@ class QuantumNovaIPAMTestCase(QuantumTestCaseBase, test.TestCase): def setUp(self): super(QuantumNovaIPAMTestCase, self).setUp() - self.net_man = quantum_manager.QuantumManager( \ - ipam_lib="nova.network.quantum.nova_ipam_lib") + + self.net_man = quantum_manager.QuantumManager( + ipam_lib="nova.network.quantum.nova_ipam_lib", + q_conn=FakeQuantumClientConnection()) # Tests seem to create some networks by default, which # we don't want. So we delete them. -- cgit