← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid-proposed

« back to all changes in this revision

Viewing changes to neutron/tests/unit/plugins/sriovnicagent/test_sriov_nic_agent.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2015-04-15 13:59:07 UTC
  • mfrom: (1.1.22)
  • Revision ID: package-import@ubuntu.com-20150415135907-z10fr18evag1ozq3
Tags: 1:2015.1~rc1-0ubuntu1
* New upstream milestone release:
  - debian/control: Update dependencies. 
  - debian/patches/disable-udev-tests.patch: Dropped no longer needed.
  - debian/patches/fixup-driver-test-execution.patch: Dropped no longer needed.
  - debian/patches/skip-iptest.patch: Skip failing test
  - debian/neutron-plugin-openvswitch-agent.install: Added neutron-ovsvapp-agent binary.
  - debian/neutron-plugin-cisco.install: Added neutron-cisco-apic-service-agent and 
    neutron-cisco-apic-host-agent

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/plugins/sriovnicagent/test_sriov_nic_agent.py
 
1
# Copyright 2014 Mellanox Technologies, Ltd
 
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
 
 
17
import mock
 
18
from oslo_config import cfg
 
19
 
 
20
from neutron.plugins.sriovnicagent.common import config  # noqa
 
21
from neutron.plugins.sriovnicagent import sriov_nic_agent
 
22
from neutron.tests import base
 
23
 
 
24
DEVICE_MAC = '11:22:33:44:55:66'
 
25
 
 
26
 
 
27
class TestSriovAgent(base.BaseTestCase):
 
28
    def setUp(self):
 
29
        super(TestSriovAgent, self).setUp()
 
30
        # disable setting up periodic state reporting
 
31
        cfg.CONF.set_override('report_interval', 0, 'AGENT')
 
32
        cfg.CONF.set_default('firewall_driver',
 
33
                             'neutron.agent.firewall.NoopFirewallDriver',
 
34
                             group='SECURITYGROUP')
 
35
        cfg.CONF.set_default('enable_security_group',
 
36
                             False,
 
37
                             group='SECURITYGROUP')
 
38
 
 
39
        class MockFixedIntervalLoopingCall(object):
 
40
            def __init__(self, f):
 
41
                self.f = f
 
42
 
 
43
            def start(self, interval=0):
 
44
                self.f()
 
45
 
 
46
        mock.patch('neutron.openstack.common.loopingcall.'
 
47
                   'FixedIntervalLoopingCall',
 
48
                   new=MockFixedIntervalLoopingCall)
 
49
 
 
50
        self.agent = sriov_nic_agent.SriovNicSwitchAgent({}, {}, 0)
 
51
 
 
52
    def test_treat_devices_removed_with_existed_device(self):
 
53
        agent = sriov_nic_agent.SriovNicSwitchAgent({}, {}, 0)
 
54
        devices = [DEVICE_MAC]
 
55
        with mock.patch.object(agent.plugin_rpc,
 
56
                               "update_device_down") as fn_udd:
 
57
            fn_udd.return_value = {'device': DEVICE_MAC,
 
58
                                   'exists': True}
 
59
            with mock.patch.object(sriov_nic_agent.LOG,
 
60
                                   'info') as log:
 
61
                resync = agent.treat_devices_removed(devices)
 
62
                self.assertEqual(2, log.call_count)
 
63
                self.assertFalse(resync)
 
64
                self.assertTrue(fn_udd.called)
 
65
 
 
66
    def test_treat_devices_removed_with_not_existed_device(self):
 
67
        agent = sriov_nic_agent.SriovNicSwitchAgent({}, {}, 0)
 
68
        devices = [DEVICE_MAC]
 
69
        with mock.patch.object(agent.plugin_rpc,
 
70
                               "update_device_down") as fn_udd:
 
71
            fn_udd.return_value = {'device': DEVICE_MAC,
 
72
                                   'exists': False}
 
73
            with mock.patch.object(sriov_nic_agent.LOG,
 
74
                                   'debug') as log:
 
75
                resync = agent.treat_devices_removed(devices)
 
76
                self.assertEqual(1, log.call_count)
 
77
                self.assertFalse(resync)
 
78
                self.assertTrue(fn_udd.called)
 
79
 
 
80
    def test_treat_devices_removed_failed(self):
 
81
        agent = sriov_nic_agent.SriovNicSwitchAgent({}, {}, 0)
 
82
        devices = [DEVICE_MAC]
 
83
        with mock.patch.object(agent.plugin_rpc,
 
84
                               "update_device_down") as fn_udd:
 
85
            fn_udd.side_effect = Exception()
 
86
            with mock.patch.object(sriov_nic_agent.LOG,
 
87
                                   'debug') as log:
 
88
                resync = agent.treat_devices_removed(devices)
 
89
                self.assertEqual(1, log.call_count)
 
90
                self.assertTrue(resync)
 
91
                self.assertTrue(fn_udd.called)
 
92
 
 
93
    def mock_scan_devices(self, expected, mock_current,
 
94
                          registered_devices, updated_devices):
 
95
        self.agent.eswitch_mgr = mock.Mock()
 
96
        self.agent.eswitch_mgr.get_assigned_devices.return_value = mock_current
 
97
 
 
98
        results = self.agent.scan_devices(registered_devices, updated_devices)
 
99
        self.assertEqual(expected, results)
 
100
 
 
101
    def test_scan_devices_returns_empty_sets(self):
 
102
        registered = set()
 
103
        updated = set()
 
104
        mock_current = set()
 
105
        expected = {'current': set(),
 
106
                    'updated': set(),
 
107
                    'added': set(),
 
108
                    'removed': set()}
 
109
        self.mock_scan_devices(expected, mock_current, registered, updated)
 
110
 
 
111
    def test_scan_devices_no_changes(self):
 
112
        registered = set(['1', '2'])
 
113
        updated = set()
 
114
        mock_current = set(['1', '2'])
 
115
        expected = {'current': set(['1', '2']),
 
116
                    'updated': set(),
 
117
                    'added': set(),
 
118
                    'removed': set()}
 
119
        self.mock_scan_devices(expected, mock_current, registered, updated)
 
120
 
 
121
    def test_scan_devices_new_and_removed(self):
 
122
        registered = set(['1', '2'])
 
123
        updated = set()
 
124
        mock_current = set(['2', '3'])
 
125
        expected = {'current': set(['2', '3']),
 
126
                    'updated': set(),
 
127
                    'added': set(['3']),
 
128
                    'removed': set(['1'])}
 
129
        self.mock_scan_devices(expected, mock_current, registered, updated)
 
130
 
 
131
    def test_scan_devices_new_updates(self):
 
132
        registered = set(['1'])
 
133
        updated = set(['2'])
 
134
        mock_current = set(['1', '2'])
 
135
        expected = {'current': set(['1', '2']),
 
136
                    'updated': set(['2']),
 
137
                    'added': set(['2']),
 
138
                    'removed': set()}
 
139
        self.mock_scan_devices(expected, mock_current, registered, updated)
 
140
 
 
141
    def test_scan_devices_updated_missing(self):
 
142
        registered = set(['1'])
 
143
        updated = set(['2'])
 
144
        mock_current = set(['1'])
 
145
        expected = {'current': set(['1']),
 
146
                    'updated': set(),
 
147
                    'added': set(),
 
148
                    'removed': set()}
 
149
        self.mock_scan_devices(expected, mock_current, registered, updated)
 
150
 
 
151
    def test_process_network_devices(self):
 
152
        agent = self.agent
 
153
        device_info = {'current': set(),
 
154
                       'added': set(['mac3', 'mac4']),
 
155
                       'updated': set(['mac2', 'mac3']),
 
156
                       'removed': set(['mac1'])}
 
157
        agent.sg_agent.prepare_devices_filter = mock.Mock()
 
158
        agent.sg_agent.refresh_firewall = mock.Mock()
 
159
        agent.treat_devices_added_updated = mock.Mock(return_value=False)
 
160
        agent.treat_devices_removed = mock.Mock(return_value=False)
 
161
 
 
162
        agent.process_network_devices(device_info)
 
163
 
 
164
        agent.sg_agent.prepare_devices_filter.assert_called_with(
 
165
                set(['mac3', 'mac4']))
 
166
        self.assertTrue(agent.sg_agent.refresh_firewall.called)
 
167
        agent.treat_devices_added_updated.assert_called_with(set(['mac2',
 
168
                                                                  'mac3',
 
169
                                                                  'mac4']))
 
170
        agent.treat_devices_removed.assert_called_with(set(['mac1']))
 
171
 
 
172
    def test_treat_devices_added_updated_admin_state_up_true(self):
 
173
        agent = self.agent
 
174
        mock_details = {'device': 'aa:bb:cc:dd:ee:ff',
 
175
                        'port_id': 'port123',
 
176
                        'network_id': 'net123',
 
177
                        'admin_state_up': True,
 
178
                        'network_type': 'vlan',
 
179
                        'segmentation_id': 100,
 
180
                        'profile': {'pci_slot': '1:2:3.0'},
 
181
                        'physical_network': 'physnet1'}
 
182
        agent.plugin_rpc = mock.Mock()
 
183
        agent.plugin_rpc.get_devices_details_list.return_value = [mock_details]
 
184
        agent.eswitch_mgr = mock.Mock()
 
185
        agent.eswitch_mgr.device_exists.return_value = True
 
186
        agent.set_device_state = mock.Mock()
 
187
        resync_needed = agent.treat_devices_added_updated(
 
188
                                    set(['aa:bb:cc:dd:ee:ff']))
 
189
 
 
190
        self.assertFalse(resync_needed)
 
191
        agent.eswitch_mgr.device_exists.assert_called_with('aa:bb:cc:dd:ee:ff',
 
192
                                                          '1:2:3.0')
 
193
        agent.eswitch_mgr.set_device_state.assert_called_with(
 
194
                                        'aa:bb:cc:dd:ee:ff',
 
195
                                        '1:2:3.0',
 
196
                                        True)
 
197
        self.assertTrue(agent.plugin_rpc.update_device_up.called)
 
198
 
 
199
    def test_treat_devices_added_updated_admin_state_up_false(self):
 
200
        agent = self.agent
 
201
        mock_details = {'device': 'aa:bb:cc:dd:ee:ff',
 
202
                        'port_id': 'port123',
 
203
                        'network_id': 'net123',
 
204
                        'admin_state_up': False,
 
205
                        'network_type': 'vlan',
 
206
                        'segmentation_id': 100,
 
207
                        'profile': {'pci_slot': '1:2:3.0'},
 
208
                        'physical_network': 'physnet1'}
 
209
        agent.plugin_rpc = mock.Mock()
 
210
        agent.plugin_rpc.get_devices_details_list.return_value = [mock_details]
 
211
        agent.remove_port_binding = mock.Mock()
 
212
        resync_needed = agent.treat_devices_added_updated(
 
213
                            set(['aa:bb:cc:dd:ee:ff']))
 
214
 
 
215
        self.assertFalse(resync_needed)
 
216
        self.assertFalse(agent.plugin_rpc.update_device_up.called)