neutron/tests/unit/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 import mock
  16
  17 from neutron.agent.common import ovs_lib
  18 from neutron.agent.linux import ovsdb_monitor
  19 from neutron.tests import base
  20
  21
  22 class TestOvsdbMonitor(base.BaseTestCase):
  23
  24     def test___init__(self):
  25         ovsdb_monitor.OvsdbMonitor('Interface')
  26
  27     def test___init___with_columns(self):
  28         columns = ['col1', 'col2']
  29         with mock.patch(
  30             'neutron.agent.linux.async_process.AsyncProcess.__init__') as init:
  31             ovsdb_monitor.OvsdbMonitor('Interface', columns=columns)
  32             cmd = init.call_args_list[0][0][0]
  33             self.assertEqual('col1,col2', cmd[-1])
  34
  35     def test___init___with_format(self):
  36         with mock.patch(
  37             'neutron.agent.linux.async_process.AsyncProcess.__init__') as init:
  38             ovsdb_monitor.OvsdbMonitor('Interface', format='blob')
  39             cmd = init.call_args_list[0][0][0]
  40             self.assertEqual('--format=blob', cmd[-1])
  41
  42
  43 class TestSimpleInterfaceMonitor(base.BaseTestCase):
  44
  45     def setUp(self):
  46         super(TestSimpleInterfaceMonitor, self).setUp()
  47         self.monitor = ovsdb_monitor.SimpleInterfaceMonitor()
  48
  49     def test_has_updates_is_false_if_active_with_no_output(self):
  50         target = ('neutron.agent.linux.ovsdb_monitor.SimpleInterfaceMonitor'
  51                   '.is_active')
  52         with mock.patch(target, return_value=True):
  53             self.assertFalse(self.monitor.has_updates)
  54
  55     def test_has_updates_after_calling_get_events_is_false(self):
  56         with mock.patch.object(
  57                 self.monitor, 'process_events') as process_events:
  58             self.monitor.new_events = {'added': ['foo'], 'removed': ['foo1']}
  59             self.assertTrue(self.monitor.has_updates)
  60             self.monitor.get_events()
  61             self.assertTrue(process_events.called)
  62             self.assertFalse(self.monitor.has_updates)
  63
  64     def process_event_unassigned_of_port(self):
  65         output = '{"data":[["e040fbec-0579-4990-8324-d338da33ae88","insert",'
  66         output += '"m50",["set",[]],["map",[]]]],"headings":["row","action",'
  67         output += '"name","ofport","external_ids"]}'
  68         with mock.patch.object(
  69                 self.monitor, 'iter_stdout', return_value=[output]):
  70             self.monitor.process_events()
  71             self.assertEqual(self.monitor.new_events['added'][0]['ofport'],
  72                              ovs_lib.UNASSIGNED_OFPORT)