← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/agent/linux/test_utils.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
https://launchpad.net/bugs/1263539
* 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:
neutron/tests/unit/agent/linux/test_utils.py
11
11
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
12
#    License for the specific language governing permissions and limitations
13
13
#    under the License.
 
14
 
 
15
import os
 
16
 
14
17
import mock
 
18
import socket
15
19
import testtools
16
20
 
17
21
from neutron.agent.linux import utils
21
25
_marker = object()
22
26
 
23
27
 
24
 
class FakeCreateProcess(object):
25
 
    class FakeStdin(object):
26
 
        def close(self):
27
 
            pass
28
 
 
29
 
    def __init__(self, returncode):
30
 
        self.returncode = returncode
31
 
        self.stdin = self.FakeStdin()
32
 
 
33
 
    def communicate(self, process_input=None):
34
 
        return '', ''
35
 
 
36
 
 
37
28
class AgentUtilsExecuteTest(base.BaseTestCase):
38
29
    def setUp(self):
39
30
        super(AgentUtilsExecuteTest, self).setUp()
40
 
        self.root_helper = "echo"
41
31
        self.test_file = self.get_temp_file_path('test_execute.tmp')
42
32
        open(self.test_file, 'w').close()
43
 
        self.mock_popen_p = mock.patch("subprocess.Popen.communicate")
44
 
        self.mock_popen = self.mock_popen_p.start()
 
33
        self.process = mock.patch('eventlet.green.subprocess.Popen').start()
 
34
        self.process.return_value.returncode = 0
 
35
        self.mock_popen = self.process.return_value.communicate
45
36
 
46
37
    def test_without_helper(self):
47
38
        expected = "%s\n" % self.test_file
52
43
    def test_with_helper(self):
53
44
        expected = "ls %s\n" % self.test_file
54
45
        self.mock_popen.return_value = [expected, ""]
55
 
        result = utils.execute(["ls", self.test_file],
56
 
                               self.root_helper)
 
46
        self.config(group='AGENT', root_helper='echo')
 
47
        result = utils.execute(["ls", self.test_file], run_as_root=True)
57
48
        self.assertEqual(result, expected)
58
49
 
59
50
    def test_stderr_true(self):
89
80
        self.assertEqual(result, expected)
90
81
 
91
82
    def test_return_code_log_error_raise_runtime(self):
92
 
        with mock.patch.object(utils, 'create_process') as create_process:
93
 
            create_process.return_value = FakeCreateProcess(1), 'ls'
94
 
            with mock.patch.object(utils, 'LOG') as log:
95
 
                self.assertRaises(RuntimeError, utils.execute,
96
 
                                  ['ls'])
97
 
                self.assertTrue(log.error.called)
 
83
        self.mock_popen.return_value = ('', '')
 
84
        self.process.return_value.returncode = 1
 
85
        with mock.patch.object(utils, 'LOG') as log:
 
86
            self.assertRaises(RuntimeError, utils.execute,
 
87
                              ['ls'])
 
88
            self.assertTrue(log.error.called)
98
89
 
99
90
    def test_return_code_log_error_no_raise_runtime(self):
100
 
        with mock.patch.object(utils, 'create_process') as create_process:
101
 
            create_process.return_value = FakeCreateProcess(1), 'ls'
102
 
            with mock.patch.object(utils, 'LOG') as log:
103
 
                utils.execute(['ls'], check_exit_code=False)
104
 
                self.assertTrue(log.error.called)
 
91
        self.mock_popen.return_value = ('', '')
 
92
        self.process.return_value.returncode = 1
 
93
        with mock.patch.object(utils, 'LOG') as log:
 
94
            utils.execute(['ls'], check_exit_code=False)
 
95
            self.assertTrue(log.error.called)
105
96
 
106
97
    def test_return_code_log_debug(self):
107
 
        with mock.patch.object(utils, 'create_process') as create_process:
108
 
            create_process.return_value = FakeCreateProcess(0), 'ls'
109
 
            with mock.patch.object(utils, 'LOG') as log:
110
 
                utils.execute(['ls'])
111
 
                self.assertTrue(log.debug.called)
 
98
        self.mock_popen.return_value = ('', '')
 
99
        with mock.patch.object(utils, 'LOG') as log:
 
100
            utils.execute(['ls'])
 
101
            self.assertTrue(log.debug.called)
112
102
 
113
103
    def test_return_code_raise_runtime_do_not_log_fail_as_error(self):
114
 
        with mock.patch.object(utils, 'create_process') as create_process:
115
 
            create_process.return_value = FakeCreateProcess(1), 'ls'
116
 
            with mock.patch.object(utils, 'LOG') as log:
117
 
                self.assertRaises(RuntimeError, utils.execute,
118
 
                                  ['ls'], log_fail_as_error=False)
119
 
                self.assertTrue(log.debug.called)
 
104
        self.mock_popen.return_value = ('', '')
 
105
        self.process.return_value.returncode = 1
 
106
        with mock.patch.object(utils, 'LOG') as log:
 
107
            self.assertRaises(RuntimeError, utils.execute,
 
108
                              ['ls'], log_fail_as_error=False)
 
109
            self.assertFalse(log.error.called)
120
110
 
121
111
 
122
112
class AgentUtilsGetInterfaceMAC(base.BaseTestCase):
172
162
 
173
163
class TestGetRoothelperChildPid(base.BaseTestCase):
174
164
    def _test_get_root_helper_child_pid(self, expected=_marker,
175
 
                                        root_helper=None, pids=None):
 
165
                                        run_as_root=False, pids=None):
176
166
        def _find_child_pids(x):
177
167
            if not pids:
178
168
                return []
182
172
        mock_pid = object()
183
173
        with mock.patch.object(utils, 'find_child_pids',
184
174
                               side_effect=_find_child_pids):
185
 
            actual = utils.get_root_helper_child_pid(mock_pid, root_helper)
 
175
            actual = utils.get_root_helper_child_pid(mock_pid, run_as_root)
186
176
        if expected is _marker:
187
177
            expected = str(mock_pid)
188
178
        self.assertEqual(expected, actual)
189
179
 
190
 
    def test_returns_process_pid_without_root_helper(self):
 
180
    def test_returns_process_pid_not_root(self):
191
181
        self._test_get_root_helper_child_pid()
192
182
 
193
 
    def test_returns_child_pid_with_root_helper(self):
 
183
    def test_returns_child_pid_as_root(self):
194
184
        self._test_get_root_helper_child_pid(expected='2', pids=['1', '2'],
195
 
                                             root_helper='a')
 
185
                                             run_as_root=True)
196
186
 
197
 
    def test_returns_last_child_pid_with_root_helper(self):
 
187
    def test_returns_last_child_pid_as_root(self):
198
188
        self._test_get_root_helper_child_pid(expected='3',
199
189
                                             pids=['1', '2', '3'],
200
 
                                             root_helper='a')
201
 
 
202
 
    def test_returns_none_with_root_helper(self):
203
 
        self._test_get_root_helper_child_pid(expected=None, root_helper='a')
 
190
                                             run_as_root=True)
 
191
 
 
192
    def test_returns_none_as_root(self):
 
193
        self._test_get_root_helper_child_pid(expected=None, run_as_root=True)
 
194
 
 
195
 
 
196
class TestPathUtilities(base.BaseTestCase):
 
197
    def test_remove_abs_path(self):
 
198
        self.assertEqual(['ping', '8.8.8.8'],
 
199
                         utils.remove_abs_path(['/usr/bin/ping', '8.8.8.8']))
 
200
 
 
201
    def test_cmdlines_are_equal(self):
 
202
        self.assertTrue(utils.cmdlines_are_equal(
 
203
            ['ping', '8.8.8.8'],
 
204
            ['/usr/bin/ping', '8.8.8.8']))
 
205
 
 
206
    def test_cmdlines_are_equal_different_commands(self):
 
207
        self.assertFalse(utils.cmdlines_are_equal(
 
208
            ['ping', '8.8.8.8'],
 
209
            ['/usr/bin/ping6', '8.8.8.8']))
 
210
 
 
211
 
 
212
class TestBaseOSUtils(base.BaseTestCase):
 
213
    @mock.patch.object(os.path, 'isdir', return_value=False)
 
214
    @mock.patch.object(os, 'makedirs')
 
215
    def test_ensure_dir_not_exist(self, makedirs, isdir):
 
216
        utils.ensure_dir('/the')
 
217
        isdir.assert_called_once_with('/the')
 
218
        makedirs.assert_called_once_with('/the', 0o755)
 
219
 
 
220
    @mock.patch.object(os.path, 'isdir', return_value=True)
 
221
    @mock.patch.object(os, 'makedirs')
 
222
    def test_ensure_dir_exist(self, makedirs, isdir):
 
223
        utils.ensure_dir('/the')
 
224
        isdir.assert_called_once_with('/the')
 
225
        self.assertFalse(makedirs.called)
 
226
 
 
227
 
 
228
class TestUnixDomainHttpConnection(base.BaseTestCase):
 
229
    def test_connect(self):
 
230
        with mock.patch.object(utils, 'cfg') as cfg:
 
231
            cfg.CONF.metadata_proxy_socket = '/the/path'
 
232
            with mock.patch('socket.socket') as socket_create:
 
233
                conn = utils.UnixDomainHTTPConnection('169.254.169.254',
 
234
                                                      timeout=3)
 
235
                conn.connect()
 
236
 
 
237
                socket_create.assert_has_calls([
 
238
                    mock.call(socket.AF_UNIX, socket.SOCK_STREAM),
 
239
                    mock.call().settimeout(3),
 
240
                    mock.call().connect('/the/path')]
 
241
                )
 
242
                self.assertEqual(conn.timeout, 3)
 
243
 
 
244
 
 
245
class TestUnixDomainHttpProtocol(base.BaseTestCase):
 
246
    def test_init_empty_client(self):
 
247
        u = utils.UnixDomainHttpProtocol(mock.Mock(), '', mock.Mock())
 
248
        self.assertEqual(u.client_address, ('<local>', 0))
 
249
 
 
250
    def test_init_with_client(self):
 
251
        u = utils.UnixDomainHttpProtocol(mock.Mock(), 'foo', mock.Mock())
 
252
        self.assertEqual(u.client_address, 'foo')
 
253
 
 
254
 
 
255
class TestUnixDomainWSGIServer(base.BaseTestCase):
 
256
    def setUp(self):
 
257
        super(TestUnixDomainWSGIServer, self).setUp()
 
258
        self.eventlet_p = mock.patch.object(utils, 'eventlet')
 
259
        self.eventlet = self.eventlet_p.start()
 
260
        self.server = utils.UnixDomainWSGIServer('test')
 
261
 
 
262
    def test_start(self):
 
263
        mock_app = mock.Mock()
 
264
        with mock.patch.object(self.server, '_launch') as launcher:
 
265
            self.server.start(mock_app, '/the/path', workers=5, backlog=128)
 
266
            self.eventlet.assert_has_calls([
 
267
                mock.call.listen(
 
268
                    '/the/path',
 
269
                    family=socket.AF_UNIX,
 
270
                    backlog=128
 
271
                )]
 
272
            )
 
273
            launcher.assert_called_once_with(mock_app, workers=5)
 
274
 
 
275
    def test_run(self):
 
276
        self.server._run('app', 'sock')
 
277
 
 
278
        self.eventlet.wsgi.server.assert_called_once_with(
 
279
            'sock',
 
280
            'app',
 
281
            protocol=utils.UnixDomainHttpProtocol,
 
282
            log=mock.ANY,
 
283
            max_size=self.server.num_threads
 
284
        )