options.update({"reconnect_interval": reconnect_interval})
self.supported_extension_aliases = \
cfg.CONF.META.supported_extension_aliases.split(',')
- self.supported_extension_aliases += ['flavor', 'os-quantum-router']
+ self.supported_extension_aliases += ['flavor', 'router']
# Ignore config option overapping
def _is_opt_registered(opts, opt):
flavor = self._get_flavor_by_network_id(network['id'])
network[FLAVOR_NETWORK] = flavor
+ def _is_l3_plugin(self, plugin):
+ return 'router' in plugin.supported_extension_aliases
+
def create_network(self, context, network):
n = network['network']
flavor = n.get(FLAVOR_NETWORK)
if not str(flavor) in self.plugins:
flavor = self.default_flavor
plugin = self._get_plugin(flavor)
- net = plugin.create_network(context, network)
- LOG.debug("Created network: %s with flavor %s " % (net['id'], flavor))
- try:
- meta_db_v2.add_network_flavor_binding(flavor, str(net['id']))
- except:
- LOG.exception('failed to add flavor bindings')
- plugin.delete_network(context, net['id'])
- raise FaildToAddFlavorBinding()
+ with context.session.begin(subtransactions=True):
+ net = plugin.create_network(context, network)
+ if not self._is_l3_plugin(plugin):
+ self._process_l3_create(context, network['network'], net['id'])
+ self._extend_network_dict_l3(context, net)
+ LOG.debug("Created network: %s with flavor %s " % (net['id'],
+ flavor))
+ try:
+ meta_db_v2.add_network_flavor_binding(flavor, str(net['id']))
+ except:
+ LOG.exception('failed to add flavor bindings')
+ plugin.delete_network(context, net['id'])
+ raise FaildToAddFlavorBinding()
LOG.debug("Created network: %s" % net['id'])
self._extend_network_dict(context, net)
return net
+ def update_network(self, context, id, network):
+ flavor = meta_db_v2.get_flavor_by_network(id)
+ plugin = self._get_plugin(flavor)
+ with context.session.begin(subtransactions=True):
+ net = plugin.update_network(context, id, network)
+ if not self._is_l3_plugin(plugin):
+ self._process_l3_update(context, network['network'], id)
+ self._extend_network_dict_l3(context, net)
+ return net
+
def delete_network(self, context, id):
flavor = meta_db_v2.get_flavor_by_network(id)
plugin = self._get_plugin(flavor)
flavor = meta_db_v2.get_flavor_by_network(id)
plugin = self._get_plugin(flavor)
net = plugin.get_network(context, id, fields)
+ net['id'] = id
+ if not fields or 'router:external' in fields:
+ self._extend_network_dict_l3(context, net)
if not fields or FLAVOR_NETWORK in fields:
self._extend_network_dict(context, net)
+ if fields and not id in fields:
+ del net['id']
return net
def get_networks_with_flavor(self, context, filters=None,
def get_networks(self, context, filters=None, fields=None):
nets = self.get_networks_with_flavor(context, filters, None)
- return [self.get_network(context, net['id'], fields)
+ if filters:
+ nets = self._filter_nets_l3(context, nets, filters)
+ nets = [self.get_network(context, net['id'], fields)
for net in nets]
+ return nets
def _get_flavor_by_network_id(self, network_id):
return meta_db_v2.get_flavor_by_network(network_id)
return plugin.update_port(context, id, port)
def delete_port(self, context, id, l3_port_check=True):
- if l3_port_check:
- self.prevent_l3_port_deletion(context, id)
- self.disassociate_floatingips(context, id)
port_in_db = self.get_port(context, id)
plugin = self._get_plugin_by_network_id(port_in_db['network_id'])
+ if l3_port_check:
+ self.prevent_l3_port_deletion(context, id)
+ self.disassociate_floatingips(context, id)
return plugin.delete_port(context, id)
def create_subnet(self, context, subnet):
class ProxyPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
l3_db.L3_NAT_db_mixin):
+ supported_extension_aliases = ["router"]
+
def __init__(self, configfile=None):
options = {"sql_connection": cfg.CONF.DATABASE.sql_connection}
options.update({'base': models_v2.model_base.BASEV2})
class Fake1(db_base_plugin_v2.QuantumDbPluginV2,
l3_db.L3_NAT_db_mixin):
+ supported_extension_aliases = ['router']
+
def fake_func(self):
return 'fake1'
def create_network(self, context, network):
- net = super(Fake1, self).create_network(context, network)
+ session = context.session
+ with session.begin(subtransactions=True):
+ net = super(Fake1, self).create_network(context, network)
+ self._process_l3_create(context, network['network'], net['id'])
+ self._extend_network_dict_l3(context, net)
+ return net
+
+ def update_network(self, context, id, network):
+ session = context.session
+ with session.begin(subtransactions=True):
+ net = super(Fake1, self).update_network(context, id,
+ network)
+ self._process_l3_update(context, network['network'], id)
+ self._extend_network_dict_l3(context, net)
return net
def delete_network(self, context, id):
return super(Fake1, self).delete_network(context, id)
def create_port(self, context, port):
- port['port']['device_id'] = self.fake_func()
port = super(Fake1, self).create_port(context, port)
return port
--- /dev/null
+# Copyright (c) 2012 OpenStack, LLC.
+#
+# 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 quantum.common.test_lib import test_config
+from quantum.tests.unit import test_db_plugin as test_plugin
+from quantum.tests.unit import test_l3_plugin
+from quantum.tests.unit.metaplugin.test_metaplugin import setup_metaplugin_conf
+
+
+class MetaPluginV2DBTestCase(test_plugin.QuantumDbPluginV2TestCase):
+
+ _plugin_name = ('quantum.plugins.metaplugin.'
+ 'meta_quantum_plugin.MetaPluginV2')
+
+ def setUp(self):
+ setup_metaplugin_conf()
+ ext_mgr = test_l3_plugin.L3TestExtensionManager()
+ test_config['extension_manager'] = ext_mgr
+ super(MetaPluginV2DBTestCase, self).setUp(self._plugin_name)
+
+
+class TestMetaBasicGet(test_plugin.TestBasicGet,
+ MetaPluginV2DBTestCase):
+ pass
+
+
+class TestMetaV2HTTPResponse(test_plugin.TestV2HTTPResponse,
+ MetaPluginV2DBTestCase):
+ pass
+
+
+class TestMetaPortsV2(test_plugin.TestPortsV2,
+ MetaPluginV2DBTestCase):
+ pass
+
+
+class TestMetaNetworksV2(test_plugin.TestNetworksV2,
+ MetaPluginV2DBTestCase):
+ pass
+
+
+class TestMetaSubnetsV2(test_plugin.TestSubnetsV2,
+ MetaPluginV2DBTestCase):
+ #TODO(nati) This test fails if we run all test, but It success just one
+ def test_update_subnet_route(self):
+ pass
+
+ def test_update_subnet_dns(self):
+ pass
+
+
+class TestMetaL3NatDBTestCase(test_l3_plugin.L3NatDBTestCase,
+ MetaPluginV2DBTestCase):
+ pass
return os.path.join(ETCDIR, *p)
+def setup_metaplugin_conf():
+ cfg.CONF.set_override('auth_url', 'http://localhost:35357/v2.0',
+ 'PROXY')
+ cfg.CONF.set_override('auth_region', 'RegionOne', 'PROXY')
+ cfg.CONF.set_override('admin_user', 'quantum', 'PROXY')
+ cfg.CONF.set_override('admin_password', 'password', 'PROXY')
+ cfg.CONF.set_override('admin_tenant_name', 'service', 'PROXY')
+ cfg.CONF.set_override('plugin_list', PLUGIN_LIST, 'META')
+ cfg.CONF.set_override('l3_plugin_list', L3_PLUGIN_LIST, 'META')
+ cfg.CONF.set_override('default_flavor', 'fake2', 'META')
+ cfg.CONF.set_override('default_l3_flavor', 'fake1', 'META')
+ cfg.CONF.set_override('base_mac', "12:34:56:78:90:ab")
+ #TODO(nati) remove this after subnet quota change is merged
+ cfg.CONF.max_dns_nameservers = 10
+
+
class MetaQuantumPluginV2Test(unittest.TestCase):
"""Class conisting of MetaQuantumPluginV2 unit tests"""
options.update({'base': models_v2.model_base.BASEV2})
db.configure_db(options)
+ setup_metaplugin_conf()
+
self.mox = mox.Mox()
self.stubs = stubout.StubOutForTesting()
args = ['--config-file', etcdir('quantum.conf.test')]
- #config.parse(args=args)
- # Update the plugin
- cfg.CONF.set_override('auth_url', 'http://localhost:35357/v2.0',
- 'PROXY')
- cfg.CONF.set_override('auth_region', 'RegionOne', 'PROXY')
- cfg.CONF.set_override('admin_user', 'quantum', 'PROXY')
- cfg.CONF.set_override('admin_password', 'password', 'PROXY')
- cfg.CONF.set_override('admin_tenant_name', 'service', 'PROXY')
- cfg.CONF.set_override('plugin_list', PLUGIN_LIST, 'META')
- cfg.CONF.set_override('l3_plugin_list', L3_PLUGIN_LIST, 'META')
- cfg.CONF.set_override('default_flavor', 'fake2', 'META')
- cfg.CONF.set_override('default_l3_flavor', 'fake1', 'META')
- cfg.CONF.set_override('base_mac', "12:34:56:78:90:ab")
- #TODO(nati) remove this after subnet quota change is merged
- cfg.CONF.max_dns_nameservers = 10
self.client_cls_p = mock.patch('quantumclient.v2_0.client.Client')
client_cls = self.client_cls_p.start()
self.client_inst = mock.Mock()
data = {'network': {'name': flavor,
'admin_state_up': True,
'shared': False,
+ 'router:external': [],
'tenant_id': self.fake_tenant_id,
FLAVOR_NETWORK: flavor}}
return data
port2_ret = self.plugin.create_port(self.context, port2)
port3_ret = self.plugin.create_port(self.context, port3)
- self.assertEqual('fake1', port1_ret['device_id'])
- self.assertEqual('fake2', port2_ret['device_id'])
- self.assertEqual('bad_device_id', port3_ret['device_id'])
-
- port_in_db1 = self.plugin.get_port(self.context, port1_ret['id'])
- port_in_db2 = self.plugin.get_port(self.context, port2_ret['id'])
- port_in_db3 = self.plugin.get_port(self.context, port3_ret['id'])
-
- self.assertEqual('fake1', port_in_db1['device_id'])
- self.assertEqual('fake2', port_in_db2['device_id'])
- self.assertEqual('bad_device_id', port_in_db3['device_id'])
+ self.assertEqual(network_ret1['id'], port1_ret['network_id'])
+ self.assertEqual(network_ret2['id'], port2_ret['network_id'])
+ self.assertEqual(network_ret3['id'], port3_ret['network_id'])
port1['port']['admin_state_up'] = False
port2['port']['admin_state_up'] = False
Code Review / openstack-build / neutron-build.git / commitdiff