← Back to branch summary

~blake-rouse/maas/osystem-preseed-cleanup

« back to all changes in this revision

Viewing changes to src/provisioningserver/import_images/tests/test_uec2roottar.py

  • Committer: Blake Rouse
  • Date: 2014-05-02 15:32:18 UTC
  • mfrom: (2290.1.7 3_add-osystem-to-node)
  • Revision ID: blake.rouse@canonical.com-20140502153218-940upl3bp5lsh50f
Merge add-osystem-to-node.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/provisioningserver/import_images/tests/test_uec2roottar.py
 
1
# Copyright 2014 Canonical Ltd.  This software is licensed under the
 
2
# GNU Affero General Public License version 3 (see the file LICENSE).
 
3
 
 
4
"""Tests for the `uec2roottar` script and its supporting module.."""
 
5
 
 
6
from __future__ import (
 
7
    absolute_import,
 
8
    print_function,
 
9
    unicode_literals,
 
10
    )
 
11
 
 
12
str = None
 
13
 
 
14
__metaclass__ = type
 
15
__all__ = []
 
16
 
 
17
import os
 
18
import os.path
 
19
from subprocess import CalledProcessError
 
20
 
 
21
from maastesting.factory import factory
 
22
from maastesting.matchers import (
 
23
    MockAnyCall,
 
24
    MockCalledOnceWith,
 
25
    MockNotCalled,
 
26
    )
 
27
from maastesting.testcase import MAASTestCase
 
28
import mock
 
29
from provisioningserver.import_images import uec2roottar
 
30
from testtools.matchers import HasLength
 
31
from testtools.testcase import ExpectedException
 
32
 
 
33
 
 
34
def make_image_name(suffix='.img'):
 
35
    """Create an image file name (but not the actual file)."""
 
36
    return factory.make_name('root') + suffix
 
37
 
 
38
 
 
39
def make_image(testcase, contents=None, suffix='.img'):
 
40
    """Create an image file."""
 
41
    name = make_image_name(suffix)
 
42
    return testcase.make_file(name=name, contents=contents)
 
43
 
 
44
 
 
45
def make_tarball_name(prefix='tarball'):
 
46
    """Create an arbitrary name for a tarball."""
 
47
    return factory.make_name(prefix) + '.tar.gz'
 
48
 
 
49
 
 
50
def make_roottar_location(testcase):
 
51
    """Create a name for an output root tarball, in an empty directory."""
 
52
    name = make_tarball_name('root')
 
53
    return os.path.join(testcase.make_dir(), name)
 
54
 
 
55
 
 
56
def patch_is_filesystem_file(testcase, answer):
 
57
    """Patch `is_filesystem_file` to return the given answer."""
 
58
    testcase.patch(uec2roottar, 'is_filesystem_file').return_value = answer
 
59
 
 
60
 
 
61
class TestMakeArgParser(MAASTestCase):
 
62
    """Tests for `make_argparser`."""
 
63
 
 
64
    def test__defines_expected_options(self):
 
65
        image = make_image(self)
 
66
        output = make_roottar_location(self)
 
67
        user = factory.make_name('user')
 
68
 
 
69
        parser = uec2roottar.make_argparser(factory.getRandomString())
 
70
        args = parser.parse_args([image, output, '--user', user])
 
71
 
 
72
        self.assertEqual(
 
73
            (
 
74
                image,
 
75
                output,
 
76
                user,
 
77
            ),
 
78
            (
 
79
                args.image,
 
80
                args.output,
 
81
                args.user,
 
82
            ))
 
83
 
 
84
    def test__user_defaults_to_None(self):
 
85
        parser = uec2roottar.make_argparser(factory.getRandomString())
 
86
        args = parser.parse_args(
 
87
            [make_image(self), make_roottar_location(self)])
 
88
        self.assertIsNone(args.user)
 
89
 
 
90
 
 
91
class TestIsFilesystemFile(MAASTestCase):
 
92
    """Tests for `is_filesystem_file`."""
 
93
 
 
94
    def test__returns_True_if_file_looks_like_filesystem(self):
 
95
        image = make_image(self, suffix='.img')
 
96
        self.patch(uec2roottar, 'check_output').return_value = (
 
97
            ("%s: filesystem data" % image).encode('utf-8'))
 
98
        self.assertTrue(uec2roottar.is_filesystem_file(image))
 
99
 
 
100
    def test__returns_False_for_tarball(self):
 
101
        image = make_image(self, suffix='.tar.gz')
 
102
        self.patch(uec2roottar, 'check_output').return_value = (
 
103
            ("%s: gzip compressed data, was ..." % image).encode('utf-8'))
 
104
        self.assertFalse(uec2roottar.is_filesystem_file(image))
 
105
 
 
106
    def test__calls_file_with_C_language_setting(self):
 
107
        env_during_invocation = {}
 
108
 
 
109
        def fake_check_output(*args, **kwargs):
 
110
            env_during_invocation.update(os.environ)
 
111
            return b''
 
112
 
 
113
        self.patch(uec2roottar, 'check_output', fake_check_output)
 
114
 
 
115
        uec2roottar.is_filesystem_file(make_image(self))
 
116
 
 
117
        self.assertEqual('C', env_during_invocation.get('LANG'))
 
118
 
 
119
 
 
120
class TestExtractImageFromTarball(MAASTestCase):
 
121
    """Tests for `extract_image_from_tarball`."""
 
122
 
 
123
    def test__extracts_image(self):
 
124
        tarball = make_tarball_name()
 
125
        self.patch(uec2roottar, 'check_call')
 
126
        # Cheat: patch away extraction of the tarball, but pass a temporary
 
127
        # directory with an image already in it.  The function will think it
 
128
        # just extracted the image from the tarball.
 
129
        image = make_image(self)
 
130
        working_dir = os.path.dirname(image)
 
131
 
 
132
        result = uec2roottar.extract_image_from_tarball(tarball, working_dir)
 
133
 
 
134
        self.assertThat(
 
135
            uec2roottar.check_call,
 
136
            MockCalledOnceWith([
 
137
                'tar',
 
138
                '-C', working_dir,
 
139
                '--wildcards', '*.img',
 
140
                '-Sxvzf',
 
141
                tarball,
 
142
                ]))
 
143
        self.assertEqual(image, result)
 
144
 
 
145
    def test__ignores_other_files(self):
 
146
        tarball = make_tarball_name()
 
147
        self.patch(uec2roottar, 'check_call')
 
148
        # Make the function think that it found two files in the tarball: an
 
149
        # image and some other file.
 
150
        image = make_image(self)
 
151
        working_dir = os.path.dirname(image)
 
152
        # This other file doesn't upset things, because it doesn't look like
 
153
        # an image file.
 
154
        factory.make_file(working_dir)
 
155
 
 
156
        self.assertEqual(
 
157
            image,
 
158
            uec2roottar.extract_image_from_tarball(tarball, working_dir))
 
159
 
 
160
    def test__fails_if_no_image_found(self):
 
161
        tarball = make_tarball_name()
 
162
        self.patch(uec2roottar, 'check_call')
 
163
        empty_dir = self.make_dir()
 
164
        error = self.assertRaises(
 
165
            uec2roottar.ImageFileError,
 
166
            uec2roottar.extract_image_from_tarball, tarball, empty_dir)
 
167
        self.assertEqual(
 
168
            "Tarball %s does not contain any *.img." % tarball,
 
169
            unicode(error))
 
170
 
 
171
    def test__fails_if_multiple_images_found(self):
 
172
        tarball = make_tarball_name()
 
173
        self.patch(uec2roottar, 'check_call')
 
174
        working_dir = self.make_dir()
 
175
        files = sorted(
 
176
            factory.make_file(working_dir, name=make_image_name())
 
177
            for _ in range(2))
 
178
        error = self.assertRaises(
 
179
            uec2roottar.ImageFileError,
 
180
            uec2roottar.extract_image_from_tarball, tarball, working_dir)
 
181
        self.assertEqual(
 
182
            "Tarball %s contains multiple image files: %s."
 
183
            % (tarball, ', '.join(files)),
 
184
            unicode(error))
 
185
 
 
186
 
 
187
class TestGetImageFile(MAASTestCase):
 
188
    """Tests for `get_image_file`."""
 
189
 
 
190
    def test__returns_actual_image_file_unchanged(self):
 
191
        patch_is_filesystem_file(self, True)
 
192
        image = make_image(self)
 
193
        self.assertEqual(
 
194
            image,
 
195
            uec2roottar.get_image_file(image, factory.make_name('dir')))
 
196
 
 
197
    def test__extracts_tarball_into_temp_dir(self):
 
198
        patch_is_filesystem_file(self, False)
 
199
        tarball = make_tarball_name()
 
200
        temp_dir = self.make_dir()
 
201
        image = make_image_name()
 
202
        patch = self.patch(uec2roottar, 'extract_image_from_tarball')
 
203
        patch.return_value = image
 
204
        result = uec2roottar.get_image_file(tarball, temp_dir)
 
205
        self.assertEqual(image, result)
 
206
        self.assertThat(patch, MockCalledOnceWith(tarball, temp_dir))
 
207
 
 
208
    def test__rejects_other_files(self):
 
209
        patch_is_filesystem_file(self, False)
 
210
        filename = factory.make_name('weird-file')
 
211
        error = self.assertRaises(
 
212
            uec2roottar.ImageFileError,
 
213
            uec2roottar.get_image_file, filename, factory.make_name('dir'))
 
214
        self.assertEqual(
 
215
            "Expected '%s' to be either a filesystem file, or a "
 
216
            "gzipped tarball containing one." % filename,
 
217
            unicode(error))
 
218
 
 
219
 
 
220
class TestUnmount(MAASTestCase):
 
221
    """Tests for `unmount`."""
 
222
 
 
223
    def test__calls_umount(self):
 
224
        self.patch(uec2roottar, 'check_call')
 
225
        mountpoint = factory.make_name('mount')
 
226
        uec2roottar.unmount(mountpoint)
 
227
        self.assertThat(
 
228
            uec2roottar.check_call,
 
229
            MockCalledOnceWith(['umount', mountpoint]))
 
230
 
 
231
    def test__propagates_failure(self):
 
232
        failure = CalledProcessError(9, factory.make_name('delibfail'))
 
233
        self.patch(uec2roottar, 'check_call').side_effect = failure
 
234
        self.patch(uec2roottar, 'logger')
 
235
        mountpoint = factory.make_name('mount')
 
236
        self.assertRaises(CalledProcessError, uec2roottar.unmount, mountpoint)
 
237
        self.assertThat(
 
238
            uec2roottar.logger.error,
 
239
            MockCalledOnceWith(
 
240
                "Could not unmount %s: %s", mountpoint, failure))
 
241
 
 
242
 
 
243
class TestLoopMount(MAASTestCase):
 
244
    """Tests for `loop_mount`."""
 
245
 
 
246
    def test__mounts_and_unmounts_image(self):
 
247
        image = make_image_name()
 
248
        self.patch(uec2roottar, 'check_call')
 
249
        mountpoint = factory.make_name('mount')
 
250
 
 
251
        calls_before = len(uec2roottar.check_call.mock_calls)
 
252
        with uec2roottar.loop_mount(image, mountpoint):
 
253
            calls_during = len(uec2roottar.check_call.mock_calls)
 
254
        calls_after = len(uec2roottar.check_call.mock_calls)
 
255
 
 
256
        self.assertEqual(
 
257
            (0, 1, 2),
 
258
            (calls_before, calls_during, calls_after))
 
259
        self.assertThat(
 
260
            uec2roottar.check_call,
 
261
            MockAnyCall(['mount', '-o', 'ro', image, mountpoint]))
 
262
        self.assertThat(
 
263
            uec2roottar.check_call,
 
264
            MockAnyCall(['umount', mountpoint]))
 
265
 
 
266
    def test__cleans_up_after_failure(self):
 
267
        class DeliberateException(Exception):
 
268
            pass
 
269
 
 
270
        self.patch(uec2roottar, 'check_call')
 
271
        image = make_image_name()
 
272
        mountpoint = factory.make_name('mount')
 
273
        with ExpectedException(DeliberateException):
 
274
            with uec2roottar.loop_mount(image, mountpoint):
 
275
                raise DeliberateException()
 
276
 
 
277
        self.assertThat(
 
278
            uec2roottar.check_call, MockAnyCall(['umount', mountpoint]))
 
279
 
 
280
 
 
281
class TestExtractImage(MAASTestCase):
 
282
    """Tests for `extract_image`."""
 
283
 
 
284
    def extract_command_line(self, call):
 
285
        """Extract the command line from a `mock.call` for `check_call`."""
 
286
        _, args, _ = call
 
287
        [command] = args
 
288
        return command
 
289
 
 
290
    def test__extracts_image(self):
 
291
        image = make_image_name()
 
292
        output = make_tarball_name()
 
293
        self.patch(uec2roottar, 'check_call')
 
294
        uec2roottar.extract_image(image, output)
 
295
        self.assertThat(uec2roottar.check_call.mock_calls, HasLength(3))
 
296
        [mount_call, tar_call, umount_call] = uec2roottar.check_call.mock_calls
 
297
        self.assertEqual('mount', self.extract_command_line(mount_call)[0])
 
298
        tar_command = self.extract_command_line(tar_call)
 
299
        self.assertEqual(['tar', '-C'], tar_command[:2])
 
300
        self.assertEqual('umount', self.extract_command_line(umount_call)[0])
 
301
 
 
302
 
 
303
class TestSetOwnership(MAASTestCase):
 
304
    """Tests for `set_ownership`."""
 
305
 
 
306
    def test__does_nothing_if_no_user_specified(self):
 
307
        self.patch(uec2roottar, 'check_call')
 
308
        uec2roottar.set_ownership(make_tarball_name(), user=None)
 
309
        self.assertThat(uec2roottar.check_call, MockNotCalled())
 
310
 
 
311
    def test__calls_chown_if_user_specified(self):
 
312
        self.patch(uec2roottar, 'check_call')
 
313
        user = factory.make_name('user')
 
314
        tarball = make_tarball_name()
 
315
        uec2roottar.set_ownership(tarball, user=user)
 
316
        self.assertThat(
 
317
            uec2roottar.check_call,
 
318
            MockCalledOnceWith(['/bin/chown', user, tarball]))
 
319
 
 
320
 
 
321
class TestUEC2RootTar(MAASTestCase):
 
322
    """Integration tests for `uec2roottar`."""
 
323
 
 
324
    def make_args(self, **kwargs):
 
325
        """Fake an `argparser` arguments object."""
 
326
        args = mock.Mock()
 
327
        for key, value in kwargs.items():
 
328
            setattr(args, key, value)
 
329
        return args
 
330
 
 
331
    def test__integrates(self):
 
332
        image_name = factory.make_name('root-image') + '.img'
 
333
        image = self.make_file(name=image_name)
 
334
        output_name = factory.make_name('root-tar') + '.tar.gz'
 
335
        output = os.path.join(self.make_dir(), output_name)
 
336
        args = self.make_args(image=image, output=output)
 
337
        self.patch(uec2roottar, 'check_call')
 
338
        patch_is_filesystem_file(self, True)
 
339
 
 
340
        uec2roottar.main(args)
 
341
 
 
342
        self.assertThat(
 
343
            uec2roottar.is_filesystem_file, MockCalledOnceWith(image))