~ubuntu-branches/ubuntu/vivid/neutron/vivid-updates

« back to all changes in this revision

Viewing changes to neutron/tests/unit/test_netns_cleanup.py

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2015-03-30 11:17:19 UTC
  • mfrom: (1.1.21)
  • Revision ID: package-import@ubuntu.com-20150330111719-h0gx7233p4jkkgfh
Tags: 1:2015.1~b3-0ubuntu1
* New upstream milestone release:
  - d/control: Align version requirements with upstream.
  - d/control: Add new dependency on oslo-log.
  - d/p/*: Rebase.
  - d/control,d/neutron-plugin-hyperv*: Dropped, decomposed into
    separate project upstream.
  - d/control,d/neutron-plugin-openflow*: Dropped, decomposed into
    separate project upstream.
  - d/neutron-common.install: Add neutron-rootwrap-daemon and 
    neutron-keepalived-state-change binaries.
  - d/rules: Ignore neutron-hyperv-agent when installing; only for Windows.
  - d/neutron-plugin-cisco.install: Drop neutron-cisco-cfg-agent as
    decomposed into separate project upstream.
  - d/neutron-plugin-vmware.install: Drop neutron-check-nsx-config and
    neutron-nsx-manage as decomposed into separate project upstream.
  - d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent.
* d/pydist-overrides: Add overrides for oslo packages.
* d/control: Fixup type in package description (LP: #1263539).
* d/p/fixup-driver-test-execution.patch: Cherry pick fix from upstream VCS
  to support unit test exection in out-of-tree vendor drivers.
* d/neutron-common.postinst: Allow general access to /etc/neutron but limit
  access to root/neutron to /etc/neutron/neutron.conf to support execution
  of unit tests in decomposed vendor drivers.
* d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent
  package.

Show diffs side-by-side

added added

removed removed

Lines of Context:
15
15
 
16
16
import mock
17
17
 
18
 
from neutron.agent.linux import interface
19
18
from neutron.cmd import netns_cleanup as util
20
19
from neutron.tests import base
21
20
 
22
21
 
23
22
class TestNetnsCleanup(base.BaseTestCase):
24
23
 
25
 
    def setup_config(self):
26
 
        # don't use default config
27
 
        pass
28
 
 
29
 
    def test_setup_conf(self):
30
 
        expected_opts = interface.OPTS
31
 
        conf = util.setup_conf()
32
 
        self.assertTrue(all([opt.name in conf for opt in expected_opts]))
33
 
 
34
24
    def test_kill_dhcp(self, dhcp_active=True):
35
25
        conf = mock.Mock()
36
 
        conf.AGENT.root_helper = 'sudo',
37
26
        conf.dhcp_driver = 'driver'
38
27
 
39
 
        method_to_patch = 'oslo.utils.importutils.import_object'
 
28
        method_to_patch = 'oslo_utils.importutils.import_object'
40
29
 
41
30
        with mock.patch(method_to_patch) as import_object:
42
31
            driver = mock.Mock()
46
35
            util.kill_dhcp(conf, 'ns')
47
36
 
48
37
            expected_params = {'conf': conf, 'network': mock.ANY,
49
 
                               'root_helper': conf.AGENT.root_helper,
50
 
                               'plugin': mock.ANY,
51
 
                               'process_monitor': mock.ANY}
 
38
                               'process_monitor': mock.ANY,
 
39
                               'plugin': mock.ANY}
52
40
            import_object.assert_called_once_with('driver', **expected_params)
53
41
 
54
42
            if dhcp_active:
73
61
            self.assertEqual(util.eligible_for_deletion(conf, ns, force),
74
62
                             expected)
75
63
 
76
 
            expected_calls = [mock.call(conf.AGENT.root_helper, ns)]
 
64
            expected_calls = [mock.call(namespace=ns)]
77
65
            if not force:
78
66
                expected_calls.append(mock.call().namespace_is_empty())
79
67
            ip_wrap.assert_has_calls(expected_calls)
113
101
                util.unplug_device(conf, device)
114
102
 
115
103
                mock_get_bridge_for_iface.assert_called_once_with('tap1')
116
 
                ovs_br_cls.assert_called_once_with('br-int',
117
 
                                                   conf.AGENT.root_helper)
 
104
                ovs_br_cls.assert_called_once_with('br-int')
118
105
                ovs_bridge.assert_has_calls(
119
106
                    [mock.call.delete_port(device.name)])
120
107
 
144
131
    def _test_destroy_namespace_helper(self, force, num_devices):
145
132
        ns = 'qrouter-6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
146
133
        conf = mock.Mock()
147
 
#        conf.AGENT.root_helper = 'sudo'
148
134
 
149
135
        lo_device = mock.Mock()
150
136
        lo_device.name = 'lo'
165
151
 
166
152
                with mock.patch.object(util, 'kill_dhcp') as kill_dhcp:
167
153
                    util.destroy_namespace(conf, ns, force)
168
 
                    expected = [mock.call(conf.AGENT.root_helper, ns)]
 
154
                    expected = [mock.call(namespace=ns)]
169
155
 
170
156
                    if force:
171
157
                        expected.extend([
191
177
    def test_destroy_namespace_exception(self):
192
178
        ns = 'qrouter-6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
193
179
        conf = mock.Mock()
194
 
        conf.AGENT.root_helper = 'sudo'
195
180
        with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip_wrap:
196
181
            ip_wrap.side_effect = Exception()
197
182
            util.destroy_namespace(conf, ns)
201
186
        with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip_wrap:
202
187
            ip_wrap.get_namespaces.return_value = namespaces
203
188
 
204
 
            with mock.patch('eventlet.sleep') as eventlet_sleep:
 
189
            with mock.patch('time.sleep') as time_sleep:
205
190
                conf = mock.Mock()
206
191
                conf.force = False
207
192
                methods_to_mock = dict(
224
209
                             mock.call(conf, 'ns2', False)])
225
210
 
226
211
                        ip_wrap.assert_has_calls(
227
 
                            [mock.call.get_namespaces(conf.AGENT.root_helper)])
 
212
                            [mock.call.get_namespaces()])
228
213
 
229
 
                        eventlet_sleep.assert_called_once_with(2)
 
214
                        time_sleep.assert_called_once_with(2)
230
215
 
231
216
    def test_main_no_candidates(self):
232
217
        namespaces = ['ns1', 'ns2']
233
218
        with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip_wrap:
234
219
            ip_wrap.get_namespaces.return_value = namespaces
235
220
 
236
 
            with mock.patch('eventlet.sleep') as eventlet_sleep:
 
221
            with mock.patch('time.sleep') as time_sleep:
237
222
                conf = mock.Mock()
238
223
                conf.force = False
239
224
                methods_to_mock = dict(
248
233
                        util.main()
249
234
 
250
235
                        ip_wrap.assert_has_calls(
251
 
                            [mock.call.get_namespaces(conf.AGENT.root_helper)])
 
236
                            [mock.call.get_namespaces()])
252
237
 
253
238
                        mocks['eligible_for_deletion'].assert_has_calls(
254
239
                            [mock.call(conf, 'ns1', False),
256
241
 
257
242
                        self.assertFalse(mocks['destroy_namespace'].called)
258
243
 
259
 
                        self.assertFalse(eventlet_sleep.called)
 
244
                        self.assertFalse(time_sleep.called)