neutron/tests/functional/agent/linux/test_ovsdb_monitor.py

   1 # Copyright 2013 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 """
  16 Tests in this module will be skipped unless:
  17
  18  - ovsdb-client is installed
  19
  20  - ovsdb-client can be invoked password-less via the configured root helper
  21
  22  - sudo testing is enabled (see neutron.tests.functional.base for details)
  23 """
  24
  25 from oslo_config import cfg
  26
  27 from neutron.agent.linux import ovsdb_monitor
  28 from neutron.agent.linux import utils
  29 from neutron.tests.common import net_helpers
  30 from neutron.tests.functional.agent.linux import base as linux_base
  31 from neutron.tests.functional import base as functional_base
  32
  33
  34 class BaseMonitorTest(linux_base.BaseOVSLinuxTestCase):
  35
  36     def setUp(self):
  37         super(BaseMonitorTest, self).setUp()
  38
  39         rootwrap_not_configured = (cfg.CONF.AGENT.root_helper ==
  40                                    functional_base.SUDO_CMD)
  41         if rootwrap_not_configured:
  42             # The monitor tests require a nested invocation that has
  43             # to be emulated by double sudo if rootwrap is not
  44             # configured.
  45             self.config(group='AGENT',
  46                         root_helper=" ".join([functional_base.SUDO_CMD] * 2))
  47
  48         self._check_test_requirements()
  49         # ovsdb-client monitor needs to have a bridge to make any output
  50         self.useFixture(net_helpers.OVSBridgeFixture())
  51
  52     def _check_test_requirements(self):
  53         self.check_command(['ovsdb-client', 'list-dbs'],
  54                            'Exit code: 1',
  55                            'password-less sudo not granted for ovsdb-client',
  56                            run_as_root=True)
  57
  58
  59 class TestOvsdbMonitor(BaseMonitorTest):
  60
  61     def setUp(self):
  62         super(TestOvsdbMonitor, self).setUp()
  63
  64         self.monitor = ovsdb_monitor.OvsdbMonitor('Bridge')
  65         self.addCleanup(self.monitor.stop)
  66         self.monitor.start()
  67
  68     def collect_monitor_output(self):
  69         output = list(self.monitor.iter_stdout())
  70         if output:
  71             # Output[0] is header row with spaces for column separation.
  72             # Use 'other_config' as an indication of the table header.
  73             self.assertIn('other_config', output[0])
  74             return True
  75
  76     def test_monitor_generates_initial_output(self):
  77         utils.wait_until_true(self.collect_monitor_output, timeout=30)
  78
  79
  80 class TestSimpleInterfaceMonitor(BaseMonitorTest):
  81
  82     def setUp(self):
  83         super(TestSimpleInterfaceMonitor, self).setUp()
  84
  85         self.monitor = ovsdb_monitor.SimpleInterfaceMonitor()
  86         self.addCleanup(self.monitor.stop)
  87         self.monitor.start(block=True, timeout=60)
  88
  89     def test_has_updates(self):
  90         utils.wait_until_true(lambda: self.monitor.has_updates)
  91         # clear the event list
  92         self.monitor.get_events()
  93         self.useFixture(net_helpers.OVSPortFixture())
  94         # has_updates after port addition should become True
  95         utils.wait_until_true(lambda: self.monitor.has_updates is True)
  96
  97     def _expected_devices_events(self, devices, state):
  98         """Helper to check that events are received for expected devices.
  99
 100         :param devices: The list of expected devices. WARNING: This list
 101           is modified by this method
 102         :param state: The state of the devices (added or removed)
 103         """
 104         events = self.monitor.get_events()
 105         event_devices = [
 106             (dev['name'], dev['external_ids']) for dev in events.get(state)]
 107         for dev in event_devices:
 108             if dev[0] in devices:
 109                 devices.remove(dev[0])
 110                 self.assertEqual(dev[1].get('iface-status'), 'active')
 111             if not devices:
 112                 return True
 113
 114     def test_get_events(self):
 115         utils.wait_until_true(lambda: self.monitor.has_updates)
 116         devices = self.monitor.get_events()
 117         self.assertTrue(devices.get('added'),
 118                         'Initial call should always be true')
 119         br = self.useFixture(net_helpers.OVSBridgeFixture())
 120         p1 = self.useFixture(net_helpers.OVSPortFixture(br.bridge))
 121         p2 = self.useFixture(net_helpers.OVSPortFixture(br.bridge))
 122         added_devices = [p1.port.name, p2.port.name]
 123         utils.wait_until_true(
 124             lambda: self._expected_devices_events(added_devices, 'added'))
 125         br.bridge.delete_port(p1.port.name)
 126         br.bridge.delete_port(p2.port.name)
 127         removed_devices = [p1.port.name, p2.port.name]
 128         utils.wait_until_true(
 129             lambda: self._expected_devices_events(removed_devices, 'removed'))
 130         # restart
 131         self.monitor.stop(block=True)
 132         self.monitor.start(block=True, timeout=60)
 133         devices = self.monitor.get_events()
 134         self.assertTrue(devices.get('added'),
 135                         'Initial call should always be true')