Merge lp:~jamesodhunt/ubuntu/vivid/ubuntu-core-upgrader/add-functional-tests into lp:ubuntu/vivid/ubuntu-core-upgrader

Proposed by James Hunt
Status: Merged
Merged at revision: 18
Proposed branch: lp:~jamesodhunt/ubuntu/vivid/ubuntu-core-upgrader/add-functional-tests
Merge into: lp:ubuntu/vivid/ubuntu-core-upgrader
Diff against target: 1174 lines (+936/-170)
5 files modified
debian/changelog (+8/-0)
functional/test_upgrader.py (+624/-0)
ubuntucoreupgrader/tests/test_upgrader.py (+5/-169)
ubuntucoreupgrader/tests/utils.py (+292/-0)
ubuntucoreupgrader/upgrader.py (+7/-1)
To merge this branch: bzr merge lp:~jamesodhunt/ubuntu/vivid/ubuntu-core-upgrader/add-functional-tests
Reviewer Review Type Date Requested Status
Michael Vogt Pending
Review via email: mp+251723@code.launchpad.net

Description of the change

* ubuntucoreupgrader/upgrader.py:
  - get_file_contents(): Fix to avoid leaking the fd.
* functional/test_upgrader.py: Basic set of functional tests.

To post a comment you must log in.
18. By James Hunt on 2015-03-04

* Fix-up changelog.

19. By James Hunt on 2015-03-04

* Moved common test code into ubuntucoreupgrader/tests/utils.py.

Preview Diff

[H/L] Next/Prev Comment, [J/K] Next/Prev File, [N/P] Next/Prev Hunk
=== modified file 'debian/changelog'
--- debian/changelog 2015-03-04 11:34:20 +0000
+++ debian/changelog 2015-03-04 13:53:47 +0000
@@ -1,3 +1,11 @@
1ubuntu-core-upgrader (0.7.6) UNRELEASED; urgency=low
2
3 * ubuntucoreupgrader/upgrader.py:
4 - get_file_contents(): Fix to avoid leaking the fd.
5 * functional/test_upgrader.py: Basic set of functional tests.
6
7 -- James Hunt <james.hunt@ubuntu.com> Wed, 04 Mar 2015 11:52:19 +0000
8
1ubuntu-core-upgrader (0.7.5) vivid; urgency=low9ubuntu-core-upgrader (0.7.5) vivid; urgency=low
210
3 [ Michael Vogt ]11 [ Michael Vogt ]
412
=== added directory 'functional'
=== added file 'functional/__init__.py'
=== added file 'functional/test_upgrader.py'
--- functional/test_upgrader.py 1970-01-01 00:00:00 +0000
+++ functional/test_upgrader.py 2015-03-04 13:53:47 +0000
@@ -0,0 +1,624 @@
1#!/usr/bin/python3
2# -*- coding: utf-8 -*-
3# --------------------------------------------------------------------
4# Copyright © 2014-2015 Canonical Ltd.
5#
6# This program is free software: you can redistribute it and/or modify
7# it under the terms of the GNU General Public License as published by
8# the Free Software Foundation, version 3 of the License, or
9# (at your option) any later version.
10#
11# This program is distributed in the hope that it will be useful,
12# but WITHOUT ANY WARRANTY; without even the implied warranty of
13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14# GNU General Public License for more details.
15#
16# You should have received a copy of the GNU General Public License
17# along with this program. If not, see <http://www.gnu.org/licenses/>.
18# --------------------------------------------------------------------
19
20# --------------------------------------------------------------------
21# Functional tests for the snappy upgrader.
22# --------------------------------------------------------------------
23
24import sys
25import os
26import logging
27import tarfile
28import unittest
29import shutil
30
31from unittest.mock import patch
32
33from ubuntucoreupgrader.upgrader import (
34 Upgrader,
35 parse_args,
36)
37
38base_dir = os.path.abspath(os.path.dirname(__file__))
39module_dir = os.path.normpath(os.path.realpath(base_dir + os.sep + '..'))
40sys.path.append(base_dir)
41
42from ubuntucoreupgrader.tests.utils import *
43
44CMD_FILE = 'ubuntu_command'
45
46
47def call_upgrader(command_file, root_dir, update):
48 '''
49 Invoke the upgrader.
50
51 :param command_file: commands file to drive the upgrader.
52 :param root_dir: Test directory to apply the upgrade to.
53 :param update: UpdateTree object.
54 '''
55
56 args = []
57 args += ['--root-dir', root_dir]
58 args += ['--debug', '1']
59
60 # don't delete the archive and command files.
61 # The tests clean up after themselves so they will get removed then,
62 # but useful to have them around to diagnose test failures.
63 args.append('--leave-files')
64
65 args.append(command_file)
66 commands = file_to_list(command_file)
67
68 cache_dir = make_tmp_dir()
69
70 def mock_get_cache_dir():
71 cache_dir = update.tmp_dir
72 sys_dir = os.path.join(cache_dir, 'system')
73 os.makedirs(sys_dir, exist_ok=True)
74 return cache_dir
75
76 upgrader = Upgrader(parse_args(args), commands, [])
77 upgrader.get_cache_dir = mock_get_cache_dir
78 upgrader.run()
79
80 shutil.rmtree(cache_dir)
81
82
83def create_device_file(path, type='c', major=-1, minor=-1):
84 '''
85 Create a device file.
86
87 :param path: full path to device file.
88 :param type: 'c' or 'b' (character or block).
89 :param major: major number.
90 :param minor: minor number.
91
92 XXX: This doesn't actually create a device node,
93 it simply creates a regular empty file whilst ensuring the filename
94 gives the impression that it is a device file.
95
96 This hackery is done for the following reasons:
97
98 - non-priv users cannot create device nodes (and the tests run as a
99 non-priv user).
100 - the removed file in the upgrade tar file does not actually specify the
101 _type_ of the files to remove. Hence, we can pretend that the file
102 to remove is a device file since the upgrader cannot know for sure
103 (it can check the existing on-disk file, but that isn't conclusive
104 since the admin may have manually modified a file, or the server
105 may have generated an invalid remove file - the upgrader cannot
106 know for sure.
107 '''
108 assert (os.path.dirname(path).endswith('/dev'))
109
110 append_file(path, 'fake-device file')
111
112
113def create_directory(path):
114 '''
115 Create a directory.
116 '''
117 os.makedirs(path, mode=TEST_DIR_MODE, exist_ok=False)
118
119
120def create_sym_link(source, dest):
121 '''
122 Create a symbolic link.
123
124 :param source: existing file to link to.
125 :param dest: name for the sym link.
126 '''
127 dirname = os.path.dirname(dest)
128 os.makedirs(dirname, mode=TEST_DIR_MODE, exist_ok=True)
129
130 os.symlink(source, dest)
131
132
133def create_hard_link(source, dest):
134 '''
135 Create a hard link.
136
137 :param source: existing file to link to.
138 :param dest: name for the hard link.
139 '''
140 os.link(source, dest)
141
142
143def is_sym_link_broken(path):
144 '''
145 :param path: symbolic link to check.
146 :return: True if the specified path is a broken symbolic link,
147 else False.
148 '''
149 try:
150 os.lstat(path)
151 os.stat(path)
152 except:
153 return True
154 return False
155
156
157def make_command_file(path, update_list):
158 '''
159 Create a command file that the upgrader processes.
160
161 :param path: full path to file to create,
162 :param update_list: list of update archives to include.
163 '''
164 l = []
165
166 for file in update_list:
167 l.append('update {} {}.asc'.format(file, file))
168
169 # flatten
170 contents = "\n".join(l) + '\n'
171
172 append_file(path, contents)
173
174
175def file_to_list(path):
176 '''
177 Convert the specified file into a list and return it.
178 '''
179 lines = []
180
181 with open(path, 'r') as f:
182 lines = f.readlines()
183
184 lines = [line.rstrip() for line in lines]
185
186 return lines
187
188
189class UpgraderFileRemovalTestCase(UbuntuCoreUpgraderTestCase):
190 '''
191 Test how the upgrader handles the removals file.
192 '''
193
194 def test_remove_file(self):
195 '''
196 Ensure the upgrader can remove a regular file.
197 '''
198
199 file = 'a-regular-file'
200
201 self.update.add_to_removed_file([file])
202
203 archive = self.update.create_archive(self.TARFILE)
204 self.assertTrue(os.path.exists(archive))
205
206 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
207 make_command_file(cmd_file, [self.TARFILE])
208
209 file_path = os.path.join(self.victim_dir, file)
210 create_file(file_path, 'foo bar')
211
212 self.assertTrue(os.path.exists(file_path))
213 self.assertTrue(os.path.isfile(file_path))
214
215 call_upgrader(cmd_file, self.victim_dir, self.update)
216
217 self.assertFalse(os.path.exists(file_path))
218
219 def test_remove_directory(self):
220 '''
221 Ensure the upgrader can remove a directory.
222 '''
223 dir = 'a-directory'
224
225 self.update.add_to_removed_file([dir])
226
227 archive = self.update.create_archive(self.TARFILE)
228 self.assertTrue(os.path.exists(archive))
229
230 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
231 make_command_file(cmd_file, [self.TARFILE])
232
233 dir_path = os.path.join(self.victim_dir, dir)
234 create_directory(dir_path)
235
236 self.assertTrue(os.path.exists(dir_path))
237 self.assertTrue(os.path.isdir(dir_path))
238
239 call_upgrader(cmd_file, self.victim_dir, self.update)
240
241 self.assertFalse(os.path.exists(dir_path))
242
243 def test_remove_sym_link_file(self):
244 '''
245 Ensure the upgrader can remove a symbolic link to a file.
246 '''
247 src = 'the-source-file'
248 link = 'the-symlink-file'
249
250 self.update.add_to_removed_file([link])
251
252 archive = self.update.create_archive(self.TARFILE)
253 self.assertTrue(os.path.exists(archive))
254
255 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
256 make_command_file(cmd_file, [self.TARFILE])
257
258 src_file_path = os.path.join(self.victim_dir, src)
259 link_file_path = os.path.join(self.victim_dir, link)
260
261 create_file(src_file_path, 'foo bar')
262
263 self.assertTrue(os.path.exists(src_file_path))
264 self.assertTrue(os.path.isfile(src_file_path))
265 self.assertFalse(os.path.islink(src_file_path))
266
267 create_sym_link(src_file_path, link_file_path)
268 self.assertTrue(os.path.exists(link_file_path))
269 self.assertTrue(os.path.islink(link_file_path))
270
271 call_upgrader(cmd_file, self.victim_dir, self.update)
272
273 # original file should still be there
274 self.assertTrue(os.path.exists(src_file_path))
275 self.assertTrue(os.path.isfile(src_file_path))
276 self.assertFalse(os.path.islink(src_file_path))
277
278 # link should have gone
279 self.assertFalse(os.path.exists(link_file_path))
280
281 def test_remove_sym_link_directory(self):
282 '''
283 Ensure the upgrader can remove a symbolic link to a directory.
284 '''
285 dir = 'the-source-directory'
286 link = 'the-symlink-file'
287
288 self.update.add_to_removed_file([link])
289
290 archive = self.update.create_archive(self.TARFILE)
291 self.assertTrue(os.path.exists(archive))
292
293 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
294 make_command_file(cmd_file, [self.TARFILE])
295
296 src_dir_path = os.path.join(self.victim_dir, dir)
297 link_file_path = os.path.join(self.victim_dir, link)
298
299 create_directory(src_dir_path)
300
301 self.assertTrue(os.path.exists(src_dir_path))
302 self.assertTrue(os.path.isdir(src_dir_path))
303 self.assertFalse(os.path.islink(src_dir_path))
304
305 create_sym_link(src_dir_path, link_file_path)
306 self.assertTrue(os.path.exists(link_file_path))
307 self.assertTrue(os.path.islink(link_file_path))
308
309 call_upgrader(cmd_file, self.victim_dir, self.update)
310
311 # original directory should still be there
312 self.assertTrue(os.path.exists(src_dir_path))
313 self.assertTrue(os.path.isdir(src_dir_path))
314 self.assertFalse(os.path.islink(src_dir_path))
315
316 # link should have gone
317 self.assertFalse(os.path.exists(link_file_path))
318
319 def test_remove_hardlink(self):
320 '''
321 Ensure the upgrader can remove a hard link to a file.
322 '''
323 src = 'the-source-file'
324 link = 'the-hardlink-file'
325
326 self.update.add_to_removed_file([link])
327
328 archive = self.update.create_archive(self.TARFILE)
329 self.assertTrue(os.path.exists(archive))
330
331 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
332 make_command_file(cmd_file, [self.TARFILE])
333
334 src_file_path = os.path.join(self.victim_dir, src)
335 link_file_path = os.path.join(self.victim_dir, link)
336
337 create_file(src_file_path, 'foo bar')
338
339 src_inode = os.stat(src_file_path).st_ino
340
341 self.assertTrue(os.path.exists(src_file_path))
342 self.assertTrue(os.path.isfile(src_file_path))
343
344 create_hard_link(src_file_path, link_file_path)
345 self.assertTrue(os.path.exists(link_file_path))
346
347 link_inode = os.stat(link_file_path).st_ino
348
349 self.assertTrue(src_inode == link_inode)
350
351 call_upgrader(cmd_file, self.victim_dir, self.update)
352
353 # original file should still be there
354 self.assertTrue(os.path.exists(src_file_path))
355 self.assertTrue(os.path.isfile(src_file_path))
356
357 # Inode should not have changed.
358 self.assertTrue(os.stat(src_file_path).st_ino == src_inode)
359
360 # link should have gone
361 self.assertFalse(os.path.exists(link_file_path))
362
363 def test_remove_device_file(self):
364 '''
365 Ensure the upgrader can deal with a device file.
366
367 XXX: Note that The upgrader currently "deals" with them by
368 XXX: ignoring them :-)
369
370 '''
371 file = '/dev/a-fake-device'
372
373 self.update.add_to_removed_file([file])
374
375 archive = self.update.create_archive(self.TARFILE)
376 self.assertTrue(os.path.exists(archive))
377
378 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
379 make_command_file(cmd_file, [self.TARFILE])
380
381 file_path = '{}{}'.format(self.victim_dir, file)
382
383 create_device_file(file_path)
384
385 self.assertTrue(os.path.exists(file_path))
386
387 # sigh - we can't assert the that filetype is a char/block
388 # device because it won't be :)
389 self.assertTrue(os.path.isfile(file_path))
390
391 call_upgrader(cmd_file, self.victim_dir, self.update)
392
393 self.assertFalse(os.path.exists(file_path))
394
395
396class UpgraderFileAddTestCase(UbuntuCoreUpgraderTestCase):
397 '''
398 Test how the upgrader handles adding new files.
399 '''
400
401 def test_create_file(self):
402 '''
403 Ensure the upgrader can create a regular file.
404 '''
405 file = 'created-regular-file'
406
407 file_path = os.path.join(self.update.system_dir, file)
408
409 create_file(file_path, 'foo bar')
410
411 archive = self.update.create_archive(self.TARFILE)
412 self.assertTrue(os.path.exists(archive))
413
414 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
415 make_command_file(cmd_file, [self.TARFILE])
416
417 file_path = os.path.join(self.victim_dir, file)
418 self.assertFalse(os.path.exists(file_path))
419
420 call_upgrader(cmd_file, self.victim_dir, self.update)
421
422 self.assertTrue(os.path.exists(file_path))
423 self.assertTrue(os.path.isfile(file_path))
424
425 def test_create_directory(self):
426 '''
427 Ensure the upgrader can create a directory.
428 '''
429 dir = 'created-directory'
430
431 dir_path = os.path.join(self.update.system_dir, dir)
432
433 create_directory(dir_path)
434
435 archive = self.update.create_archive(self.TARFILE)
436 self.assertTrue(os.path.exists(archive))
437
438 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
439 make_command_file(cmd_file, [self.TARFILE])
440
441 dir_path = os.path.join(self.victim_dir, dir)
442 self.assertFalse(os.path.exists(dir_path))
443
444 call_upgrader(cmd_file, self.victim_dir, self.update)
445
446 self.assertTrue(os.path.exists(dir_path))
447 self.assertTrue(os.path.isdir(dir_path))
448
449 def test_create_absolute_sym_link_to_file(self):
450 '''
451 Ensure the upgrader can create a symbolic link to a file (which
452 already exists and is not included in the update archive).
453 '''
454 src = 'the-source-file'
455 link = 'the-symlink-file'
456
457 # the file the link points to should *NOT* be below the 'system/'
458 # directory (since there isn't one post-unpack).
459
460 # an absolute sym-link target path
461 src_file_path = '/{}'.format(src)
462 link_file_path = os.path.join(self.update.system_dir, link)
463
464 victim_src_file_path = os.path.normpath('{}/{}'
465 .format(self.victim_dir,
466 src_file_path))
467 victim_link_file_path = os.path.join(self.victim_dir, link)
468
469 # Create a broken sym link ('/system/<link> -> /<src>')
470 create_sym_link(src_file_path, link_file_path)
471
472 self.assertTrue(os.path.lexists(link_file_path))
473 self.assertTrue(os.path.islink(link_file_path))
474 self.assertTrue(is_sym_link_broken(link_file_path))
475
476 archive = self.update.create_archive(self.TARFILE)
477 self.assertTrue(os.path.exists(archive))
478
479 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
480 make_command_file(cmd_file, [self.TARFILE])
481
482 create_file(victim_src_file_path, 'foo')
483
484 self.assertTrue(os.path.exists(victim_src_file_path))
485 self.assertTrue(os.path.isfile(victim_src_file_path))
486
487 self.assertFalse(os.path.exists(victim_link_file_path))
488
489 call_upgrader(cmd_file, self.victim_dir, self.update)
490
491 self.assertTrue(os.path.exists(victim_src_file_path))
492 self.assertTrue(os.path.isfile(victim_src_file_path))
493
494 # upgrader should have created the link in the victim directory
495 self.assertTrue(os.path.lexists(victim_link_file_path))
496 self.assertTrue(os.path.islink(victim_link_file_path))
497 self.assertFalse(is_sym_link_broken(victim_link_file_path))
498
499 def test_create_relative_sym_link_to_file(self):
500 '''
501 Ensure the upgrader can create a symbolic link to a file (which
502 already exists and is not included in the update archive).
503 '''
504 src = 'a/b/c/the-source-file'
505 link = 'a/d/e/the-symlink-file'
506
507 # a relative sym-link target path
508 # ##src_file_path = '../../b/c/{}'.format(src)
509 src_file_path = '../../b/c/the-source-file'.format(src)
510
511 # the file the link points to should *NOT* be below the 'system/'
512 # directory (since there isn't one post-unpack).
513
514 link_file_path = os.path.join(self.update.system_dir, link)
515
516 victim_src_file_path = os.path.normpath('{}/{}'
517 .format(self.victim_dir,
518 src))
519 victim_link_file_path = os.path.join(self.victim_dir, link)
520
521 create_sym_link(src_file_path, link_file_path)
522
523 self.assertTrue(os.path.lexists(link_file_path))
524 self.assertTrue(os.path.islink(link_file_path))
525 self.assertTrue(is_sym_link_broken(link_file_path))
526
527 archive = self.update.create_archive(self.TARFILE)
528 self.assertTrue(os.path.exists(archive))
529
530 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
531 make_command_file(cmd_file, [self.TARFILE])
532
533 create_file(victim_src_file_path, 'foo')
534
535 self.assertTrue(os.path.exists(victim_src_file_path))
536 self.assertTrue(os.path.isfile(victim_src_file_path))
537
538 self.assertFalse(os.path.exists(victim_link_file_path))
539
540 call_upgrader(cmd_file, self.victim_dir, self.update)
541
542 self.assertTrue(os.path.exists(victim_src_file_path))
543 self.assertTrue(os.path.isfile(victim_src_file_path))
544
545 # upgrader should have created the link in the victim directory
546 self.assertTrue(os.path.lexists(victim_link_file_path))
547 self.assertTrue(os.path.islink(victim_link_file_path))
548 self.assertFalse(is_sym_link_broken(victim_link_file_path))
549
550 def test_create_broken_sym_link_file(self):
551 '''
552 Ensure the upgrader can create a broken symbolic link
553 (one that points to a non-existent file).
554 '''
555 src = 'the-source-file'
556 link = 'the-symlink-file'
557
558 # the file the link points to should *NOT* be below the 'system/'
559 # directory (since there isn't one post-unpack).
560 src_file_path = src
561
562 link_file_path = os.path.join(self.update.system_dir, link)
563
564 # Create a broken sym link ('/system/<link> -> /<src>')
565 create_sym_link(src_file_path, link_file_path)
566
567 self.assertTrue(os.path.lexists(link_file_path))
568 self.assertTrue(os.path.islink(link_file_path))
569 self.assertTrue(is_sym_link_broken(link_file_path))
570
571 archive = self.update.create_archive(self.TARFILE)
572 self.assertTrue(os.path.exists(archive))
573
574 cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
575 make_command_file(cmd_file, [self.TARFILE])
576
577 victim_src_file_path = os.path.join(self.victim_dir, src)
578 victim_link_file_path = os.path.join(self.victim_dir, link)
579
580 self.assertFalse(os.path.exists(victim_src_file_path))
581 self.assertFalse(os.path.exists(victim_link_file_path))
582
583 call_upgrader(cmd_file, self.victim_dir, self.update)
584
585 # source still shouldn't exist
586 self.assertFalse(os.path.exists(victim_src_file_path))
587
588 # upgrader should have created the link in the victim directory
589 self.assertTrue(os.path.lexists(victim_link_file_path))
590 self.assertTrue(os.path.islink(victim_link_file_path))
591 self.assertTrue(is_sym_link_broken(victim_link_file_path))
592
593
594def main():
595 kwargs = {}
596 format = \
597 '%(asctime)s:' \
598 '%(filename)s:' \
599 '%(name)s:' \
600 '%(funcName)s:' \
601 '%(levelname)s:' \
602 '%(message)s'
603
604 kwargs['format'] = format
605
606 # We want to see what's happening
607 kwargs['level'] = logging.DEBUG
608
609 logging.basicConfig(**kwargs)
610
611 unittest.main(
612 testRunner=unittest.TextTestRunner(
613 stream=sys.stdout,
614 verbosity=2,
615
616 # don't keep running tests if one fails
617 # (... who _wouldn't_ want this???)
618 failfast=True
619 ),
620
621 )
622
623if __name__ == '__main__':
624 main()
0625
=== modified file 'ubuntucoreupgrader/tests/test_upgrader.py'
--- ubuntucoreupgrader/tests/test_upgrader.py 2015-03-04 11:27:28 +0000
+++ ubuntucoreupgrader/tests/test_upgrader.py 2015-03-04 13:53:47 +0000
@@ -29,6 +29,7 @@
29import unittest29import unittest
30import os30import os
31import shutil31import shutil
32import sys
3233
33from unittest.mock import patch34from unittest.mock import patch
3435
@@ -38,7 +39,10 @@
38 parse_args,39 parse_args,
39)40)
4041
41script_name = os.path.basename(__file__)42base_dir = os.path.abspath(os.path.dirname(__file__))
43sys.path.append(base_dir)
44
45from ubuntucoreupgrader.tests.utils import *
4246
43# file mode to use for creating test directories.47# file mode to use for creating test directories.
44TEST_DIR_MODE = 0o75048TEST_DIR_MODE = 0o750
@@ -50,19 +54,6 @@
50 return parse_args([])54 return parse_args([])
5155
5256
53def make_tmp_dir(tag=None):
54 '''
55 Create a temporary directory.
56 '''
57
58 if tag:
59 prefix = '{}-{}-'.format(script_name, tag)
60 else:
61 prefix = script_name
62
63 return tempfile.mkdtemp(prefix=prefix)
64
65
66class UpgradeTestCase(unittest.TestCase):57class UpgradeTestCase(unittest.TestCase):
6758
68 def test_tar_generator_unpack_assets(self):59 def test_tar_generator_unpack_assets(self):
@@ -172,28 +163,6 @@
172 shutil.rmtree(cache_dir)163 shutil.rmtree(cache_dir)
173164
174165
175def append_file(path, contents):
176 '''
177 Append to a regular file (create it doesn't exist).
178 '''
179
180 dirname = os.path.dirname(path)
181 os.makedirs(dirname, mode=TEST_DIR_MODE, exist_ok=True)
182
183 with open(path, 'a') as fh:
184 fh.writelines(contents)
185
186 if not contents.endswith('\n'):
187 fh.write('\n')
188
189
190def create_file(path, contents):
191 '''
192 Create a regular file.
193 '''
194 append_file(path, contents)
195
196
197def touch_file(path):166def touch_file(path):
198 '''167 '''
199 Create an empty file (creating any necessary intermediate168 Create an empty file (creating any necessary intermediate
@@ -223,139 +192,6 @@
223 return l192 return l
224193
225194
226class UpdateTree():
227 '''
228 Representation of a directory tree that will be converted into an
229 update archive.
230 '''
231 TEST_REMOVED_FILE = 'removed'
232 TEST_SYSTEM_DIR = 'system/'
233
234 def __init__(self):
235
236 # Directory tree used to construct the tar file from.
237 # Also used to hold the TEST_REMOVED_FILE file.
238 self.dir = make_tmp_dir(tag='UpdateTree-tar-source')
239
240 self.removed_file = os.path.join(self.dir, self.TEST_REMOVED_FILE)
241
242 # Directory to place create/modify files into.
243 self.system_dir = os.path.join(self.dir, self.TEST_SYSTEM_DIR)
244
245 # Directory used to write the generated tarfile to.
246 # This directory should also be used to write the command file
247 # to.
248 self.tmp_dir = make_tmp_dir(tag='UpdateTree-cache')
249
250 def destroy(self):
251 if os.path.exists(self.dir):
252 shutil.rmtree(self.dir)
253
254 if os.path.exists(self.tmp_dir):
255 shutil.rmtree(self.tmp_dir)
256
257 def tar_filter(self, member):
258 '''
259 Function to filter the tarinfo members before creating the
260 archive.
261 '''
262 # members are created with relative paths (no leading slash)
263 path = os.sep + member.name
264
265 if member.name == '/.':
266 return None
267
268 i = path.find(self.dir)
269 assert(i == 0)
270
271 # remove the temporary directory elements
272 # (+1 for the os.sep we added above)
273 member.name = path[len(self.dir)+1:]
274
275 return member
276
277 def create_archive(self, name):
278 '''
279 Create an archive with the specified name from the UpdateTree
280 object. Also creates a fake signature file alongside the archive
281 file since this is currently required by the upgrader (although
282 it is not validated).
283
284 :param name: name of tarfile.
285 :param name: full path to xz archive to create.
286 :return full path to tar file with name @name.
287 '''
288
289 tar_path = os.path.join(self.tmp_dir, name)
290 tar = tarfile.open(tar_path, 'w:xz')
291
292 # We can't just add recursively since that would attempt to add
293 # the parent directory. However, the real update tars don't
294 # include that, and attempting to ignore the parent directory
295 # results in an empty archive. So, walk the tree and add
296 # file-by-file.
297 for path, names, files in os.walk(self.dir):
298 for file in files:
299 full = os.path.join(path, file)
300 tar.add(full, recursive=False, filter=self.tar_filter)
301 if not files and not names:
302 # add (empty) directories
303 tar.add(path, recursive=False, filter=self.tar_filter)
304
305 tar.close()
306
307 signature = '{}.asc'.format(tar_path)
308
309 with open(signature, 'w') as fh:
310 fh.write('fake signature file')
311
312 return tar_path
313
314
315class UbuntuCoreUpgraderTestCase(unittest.TestCase):
316 '''
317 Base class for Upgrader tests.
318 '''
319
320 TARFILE = 'update.tar.xz'
321
322 # result of last test run. Hack to deal with fact that even if a
323 # test fails, unittest still calls .tearDown() (whomever thought
324 # that was a good idea...?)
325 currentResult = None
326
327 def setUp(self):
328 '''
329 Test setup.
330 '''
331 # Create an object to hold the tree that will be converted into
332 # an upgrade archive.
333 self.update = UpdateTree()
334
335 # The directory which will have the update archive applied to
336 # it.
337 self.victim_dir = make_tmp_dir(tag='victim')
338
339 def tearDown(self):
340 '''
341 Test cleanup.
342 '''
343
344 if not self.currentResult.wasSuccessful():
345 # Do not clean up - the only sane option if a test fails.
346 return
347
348 self.update.destroy()
349 self.update = None
350
351 shutil.rmtree(self.victim_dir)
352 self.victim_dir = None
353
354 def run(self, result=None):
355 self.currentResult = result
356 unittest.TestCase.run(self, result)
357
358
359def mock_get_root_partitions_by_label():195def mock_get_root_partitions_by_label():
360 matches = []196 matches = []
361197
362198
=== added file 'ubuntucoreupgrader/tests/utils.py'
--- ubuntucoreupgrader/tests/utils.py 1970-01-01 00:00:00 +0000
+++ ubuntucoreupgrader/tests/utils.py 2015-03-04 13:53:47 +0000
@@ -0,0 +1,292 @@
1# -*- coding: utf-8 -*-
2# --------------------------------------------------------------------
3# Copyright © 2014-2015 Canonical Ltd.
4#
5# This program is free software: you can redistribute it and/or modify
6# it under the terms of the GNU General Public License as published by
7# the Free Software Foundation, version 3 of the License, or
8# (at your option) any later version.
9#
10# This program is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with this program. If not, see <http://www.gnu.org/licenses/>.
17# --------------------------------------------------------------------
18
19import os
20import tempfile
21import tarfile
22import shutil
23import unittest
24
25from unittest.mock import patch
26
27# file mode to use for creating test directories.
28TEST_DIR_MODE = 0o750
29
30script_name = os.path.basename(__file__)
31
32
33def make_tmp_dir(tag=None):
34 '''
35 Create a temporary directory.
36 '''
37
38 if tag:
39 prefix = '{}-{}-'.format(script_name, tag)
40 else:
41 prefix = script_name
42
43 return tempfile.mkdtemp(prefix=prefix)
44
45
46def append_file(path, contents):
47 '''
48 Append to a regular file (create it doesn't exist).
49 '''
50
51 dirname = os.path.dirname(path)
52 os.makedirs(dirname, mode=TEST_DIR_MODE, exist_ok=True)
53
54 with open(path, 'a') as fh:
55 fh.writelines(contents)
56
57 if not contents.endswith('\n'):
58 fh.write('\n')
59
60
61def create_file(path, contents):
62 '''
63 Create a regular file.
64 '''
65 append_file(path, contents)
66
67
68def mock_get_root_partitions_by_label():
69 '''
70 Fake disk partition details for testing.
71 '''
72 matches = []
73
74 matches.append(('system-a', '/dev/sda3', '/'))
75 matches.append(('system-b', '/dev/sda4', '/writable/cache/system'))
76
77 return matches
78
79
80def mock_make_mount_private(target):
81 '''
82 NOP implementation.
83 '''
84 pass
85
86
87class UpdateTree():
88 '''
89 Representation of a directory tree that will be converted into an
90 update archive.
91 '''
92 TEST_REMOVED_FILE = 'removed'
93 TEST_SYSTEM_DIR = 'system/'
94
95 def __init__(self):
96
97 # Directory tree used to construct the tar file from.
98 # Also used to hold the TEST_REMOVED_FILE file.
99 self.dir = make_tmp_dir(tag='UpdateTree-tar-source')
100
101 self.removed_file = os.path.join(self.dir, self.TEST_REMOVED_FILE)
102
103 # Directory to place create/modify files into.
104 self.system_dir = os.path.join(self.dir, self.TEST_SYSTEM_DIR)
105
106 # Directory used to write the generated tarfile to.
107 # This directory should also be used to write the command file
108 # to.
109 self.tmp_dir = make_tmp_dir(tag='UpdateTree-cache')
110
111 def destroy(self):
112 if os.path.exists(self.dir):
113 shutil.rmtree(self.dir)
114
115 if os.path.exists(self.tmp_dir):
116 shutil.rmtree(self.tmp_dir)
117
118 def add_to_removed_file(self, removed_files):
119 '''
120 Add the specified list of files to the removed file.
121
122 The 'removed' file is simply a file with a well-known name that
123 contains a list of files (one per line) to be removed from a
124 system before the rest of the update archive is unpacked.
125
126 :param removed_files: list of file names to add to the removed file.
127
128 '''
129 # all files listed in the removed list must be system files
130 final = list(map(lambda a:
131 '{}{}'.format(self.TEST_SYSTEM_DIR, a), removed_files))
132
133 contents = "".join(final)
134 append_file(self.removed_file, contents)
135
136 def tar_filter(self, member):
137 '''
138 Function to filter the tarinfo members before creating the
139 archive.
140 '''
141 # members are created with relative paths (no leading slash)
142 path = os.sep + member.name
143
144 if member.name == '/.':
145 return None
146
147 i = path.find(self.dir)
148 assert(i == 0)
149
150 # remove the temporary directory elements
151 # (+1 for the os.sep we added above)
152 member.name = path[len(self.dir)+1:]
153
154 return member
155
156 def create_archive(self, name):
157 '''
158 Create an archive with the specified name from the UpdateTree
159 object. Also creates a fake signature file alongside the archive
160 file since this is currently required by the upgrader (although
161 it is not validated).
162
163 :param name: name of tarfile.
164 :param name: full path to xz archive to create.
165 :return full path to tar file with name @name.
166 '''
167
168 self.tar_path = os.path.join(self.tmp_dir, name)
169 tar = tarfile.open(self.tar_path, 'w:xz')
170
171 # We can't just add recursively since that would attempt to add
172 # the parent directory. However, the real update tars don't
173 # include that, and attempting to ignore the parent directory
174 # results in an empty archive. So, walk the tree and add
175 # file-by-file.
176 for path, names, files in os.walk(self.dir):
177 for file in files:
178 full = os.path.join(path, file)
179 tar.add(full, recursive=False, filter=self.tar_filter)
180 if not files and not names:
181 # add (empty) directories
182 tar.add(path, recursive=False, filter=self.tar_filter)
183
184 tar.close()
185
186 signature = '{}.asc'.format(self.tar_path)
187
188 with open(signature, 'w') as fh:
189 fh.write('fake signature file')
190
191 return self.tar_path
192
193
194class UbuntuCoreUpgraderTestCase(unittest.TestCase):
195 '''
196 Base class for Upgrader tests.
197
198 Most of the tests follow a standard pattern:
199
200 1) Create an UpdateTree object:
201
202 update = UpdateTree()
203
204 This creates 2 temporary directories:
205
206 - self.system dir: Used as to generate an update archive from.
207
208 - self.tmp_dir: Used to write the generated archive file to. The
209 intention is that this directory should also be used to hold
210 the command file.
211
212 2) Removal tests call update.add_to_removed_file(file) to add a
213 particular file to the removals file in the update archive.
214
215 3) Create/Modify tests create files below update.system_dir.
216
217 4) Create an update archive (which includes the removals file
218 and all files below update.system_dir):
219
220 archive = update.create_archive(self.TARFILE)
221
222 5) Create a command file (which tells the upgrader what to do
223 and which archive files to apply):
224
225 make_command_file(...)
226
227 6) Create a victim directory. This is a temporary directory where
228 the upgrade will happen.
229
230 7) Start the upgrade:
231
232 call_upgrader(...)
233
234 8) Perform checks on the victim directory to ensure that upgrade
235 did what was expected.
236
237 '''
238
239 TARFILE = 'update.tar.xz'
240
241 # result of last test run. Hack to deal with fact that even if a
242 # test fails, unittest still calls .tearDown() (whomever thought
243 # that was a good idea...?)
244 currentResult = None
245
246 def setUp(self):
247 '''
248 Test setup.
249 '''
250 # Create an object to hold the tree that will be converted into
251 # an upgrade archive.
252 self.update = UpdateTree()
253
254 # The directory which will have the update archive applied to
255 # it.
256 self.victim_dir = make_tmp_dir(tag='victim')
257
258 self.patch_get_root_partitions_by_label = \
259 patch('ubuntucoreupgrader.upgrader.get_root_partitions_by_label',
260 mock_get_root_partitions_by_label)
261
262 self.patch_make_mount_private = \
263 patch('ubuntucoreupgrader.upgrader.make_mount_private',
264 mock_make_mount_private)
265
266 self.mock_get_root_partitions_by_label = \
267 self.patch_get_root_partitions_by_label.start()
268
269 self.mock_make_mount_private = \
270 self.patch_make_mount_private.start()
271
272 def tearDown(self):
273 '''
274 Test cleanup.
275 '''
276
277 if not self.currentResult.wasSuccessful():
278 # Do not clean up - the only sane option if a test fails.
279 return
280
281 self.update.destroy()
282 self.update = None
283
284 shutil.rmtree(self.victim_dir)
285 self.victim_dir = None
286
287 self.patch_get_root_partitions_by_label.stop()
288 self.patch_make_mount_private.stop()
289
290 def run(self, result=None):
291 self.currentResult = result
292 unittest.TestCase.run(self, result)
0293
=== modified file 'ubuntucoreupgrader/upgrader.py'
--- ubuntucoreupgrader/upgrader.py 2015-03-04 11:27:28 +0000
+++ ubuntucoreupgrader/upgrader.py 2015-03-04 13:53:47 +0000
@@ -1455,7 +1455,13 @@
1455 tar.extract(path=tmpdir, member=tar.getmember(file))1455 tar.extract(path=tmpdir, member=tar.getmember(file))
14561456
1457 path = os.path.join(tmpdir, file)1457 path = os.path.join(tmpdir, file)
1458 lines = [line.rstrip() for line in open(path, 'r')]1458
1459 lines = []
1460
1461 with open(path, 'r') as f:
1462 lines = f.readlines()
1463
1464 lines = [line.rstrip() for line in lines]
14591465
1460 shutil.rmtree(tmpdir)1466 shutil.rmtree(tmpdir)
14611467

Subscribers

People subscribed via source and target branches

to all changes: