207 lines
8.8 KiB
Python
207 lines
8.8 KiB
Python
# Copyright (c) 2016 Hewlett Packard Enterprise Development Company, L.P.
|
|
#
|
|
# 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 oslo_config import cfg
|
|
from oslo_utils import importutils
|
|
|
|
from neutron.api.v2 import attributes
|
|
from neutron import context
|
|
from neutron.extensions import agent
|
|
from neutron import manager
|
|
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_base_plugin
|
|
from neutron.tests.unit.extensions import test_agent
|
|
|
|
from neutron_dynamic_routing.db import bgp_db
|
|
from neutron_dynamic_routing.db import bgp_dragentscheduler_db as bgp_dras_db
|
|
from neutron_dynamic_routing.extensions import bgp
|
|
from neutron_dynamic_routing.extensions import bgp_dragentscheduler as bgp_dras_ext # noqa
|
|
from neutron_dynamic_routing.tests.common import helpers
|
|
from neutron_dynamic_routing.tests.unit.db import test_bgp_db
|
|
|
|
from webob import exc
|
|
|
|
|
|
class BgpDrSchedulerTestExtensionManager(object):
|
|
|
|
def get_resources(self):
|
|
attributes.RESOURCE_ATTRIBUTE_MAP.update(
|
|
agent.RESOURCE_ATTRIBUTE_MAP)
|
|
resources = agent.Agent.get_resources()
|
|
resources.extend(bgp_dras_ext.Bgp_dragentscheduler.get_resources())
|
|
return resources
|
|
|
|
def get_actions(self):
|
|
return []
|
|
|
|
def get_request_extensions(self):
|
|
return []
|
|
|
|
|
|
class TestBgpDrSchedulerPlugin(bgp_db.BgpDbMixin,
|
|
bgp_dras_db.BgpDrAgentSchedulerDbMixin):
|
|
|
|
bgp_drscheduler = importutils.import_object(
|
|
cfg.CONF.bgp_drscheduler_driver)
|
|
|
|
supported_extension_aliases = ["bgp_dragent_scheduler"]
|
|
|
|
def get_plugin_description(self):
|
|
return ("BGP dynamic routing service Plugin test class that test "
|
|
"BGP speaker functionality, with scheduler.")
|
|
|
|
|
|
class BgpDrSchedulingTestCase(test_agent.AgentDBTestMixIn,
|
|
test_bgp_db.BgpEntityCreationMixin):
|
|
|
|
def test_schedule_bgp_speaker(self):
|
|
"""Test happy path over full scheduling cycle."""
|
|
with self.bgp_speaker(4, 1234) as ri:
|
|
bgp_speaker_id = ri['id']
|
|
helpers.register_bgp_dragent(host='host1')
|
|
agent = self._list('agents')['agents'][0]
|
|
agent_id = agent['id']
|
|
|
|
data = {'bgp_speaker_id': bgp_speaker_id}
|
|
req = self.new_create_request('agents', data, self.fmt,
|
|
agent_id, 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPCreated.code, res.status_int)
|
|
|
|
req_show = self.new_show_request('agents', agent_id, self.fmt,
|
|
'bgp-drinstances')
|
|
res = req_show.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPOk.code, res.status_int)
|
|
res = self.deserialize(self.fmt, res)
|
|
self.assertIn('bgp_speakers', res)
|
|
self.assertTrue(bgp_speaker_id,
|
|
res['bgp_speakers'][0]['id'])
|
|
|
|
req = self.new_delete_request('agents',
|
|
agent_id,
|
|
self.fmt,
|
|
'bgp-drinstances',
|
|
bgp_speaker_id)
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPNoContent.code, res.status_int)
|
|
|
|
res = req_show.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPOk.code, res.status_int)
|
|
res = self.deserialize(self.fmt, res)
|
|
self.assertIn('bgp_speakers', res)
|
|
self.assertEqual([], res['bgp_speakers'])
|
|
|
|
def test_schedule_bgp_speaker_on_invalid_agent(self):
|
|
"""Test error while scheduling BGP speaker on an invalid agent."""
|
|
with self.bgp_speaker(4, 1234) as ri:
|
|
bgp_speaker_id = ri['id']
|
|
self._register_l3_agent(host='host1') # Register wrong agent
|
|
agent = self._list('agents')['agents'][0]
|
|
data = {'bgp_speaker_id': bgp_speaker_id}
|
|
req = self.new_create_request(
|
|
'agents', data, self.fmt,
|
|
agent['id'], 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
|
|
# Raises an AgentNotFound exception if the agent is invalid
|
|
self.assertEqual(exc.HTTPNotFound.code, res.status_int)
|
|
|
|
def test_schedule_bgp_speaker_twice_on_same_agent(self):
|
|
"""Test error if a BGP speaker is scheduled twice on same agent"""
|
|
with self.bgp_speaker(4, 1234) as ri:
|
|
bgp_speaker_id = ri['id']
|
|
helpers.register_bgp_dragent(host='host1')
|
|
agent = self._list('agents')['agents'][0]
|
|
data = {'bgp_speaker_id': bgp_speaker_id}
|
|
req = self.new_create_request(
|
|
'agents', data, self.fmt,
|
|
agent['id'], 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPCreated.code, res.status_int)
|
|
|
|
# Try second time, should raise conflict
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPConflict.code, res.status_int)
|
|
|
|
def test_schedule_bgp_speaker_on_two_different_agents(self):
|
|
"""Test that a BGP speaker can be associated to two agents."""
|
|
with self.bgp_speaker(4, 1234) as ri:
|
|
bgp_speaker_id = ri['id']
|
|
helpers.register_bgp_dragent(host='host1')
|
|
helpers.register_bgp_dragent(host='host2')
|
|
data = {'bgp_speaker_id': bgp_speaker_id}
|
|
|
|
agent1 = self._list('agents')['agents'][0]
|
|
req = self.new_create_request(
|
|
'agents', data, self.fmt,
|
|
agent1['id'], 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPCreated.code, res.status_int)
|
|
|
|
agent2 = self._list('agents')['agents'][1]
|
|
req = self.new_create_request(
|
|
'agents', data, self.fmt,
|
|
agent2['id'], 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPCreated.code, res.status_int)
|
|
|
|
def test_schedule_multi_bgp_speaker_on_one_dragent(self):
|
|
"""Test only one BGP speaker can be associated to one dragent."""
|
|
with self.bgp_speaker(4, 1) as ri1, self.bgp_speaker(4, 2) as ri2:
|
|
helpers.register_bgp_dragent(host='host1')
|
|
|
|
agent = self._list('agents')['agents'][0]
|
|
data = {'bgp_speaker_id': ri1['id']}
|
|
req = self.new_create_request(
|
|
'agents', data, self.fmt,
|
|
agent['id'], 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPCreated.code, res.status_int)
|
|
|
|
data = {'bgp_speaker_id': ri2['id']}
|
|
req = self.new_create_request(
|
|
'agents', data, self.fmt,
|
|
agent['id'], 'bgp-drinstances')
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(exc.HTTPConflict.code, res.status_int)
|
|
|
|
def test_non_scheduled_bgp_speaker_binding_removal(self):
|
|
"""Test exception while removing an invalid binding."""
|
|
with self.bgp_speaker(4, 1234) as ri1:
|
|
helpers.register_bgp_dragent(host='host1')
|
|
agent = self._list('agents')['agents'][0]
|
|
agent_id = agent['id']
|
|
self.assertRaises(bgp_dras_ext.DrAgentNotHostingBgpSpeaker,
|
|
self.bgp_plugin.remove_bgp_speaker_from_dragent,
|
|
self.context, agent_id, ri1['id'])
|
|
|
|
|
|
class BgpDrPluginSchedulerTests(test_db_base_plugin.NeutronDbPluginV2TestCase,
|
|
BgpDrSchedulingTestCase):
|
|
|
|
def setUp(self, plugin=None, ext_mgr=None, service_plugins=None):
|
|
if not plugin:
|
|
plugin = ('neutron_dynamic_routing.tests.unit.db.'
|
|
'test_bgp_dragentscheduler_db.TestBgpDrSchedulerPlugin')
|
|
if not service_plugins:
|
|
service_plugins = {bgp.BGP_EXT_ALIAS:
|
|
'neutron_dynamic_routing.services.bgp.'
|
|
'bgp_plugin.BgpPlugin'}
|
|
|
|
ext_mgr = ext_mgr or BgpDrSchedulerTestExtensionManager()
|
|
super(BgpDrPluginSchedulerTests, self).setUp(
|
|
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
|
|
self.bgp_plugin = manager.NeutronManager.get_service_plugins().get(
|
|
bgp.BGP_EXT_ALIAS)
|
|
self.context = context.get_admin_context()
|