import os
import neutron
+from neutron.common import constants
+from neutron.common import topics
+from neutron import context
+from neutron import manager
+
+HOST = 'localhost'
def find_file(filename, path):
return find_file(
filename,
path=os.path.join(neutron.__path__[0], '..', 'etc'))
+
+
+def _get_l3_agent_dict(host, agent_mode, internal_only=True,
+ ext_net_id='', ext_bridge='', router_id=None):
+ return {
+ 'agent_type': constants.AGENT_TYPE_L3,
+ 'binary': 'neutron-l3-agent',
+ 'host': host,
+ 'topic': topics.L3_AGENT,
+ 'configurations': {'agent_mode': agent_mode,
+ 'handle_internal_only_routers': internal_only,
+ 'external_network_bridge': ext_bridge,
+ 'gateway_external_network_id': ext_net_id,
+ 'router_id': router_id,
+ 'use_namespaces': router_id is None}}
+
+
+def _register_agent(agent):
+ core_plugin = manager.NeutronManager.get_plugin()
+ admin_context = context.get_admin_context()
+ core_plugin.create_or_update_agent(admin_context, agent)
+ return core_plugin.get_agents_db(
+ admin_context,
+ filters={'host': [agent['host']],
+ 'agent_type': [agent['agent_type']]})[0]
+
+
+def register_l3_agent(host=HOST, agent_mode=constants.L3_AGENT_MODE_LEGACY,
+ internal_only=True, ext_net_id='', ext_bridge='',
+ router_id=None):
+ agent = _get_l3_agent_dict(host, agent_mode, internal_only, ext_net_id,
+ ext_bridge, router_id)
+ return _register_agent(agent)
from neutron import manager
from neutron.openstack.common import uuidutils
from neutron.scheduler import l3_agent_scheduler
+from neutron.tests.common import helpers
from neutron.tests.unit import testlib_api
_uuid = uuidutils.generate_uuid
cfg.CONF.set_override('allow_overlapping_ips', True)
self.plugin = FakeL3PluginWithAgents()
- self.agent1 = self._register_agent('legacy', 'l3host')
- self.agent2 = self._register_agent('dvr_snat', 'l3host_2')
-
- def _register_agent(self, agent_mode, host):
- agent_status = {
- 'agent_type': constants.AGENT_TYPE_L3,
- 'binary': 'neutron-l3-agent',
- 'host': host,
- 'topic': 'N/A',
- 'configurations': {'agent_mode': agent_mode}
- }
-
- self.plugin.create_or_update_agent(self.admin_ctx, agent_status)
- return self.plugin.get_agents(
- self.admin_ctx, filters={'host': [host]})[0]
+ self.agent1 = helpers.register_l3_agent()
+ self.agent2 = helpers.register_l3_agent(
+ 'host_2', constants.L3_AGENT_MODE_DVR_SNAT)
def _bring_down_agent(self, agent_id):
update = {
# Test setup registers two l3 agents.
# Register another l3 agent with dvr mode and assert that
# get_number_of_ha_agent_candidates return 2.
- dvr_agent_status = {
- 'agent_type': constants.AGENT_TYPE_L3,
- 'binary': 'neutron-l3-agent',
- 'host': 'l3host_3',
- 'topic': 'N/A',
- 'configurations': {'agent_mode': 'dvr'}
- }
- self.plugin.create_or_update_agent(self.admin_ctx, dvr_agent_status)
+ helpers.register_l3_agent('host_3', constants.L3_AGENT_MODE_DVR)
num_ha_candidates = self.plugin.get_number_of_agents_for_scheduling(
self.admin_ctx)
self.assertEqual(2, num_ha_candidates)
def test_get_number_of_agents_for_scheduling_not_enough_agents(self):
cfg.CONF.set_override('min_l3_agents_per_router', 3)
- agent_to_bring_down = self._register_agent('legacy', 'l3host_3')
+ agent_to_bring_down = helpers.register_l3_agent(host='l3host_3')
self._bring_down_agent(agent_to_bring_down['id'])
self.assertRaises(l3_ext_ha_mode.HANotEnoughAvailableAgents,
self.plugin.get_number_of_agents_for_scheduling,
from neutron.api.v2 import attributes
from neutron.common import constants
-from neutron.common import topics
from neutron import context
from neutron.db import agents_db
from neutron.db import db_base_plugin_v2
from neutron.extensions import agent
from neutron.openstack.common import uuidutils
+from neutron.tests.common import helpers
from neutron.tests import tools
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2
def _register_agent_states(self, lbaas_agents=False):
"""Register two L3 agents and two DHCP agents."""
- l3_hosta = {
- 'binary': 'neutron-l3-agent',
- 'host': L3_HOSTA,
- 'topic': topics.L3_AGENT,
- 'configurations': {'use_namespaces': True,
- 'router_id': None,
- 'handle_internal_only_routers':
- True,
- 'gateway_external_network_id':
- None,
- 'interface_driver': 'interface_driver',
- },
- 'agent_type': constants.AGENT_TYPE_L3}
- l3_hostb = copy.deepcopy(l3_hosta)
- l3_hostb['host'] = L3_HOSTB
+ l3_hosta = helpers._get_l3_agent_dict(
+ L3_HOSTA, constants.L3_AGENT_MODE_LEGACY)
+ l3_hostb = helpers._get_l3_agent_dict(
+ L3_HOSTB, constants.L3_AGENT_MODE_LEGACY)
dhcp_hosta = {
'binary': 'neutron-dhcp-agent',
'host': DHCP_HOSTA,
lbaas_hostb = copy.deepcopy(lbaas_hosta)
lbaas_hostb['host'] = LBAAS_HOSTB
callback = agents_db.AgentExtRpcCallback()
- callback.report_state(self.adminContext,
- agent_state={'agent_state': l3_hosta},
- time=timeutils.strtime())
- callback.report_state(self.adminContext,
- agent_state={'agent_state': l3_hostb},
- time=timeutils.strtime())
+ helpers.register_l3_agent(host=L3_HOSTA)
+ helpers.register_l3_agent(host=L3_HOSTB)
callback.report_state(self.adminContext,
agent_state={'agent_state': dhcp_hosta},
time=timeutils.strtime())
time=timeutils.strtime())
return [dhcp_host]
- def _register_one_l3_agent(self, host=L3_HOSTA, internal_only=True,
- ext_net_id='', ext_bridge=''):
- l3 = {
- 'binary': 'neutron-l3-agent',
- 'host': host,
- 'topic': topics.L3_AGENT,
- 'configurations': {'use_namespaces': True,
- 'router_id': None,
- 'handle_internal_only_routers': internal_only,
- 'external_network_bridge': ext_bridge,
- 'gateway_external_network_id': ext_net_id,
- 'interface_driver': 'interface_driver',
- },
- 'agent_type': constants.AGENT_TYPE_L3}
- callback = agents_db.AgentExtRpcCallback()
- callback.report_state(self.adminContext,
- agent_state={'agent_state': l3},
- time=timeutils.strtime())
-
class AgentDBTestCase(AgentDBTestMixIn,
test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants as service_constants
from neutron.tests import base
+from neutron.tests.common import helpers
from neutron.tests import fake_notifier
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2
self.subnet()) as (r, s1, s2):
self._set_net_external(s1['subnet']['network_id'])
l3_rpc_cb = l3_rpc.L3RpcCallback()
- self._register_one_l3_agent(
+ helpers.register_l3_agent(
host='host1',
ext_net_id=s1['subnet']['network_id'])
- self._register_one_l3_agent(
+ helpers.register_l3_agent(
host='host2', internal_only=False,
ext_net_id=s2['subnet']['network_id'])
l3_rpc_cb.sync_routers(self.adminContext,
self.subnet()) as (r, s1, s2):
self._set_net_external(s1['subnet']['network_id'])
l3_rpc_cb = l3_rpc.L3RpcCallback()
- self._register_one_l3_agent(
+ helpers.register_l3_agent(
host='host1',
ext_net_id=s1['subnet']['network_id'])
- self._register_one_l3_agent(
+ helpers.register_l3_agent(
host='host2', internal_only=False,
ext_net_id='', ext_bridge='')
l3_rpc_cb.sync_routers(self.adminContext,
from neutron import manager
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants as service_constants
+from neutron.tests.common import helpers
from neutron.tests import fake_notifier
from neutron.tests import tools
from neutron.tests.unit.api import test_extensions
def test_router_auto_schedule_with_hosted_2(self):
# one agent hosts one router
l3_rpc_cb = l3_rpc.L3RpcCallback()
- l3_hosta = {
- 'binary': 'neutron-l3-agent',
- 'host': L3_HOSTA,
- 'topic': 'L3_AGENT',
- 'configurations': {'use_namespaces': True,
- 'router_id': None,
- 'handle_internal_only_routers':
- True,
- 'gateway_external_network_id':
- None,
- 'interface_driver': 'interface_driver',
- },
- 'agent_type': constants.AGENT_TYPE_L3}
- l3_hostb = copy.deepcopy(l3_hosta)
- l3_hostb['host'] = L3_HOSTB
with self.router() as router1:
- self._register_one_agent_state(l3_hosta)
+ hosta_id = helpers.register_l3_agent(host=L3_HOSTA).id
l3_rpc_cb.sync_routers(self.adminContext, host=L3_HOSTA)
- hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3,
- L3_HOSTA)
self._disable_agent(hosta_id, admin_state_up=False)
with self.router() as router2:
- self._register_one_agent_state(l3_hostb)
+ hostb_id = helpers.register_l3_agent(host=L3_HOSTB).id
l3_rpc_cb.sync_routers(self.adminContext, host=L3_HOSTB)
l3_agents_1 = self._list_l3_agents_hosting_router(
router1['router']['id'])
router2['router']['id'])
hosta_routers = self._list_routers_hosted_by_l3_agent(hosta_id)
num_hosta_routers = len(hosta_routers['routers'])
- hostb_id = self._get_agent_id(
- constants.AGENT_TYPE_L3,
- L3_HOSTB)
hostb_routers = self._list_routers_hosted_by_l3_agent(hostb_id)
num_hostb_routers = len(hostb_routers['routers'])
self.assertEqual(0, num_hosta_routers)
def test_router_auto_schedule_with_candidates(self):
- l3_hosta = {
- 'binary': 'neutron-l3-agent',
- 'host': L3_HOSTA,
- 'topic': 'L3_AGENT',
- 'configurations': {'use_namespaces': False,
- 'router_id': None,
- 'handle_internal_only_routers':
- True,
- 'gateway_external_network_id':
- None,
- 'interface_driver': 'interface_driver',
- },
- 'agent_type': constants.AGENT_TYPE_L3}
with contextlib.nested(self.router(),
self.router()) as (router1, router2):
l3_rpc_cb = l3_rpc.L3RpcCallback()
- l3_hosta['configurations']['router_id'] = router1['router']['id']
- self._register_one_agent_state(l3_hosta)
- hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3,
- L3_HOSTA)
+ agent = helpers.register_l3_agent(
+ host=L3_HOSTA, router_id=router1['router']['id'])
l3_rpc_cb.sync_routers(self.adminContext, host=L3_HOSTA)
- hosta_routers = self._list_routers_hosted_by_l3_agent(hosta_id)
+ hosta_routers = self._list_routers_hosted_by_l3_agent(agent.id)
num_hosta_routers = len(hosta_routers['routers'])
l3_agents_1 = self._list_l3_agents_hosting_router(
router1['router']['id'])
_sync_router_with_ids(router_ids, 4, 4, hosta_id)
def test_router_schedule_with_candidates(self):
- l3_hosta = {
- 'binary': 'neutron-l3-agent',
- 'host': L3_HOSTA,
- 'topic': 'L3_AGENT',
- 'configurations': {'use_namespaces': False,
- 'router_id': None,
- 'handle_internal_only_routers':
- True,
- 'gateway_external_network_id':
- None,
- 'interface_driver': 'interface_driver',
- },
- 'agent_type': constants.AGENT_TYPE_L3}
with contextlib.nested(self.router(),
self.router(),
self.subnet(),
router2,
subnet1,
subnet2):
- l3_hosta['configurations']['router_id'] = router1['router']['id']
- self._register_one_agent_state(l3_hosta)
- hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3,
- L3_HOSTA)
+ agent = helpers.register_l3_agent(
+ host=L3_HOSTA, router_id=router1['router']['id'])
self._router_interface_action('add',
router1['router']['id'],
subnet1['subnet']['id'],
router2['router']['id'],
subnet2['subnet']['id'],
None)
- hosta_routers = self._list_routers_hosted_by_l3_agent(hosta_id)
+ hosta_routers = self._list_routers_hosted_by_l3_agent(agent.id)
num_hosta_routers = len(hosta_routers['routers'])
l3_agents_1 = self._list_l3_agents_hosting_router(
router1['router']['id'])
from sqlalchemy.orm import query
from neutron.common import constants
-from neutron.common import topics
from neutron import context as q_context
from neutron.db import agents_db
from neutron.db import common_db_mixin
from neutron import manager
from neutron.scheduler import l3_agent_scheduler
from neutron.tests import base
+from neutron.tests.common import helpers
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit.extensions import test_l3
from neutron.tests.unit import testlib_api
load_tests = testscenarios.load_tests_apply_scenarios
HOST_DVR = 'my_l3_host_dvr'
-DVR_L3_AGENT = {
- 'binary': 'neutron-l3-agent',
- 'host': HOST_DVR,
- 'topic': topics.L3_AGENT,
- 'configurations': {'agent_mode': 'dvr'},
- 'agent_type': constants.AGENT_TYPE_L3,
- 'start_flag': True
-}
-
HOST_DVR_SNAT = 'my_l3_host_dvr_snat'
-DVR_SNAT_L3_AGENT = {
- 'binary': 'neutron-l3-agent',
- 'host': HOST_DVR_SNAT,
- 'topic': topics.L3_AGENT,
- 'configurations': {'agent_mode': 'dvr_snat'},
- 'agent_type': constants.AGENT_TYPE_L3,
- 'start_flag': True
-}
class FakeL3Scheduler(l3_agent_scheduler.L3Scheduler):
class L3SchedulerBaseMixin(object):
- def _register_l3_agent(self, host, agent_mode='legacy', plugin=None):
- if not plugin:
- plugin = self.plugin
-
- agent = {
- 'binary': 'neutron-l3-agent',
- 'host': host,
- 'topic': topics.L3_AGENT,
- 'configurations': {'agent_mode': agent_mode},
- 'agent_type': constants.AGENT_TYPE_L3,
- 'start_flag': True
- }
- callback = agents_db.AgentExtRpcCallback()
- callback.report_state(self.adminContext,
- agent_state={'agent_state': agent},
- time=timeutils.strtime())
- agent_db = plugin.get_agents_db(self.adminContext,
- filters={'host': [agent['host']]})
- return agent_db[0]
-
def _register_l3_agents(self, plugin=None):
- self.agent1 = self._register_l3_agent('host_1', plugin=plugin)
+ self.agent1 = helpers.register_l3_agent(
+ 'host_1', constants.L3_AGENT_MODE_LEGACY)
self.agent_id1 = self.agent1.id
-
- self.agent2 = self._register_l3_agent('host_2', plugin=plugin)
+ self.agent2 = helpers.register_l3_agent(
+ 'host_2', constants.L3_AGENT_MODE_LEGACY)
self.agent_id2 = self.agent2.id
def _register_l3_dvr_agents(self):
- callback = agents_db.AgentExtRpcCallback()
- callback.report_state(self.adminContext,
- agent_state={'agent_state': DVR_L3_AGENT},
- time=timeutils.strtime())
- agent_db = self.plugin.get_agents_db(self.adminContext,
- filters={'host': [HOST_DVR]})
- self.l3_dvr_agent = agent_db[0]
- self.l3_dvr_agent_id = agent_db[0].id
- callback.report_state(self.adminContext,
- agent_state={'agent_state': DVR_SNAT_L3_AGENT},
- time=timeutils.strtime())
- agent_db = self.plugin.get_agents_db(self.adminContext,
- filters={'host': [HOST_DVR_SNAT]})
- self.l3_dvr_snat_id = agent_db[0].id
- self.l3_dvr_snat_agent = agent_db[0]
+ self.l3_dvr_agent = helpers.register_l3_agent(
+ HOST_DVR, constants.L3_AGENT_MODE_DVR)
+ self.l3_dvr_agent_id = self.l3_dvr_agent.id
+ self.l3_dvr_snat_agent = helpers.register_l3_agent(
+ HOST_DVR_SNAT, constants.L3_AGENT_MODE_DVR_SNAT)
+ self.l3_dvr_snat_id = self.l3_dvr_snat_agent.id
def _set_l3_agent_admin_state(self, context, agent_id, state=True):
update = {'agent': {'admin_state_up': state}}
}
self._register_l3_dvr_agents()
agent_id = self.l3_dvr_snat_id
- agent = self.l3_dvr_snat_agent
router = self._create_router_for_l3_agent_dvr_test(
distributed=True,
external_gw=external_gw_info)
self.add_router_to_l3_agent(self.adminContext, agent_id,
router['router']['id'])
rtr_agent_binding.assert_called_once_with(
- self.adminContext, agent, router['router'])
+ self.adminContext, mock.ANY, router['router'])
def test_add_router_to_l3_agent(self):
self._test_add_router_to_l3_agent()
agent_list = [self.agent1, self.l3_dvr_agent]
# test dvr agent_mode case only dvr agent should be candidate
router['distributed'] = True
- exp_host = DVR_L3_AGENT.get('host')
self.check_ports_exist_on_l3agent = mock.Mock(return_value=True)
- self._check_get_l3_agent_candidates(router, agent_list, exp_host)
+ self._check_get_l3_agent_candidates(router, agent_list, HOST_DVR)
def test_get_l3_agent_candidates_dvr_no_vms(self):
self._register_l3_dvr_agents()
router['external_gateway_info'] = None
router['id'] = str(uuid.uuid4())
agent_list = [self.agent1, self.l3_dvr_agent]
- exp_host = DVR_L3_AGENT.get('host')
router['distributed'] = True
# Test no VMs present case
self.check_ports_exist_on_l3agent = mock.Mock(return_value=False)
self._check_get_l3_agent_candidates(
- router, agent_list, exp_host, count=0)
+ router, agent_list, HOST_DVR, count=0)
def test_get_l3_agent_candidates_dvr_snat(self):
self._register_l3_dvr_agents()
router['distributed'] = True
agent_list = [self.l3_dvr_snat_agent]
- exp_host = DVR_SNAT_L3_AGENT.get('host')
self.check_ports_exist_on_l3agent = mock.Mock(return_value=True)
- self._check_get_l3_agent_candidates(router, agent_list, exp_host)
+ self._check_get_l3_agent_candidates(router, agent_list, HOST_DVR_SNAT)
def test_get_l3_agent_candidates_dvr_snat_no_vms(self):
self._register_l3_dvr_agents()
router['distributed'] = True
agent_list = [self.l3_dvr_snat_agent]
- exp_host = DVR_SNAT_L3_AGENT.get('host')
self.check_ports_exist_on_l3agent = mock.Mock(return_value=False)
# Test no VMs present case
self.check_ports_exist_on_l3agent.return_value = False
self._check_get_l3_agent_candidates(
- router, agent_list, exp_host, count=0)
+ router, agent_list, HOST_DVR_SNAT, count=0)
def test_get_l3_agent_candidates_centralized(self):
self._register_l3_dvr_agents()
router['id'] = str(uuid.uuid4())
# check centralized test case
router['distributed'] = False
- exp_host = DVR_SNAT_L3_AGENT.get('host')
agent_list = [self.l3_dvr_snat_agent]
- self._check_get_l3_agent_candidates(router, agent_list, exp_host)
+ self._check_get_l3_agent_candidates(router, agent_list, HOST_DVR_SNAT)
def _prepare_check_ports_exist_tests(self):
l3_agent = agents_db.Agent()
self.assertTrue(val)
def test_get_l3_agents_hosting_routers(self):
- agent = self._register_l3_agent('host_6')
+ agent = helpers.register_l3_agent('host_6')
router = self._make_router(self.fmt,
tenant_id=str(uuid.uuid4()),
name='r1')
super(L3_HA_scheduler_db_mixinTestCase,
self)._register_l3_agents(plugin=plugin)
- self.agent3 = self._register_l3_agent('host_3', plugin=plugin)
+ self.agent3 = helpers.register_l3_agent(host='host_3')
self.agent_id3 = self.agent3.id
- self.agent4 = self._register_l3_agent('host_4', plugin=plugin)
+ self.agent4 = helpers.register_l3_agent(host='host_4')
self.agent_id4 = self.agent4.id
def test_get_ha_routers_l3_agents_count(self):
self.assertIn(self.agent_id1, agent_ids)
self.assertIn(self.agent_id2, agent_ids)
- agent = self._register_l3_agent('host_3')
+ agent = helpers.register_l3_agent(host='host_3')
self.agent_id3 = agent.id
routers_to_auto_schedule = [router['id']] if specific_router else []
self.plugin.auto_schedule_routers(self.adminContext,
super(L3HALeastRoutersSchedulerTestCase,
self)._register_l3_agents(plugin=plugin)
- agent = self._register_l3_agent('host_3', plugin=plugin)
+ agent = helpers.register_l3_agent(host='host_3')
self.agent_id3 = agent.id
- agent = self._register_l3_agent('host_4', plugin=plugin)
+ agent = helpers.register_l3_agent(host='host_4')
self.agent_id4 = agent.id
def setUp(self):
hosts = ['host_1', 'host_2', 'host_3', 'host_4', 'host_5']
agent_modes = ['legacy', 'dvr_snat', 'dvr', 'fake_mode', 'legacy']
for host, agent_mode in zip(hosts, agent_modes):
- self._register_l3_agent(host, agent_mode, self.plugin)
+ helpers.register_l3_agent(host, agent_mode)
def _get_agent_mode(self, agent):
agent_conf = self.plugin.get_configuration_dict(agent)
# under the License.
import mock
-from oslo_utils import timeutils
from neutron.api.v2 import attributes as attr
-from neutron.common import constants as n_constants
-from neutron.common import topics
from neutron import context
from neutron.db import agents_db
from neutron.db import l3_agentschedulers_db
from neutron import manager
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants
+from neutron.tests.common import helpers
from neutron.tests.unit.db.metering import test_metering_db
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit.extensions import test_l3
self.tenant_id_2 = 'tenant_id_2'
self.adminContext = context.get_admin_context()
- self._register_l3_agent('agent1')
-
- def _register_l3_agent(self, host):
- agent = {
- 'binary': 'neutron-l3-agent',
- 'host': host,
- 'topic': topics.L3_AGENT,
- 'configurations': {},
- 'agent_type': n_constants.AGENT_TYPE_L3,
- 'start_flag': True
- }
- callback = agents_db.AgentExtRpcCallback()
- callback.report_state(self.adminContext,
- agent_state={'agent_state': agent},
- time=timeutils.strtime())
+ helpers.register_l3_agent(host='agent1')
def test_get_sync_data_metering(self):
with self.subnet() as subnet:
host='agent1')
self.assertEqual('router1', data[0]['name'])
- self._register_l3_agent('agent2')
+ helpers.register_l3_agent(host='agent2')
data = callbacks.get_sync_data_metering(self.adminContext,
host='agent2')
self.assertFalse(data)
Code Review / openstack-build / neutron-build.git / commitdiff