← Back to branch summary

~abentley/juju-ci-tools/client-from-config-4

« back to all changes in this revision

Viewing changes to tests/test_utility.py

  • Committer: Martin Packman
  • Date: 2015-11-17 14:19:06 UTC
  • mto: This revision was merged to the branch mainline in revision 1155.
  • Revision ID: martin.packman@canonical.com-20151117141906-a4zmuqre72s7fyf1
Sample script for generating image streams for rackspace

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_utility.py
12
12
import os
13
13
import socket
14
14
from time import time
15
 
import warnings
 
15
from unittest import TestCase
16
16
 
17
17
from mock import (
18
18
    call,
19
19
    patch,
20
20
    )
21
21
 
22
 
from tests import (
23
 
    TestCase,
24
 
)
25
22
from utility import (
26
23
    add_basic_testing_arguments,
27
 
    as_literal_address,
28
24
    extract_deb,
29
 
    _find_candidates,
30
25
    find_candidates,
31
26
    find_latest_branch_candidates,
32
27
    get_auth_token,
33
28
    get_candidates_path,
34
29
    get_deb_arch,
35
30
    get_winrm_certs,
36
 
    is_ipv6_address,
37
31
    quote,
38
32
    run_command,
39
33
    scoped_environ,
40
 
    split_address_port,
41
34
    temp_dir,
42
35
    until_timeout,
43
36
    wait_for_port,
108
101
 
109
102
class TestFindCandidates(TestCase):
110
103
 
111
 
    def test__find_candidates_artifacts_default(self):
112
 
        with temp_dir() as root:
113
 
            make_candidate_dir(root, 'master-artifacts')
114
 
            make_candidate_dir(root, '1.25')
115
 
            candidate = os.path.join(root, 'candidate', '1.25')
116
 
            self.assertEqual(list(_find_candidates(root)), [
117
 
                (candidate, os.path.join(candidate, 'buildvars.json'))])
118
 
 
119
 
    def test__find_candidates_artifacts_enabled(self):
120
 
        with temp_dir() as root:
121
 
            make_candidate_dir(root, 'master-artifacts')
122
 
            make_candidate_dir(root, '1.25')
123
 
            candidate = os.path.join(root, 'candidate', 'master-artifacts')
124
 
            self.assertEqual(list(_find_candidates(root, artifacts=True)), [
125
 
                (candidate, os.path.join(candidate, 'buildvars.json'))])
126
 
 
127
104
    def test_find_candidates(self):
128
105
        with temp_dir() as root:
129
106
            master_path = make_candidate_dir(root, 'master')
175
152
 
176
153
    def test_find_latest_branch_candidates(self):
177
154
        with temp_dir() as root:
178
 
            master_path = make_candidate_dir(root, 'master-artifacts')
 
155
            master_path = make_candidate_dir(root, 'master')
179
156
            self.assertEqual(find_latest_branch_candidates(root),
180
 
                             [(master_path, 1234)])
 
157
                             [master_path])
181
158
 
182
159
    def test_find_latest_branch_candidates_old_buildvars(self):
183
160
        with temp_dir() as root:
184
161
            a_week_ago = time() - timedelta(days=7, seconds=1).total_seconds()
185
 
            make_candidate_dir(root, 'master-artifacts', modified=a_week_ago)
 
162
            make_candidate_dir(root, 'master', modified=a_week_ago)
186
163
            self.assertEqual(find_latest_branch_candidates(root), [])
187
164
 
188
165
    def test_ignore_older_revision_build(self):
189
166
        with temp_dir() as root:
190
167
            path_1234 = make_candidate_dir(
191
 
                root, '1234-artifacts', 'mybranch', '1234')
 
168
                root, '1234', 'mybranch', '1234')
192
169
            make_candidate_dir(root, '1233', 'mybranch', '1233')
193
 
            self.assertEqual(find_latest_branch_candidates(root), [
194
 
                (path_1234, 1234)])
 
170
            self.assertEqual(find_latest_branch_candidates(root), [path_1234])
195
171
 
196
172
    def test_include_older_revision_build_different_branch(self):
197
173
        with temp_dir() as root:
198
174
            path_1234 = make_candidate_dir(
199
 
                root, '1234-artifacts', 'branch_foo', '1234')
 
175
                root, '1234', 'branch_foo', '1234')
200
176
            path_1233 = make_candidate_dir(
201
 
                root, '1233-artifacts', 'branch_bar', '1233')
 
177
                root, '1233', 'branch_bar', '1233')
202
178
            self.assertItemsEqual(
203
 
                find_latest_branch_candidates(root), [
204
 
                    (path_1233, 1233), (path_1234, 1234)])
205
 
 
206
 
 
207
 
class TestAsLiteralAddress(TestCase):
208
 
 
209
 
    def test_hostname(self):
210
 
        self.assertEqual("name.testing", as_literal_address("name.testing"))
211
 
 
212
 
    def test_ipv4(self):
213
 
        self.assertEqual("127.0.0.2", as_literal_address("127.0.0.2"))
214
 
 
215
 
    def test_ipv6(self):
216
 
        self.assertEqual("[2001:db8::7]", as_literal_address("2001:db8::7"))
217
 
 
218
 
 
219
 
class TestIsIPv6Address(TestCase):
220
 
 
221
 
    def test_hostname(self):
222
 
        self.assertIs(False, is_ipv6_address("name.testing"))
223
 
 
224
 
    def test_ipv4(self):
225
 
        self.assertIs(False, is_ipv6_address("127.0.0.2"))
226
 
 
227
 
    def test_ipv6(self):
228
 
        self.assertIs(True, is_ipv6_address("2001:db8::4"))
229
 
 
230
 
    def test_ipv6_missing_support(self):
231
 
        with patch('utility.socket', wraps=socket) as wrapped_socket:
232
 
            del wrapped_socket.inet_pton
233
 
            result = is_ipv6_address("2001:db8::4")
234
 
        # Would use expectedFailure here, but instead just assert wrong result.
235
 
        self.assertIs(False, result)
236
 
 
237
 
 
238
 
class TestSplitAddressPort(TestCase):
239
 
 
240
 
    def test_hostname(self):
241
 
        self.assertEqual(
242
 
            ("name.testing", None), split_address_port("name.testing"))
243
 
 
244
 
    def test_ipv4(self):
245
 
        self.assertEqual(
246
 
            ("127.0.0.2", "17017"), split_address_port("127.0.0.2:17017"))
247
 
 
248
 
    def test_ipv6(self):
249
 
        self.assertEqual(
250
 
            ("2001:db8::7", "17017"), split_address_port("2001:db8::7:17017"))
 
179
                find_latest_branch_candidates(root), [path_1233, path_1234])
251
180
 
252
181
 
253
182
class TestWaitForPort(TestCase):
298
227
        connect_mock.assert_called_once_with(('192.168.8.3', 27))
299
228
 
300
229
    def test_wait_for_port_no_address_closed(self):
301
 
        error = socket.gaierror(socket.EAI_NODATA, 'What address?')
302
230
        with patch('socket.getaddrinfo', autospec=True,
303
 
                   side_effect=error) as gai_mock:
 
231
                   side_effect=socket.error(-5, None)) as gai_mock:
304
232
            with patch('socket.socket') as socket_mock:
305
233
                wait_for_port('asdf', 26, closed=True)
306
234
        gai_mock.assert_called_once_with('asdf', 26, socket.AF_INET,
315
243
            if loc['stub_called']:
316
244
                raise ValueError()
317
245
            loc['stub_called'] = True
318
 
            raise socket.error(socket.EAI_NODATA, 'Err, address?')
 
246
            raise socket.error(-5, None)
319
247
 
320
248
        with patch('socket.getaddrinfo', autospec=True, side_effect=gai_stub,
321
249
                   ) as gai_mock:
328
256
            ])
329
257
        self.assertEqual(socket_mock.call_count, 0)
330
258
 
331
 
    def test_ipv6_open(self):
332
 
        gai_result = [(23, 0, 0, '', ('2001:db8::2', 22, 0, 0))]
333
 
        with patch('socket.getaddrinfo', autospec=True,
334
 
                   return_value=gai_result) as gai_mock:
335
 
            with patch('socket.socket') as socket_mock:
336
 
                wait_for_port('2001:db8::2', 22, closed=False)
337
 
        gai_mock.assert_called_once_with(
338
 
            '2001:db8::2', 22, socket.AF_INET6, socket.SOCK_STREAM)
339
 
        socket_mock.assert_called_once_with(23, 0, 0)
340
 
        connect_mock = socket_mock.return_value.connect
341
 
        connect_mock.assert_called_once_with(('2001:db8::2', 22, 0, 0))
342
 
 
343
259
 
344
260
class TestExtractDeb(TestCase):
345
261
 
372
288
            upload_tools=False, bootstrap_host=None, machine=[], region=None)
373
289
        self.assertEqual(args, expected)
374
290
 
375
 
    def test_positional_args_add_juju_bin_name(self):
376
 
        cmd_line = ['local', '/juju', '/tmp/logs', 'testtest']
377
 
        parser = add_basic_testing_arguments(ArgumentParser())
378
 
        args = parser.parse_args(cmd_line)
379
 
        self.assertEqual(args.juju_bin, '/juju')
380
 
 
381
 
    def test_positional_args_accepts_juju_exe(self):
382
 
        cmd_line = ['local', 'c:\\juju.exe', '/tmp/logs', 'testtest']
383
 
        parser = add_basic_testing_arguments(ArgumentParser())
384
 
        args = parser.parse_args(cmd_line)
385
 
        self.assertEqual(args.juju_bin, 'c:\\juju.exe')
386
 
 
387
 
    def test_warns_on_dirty_logs(self):
388
 
        with warnings.catch_warnings(record=True) as warned:
389
 
            with temp_dir() as log_dir:
390
 
                open(os.path.join(log_dir, "existing.log"), "w").close()
391
 
                cmd_line = ['local', '/a/juju', log_dir, 'testtest']
392
 
                parser = add_basic_testing_arguments(ArgumentParser())
393
 
                parser.parse_args(cmd_line)
394
 
            self.assertEqual(len(warned), 1)
395
 
            self.assertRegexpMatches(
396
 
                str(warned[0].message),
397
 
                r"^Directory '.*' has existing contents.$")
398
 
        self.assertEqual("", self.log_stream.getvalue())
399
 
 
400
 
    def test_no_warn_on_empty_logs(self):
401
 
        """Special case a file named 'empty' doesn't make log dir dirty"""
402
 
        with warnings.catch_warnings(record=True) as warned:
403
 
            with temp_dir() as log_dir:
404
 
                open(os.path.join(log_dir, "empty"), "w").close()
405
 
                cmd_line = ['local', '/a/juju', log_dir, 'testtest']
406
 
                parser = add_basic_testing_arguments(ArgumentParser())
407
 
                parser.parse_args(cmd_line)
408
 
            self.assertEqual(warned, [])
409
 
        self.assertEqual("", self.log_stream.getvalue())
410
 
 
411
291
    def test_debug(self):
412
292
        cmd_line = ['local', '/foo/juju', '/tmp/logs', 'testtest', '--debug']
413
293
        parser = add_basic_testing_arguments(ArgumentParser())