neutron/tests/unit/extensions/test_agent.py

   1 # Copyright (c) 2013 OpenStack Foundation.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import copy
  17 from datetime import datetime
  18 import time
  19
  20 from oslo_config import cfg
  21 from oslo_utils import uuidutils
  22 from webob import exc
  23
  24 from neutron.api.v2 import attributes
  25 from neutron.common import constants
  26 from neutron import context
  27 from neutron.db import agents_db
  28 from neutron.db import db_base_plugin_v2
  29 from neutron.extensions import agent
  30 from neutron.tests.common import helpers
  31 from neutron.tests import tools
  32 from neutron.tests.unit.api.v2 import test_base
  33 from neutron.tests.unit.db import test_db_base_plugin_v2
  34
  35
  36 _uuid = uuidutils.generate_uuid
  37 _get_path = test_base._get_path
  38 L3_HOSTA = 'hosta'
  39 DHCP_HOSTA = 'hosta'
  40 L3_HOSTB = 'hostb'
  41 DHCP_HOSTC = 'hostc'
  42 LBAAS_HOSTA = 'hosta'
  43 LBAAS_HOSTB = 'hostb'
  44
  45
  46 class AgentTestExtensionManager(object):
  47
  48     def get_resources(self):
  49         # Add the resources to the global attribute map
  50         # This is done here as the setup process won't
  51         # initialize the main API router which extends
  52         # the global attribute map
  53         attributes.RESOURCE_ATTRIBUTE_MAP.update(
  54             agent.RESOURCE_ATTRIBUTE_MAP)
  55         return agent.Agent.get_resources()
  56
  57     def get_actions(self):
  58         return []
  59
  60     def get_request_extensions(self):
  61         return []
  62
  63
  64 # This plugin class is just for testing
  65 class TestAgentPlugin(db_base_plugin_v2.NeutronDbPluginV2,
  66                       agents_db.AgentDbMixin):
  67     supported_extension_aliases = ["agent"]
  68
  69
  70 class AgentDBTestMixIn(object):
  71
  72     def _list_agents(self, expected_res_status=None,
  73                      neutron_context=None,
  74                      query_string=None):
  75         agent_res = self._list('agents',
  76                                neutron_context=neutron_context,
  77                                query_params=query_string)
  78         if expected_res_status:
  79             self.assertEqual(agent_res.status_int, expected_res_status)
  80         return agent_res
  81
  82     def _register_agent_states(self, lbaas_agents=False):
  83         """Register two L3 agents and two DHCP agents."""
  84         l3_hosta = helpers._get_l3_agent_dict(
  85             L3_HOSTA, constants.L3_AGENT_MODE_LEGACY)
  86         l3_hostb = helpers._get_l3_agent_dict(
  87             L3_HOSTB, constants.L3_AGENT_MODE_LEGACY)
  88         dhcp_hosta = helpers._get_dhcp_agent_dict(DHCP_HOSTA)
  89         dhcp_hostc = helpers._get_dhcp_agent_dict(DHCP_HOSTC)
  90         helpers.register_l3_agent(host=L3_HOSTA)
  91         helpers.register_l3_agent(host=L3_HOSTB)
  92         helpers.register_dhcp_agent(host=DHCP_HOSTA)
  93         helpers.register_dhcp_agent(host=DHCP_HOSTC)
  94
  95         res = [l3_hosta, l3_hostb, dhcp_hosta, dhcp_hostc]
  96         if lbaas_agents:
  97             lbaas_hosta = {
  98                 'binary': 'neutron-loadbalancer-agent',
  99                 'host': LBAAS_HOSTA,
 100                 'topic': 'LOADBALANCER_AGENT',
 101                 'configurations': {'device_drivers': ['haproxy_ns']},
 102                 'agent_type': constants.AGENT_TYPE_LOADBALANCER}
 103             lbaas_hostb = copy.deepcopy(lbaas_hosta)
 104             lbaas_hostb['host'] = LBAAS_HOSTB
 105             callback = agents_db.AgentExtRpcCallback()
 106             callback.report_state(
 107                 self.adminContext,
 108                 agent_state={'agent_state': lbaas_hosta},
 109                 time=datetime.utcnow().strftime(constants.ISO8601_TIME_FORMAT))
 110             callback.report_state(
 111                 self.adminContext,
 112                 agent_state={'agent_state': lbaas_hostb},
 113                 time=datetime.utcnow().strftime(constants.ISO8601_TIME_FORMAT))
 114             res += [lbaas_hosta, lbaas_hostb]
 115
 116         return res
 117
 118     def _register_dvr_agents(self):
 119         dvr_snat_agent = helpers.register_l3_agent(
 120             host=L3_HOSTA, agent_mode=constants.L3_AGENT_MODE_DVR_SNAT)
 121         dvr_agent = helpers.register_l3_agent(
 122             host=L3_HOSTB, agent_mode=constants.L3_AGENT_MODE_DVR)
 123         return [dvr_snat_agent, dvr_agent]
 124
 125
 126 class AgentDBTestCase(AgentDBTestMixIn,
 127                       test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
 128     fmt = 'json'
 129
 130     def setUp(self):
 131         plugin = 'neutron.tests.unit.extensions.test_agent.TestAgentPlugin'
 132         # for these tests we need to enable overlapping ips
 133         cfg.CONF.set_default('allow_overlapping_ips', True)
 134         self.useFixture(tools.AttributeMapMemento())
 135         ext_mgr = AgentTestExtensionManager()
 136         super(AgentDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
 137         self.adminContext = context.get_admin_context()
 138
 139     def test_create_agent(self):
 140         data = {'agent': {}}
 141         _req = self.new_create_request('agents', data, self.fmt)
 142         _req.environ['neutron.context'] = context.Context(
 143             '', 'tenant_id')
 144         res = _req.get_response(self.ext_api)
 145         self.assertEqual(res.status_int, exc.HTTPBadRequest.code)
 146
 147     def test_list_agent(self):
 148         agents = self._register_agent_states()
 149         res = self._list('agents')
 150         self.assertEqual(len(agents), len(res['agents']))
 151
 152     def test_show_agent(self):
 153         self._register_agent_states()
 154         agents = self._list_agents(
 155             query_string='binary=neutron-l3-agent')
 156         self.assertEqual(2, len(agents['agents']))
 157         agent = self._show('agents', agents['agents'][0]['id'])
 158         self.assertEqual('neutron-l3-agent', agent['agent']['binary'])
 159
 160     def test_update_agent(self):
 161         self._register_agent_states()
 162         agents = self._list_agents(
 163             query_string='binary=neutron-l3-agent&host=' + L3_HOSTB)
 164         self.assertEqual(1, len(agents['agents']))
 165         com_id = agents['agents'][0]['id']
 166         agent = self._show('agents', com_id)
 167         new_agent = {}
 168         new_agent['agent'] = {}
 169         new_agent['agent']['admin_state_up'] = False
 170         new_agent['agent']['description'] = 'description'
 171         self._update('agents', com_id, new_agent)
 172         agent = self._show('agents', com_id)
 173         self.assertFalse(agent['agent']['admin_state_up'])
 174         self.assertEqual('description', agent['agent']['description'])
 175
 176     def test_dead_agent(self):
 177         cfg.CONF.set_override('agent_down_time', 1)
 178         self._register_agent_states()
 179         time.sleep(1.5)
 180         agents = self._list_agents(
 181             query_string='binary=neutron-l3-agent&host=' + L3_HOSTB)
 182         self.assertFalse(agents['agents'][0]['alive'])