13
13
# under the License.
16
from oslo.config import cfg
16
from oslo_config import cfg
18
from neutron.api.v2 import attributes
18
19
from neutron.common import constants
19
20
from neutron import context
20
21
from neutron.db import agents_db
21
22
from neutron.db import common_db_mixin
22
23
from neutron.db import l3_agentschedulers_db
23
24
from neutron.db import l3_hamode_db
25
from neutron.extensions import l3
24
26
from neutron.extensions import l3_ext_ha_mode
25
27
from neutron import manager
26
28
from neutron.openstack.common import uuidutils
195
197
interface['device_owner'])
196
198
self.assertEqual(cfg.CONF.l3_ha_net_cidr, interface['subnet']['cidr'])
198
def test_update_state(self):
199
router = self._create_router()
200
self._bind_router(router['id'])
201
routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx,
203
state = routers[0].get(constants.HA_ROUTER_STATE_KEY)
204
self.assertEqual('standby', state)
206
self.plugin.update_router_state(self.admin_ctx, router['id'], 'active',
209
routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx,
212
state = routers[0].get(constants.HA_ROUTER_STATE_KEY)
213
self.assertEqual('active', state)
215
200
def test_unique_ha_network_per_tenant(self):
216
201
tenant1 = _uuid()
217
202
tenant2 = _uuid()
424
409
self.assertEqual(2, num_ha_candidates)
412
class L3HAModeDbTestCase(L3HATestFramework):
414
def _create_network(self, plugin, ctx, name='net',
415
tenant_id='tenant1'):
416
network = {'network': {'name': name,
418
'admin_state_up': True,
419
'tenant_id': tenant_id}}
420
return plugin.create_network(ctx, network)['id']
422
def _create_subnet(self, plugin, ctx, network_id, cidr='10.0.0.0/8',
423
name='subnet', tenant_id='tenant1'):
424
subnet = {'subnet': {'name': name,
426
'network_id': network_id,
428
'gateway_ip': attributes.ATTR_NOT_SPECIFIED,
429
'allocation_pools': attributes.ATTR_NOT_SPECIFIED,
430
'dns_nameservers': attributes.ATTR_NOT_SPECIFIED,
431
'host_routes': attributes.ATTR_NOT_SPECIFIED,
432
'tenant_id': tenant_id,
434
'ipv6_ra_mode': attributes.ATTR_NOT_SPECIFIED}}
435
created_subnet = plugin.create_subnet(ctx, subnet)
436
return created_subnet
438
def test_remove_ha_in_use(self):
439
router = self._create_router(ctx=self.admin_ctx)
440
network_id = self._create_network(self.core_plugin, self.admin_ctx)
441
subnet = self._create_subnet(self.core_plugin, self.admin_ctx,
443
interface_info = {'subnet_id': subnet['id']}
444
self.plugin.add_router_interface(self.admin_ctx,
447
self.assertRaises(l3.RouterInUse, self.plugin.delete_router,
448
self.admin_ctx, router['id'])
449
bindings = self.plugin.get_ha_router_port_bindings(
450
self.admin_ctx, [router['id']])
451
self.assertEqual(2, len(bindings))
427
454
class L3HAUserTestCase(L3HATestFramework):