neutron/tests/functional/agent/l3/test_keepalived_state_change.py

   1 # Copyright (c) 2015 Red Hat Inc.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    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, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 import os
  16
  17 import mock
  18 from oslo_config import cfg
  19 from oslo_config import fixture as fixture_config
  20 from oslo_utils import uuidutils
  21
  22 from neutron._i18n import _
  23 from neutron.agent.l3 import keepalived_state_change
  24 from neutron.tests.functional import base
  25
  26
  27 class TestKeepalivedStateChange(base.BaseSudoTestCase):
  28     def setUp(self):
  29         super(TestKeepalivedStateChange, self).setUp()
  30         self.conf_fixture = self.useFixture(fixture_config.Config())
  31         self.conf_fixture.register_opt(
  32             cfg.StrOpt('metadata_proxy_socket',
  33                        default='$state_path/metadata_proxy',
  34                        help=_('Location of Metadata Proxy UNIX domain '
  35                               'socket')))
  36
  37         self.router_id = uuidutils.generate_uuid()
  38         self.conf_dir = self.get_default_temp_dir().path
  39         self.cidr = '169.254.128.1/24'
  40         self.interface_name = 'interface'
  41         self.monitor = keepalived_state_change.MonitorDaemon(
  42             self.get_temp_file_path('monitor.pid'),
  43             self.router_id,
  44             1,
  45             2,
  46             'namespace',
  47             self.conf_dir,
  48             self.interface_name,
  49             self.cidr)
  50         mock.patch.object(self.monitor, 'notify_agent').start()
  51         self.line = '1: %s    inet %s' % (self.interface_name, self.cidr)
  52
  53     def test_parse_and_handle_event_wrong_device_completes_without_error(self):
  54         self.monitor.parse_and_handle_event(
  55             '1: wrong_device    inet wrong_cidr')
  56
  57     def _get_state(self):
  58         with open(os.path.join(self.monitor.conf_dir, 'state')) as state_file:
  59             return state_file.read()
  60
  61     def test_parse_and_handle_event_writes_to_file(self):
  62         self.monitor.parse_and_handle_event('Deleted %s' % self.line)
  63         self.assertEqual('backup', self._get_state())
  64
  65         self.monitor.parse_and_handle_event(self.line)
  66         self.assertEqual('master', self._get_state())
  67
  68     def test_parse_and_handle_event_fails_writing_state(self):
  69         with mock.patch.object(
  70                 self.monitor, 'write_state_change', side_effect=OSError):
  71             self.monitor.parse_and_handle_event(self.line)
  72
  73     def test_parse_and_handle_event_fails_notifying_agent(self):
  74         with mock.patch.object(
  75                 self.monitor, 'notify_agent', side_effect=Exception):
  76             self.monitor.parse_and_handle_event(self.line)