1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright (c) 2012 OpenStack, LLC.
5
# Licensed under the Apache License, Version 2.0 (the "License"); you may
6
# not use this file except in compliance with the License. You may obtain
7
# a copy of the License at
9
# http://www.apache.org/licenses/LICENSE-2.0
11
# Unless required by applicable law or agreed to in writing, software
12
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14
# License for the specific language governing permissions and limitations
18
import unittest2 as unittest
20
from quantum.openstack.common import cfg
21
from quantum.plugins.openvswitch.agent import ovs_quantum_agent
22
from quantum.plugins.openvswitch.common import config
25
class CreateAgentConfigMap(unittest.TestCase):
27
def test_create_agent_config_map_succeeds(self):
28
self.assertTrue(ovs_quantum_agent.create_agent_config_map(cfg.CONF))
30
def test_create_agent_config_map_fails_for_invalid_tunnel_config(self):
31
self.addCleanup(cfg.CONF.reset)
32
# An ip address is required for tunneling but there is no default
33
cfg.CONF.set_override('enable_tunneling', True, group='OVS')
34
with self.assertRaises(ValueError):
35
ovs_quantum_agent.create_agent_config_map(cfg.CONF)
38
class TestOvsQuantumAgent(unittest.TestCase):
41
self.addCleanup(cfg.CONF.reset)
42
# Avoid rpc initialization for unit tests
43
cfg.CONF.set_override('rpc_backend',
44
'quantum.openstack.common.rpc.impl_fake')
45
kwargs = ovs_quantum_agent.create_agent_config_map(cfg.CONF)
46
with mock.patch('quantum.plugins.openvswitch.agent.ovs_quantum_agent.'
47
'OVSQuantumAgent.setup_integration_br',
48
return_value=mock.Mock()):
49
with mock.patch('quantum.agent.linux.utils.get_interface_mac',
50
return_value='000000000001'):
51
self.agent = ovs_quantum_agent.OVSQuantumAgent(**kwargs)
52
self.agent.plugin_rpc = mock.Mock()
53
self.agent.context = mock.Mock()
54
self.agent.agent_id = mock.Mock()
56
def mock_port_bound(self, ofport=None):
59
net_uuid = 'my-net-uuid'
60
with mock.patch.object(self.agent.int_br,
61
'delete_flows') as delete_flows_func:
62
self.agent.port_bound(port, net_uuid, 'local', None, None)
63
self.assertEqual(delete_flows_func.called, ofport != -1)
65
def test_port_bound_deletes_flows_for_valid_ofport(self):
66
self.mock_port_bound(ofport=1)
68
def test_port_bound_ignores_flows_for_invalid_ofport(self):
69
self.mock_port_bound(ofport=-1)
71
def test_port_dead(self):
72
with mock.patch.object(self.agent.int_br,
73
'add_flow') as add_flow_func:
74
self.agent.port_dead(mock.Mock())
75
self.assertTrue(add_flow_func.called)
77
def mock_update_ports(self, vif_port_set=None, registered_ports=None):
78
with mock.patch.object(self.agent.int_br, 'get_vif_port_set',
79
return_value=vif_port_set):
80
return self.agent.update_ports(registered_ports)
82
def test_update_ports_returns_none_for_unchanged_ports(self):
83
self.assertIsNone(self.mock_update_ports())
85
def test_update_ports_returns_port_changes(self):
86
vif_port_set = set([1, 3])
87
registered_ports = set([1, 2])
88
expected = dict(current=vif_port_set, added=set([3]), removed=set([2]))
89
actual = self.mock_update_ports(vif_port_set, registered_ports)
90
self.assertEqual(expected, actual)
92
def test_treat_devices_added_returns_true_for_missing_device(self):
93
with mock.patch.object(self.agent.plugin_rpc, 'get_device_details',
94
side_effect=Exception()):
95
self.assertTrue(self.agent.treat_devices_added([{}]))
97
def mock_treat_devices_added(self, details, port, func_name):
100
:param details: the details to return for the device
101
:param port: the port that get_vif_port_by_id should return
102
:param func_name: the function that should be called
103
:returns: whether the named function was called
105
with mock.patch.object(self.agent.plugin_rpc, 'get_device_details',
106
return_value=details):
107
with mock.patch.object(self.agent.int_br, 'get_vif_port_by_id',
109
with mock.patch.object(self.agent, func_name) as func:
110
self.assertFalse(self.agent.treat_devices_added([{}]))
113
def test_treat_devices_added_ignores_invalid_ofport(self):
116
self.assertFalse(self.mock_treat_devices_added(mock.MagicMock(), port,
119
def test_treat_devices_added_marks_unknown_port_as_dead(self):
122
self.assertTrue(self.mock_treat_devices_added(mock.MagicMock(), port,
125
def test_treat_devices_added_updates_known_port(self):
126
details = mock.MagicMock()
127
details.__contains__.side_effect = lambda x: True
128
self.assertTrue(self.mock_treat_devices_added(details,
132
def test_treat_devices_removed_returns_true_for_missing_device(self):
133
with mock.patch.object(self.agent.plugin_rpc, 'update_device_down',
134
side_effect=Exception()):
135
self.assertTrue(self.agent.treat_devices_removed([{}]))
137
def mock_treat_devices_removed(self, port_exists):
138
details = dict(exists=port_exists)
139
with mock.patch.object(self.agent.plugin_rpc, 'update_device_down',
140
return_value=details):
141
with mock.patch.object(self.agent, 'port_unbound') as func:
142
self.assertFalse(self.agent.treat_devices_removed([{}]))
143
self.assertEqual(func.called, not port_exists)
145
def test_treat_devices_removed_unbinds_port(self):
146
self.mock_treat_devices_removed(False)
148
def test_treat_devices_removed_ignores_missing_port(self):
149
self.mock_treat_devices_removed(False)
added
removed
quantum/tests/unit/openvswitch/test_ovs_quantum_agent.py
