neutron/tests/unit/api/rpc/agentnotifiers/test_dhcp_rpc_agent_api.py

   1 # Copyright (c) 2013 Red Hat, Inc.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import datetime
  17 import mock
  18
  19 from oslo_utils import timeutils
  20
  21 from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
  22 from neutron.common import utils
  23 from neutron.db import agents_db
  24 from neutron.db.agentschedulers_db import cfg
  25 from neutron.tests import base
  26
  27
  28 class TestDhcpAgentNotifyAPI(base.BaseTestCase):
  29
  30     def setUp(self):
  31         super(TestDhcpAgentNotifyAPI, self).setUp()
  32         self.notifier = (
  33             dhcp_rpc_agent_api.DhcpAgentNotifyAPI(plugin=mock.Mock()))
  34
  35         mock_util_p = mock.patch.object(utils, 'is_extension_supported')
  36         mock_log_p = mock.patch.object(dhcp_rpc_agent_api, 'LOG')
  37         mock_fanout_p = mock.patch.object(self.notifier, '_fanout_message')
  38         mock_cast_p = mock.patch.object(self.notifier, '_cast_message')
  39         self.mock_util = mock_util_p.start()
  40         self.mock_log = mock_log_p.start()
  41         self.mock_fanout = mock_fanout_p.start()
  42         self.mock_cast = mock_cast_p.start()
  43
  44     def _test__schedule_network(self, network,
  45                                 new_agents=None, existing_agents=None,
  46                                 expected_casts=0, expected_warnings=0):
  47         self.notifier.plugin.schedule_network.return_value = new_agents
  48         agents = self.notifier._schedule_network(
  49             mock.ANY, network, existing_agents)
  50         if new_agents is None:
  51             new_agents = []
  52         self.assertEqual(new_agents + existing_agents, agents)
  53         self.assertEqual(expected_casts, self.mock_cast.call_count)
  54         self.assertEqual(expected_warnings, self.mock_log.warn.call_count)
  55
  56     def test__schedule_network(self):
  57         agent = agents_db.Agent()
  58         agent.admin_state_up = True
  59         agent.heartbeat_timestamp = timeutils.utcnow()
  60         network = {'id': 'foo_net_id'}
  61         self._test__schedule_network(network,
  62                                      new_agents=[agent], existing_agents=[],
  63                                      expected_casts=1, expected_warnings=0)
  64
  65     def test__schedule_network_no_existing_agents(self):
  66         agent = agents_db.Agent()
  67         agent.admin_state_up = True
  68         agent.heartbeat_timestamp = timeutils.utcnow()
  69         network = {'id': 'foo_net_id'}
  70         self._test__schedule_network(network,
  71                                      new_agents=None, existing_agents=[agent],
  72                                      expected_casts=0, expected_warnings=0)
  73
  74     def test__schedule_network_no_new_agents(self):
  75         network = {'id': 'foo_net_id'}
  76         self._test__schedule_network(network,
  77                                      new_agents=None, existing_agents=[],
  78                                      expected_casts=0, expected_warnings=1)
  79
  80     def _test__get_enabled_agents(self, network,
  81                                   agents=None, port_count=0,
  82                                   expected_warnings=0, expected_errors=0):
  83         self.notifier.plugin.get_ports_count.return_value = port_count
  84         enabled_agents = self.notifier._get_enabled_agents(
  85             mock.ANY, network, agents, mock.ANY, mock.ANY)
  86         if not cfg.CONF.enable_services_on_agents_with_admin_state_down:
  87             agents = [x for x in agents if x.admin_state_up]
  88         self.assertEqual(agents, enabled_agents)
  89         self.assertEqual(expected_warnings, self.mock_log.warn.call_count)
  90         self.assertEqual(expected_errors, self.mock_log.error.call_count)
  91
  92     def test__get_enabled_agents(self):
  93         agent1 = agents_db.Agent()
  94         agent1.admin_state_up = True
  95         agent1.heartbeat_timestamp = timeutils.utcnow()
  96         agent2 = agents_db.Agent()
  97         agent2.admin_state_up = False
  98         agent2.heartbeat_timestamp = timeutils.utcnow()
  99         network = {'id': 'foo_network_id'}
 100         self._test__get_enabled_agents(network, agents=[agent1])
 101
 102     def test__get_enabled_agents_with_inactive_ones(self):
 103         agent1 = agents_db.Agent()
 104         agent1.admin_state_up = True
 105         agent1.heartbeat_timestamp = timeutils.utcnow()
 106         agent2 = agents_db.Agent()
 107         agent2.admin_state_up = True
 108         # This is effectively an inactive agent
 109         agent2.heartbeat_timestamp = datetime.datetime(2000, 1, 1, 0, 0)
 110         network = {'id': 'foo_network_id'}
 111         self._test__get_enabled_agents(network,
 112                                        agents=[agent1, agent2],
 113                                        expected_warnings=1, expected_errors=0)
 114
 115     def test__get_enabled_agents_with_notification_required(self):
 116         network = {'id': 'foo_network_id', 'subnets': ['foo_subnet_id']}
 117         agent = agents_db.Agent()
 118         agent.admin_state_up = False
 119         agent.heartbeat_timestamp = timeutils.utcnow()
 120         self._test__get_enabled_agents(network, [agent], port_count=20,
 121                                        expected_warnings=0, expected_errors=1)
 122
 123     def test__get_enabled_agents_with_admin_state_down(self):
 124         cfg.CONF.set_override(
 125             'enable_services_on_agents_with_admin_state_down', True)
 126         agent1 = agents_db.Agent()
 127         agent1.admin_state_up = True
 128         agent1.heartbeat_timestamp = timeutils.utcnow()
 129         agent2 = agents_db.Agent()
 130         agent2.admin_state_up = False
 131         agent2.heartbeat_timestamp = timeutils.utcnow()
 132         network = {'id': 'foo_network_id'}
 133         self._test__get_enabled_agents(network, agents=[agent1, agent2])
 134
 135     def test__notify_agents_fanout_required(self):
 136         self.notifier._notify_agents(mock.ANY,
 137                                      'network_delete_end',
 138                                      mock.ANY, 'foo_network_id')
 139         self.assertEqual(1, self.mock_fanout.call_count)
 140
 141     def _test__notify_agents(self, method,
 142                              expected_scheduling=0, expected_casts=0):
 143         with mock.patch.object(self.notifier, '_schedule_network') as f:
 144             with mock.patch.object(self.notifier, '_get_enabled_agents') as g:
 145                 agent = agents_db.Agent()
 146                 agent.admin_state_up = True
 147                 agent.heartbeat_timestamp = timeutils.utcnow()
 148                 g.return_value = [agent]
 149                 dummy_payload = {'port': {}}
 150                 self.notifier._notify_agents(mock.Mock(), method,
 151                                              dummy_payload, 'foo_network_id')
 152                 self.assertEqual(expected_scheduling, f.call_count)
 153                 self.assertEqual(expected_casts, self.mock_cast.call_count)
 154
 155     def test__notify_agents_cast_required_with_scheduling(self):
 156         self._test__notify_agents('port_create_end',
 157                                   expected_scheduling=1, expected_casts=1)
 158
 159     def test__notify_agents_cast_required_wo_scheduling_on_port_update(self):
 160         self._test__notify_agents('port_update_end',
 161                                   expected_scheduling=0, expected_casts=1)
 162
 163     def test__notify_agents_cast_required_with_scheduling_subnet_create(self):
 164         self._test__notify_agents('subnet_create_end',
 165                                   expected_scheduling=1, expected_casts=1)
 166
 167     def test__notify_agents_no_action(self):
 168         self._test__notify_agents('network_create_end',
 169                                   expected_scheduling=0, expected_casts=0)
 170
 171     def test__fanout_message(self):
 172         self.notifier._fanout_message(mock.ANY, mock.ANY, mock.ANY)
 173         self.assertEqual(1, self.mock_fanout.call_count)
 174
 175     def test__cast_message(self):
 176         self.notifier._cast_message(mock.ANY, mock.ANY, mock.ANY)
 177         self.assertEqual(1, self.mock_cast.call_count)