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

[James Hunt]

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

[lp:~jamesodhunt/ubuntu/vivid/ubuntu-core-upgrader/add-functional-tests]

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.

=== 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 @@
+ubuntu-core-upgrader (0.7.6) UNRELEASED; urgency=low
+
+  * ubuntucoreupgrader/upgrader.py:
+    - get_file_contents(): Fix to avoid leaking the fd.
+  * functional/test_upgrader.py: Basic set of functional tests. 
+
+ -- James Hunt <[e-mail]>  Wed, 04 Mar 2015 11:52:19 +0000
+
 ubuntu-core-upgrader (0.7.5) vivid; urgency=low
 
   [ Michael Vogt ]

=== 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 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# --------------------------------------------------------------------
+# Copyright © 2014-2015 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+# --------------------------------------------------------------------
+
+# --------------------------------------------------------------------
+# Functional tests for the snappy upgrader.
+# --------------------------------------------------------------------
+
+import sys
+import os
+import logging
+import tarfile
+import unittest
+import shutil
+
+from unittest.mock import patch
+
+from ubuntucoreupgrader.upgrader import (
+    Upgrader,
+    parse_args,
+)
+
+base_dir = os.path.abspath(os.path.dirname(__file__))
+module_dir = os.path.normpath(os.path.realpath(base_dir + os.sep + '..'))
+sys.path.append(base_dir)
+
+from ubuntucoreupgrader.tests.utils import *
+
+CMD_FILE = 'ubuntu_command'
+
+
+def call_upgrader(command_file, root_dir, update):
+    '''
+    Invoke the upgrader.
+
+    :param command_file: commands file to drive the upgrader.
+    :param root_dir: Test directory to apply the upgrade to.
+    :param update: UpdateTree object.
+    '''
+
+    args = []
+    args += ['--root-dir', root_dir]
+    args += ['--debug', '1']
+
+    # don't delete the archive and command files.
+    # The tests clean up after themselves so they will get removed then,
+    # but useful to have them around to diagnose test failures.
+    args.append('--leave-files')
+
+    args.append(command_file)
+    commands = file_to_list(command_file)
+
+    cache_dir = make_tmp_dir()
+
+    def mock_get_cache_dir():
+        cache_dir = update.tmp_dir
+        sys_dir = os.path.join(cache_dir, 'system')
+        os.makedirs(sys_dir, exist_ok=True)
+        return cache_dir
+
+    upgrader = Upgrader(parse_args(args), commands, [])
+    upgrader.get_cache_dir = mock_get_cache_dir
+    upgrader.run()
+
+    shutil.rmtree(cache_dir)
+
+
+def create_device_file(path, type='c', major=-1, minor=-1):
+    '''
+    Create a device file.
+
+    :param path: full path to device file.
+    :param type: 'c' or 'b' (character or block).
+    :param major: major number.
+    :param minor: minor number.
+
+    XXX: This doesn't actually create a device node,
+    it simply creates a regular empty file whilst ensuring the filename
+    gives the impression that it is a device file.
+
+    This hackery is done for the following reasons:
+
+    - non-priv users cannot create device nodes (and the tests run as a
+      non-priv user).
+    - the removed file in the upgrade tar file does not actually specify the
+      _type_ of the files to remove. Hence, we can pretend that the file
+      to remove is a device file since the upgrader cannot know for sure
+      (it can check the existing on-disk file, but that isn't conclusive
+      since the admin may have manually modified a file, or the server
+      may have generated an invalid remove file - the upgrader cannot
+      know for sure.
+    '''
+    assert (os.path.dirname(path).endswith('/dev'))
+
+    append_file(path, 'fake-device file')
+
+
+def create_directory(path):
+    '''
+    Create a directory.
+    '''
+    os.makedirs(path, mode=TEST_DIR_MODE, exist_ok=False)
+
+
+def create_sym_link(source, dest):
+    '''
+    Create a symbolic link.
+
+    :param source: existing file to link to.
+    :param dest: name for the sym link.
+    '''
+    dirname = os.path.dirname(dest)
+    os.makedirs(dirname, mode=TEST_DIR_MODE, exist_ok=True)
+
+    os.symlink(source, dest)
+
+
+def create_hard_link(source, dest):
+    '''
+    Create a hard link.
+
+    :param source: existing file to link to.
+    :param dest: name for the hard link.
+    '''
+    os.link(source, dest)
+
+
+def is_sym_link_broken(path):
+    '''
+    :param path: symbolic link to check.
+    :return: True if the specified path is a broken symbolic link,
+     else False.
+    '''
+    try:
+        os.lstat(path)
+        os.stat(path)
+    except:
+        return True
+    return False
+
+
+def make_command_file(path, update_list):
+    '''
+    Create a command file that the upgrader processes.
+
+    :param path: full path to file to create,
+    :param update_list: list of update archives to include.
+    '''
+    l = []
+
+    for file in update_list:
+        l.append('update {} {}.asc'.format(file, file))
+
+    # flatten
+    contents = "\n".join(l) + '\n'
+
+    append_file(path, contents)
+
+
+def file_to_list(path):
+    '''
+    Convert the specified file into a list and return it.
+    '''
+    lines = []
+
+    with open(path, 'r') as f:
+        lines = f.readlines()
+
+    lines = [line.rstrip() for line in lines]
+
+    return lines
+
+
+class UpgraderFileRemovalTestCase(UbuntuCoreUpgraderTestCase):
+    '''
+    Test how the upgrader handles the removals file.
+    '''
+
+    def test_remove_file(self):
+        '''
+        Ensure the upgrader can remove a regular file.
+        '''
+
+        file = 'a-regular-file'
+
+        self.update.add_to_removed_file([file])
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        file_path = os.path.join(self.victim_dir, file)
+        create_file(file_path, 'foo bar')
+
+        self.assertTrue(os.path.exists(file_path))
+        self.assertTrue(os.path.isfile(file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertFalse(os.path.exists(file_path))
+
+    def test_remove_directory(self):
+        '''
+        Ensure the upgrader can remove a directory.
+        '''
+        dir = 'a-directory'
+
+        self.update.add_to_removed_file([dir])
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        dir_path = os.path.join(self.victim_dir, dir)
+        create_directory(dir_path)
+
+        self.assertTrue(os.path.exists(dir_path))
+        self.assertTrue(os.path.isdir(dir_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertFalse(os.path.exists(dir_path))
+
+    def test_remove_sym_link_file(self):
+        '''
+        Ensure the upgrader can remove a symbolic link to a file.
+        '''
+        src = 'the-source-file'
+        link = 'the-symlink-file'
+
+        self.update.add_to_removed_file([link])
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        src_file_path = os.path.join(self.victim_dir, src)
+        link_file_path = os.path.join(self.victim_dir, link)
+
+        create_file(src_file_path, 'foo bar')
+
+        self.assertTrue(os.path.exists(src_file_path))
+        self.assertTrue(os.path.isfile(src_file_path))
+        self.assertFalse(os.path.islink(src_file_path))
+
+        create_sym_link(src_file_path, link_file_path)
+        self.assertTrue(os.path.exists(link_file_path))
+        self.assertTrue(os.path.islink(link_file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        # original file should still be there
+        self.assertTrue(os.path.exists(src_file_path))
+        self.assertTrue(os.path.isfile(src_file_path))
+        self.assertFalse(os.path.islink(src_file_path))
+
+        # link should have gone
+        self.assertFalse(os.path.exists(link_file_path))
+
+    def test_remove_sym_link_directory(self):
+        '''
+        Ensure the upgrader can remove a symbolic link to a directory.
+        '''
+        dir = 'the-source-directory'
+        link = 'the-symlink-file'
+
+        self.update.add_to_removed_file([link])
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        src_dir_path = os.path.join(self.victim_dir, dir)
+        link_file_path = os.path.join(self.victim_dir, link)
+
+        create_directory(src_dir_path)
+
+        self.assertTrue(os.path.exists(src_dir_path))
+        self.assertTrue(os.path.isdir(src_dir_path))
+        self.assertFalse(os.path.islink(src_dir_path))
+
+        create_sym_link(src_dir_path, link_file_path)
+        self.assertTrue(os.path.exists(link_file_path))
+        self.assertTrue(os.path.islink(link_file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        # original directory should still be there
+        self.assertTrue(os.path.exists(src_dir_path))
+        self.assertTrue(os.path.isdir(src_dir_path))
+        self.assertFalse(os.path.islink(src_dir_path))
+
+        # link should have gone
+        self.assertFalse(os.path.exists(link_file_path))
+
+    def test_remove_hardlink(self):
+        '''
+        Ensure the upgrader can remove a hard link to a file.
+        '''
+        src = 'the-source-file'
+        link = 'the-hardlink-file'
+
+        self.update.add_to_removed_file([link])
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        src_file_path = os.path.join(self.victim_dir, src)
+        link_file_path = os.path.join(self.victim_dir, link)
+
+        create_file(src_file_path, 'foo bar')
+
+        src_inode = os.stat(src_file_path).st_ino
+
+        self.assertTrue(os.path.exists(src_file_path))
+        self.assertTrue(os.path.isfile(src_file_path))
+
+        create_hard_link(src_file_path, link_file_path)
+        self.assertTrue(os.path.exists(link_file_path))
+
+        link_inode = os.stat(link_file_path).st_ino
+
+        self.assertTrue(src_inode == link_inode)
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        # original file should still be there
+        self.assertTrue(os.path.exists(src_file_path))
+        self.assertTrue(os.path.isfile(src_file_path))
+
+        # Inode should not have changed.
+        self.assertTrue(os.stat(src_file_path).st_ino == src_inode)
+
+        # link should have gone
+        self.assertFalse(os.path.exists(link_file_path))
+
+    def test_remove_device_file(self):
+        '''
+        Ensure the upgrader can deal with a device file.
+
+        XXX: Note that The upgrader currently "deals" with them by
+        XXX: ignoring them :-)
+
+        '''
+        file = '/dev/a-fake-device'
+
+        self.update.add_to_removed_file([file])
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        file_path = '{}{}'.format(self.victim_dir, file)
+
+        create_device_file(file_path)
+
+        self.assertTrue(os.path.exists(file_path))
+
+        # sigh - we can't assert the that filetype is a char/block
+        # device because it won't be :)
+        self.assertTrue(os.path.isfile(file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertFalse(os.path.exists(file_path))
+
+
+class UpgraderFileAddTestCase(UbuntuCoreUpgraderTestCase):
+    '''
+    Test how the upgrader handles adding new files.
+    '''
+
+    def test_create_file(self):
+        '''
+        Ensure the upgrader can create a regular file.
+        '''
+        file = 'created-regular-file'
+
+        file_path = os.path.join(self.update.system_dir, file)
+
+        create_file(file_path, 'foo bar')
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        file_path = os.path.join(self.victim_dir, file)
+        self.assertFalse(os.path.exists(file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertTrue(os.path.exists(file_path))
+        self.assertTrue(os.path.isfile(file_path))
+
+    def test_create_directory(self):
+        '''
+        Ensure the upgrader can create a directory.
+        '''
+        dir = 'created-directory'
+
+        dir_path = os.path.join(self.update.system_dir, dir)
+
+        create_directory(dir_path)
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        dir_path = os.path.join(self.victim_dir, dir)
+        self.assertFalse(os.path.exists(dir_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertTrue(os.path.exists(dir_path))
+        self.assertTrue(os.path.isdir(dir_path))
+
+    def test_create_absolute_sym_link_to_file(self):
+        '''
+        Ensure the upgrader can create a symbolic link to a file (which
+        already exists and is not included in the update archive).
+        '''
+        src = 'the-source-file'
+        link = 'the-symlink-file'
+
+        # the file the link points to should *NOT* be below the 'system/'
+        # directory (since there isn't one post-unpack).
+
+        # an absolute sym-link target path
+        src_file_path = '/{}'.format(src)
+        link_file_path = os.path.join(self.update.system_dir, link)
+
+        victim_src_file_path = os.path.normpath('{}/{}'
+                                                .format(self.victim_dir,
+                                                        src_file_path))
+        victim_link_file_path = os.path.join(self.victim_dir, link)
+
+        # Create a broken sym link ('/system/<link> -> /<src>')
+        create_sym_link(src_file_path, link_file_path)
+
+        self.assertTrue(os.path.lexists(link_file_path))
+        self.assertTrue(os.path.islink(link_file_path))
+        self.assertTrue(is_sym_link_broken(link_file_path))
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        create_file(victim_src_file_path, 'foo')
+
+        self.assertTrue(os.path.exists(victim_src_file_path))
+        self.assertTrue(os.path.isfile(victim_src_file_path))
+
+        self.assertFalse(os.path.exists(victim_link_file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertTrue(os.path.exists(victim_src_file_path))
+        self.assertTrue(os.path.isfile(victim_src_file_path))
+
+        # upgrader should have created the link in the victim directory
+        self.assertTrue(os.path.lexists(victim_link_file_path))
+        self.assertTrue(os.path.islink(victim_link_file_path))
+        self.assertFalse(is_sym_link_broken(victim_link_file_path))
+
+    def test_create_relative_sym_link_to_file(self):
+        '''
+        Ensure the upgrader can create a symbolic link to a file (which
+        already exists and is not included in the update archive).
+        '''
+        src = 'a/b/c/the-source-file'
+        link = 'a/d/e/the-symlink-file'
+
+        # a relative sym-link target path
+        # ##src_file_path = '../../b/c/{}'.format(src)
+        src_file_path = '../../b/c/the-source-file'.format(src)
+
+        # the file the link points to should *NOT* be below the 'system/'
+        # directory (since there isn't one post-unpack).
+
+        link_file_path = os.path.join(self.update.system_dir, link)
+
+        victim_src_file_path = os.path.normpath('{}/{}'
+                                                .format(self.victim_dir,
+                                                        src))
+        victim_link_file_path = os.path.join(self.victim_dir, link)
+
+        create_sym_link(src_file_path, link_file_path)
+
+        self.assertTrue(os.path.lexists(link_file_path))
+        self.assertTrue(os.path.islink(link_file_path))
+        self.assertTrue(is_sym_link_broken(link_file_path))
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        create_file(victim_src_file_path, 'foo')
+
+        self.assertTrue(os.path.exists(victim_src_file_path))
+        self.assertTrue(os.path.isfile(victim_src_file_path))
+
+        self.assertFalse(os.path.exists(victim_link_file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        self.assertTrue(os.path.exists(victim_src_file_path))
+        self.assertTrue(os.path.isfile(victim_src_file_path))
+
+        # upgrader should have created the link in the victim directory
+        self.assertTrue(os.path.lexists(victim_link_file_path))
+        self.assertTrue(os.path.islink(victim_link_file_path))
+        self.assertFalse(is_sym_link_broken(victim_link_file_path))
+
+    def test_create_broken_sym_link_file(self):
+        '''
+        Ensure the upgrader can create a broken symbolic link
+        (one that points to a non-existent file).
+        '''
+        src = 'the-source-file'
+        link = 'the-symlink-file'
+
+        # the file the link points to should *NOT* be below the 'system/'
+        # directory (since there isn't one post-unpack).
+        src_file_path = src
+
+        link_file_path = os.path.join(self.update.system_dir, link)
+
+        # Create a broken sym link ('/system/<link> -> /<src>')
+        create_sym_link(src_file_path, link_file_path)
+
+        self.assertTrue(os.path.lexists(link_file_path))
+        self.assertTrue(os.path.islink(link_file_path))
+        self.assertTrue(is_sym_link_broken(link_file_path))
+
+        archive = self.update.create_archive(self.TARFILE)
+        self.assertTrue(os.path.exists(archive))
+
+        cmd_file = os.path.join(self.update.tmp_dir, CMD_FILE)
+        make_command_file(cmd_file, [self.TARFILE])
+
+        victim_src_file_path = os.path.join(self.victim_dir, src)
+        victim_link_file_path = os.path.join(self.victim_dir, link)
+
+        self.assertFalse(os.path.exists(victim_src_file_path))
+        self.assertFalse(os.path.exists(victim_link_file_path))
+
+        call_upgrader(cmd_file, self.victim_dir, self.update)
+
+        # source still shouldn't exist
+        self.assertFalse(os.path.exists(victim_src_file_path))
+
+        # upgrader should have created the link in the victim directory
+        self.assertTrue(os.path.lexists(victim_link_file_path))
+        self.assertTrue(os.path.islink(victim_link_file_path))
+        self.assertTrue(is_sym_link_broken(victim_link_file_path))
+
+
+def main():
+    kwargs = {}
+    format =             \
+        '%(asctime)s:'   \
+        '%(filename)s:'  \
+        '%(name)s:'      \
+        '%(funcName)s:'  \
+        '%(levelname)s:' \
+        '%(message)s'
+
+    kwargs['format'] = format
+
+    # We want to see what's happening
+    kwargs['level'] = logging.DEBUG
+
+    logging.basicConfig(**kwargs)
+
+    unittest.main(
+        testRunner=unittest.TextTestRunner(
+            stream=sys.stdout,
+            verbosity=2,
+
+            # don't keep running tests if one fails
+            # (... who _wouldn't_ want this???)
+            failfast=True
+        ),
+
+    )
+
+if __name__ == '__main__':
+    main()

=== 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 @@
 import unittest
 import os
 import shutil
+import sys
 
 from unittest.mock import patch
 
@@ -38,7 +39,10 @@
     parse_args,
 )
 
-script_name = os.path.basename(__file__)
+base_dir = os.path.abspath(os.path.dirname(__file__))
+sys.path.append(base_dir)
+
+from ubuntucoreupgrader.tests.utils import *
 
 # file mode to use for creating test directories.
 TEST_DIR_MODE = 0o750
@@ -50,19 +54,6 @@
     return parse_args([])
 
 
-def make_tmp_dir(tag=None):
-    '''
-    Create a temporary directory.
-    '''
-
-    if tag:
-        prefix = '{}-{}-'.format(script_name, tag)
-    else:
-        prefix = script_name
-
-    return tempfile.mkdtemp(prefix=prefix)
-
-
 class UpgradeTestCase(unittest.TestCase):
 
     def test_tar_generator_unpack_assets(self):
@@ -172,28 +163,6 @@
         shutil.rmtree(cache_dir)
 
 
-def append_file(path, contents):
-    '''
-    Append to a regular file (create it doesn't exist).
-    '''
-
-    dirname = os.path.dirname(path)
-    os.makedirs(dirname, mode=TEST_DIR_MODE, exist_ok=True)
-
-    with open(path, 'a') as fh:
-        fh.writelines(contents)
-
-        if not contents.endswith('\n'):
-            fh.write('\n')
-
-
-def create_file(path, contents):
-    '''
-    Create a regular file.
-    '''
-    append_file(path, contents)
-
-
 def touch_file(path):
     '''
     Create an empty file (creating any necessary intermediate
@@ -223,139 +192,6 @@
     return l
 
 
-class UpdateTree():
-    '''
-    Representation of a directory tree that will be converted into an
-    update archive.
-    '''
-    TEST_REMOVED_FILE = 'removed'
-    TEST_SYSTEM_DIR = 'system/'
-
-    def __init__(self):
-
-        # Directory tree used to construct the tar file from.
-        # Also used to hold the TEST_REMOVED_FILE file.
-        self.dir = make_tmp_dir(tag='UpdateTree-tar-source')
-
-        self.removed_file = os.path.join(self.dir, self.TEST_REMOVED_FILE)
-
-        # Directory to place create/modify files into.
-        self.system_dir = os.path.join(self.dir, self.TEST_SYSTEM_DIR)
-
-        # Directory used to write the generated tarfile to.
-        # This directory should also be used to write the command file
-        # to.
-        self.tmp_dir = make_tmp_dir(tag='UpdateTree-cache')
-
-    def destroy(self):
-        if os.path.exists(self.dir):
-            shutil.rmtree(self.dir)
-
-        if os.path.exists(self.tmp_dir):
-            shutil.rmtree(self.tmp_dir)
-
-    def tar_filter(self, member):
-        '''
-        Function to filter the tarinfo members before creating the
-        archive.
-        '''
-        # members are created with relative paths (no leading slash)
-        path = os.sep + member.name
-
-        if member.name == '/.':
-            return None
-
-        i = path.find(self.dir)
-        assert(i == 0)
-
-        # remove the temporary directory elements
-        # (+1 for the os.sep we added above)
-        member.name = path[len(self.dir)+1:]
-
-        return member
-
-    def create_archive(self, name):
-        '''
-        Create an archive with the specified name from the UpdateTree
-        object. Also creates a fake signature file alongside the archive
-        file since this is currently required by the upgrader (although
-        it is not validated).
-
-        :param name: name of tarfile.
-        :param name: full path to xz archive to create.
-        :return full path to tar file with name @name.
-        '''
-
-        tar_path = os.path.join(self.tmp_dir, name)
-        tar = tarfile.open(tar_path, 'w:xz')
-
-        # We can't just add recursively since that would attempt to add
-        # the parent directory. However, the real update tars don't
-        # include that, and attempting to ignore the parent directory
-        # results in an empty archive. So, walk the tree and add
-        # file-by-file.
-        for path, names, files in os.walk(self.dir):
-            for file in files:
-                full = os.path.join(path, file)
-                tar.add(full, recursive=False, filter=self.tar_filter)
-            if not files and not names:
-                # add (empty) directories
-                tar.add(path, recursive=False, filter=self.tar_filter)
-
-        tar.close()
-
-        signature = '{}.asc'.format(tar_path)
-
-        with open(signature, 'w') as fh:
-            fh.write('fake signature file')
-
-        return tar_path
-
-
-class UbuntuCoreUpgraderTestCase(unittest.TestCase):
-    '''
-    Base class for Upgrader tests.
-    '''
-
-    TARFILE = 'update.tar.xz'
-
-    # result of last test run. Hack to deal with fact that even if a
-    # test fails, unittest still calls .tearDown() (whomever thought
-    # that was a good idea...?)
-    currentResult = None
-
-    def setUp(self):
-        '''
-        Test setup.
-        '''
-        # Create an object to hold the tree that will be converted into
-        # an upgrade archive.
-        self.update = UpdateTree()
-
-        # The directory which will have the update archive applied to
-        # it.
-        self.victim_dir = make_tmp_dir(tag='victim')
-
-    def tearDown(self):
-        '''
-        Test cleanup.
-        '''
-
-        if not self.currentResult.wasSuccessful():
-            # Do not clean up - the only sane option if a test fails.
-            return
-
-        self.update.destroy()
-        self.update = None
-
-        shutil.rmtree(self.victim_dir)
-        self.victim_dir = None
-
-    def run(self, result=None):
-        self.currentResult = result
-        unittest.TestCase.run(self, result)
-
-
 def mock_get_root_partitions_by_label():
     matches = []
 

=== 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 @@
+# -*- coding: utf-8 -*-
+# --------------------------------------------------------------------
+# Copyright © 2014-2015 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+# --------------------------------------------------------------------
+
+import os
+import tempfile
+import tarfile
+import shutil
+import unittest
+
+from unittest.mock import patch
+
+# file mode to use for creating test directories.
+TEST_DIR_MODE = 0o750
+
+script_name = os.path.basename(__file__)
+
+
+def make_tmp_dir(tag=None):
+    '''
+    Create a temporary directory.
+    '''
+
+    if tag:
+        prefix = '{}-{}-'.format(script_name, tag)
+    else:
+        prefix = script_name
+
+    return tempfile.mkdtemp(prefix=prefix)
+
+
+def append_file(path, contents):
+    '''
+    Append to a regular file (create it doesn't exist).
+    '''
+
+    dirname = os.path.dirname(path)
+    os.makedirs(dirname, mode=TEST_DIR_MODE, exist_ok=True)
+
+    with open(path, 'a') as fh:
+        fh.writelines(contents)
+
+        if not contents.endswith('\n'):
+            fh.write('\n')
+
+
+def create_file(path, contents):
+    '''
+    Create a regular file.
+    '''
+    append_file(path, contents)
+
+
+def mock_get_root_partitions_by_label():
+    '''
+    Fake disk partition details for testing.
+    '''
+    matches = []
+
+    matches.append(('system-a', '/dev/sda3', '/'))
+    matches.append(('system-b', '/dev/sda4', '/writable/cache/system'))
+
+    return matches
+
+
+def mock_make_mount_private(target):
+    '''
+    NOP implementation.
+    '''
+    pass
+
+
+class UpdateTree():
+    '''
+    Representation of a directory tree that will be converted into an
+    update archive.
+    '''
+    TEST_REMOVED_FILE = 'removed'
+    TEST_SYSTEM_DIR = 'system/'
+
+    def __init__(self):
+
+        # Directory tree used to construct the tar file from.
+        # Also used to hold the TEST_REMOVED_FILE file.
+        self.dir = make_tmp_dir(tag='UpdateTree-tar-source')
+
+        self.removed_file = os.path.join(self.dir, self.TEST_REMOVED_FILE)
+
+        # Directory to place create/modify files into.
+        self.system_dir = os.path.join(self.dir, self.TEST_SYSTEM_DIR)
+
+        # Directory used to write the generated tarfile to.
+        # This directory should also be used to write the command file
+        # to.
+        self.tmp_dir = make_tmp_dir(tag='UpdateTree-cache')
+
+    def destroy(self):
+        if os.path.exists(self.dir):
+            shutil.rmtree(self.dir)
+
+        if os.path.exists(self.tmp_dir):
+            shutil.rmtree(self.tmp_dir)
+
+    def add_to_removed_file(self, removed_files):
+        '''
+        Add the specified list of files to the removed file.
+
+        The 'removed' file is simply a file with a well-known name that
+        contains a list of files (one per line) to be removed from a
+        system before the rest of the update archive is unpacked.
+
+        :param removed_files: list of file names to add to the removed file.
+
+        '''
+        # all files listed in the removed list must be system files
+        final = list(map(lambda a:
+                     '{}{}'.format(self.TEST_SYSTEM_DIR, a), removed_files))
+
+        contents = "".join(final)
+        append_file(self.removed_file, contents)
+
+    def tar_filter(self, member):
+        '''
+        Function to filter the tarinfo members before creating the
+        archive.
+        '''
+        # members are created with relative paths (no leading slash)
+        path = os.sep + member.name
+
+        if member.name == '/.':
+            return None
+
+        i = path.find(self.dir)
+        assert(i == 0)
+
+        # remove the temporary directory elements
+        # (+1 for the os.sep we added above)
+        member.name = path[len(self.dir)+1:]
+
+        return member
+
+    def create_archive(self, name):
+        '''
+        Create an archive with the specified name from the UpdateTree
+        object. Also creates a fake signature file alongside the archive
+        file since this is currently required by the upgrader (although
+        it is not validated).
+
+        :param name: name of tarfile.
+        :param name: full path to xz archive to create.
+        :return full path to tar file with name @name.
+        '''
+
+        self.tar_path = os.path.join(self.tmp_dir, name)
+        tar = tarfile.open(self.tar_path, 'w:xz')
+
+        # We can't just add recursively since that would attempt to add
+        # the parent directory. However, the real update tars don't
+        # include that, and attempting to ignore the parent directory
+        # results in an empty archive. So, walk the tree and add
+        # file-by-file.
+        for path, names, files in os.walk(self.dir):
+            for file in files:
+                full = os.path.join(path, file)
+                tar.add(full, recursive=False, filter=self.tar_filter)
+            if not files and not names:
+                # add (empty) directories
+                tar.add(path, recursive=False, filter=self.tar_filter)
+
+        tar.close()
+
+        signature = '{}.asc'.format(self.tar_path)
+
+        with open(signature, 'w') as fh:
+            fh.write('fake signature file')
+
+        return self.tar_path
+
+
+class UbuntuCoreUpgraderTestCase(unittest.TestCase):
+    '''
+    Base class for Upgrader tests.
+
+    Most of the tests follow a standard pattern:
+
+    1) Create an UpdateTree object:
+
+         update = UpdateTree()
+
+       This creates 2 temporary directories:
+
+       - self.system dir: Used as to generate an update archive from.
+
+       - self.tmp_dir: Used to write the generated archive file to. The
+         intention is that this directory should also be used to hold
+         the command file.
+
+    2) Removal tests call update.add_to_removed_file(file) to add a
+       particular file to the removals file in the update archive.
+
+    3) Create/Modify tests create files below update.system_dir.
+
+    4) Create an update archive (which includes the removals file
+       and all files below update.system_dir):
+
+         archive = update.create_archive(self.TARFILE)
+
+    5) Create a command file (which tells the upgrader what to do
+       and which archive files to apply):
+
+         make_command_file(...)
+
+    6) Create a victim directory. This is a temporary directory where
+       the upgrade will happen.
+
+    7) Start the upgrade:
+
+         call_upgrader(...)
+
+    8) Perform checks on the victim directory to ensure that upgrade
+       did what was expected.
+
+    '''
+
+    TARFILE = 'update.tar.xz'
+
+    # result of last test run. Hack to deal with fact that even if a
+    # test fails, unittest still calls .tearDown() (whomever thought
+    # that was a good idea...?)
+    currentResult = None
+
+    def setUp(self):
+        '''
+        Test setup.
+        '''
+        # Create an object to hold the tree that will be converted into
+        # an upgrade archive.
+        self.update = UpdateTree()
+
+        # The directory which will have the update archive applied to
+        # it.
+        self.victim_dir = make_tmp_dir(tag='victim')
+
+        self.patch_get_root_partitions_by_label = \
+            patch('ubuntucoreupgrader.upgrader.get_root_partitions_by_label',
+                  mock_get_root_partitions_by_label)
+
+        self.patch_make_mount_private = \
+            patch('ubuntucoreupgrader.upgrader.make_mount_private',
+                  mock_make_mount_private)
+
+        self.mock_get_root_partitions_by_label = \
+            self.patch_get_root_partitions_by_label.start()
+
+        self.mock_make_mount_private = \
+            self.patch_make_mount_private.start()
+
+    def tearDown(self):
+        '''
+        Test cleanup.
+        '''
+
+        if not self.currentResult.wasSuccessful():
+            # Do not clean up - the only sane option if a test fails.
+            return
+
+        self.update.destroy()
+        self.update = None
+
+        shutil.rmtree(self.victim_dir)
+        self.victim_dir = None
+
+        self.patch_get_root_partitions_by_label.stop()
+        self.patch_make_mount_private.stop()
+
+    def run(self, result=None):
+        self.currentResult = result
+        unittest.TestCase.run(self, result)

=== 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 @@
         tar.extract(path=tmpdir, member=tar.getmember(file))
 
         path = os.path.join(tmpdir, file)
-        lines = [line.rstrip() for line in open(path, 'r')]
+
+        lines = []
+
+        with open(path, 'r') as f:
+            lines = f.readlines()
+
+        lines = [line.rstrip() for line in lines]
 
         shutil.rmtree(tmpdir)