[rhel7-branch 8/9] driver_updates: fixes from patch review

wgwoods installerbot-noreply at redhat.com
Thu Jul 9 15:18:36 UTC 2015


From: Will Woods <wwoods at redhat.com>

- use constant for ANACONDAVER

- rename constants to uppercase
  (ARCH, KERNELVER, MODULE_UPDATES_DIR, FIRMWARE_UPDATES_DIR)

- extract_drivers(): explicitly refuse to be called with drivers=..
  and repos=.. so we don't construct a temporary list

- add list_drivers(), use that wherever we were doing
  [d for r in repos for d in dd_list(r)]

- read_lines(): refactor and add comment to be clearer about what it's doing,
  add unit test

- DeviceInfo.shortdev: use os.path.realpath instead of manually
  resolving symlinks, add comment about why we don't use basename(),
  add unit test

- Add extra explanatory comments

- TextMenu.format_header(): use 4*' ' instead of '    '

- Move test_driver_updates.py to tests/dracut_tests/

- Fix ExtractDriversTestCase trying to create /updates

Related: RHBZ#1084197
---
 dracut/Makefile.am                        |   3 +-
 dracut/driver_updates.py                  |  86 ++--
 dracut/test_driver_updates.py             | 624 -----------------------------
 tests/dracut_tests/test_driver_updates.py | 631 ++++++++++++++++++++++++++++++
 tests/nosetests.sh                        |   2 +-
 5 files changed, 689 insertions(+), 657 deletions(-)
 delete mode 100644 dracut/test_driver_updates.py
 create mode 100644 tests/dracut_tests/test_driver_updates.py

diff --git a/dracut/Makefile.am b/dracut/Makefile.am
index 44a5b46..71d6bed 100644
--- a/dracut/Makefile.am
+++ b/dracut/Makefile.am
@@ -46,7 +46,6 @@ dist_dracut_SCRIPTS = module-setup.sh \
 		      driver-updates at .service \
 		      driver-updates-genrules.sh \
 		      anaconda-depmod.sh \
-		      driver_updates.py \
-		      test_driver_updates.py
+		      driver_updates.py
 
 MAINTAINERCLEANFILES = Makefile.in
diff --git a/dracut/driver_updates.py b/dracut/driver_updates.py
index 4c331d5..0bf9648 100755
--- a/dracut/driver_updates.py
+++ b/dracut/driver_updates.py
@@ -49,13 +49,18 @@
 /tmp/dd.done should be created when all the user-requested stuff above has been
 handled; the installer won't start up until this file is created.
 
-Repositories for installed drivers are copied into /run/install/DD-X where X
-starts at 1 and increments for each repository.
+Packages will be extracted to /updates, which gets overlaid on top
+of the installer's filesystem when we leave the initramfs.
 
-Selected driver package names are saved in /run/install/dd_packages.
+Modules and firmware get moved to /lib/modules/`uname -r`/updates and
+/lib/firmware/updates (under /updates, as above). They also get copied into the
+corresponding paths in the initramfs, so we can load them immediately.
 
-Anaconda uses the repository and package list to install the same set of drivers
-to the target system.
+The repositories get copied into /run/install/DD-1, /run/install/DD-2, etc.
+Driver package names are saved in /run/install/dd_packages.
+
+During system installation, anaconda will install the packages listed in
+/run/install/dd_packages to the target system.
 """
 
 import logging
@@ -80,8 +85,19 @@
 
 log = logging.getLogger("DD")
 
-arch = os.uname()[4]
-kernelver = os.uname()[2]
+# NOTE: Yes, the version is wrong, but previous versions of this utility also
+# hardcoded this value, because changing it will break any driver disk that has
+# binary/library packages with "installer-enhancement = 19.0"..
+# If we *need* to break compatibility, this should definitely get changed, but
+# otherwise we probably shouldn't change this unless/until we're sure that
+# everyone is using something like "installer-enhancement >= 19.0" instead..
+ANACONDAVER = "19.0"
+
+ARCH = os.uname()[4]
+KERNELVER = os.uname()[2]
+
+MODULE_UPDATES_DIR = "/lib/modules/%s/updates" % ARCH
+FIRMWARE_UPDATES_DIR = "/lib/firmware/updates"
 
 def mkdir_seq(stem):
     """
@@ -108,7 +124,7 @@ def find_repos(mnt):
     """find any valid driverdisk repos that exist under mnt."""
     dd_repos = []
     for root, dirs, files in os.walk(mnt, followlinks=True):
-        repo = root+"/rpms/"+arch
+        repo = root+"/rpms/"+ARCH
         if "rhdd3" in files and "rpms" in dirs and os.path.isdir(repo):
             log.debug("found repo: %s", repo)
             dd_repos.append(repo)
@@ -133,9 +149,9 @@ def __init__(self, source="", name="", flags="", description="", repo=""):
 def dd_list(dd_path, anaconda_ver=None, kernel_ver=None):
     log.debug("dd_list: listing %s", dd_path)
     if not anaconda_ver:
-        anaconda_ver = '19.0'
+        anaconda_ver = ANACONDAVER
     if not kernel_ver:
-        kernel_ver = kernelver
+        kernel_ver = KERNELVER
     cmd = ["dd_list", '-d', dd_path, '-k', kernel_ver, '-a', anaconda_ver]
     out = subprocess.check_output(cmd, stderr=DEVNULL)
     out = out.decode('utf-8')
@@ -147,10 +163,13 @@ def dd_list(dd_path, anaconda_ver=None, kernel_ver=None):
 def dd_extract(rpm_path, outdir, kernel_ver=None, flags='-blmf'):
     log.debug("dd_extract: extracting %s", rpm_path)
     if not kernel_ver:
-        kernel_ver = kernelver
+        kernel_ver = KERNELVER
     cmd = ["dd_extract", flags, '-r', rpm_path, '-d', outdir, '-k', kernel_ver]
     subprocess.check_output(cmd, stderr=DEVNULL) # discard stdout
 
+def list_drivers(repos, anaconda_ver=None, kernel_ver=None):
+    return [d for r in repos for d in dd_list(r, anaconda_ver, kernel_ver)]
+
 def mount(dev, mnt=None):
     """Mount the given dev at the mountpoint given by mnt."""
     # NOTE: dev may be a filesystem image - "-o loop" is not necessary anymore
@@ -173,9 +192,6 @@ def mounted(dev, mnt=None):
     finally:
         umount(mnt)
 
-module_updates_dir = '/lib/modules/%s/updates' % os.uname()[2]
-firmware_updates_dir = '/lib/firmware/updates'
-
 def iter_files(topdir, pattern=None):
     """iterator; yields full paths to files under topdir that match pattern."""
     for head, _, files in os.walk(topdir):
@@ -210,9 +226,12 @@ def append_line(filename, line):
     with open(filename, 'a') as outf:
         outf.write(line)
 
+# NOTE: items returned by read_lines should match items passed to append_line,
+#       which is why we remove the newlines
 def read_lines(filename):
+    """return a list containing each line in filename, with newlines removed."""
     try:
-        return open(filename).read().splitlines()
+        return [line.rstrip('\n') for line in open(filename)]
     except IOError:
         return []
 
@@ -230,7 +249,7 @@ def extract_drivers(drivers=None, repos=None, outdir="/updates",
 
     drivers should be a list of Drivers to extract, or None.
     repos should be a list of repo paths to extract, or None.
-    (If both are empty, nothing happens..)
+    Raises ValueError if you pass both.
 
     If any packages containing modules or firmware are extracted, also:
     * call save_repo for that package's repo
@@ -240,8 +259,10 @@ def extract_drivers(drivers=None, repos=None, outdir="/updates",
     """
     if not drivers:
         drivers = []
+    if drivers and repos:
+        raise ValueError("extract_drivers: drivers or repos, not both")
     if repos:
-        drivers += [d for repo in repos for d in dd_list(repo)]
+        drivers = list_drivers(repos)
 
     save_repos = set()
     new_drivers = False
@@ -270,10 +291,10 @@ def grab_driver_files(outdir="/updates"):
     """
     modules = list(iter_files(outdir+'/lib/modules',"*.ko*"))
     firmware = list(iter_files(outdir+'/lib/firmware'))
-    copy_files(modules, module_updates_dir)
-    copy_files(firmware, firmware_updates_dir)
-    move_files(modules, outdir+module_updates_dir)
-    move_files(firmware, outdir+firmware_updates_dir)
+    copy_files(modules, MODULE_UPDATES_DIR)
+    copy_files(firmware, FIRMWARE_UPDATES_DIR)
+    move_files(modules, outdir+MODULE_UPDATES_DIR)
+    move_files(firmware, outdir+FIRMWARE_UPDATES_DIR)
     return [os.path.basename(m).split('.ko')[0] for m in modules]
 
 def load_drivers(modnames):
@@ -282,6 +303,8 @@ def load_drivers(modnames):
     subprocess.call(["depmod", "-a"])
     subprocess.call(["modprobe", "-a"] + modnames)
 
+# We *could* pass in "outdir" if we wanted to extract things somewhere else,
+# but right now the only use case is running inside the initramfs, so..
 def process_driver_disk(dev, interactive=False):
     try:
         _process_driver_disk(dev, interactive=interactive)
@@ -350,12 +373,13 @@ def __repr__(self):
 
     @property
     def shortdev(self):
-        if os.path.islink(self.device):
-            return os.path.basename(os.readlink(self.device))
-        elif self.device.startswith('/dev/'):
-            return self.device[5:]
-        else:
-            return self.device
+        # resolve any symlinks (/dev/disk/by-label/OEMDRV -> /dev/sr0)
+        dev = os.path.realpath(self.device)
+        # NOTE: not os.path.basename 'cuz some devices legitimately have
+        # a '/' in their name: /dev/cciss/c0d0, /dev/i2o/hda, etc.
+        if dev.startswith('/dev/'):
+            dev = dev[5:]
+        return dev
 
 def blkid():
     out = subprocess.check_output("blkid -o export -s UUID -s TYPE".split())
@@ -363,6 +387,8 @@ def blkid():
     return [dict(kv.split('=',1) for kv in block.splitlines())
                                  for block in out.split('\n\n')]
 
+# We use this to get disk labels because blkid's encoding of non-printable and
+# non-ascii characters is weird and doesn't match what you'd expect to see.
 def get_disk_labels():
     return {os.path.realpath(s):os.path.basename(s)
             for s in iter_files("/dev/disk/by-label")}
@@ -451,9 +477,9 @@ def format_items(self):
 
     def format_header(self):
         if self.multi:
-            return '        '+self.format_item(self.headeritem)
+            return (8*' ')+self.format_item(self.headeritem)
         else:
-            return '    '+self.format_item(self.headeritem)
+            return (4*' ')+self.format_item(self.headeritem)
 
     def action_dict(self):
         actions = {
@@ -498,7 +524,7 @@ def run(self):
         return self.selected_items
 
 def repo_menu(repos):
-    drivers = [d for r in repos for d in dd_list(r)]
+    drivers = list_drivers(repos)
     if not drivers:
         log.info("No suitable drivers found.")
         return []
diff --git a/dracut/test_driver_updates.py b/dracut/test_driver_updates.py
deleted file mode 100644
index 2aaa993..0000000
--- a/dracut/test_driver_updates.py
+++ /dev/null
@@ -1,624 +0,0 @@
-# test_driver_updates.py - unittests for driver_updates.py
-
-import unittest
-try:
-    import unittest.mock as mock
-except ImportError:
-    import mock
-
-import os
-import tempfile
-import shutil
-
-from driver_updates import copy_files, move_files, iter_files, ensure_dir
-from driver_updates import append_line, mkdir_seq
-
-def touch(path):
-    try:
-        open(path, 'a')
-    except IOError as e:
-        if e.errno != 17: raise
-
-def makedir(path):
-    ensure_dir(path)
-    return path
-
-def makefile(path):
-    makedir(os.path.dirname(path))
-    touch(path)
-    return path
-
-def makefiles(*paths):
-    return [makefile(p) for p in paths]
-
-class FileTestCaseBase(unittest.TestCase):
-    def setUp(self):
-        self.tmpdir = tempfile.mkdtemp(prefix="test_driver_updates.")
-        self.srcdir = self.tmpdir+'/src/'
-        self.destdir = self.tmpdir+'/dest/'
-
-    def tearDown(self):
-        shutil.rmtree(self.tmpdir, ignore_errors=True)
-
-    def makefiles(self, *paths):
-        return [makefile(os.path.normpath(self.tmpdir+'/'+p)) for p in paths]
-
-class SelfTestCase(FileTestCaseBase):
-    def test_makefiles(self):
-        """check test helpers"""
-        filepaths = ["sub/dir/test.file", "testfile"]
-        self.makefiles(*filepaths)
-        for f in filepaths:
-            self.assertTrue(os.path.exists(self.tmpdir+'/'+f))
-
-class TestCopyFiles(FileTestCaseBase):
-    def test_basic(self):
-        """copy_file: copy files into destdir, leaving existing contents"""
-        files = self.makefiles("src/file1", "src/subdir/file2")
-        self.makefiles("dest/file3")
-        copy_files(files, self.destdir)
-        result = set(os.listdir(self.destdir))
-        self.assertEqual(result, set(["file1", "file2", "file3"]))
-
-    def test_overwrite(self):
-        """copy_file: overwrite files in destdir if they have the same name"""
-        src, dest = self.makefiles("src/file1", "dest/file1")
-        with open(src, 'w') as outf:
-            outf.write("srcfile")
-        with open(dest, 'w') as outf:
-            outf.write("destfile")
-        copy_files([src], self.destdir)
-        self.assertEqual(os.listdir(self.destdir), ["file1"])
-        self.assertEqual(open(dest).read(), "srcfile")
-
-    def test_samefile(self):
-        """copy_file: skip files already in destdir"""
-        (dest,) = self.makefiles("dest/file1")
-        with open(dest, 'w') as outf:
-            outf.write("destfile")
-        copy_files([dest], self.destdir)
-        self.assertEqual(os.listdir(self.destdir), ["file1"])
-        self.assertEqual(open(dest).read(), "destfile")
-
-    def test_copy_to_parent(self):
-        """copy_file: skip files in subdirs of destdir"""
-        files = self.makefiles("dest/subdir/file1")
-        copy_files(files, self.destdir)
-        self.assertEqual(list(iter_files(self.destdir)), files)
-
-class TestIterFiles(FileTestCaseBase):
-    def test_basic(self):
-        """iter_files: iterates over full paths to files under topdir"""
-        files = set(self.makefiles("src/file1", "dest/file2", "src/sub/file3"))
-        makedir(self.tmpdir+'/empty/dir')
-        result = set(iter_files(self.tmpdir))
-        self.assertEqual(files, result)
-
-    def test_pattern(self):
-        """iter_files: match filename against glob pattern"""
-        self.makefiles("src/file1.so", "src/sub.ko/file2")
-        goodfiles = set(self.makefiles("src/sub/file1.ko", "src/file2.ko.xz"))
-        result = set(iter_files(self.tmpdir, pattern="*.ko*"))
-        self.assertEqual(result, goodfiles)
-
-class TestMoveFiles(FileTestCaseBase):
-    def test_basic(self):
-        """move_files: move files to destdir"""
-        files = self.makefiles("src/file1", "src/subdir/file2")
-        move_files(files, self.destdir)
-        self.assertEqual(set(os.listdir(self.destdir)), set(["file1", "file2"]))
-        self.assertEqual(list(iter_files(self.srcdir)), [])
-
-    def test_overwrite(self):
-        """move_files: overwrite files with the same name"""
-        src, dest = self.makefiles("src/file1", "dest/file1")
-        with open(src, 'w') as outf:
-            outf.write("srcfile")
-        with open(dest, 'w') as outf:
-            outf.write("destfile")
-        move_files([src], self.destdir)
-        self.assertEqual(os.listdir(self.destdir), ["file1"])
-        self.assertEqual(open(dest).read(), "srcfile")
-        self.assertEqual(list(iter_files(self.srcdir)), [])
-
-    def test_samefile(self):
-        """move_files: leave files alone if they're already in destdir"""
-        (dest,) = self.makefiles("dest/file1")
-        with open(dest, 'w') as outf:
-            outf.write("destfile")
-        move_files([dest], self.destdir)
-        self.assertEqual(os.listdir(self.destdir), ["file1"])
-        self.assertEqual(open(dest).read(), "destfile")
-
-    def test_move_to_parent(self):
-        """move_files: leave files alone if they're in a subdir of destdir"""
-        files = set(self.makefiles("dest/subdir/file1", "dest/file2"))
-        move_files(files, self.destdir)
-        self.assertEqual(set(iter_files(self.destdir)), files)
-
-class TestAppendLine(FileTestCaseBase):
-    def test_empty(self):
-        """append_line: create file + append \\n when needed"""
-        line = "this is a line of text with no newline"
-        outfile = self.tmpdir+'/outfile'
-        append_line(outfile, line)
-        self.assertEqual(open(outfile).read(), line+'\n')
-
-    def test_append(self):
-        """append_line: adds a line to the end of an existing file"""
-        oldlines = ["line one", "line two", "and I'm line three"]
-        outfile = self.tmpdir+'/outfile'
-        with open(outfile, 'w') as outf:
-            for line in oldlines:
-                outf.write(line+'\n')
-        line = "this line contains a newline already\n"
-        append_line(outfile, line)
-        self.assertEqual(open(outfile).read(), '\n'.join(oldlines+[line]))
-
-from driver_updates import read_lines
-class TestReadLine(FileTestCaseBase):
-    def test_empty(self):
-        """read_lines: return [] for empty file"""
-        [empty] = self.makefiles("emptyfile")
-        self.assertEqual(read_lines(empty), [])
-
-    def test_missing(self):
-        """read_lines: return [] for missing file"""
-        self.assertEqual(read_lines(self.tmpdir+'/no-such-file'),[])
-
-    def test_readlines(self):
-        """read_lines: returns a list of lines without trailing newlines"""
-        filedata = 'line one\nline two\n\nline four\n'
-        outfile = self.tmpdir+'/outfile'
-        with open(outfile, 'w') as outf:
-            outf.write(filedata)
-        lines = read_lines(outfile)
-        self.assertEqual(lines, ['line one', 'line two','','line four'])
-
-class TestMkdirSeq(FileTestCaseBase):
-    def test_basic(self):
-        """mkdir_seq: first dir ends with 1"""
-        newdir = mkdir_seq(self.srcdir+'/DD-')
-        self.assertEqual(newdir, self.srcdir+'/DD-1')
-        self.assertTrue(os.path.isdir(newdir))
-
-    def test_one_exists(self):
-        """mkdir_seq: increment number if file exists"""
-        firstdir = mkdir_seq(self.srcdir+'/DD-')
-        newdir = mkdir_seq(self.srcdir+'/DD-')
-        self.assertEqual(newdir, self.srcdir+'/DD-2')
-        self.assertTrue(os.path.isdir(newdir))
-        self.assertTrue(os.path.isdir(firstdir))
-
-from driver_updates import find_repos, save_repo, arch
-# As far as we know, this is what makes a valid repo: rhdd3 + rpms/`uname -m`/
-def makerepo(topdir, desc=None):
-    descfile = makefile(topdir+'/rhdd3')
-    if not desc:
-        desc = os.path.basename(topdir)
-    with open(descfile, "w") as outf:
-        outf.write(desc+"\n")
-    makedir(topdir+'/rpms/'+arch)
-
-class TestFindRepos(FileTestCaseBase):
-    def test_basic(self):
-        """find_repos: return RPM dir if a valid repo is found"""
-        makerepo(self.tmpdir)
-        repos = find_repos(self.tmpdir)
-        self.assertEqual(repos, [self.tmpdir+'/rpms/'+arch])
-        self.assertTrue(os.path.isdir(repos[0]))
-
-    def test_multiple_subdirs(self):
-        """find_repos: descend multiple subdirs if needed"""
-        makerepo(self.tmpdir+'/driver1')
-        makerepo(self.tmpdir+'/sub/driver1')
-        makerepo(self.tmpdir+'/sub/driver2')
-        repos = find_repos(self.tmpdir)
-        self.assertEqual(len(repos),3)
-
-class TestSaveRepo(FileTestCaseBase):
-    def test_basic(self):
-        """save_repo: copies a directory to /run/install/DD-X"""
-        makerepo(self.srcdir)
-        [repo] = find_repos(self.srcdir)
-        makefile(repo+'/fake-something.rpm')
-        saved = save_repo(repo, target=self.destdir)
-        self.assertEqual(set(os.listdir(saved)), set(["fake-something.rpm"]))
-        self.assertEqual(saved, os.path.join(self.destdir, "DD-1"))
-
-from driver_updates import mount, umount, mounted
-class MountTestCase(unittest.TestCase):
-    @mock.patch('driver_updates.mkdir_seq')
-    @mock.patch('driver_updates.subprocess.check_call')
-    def test_mkdir(self, check_call, mkdir):
-        """mount: makes mountpoint if needed"""
-        dev, mnt = '/dev/fake', '/media/DD-1'
-        mkdir.return_value = mnt
-        mountpoint = mount(dev)
-        mkdir.assert_called_once_with('/media/DD-')
-        check_call.assert_called_once_with(["mount", dev, mnt])
-        self.assertEqual(mnt, mountpoint)
-
-    @mock.patch('driver_updates.mkdir_seq')
-    @mock.patch('driver_updates.subprocess.check_call')
-    def test_basic(self, check_call, mkdir):
-        """mount: calls mount(8) to mount a device/image"""
-        dev, mnt = '/dev/fake', '/media/fake'
-        mount(dev, mnt)
-        check_call.assert_called_once_with(["mount", dev, mnt])
-        self.assertFalse(mkdir.called)
-
-    @mock.patch('driver_updates.subprocess.call')
-    def test_umount(self, call):
-        """umount: calls umount(8)"""
-        mnt = '/mnt/fake'
-        umount(mnt)
-        call.assert_called_once_with(["umount", mnt])
-
-    @mock.patch('driver_updates.mount')
-    @mock.patch('driver_updates.umount')
-    def test_mount_manager(self, mock_umount, mock_mount):
-        """mounted: context manager mounts/umounts as expected"""
-        dev, mnt = '/dev/fake', '/media/fake'
-        mock_mount.return_value = mnt
-        with mounted(dev, mnt) as mountpoint:
-            mock_mount.assert_called_once_with(dev, mnt)
-            self.assertFalse(mock_umount.called)
-            self.assertEqual(mountpoint, mnt)
-        mock_umount.assert_called_once_with(mnt)
-
-# NOTE: dd_list and dd_extract get tested pretty thoroughly in tests/dd_tests,
-# so this is a slightly higher-level test case
-from driver_updates import dd_list, dd_extract, Driver
-fake_module = Driver(
-    source='/repo/path/to/fake-driver-1.0-1.rpm',
-    name='fake-driver',
-    flags='modules firmwares',
-    description='Wow this is totally a fake driver.\nHooray for this',
-    repo='/repo/path/to'
-)
-fake_enhancement = Driver(
-    source='/repo/path/to/fake-enhancement-1.0-1.rpm',
-    name='fake-enhancement',
-    flags='binaries libraries',
-    description='This is enhancing the crap out of the installer.\n\nYeah.',
-    repo=fake_module.repo
-)
-def dd_list_output(driver):
-    out='{0.source}\n{0.name}\n{0.flags}\n{0.description}\n---\n'.format(driver)
-    return out.encode('utf-8')
-
-class DDUtilsTestCase(unittest.TestCase):
-    @mock.patch("driver_updates.subprocess.check_output")
-    def test_dd_list(self, check_output):
-        """dd_list: returns a list of Driver objects parsed from output"""
-        output = dd_list_output(fake_module)+dd_list_output(fake_enhancement)
-        check_output.return_value = output
-        anaconda, kernel = '19.0', os.uname()[2]
-        result = dd_list(fake_module.repo)
-        cmd = check_output.call_args[0][0]
-        self.assertIn(kernel, cmd)
-        self.assertIn(anaconda, cmd)
-        self.assertIn(fake_module.repo, cmd)
-        self.assertTrue(cmd[0].endswith("dd_list"))
-        self.assertEqual(len(result), 2)
-        mod, enh = sorted(result, key=lambda d: d.name)
-        self.assertEqual(mod.__dict__, fake_module.__dict__)
-        self.assertEqual(enh.__dict__, fake_enhancement.__dict__)
-
-    @mock.patch("driver_updates.subprocess.check_output")
-    def test_dd_extract(self, check_output):
-        """dd_extract: call binary with expected arguments"""
-        rpm = "/some/kind/of/path.rpm"
-        outdir = "/output/dir"
-        dd_extract(rpm, outdir)
-        cmd = check_output.call_args[0][0]
-        self.assertIn(os.uname()[2], cmd)
-        self.assertIn(rpm, cmd)
-        self.assertIn(outdir, cmd)
-        self.assertIn("-blmf", cmd)
-        self.assertTrue(cmd[0].endswith("dd_extract"))
-
-from driver_updates import extract_drivers, grab_driver_files, load_drivers
-
-class ExtractDriversTestCase(unittest.TestCase):
-    @mock.patch("driver_updates.save_repo")
-    @mock.patch("driver_updates.append_line")
-    @mock.patch("driver_updates.dd_extract")
-    def test_drivers(self, mock_extract, mock_append, mock_save):
-        """extract_drivers: save repo, write pkglist"""
-        extract_drivers(drivers=[fake_enhancement, fake_module])
-        # extracts all listed modules
-        mock_extract.assert_has_calls([
-            mock.call(fake_enhancement.source, "/updates"),
-            mock.call(fake_module.source, "/updates")
-        ], any_order=True)
-        pkglist = "/run/install/dd_packages"
-        mock_append.assert_called_once_with(pkglist, fake_module.name)
-        mock_save.assert_called_once_with(fake_module.repo)
-
-    @mock.patch("driver_updates.save_repo")
-    @mock.patch("driver_updates.append_line")
-    @mock.patch("driver_updates.dd_extract")
-    def test_enhancements(self, mock_extract, mock_append, mock_save):
-        """extract_drivers: extract selected drivers, don't save enhancements"""
-        extract_drivers(drivers=[fake_enhancement])
-        mock_extract.assert_called_once_with(
-            fake_enhancement.source, "/updates"
-        )
-        self.assertFalse(mock_append.called)
-        self.assertFalse(mock_save.called)
-
-    @mock.patch("driver_updates.save_repo")
-    @mock.patch("driver_updates.append_line")
-    @mock.patch("driver_updates.dd_extract")
-    def test_repo(self, mock_extract, mock_append, mock_save):
-        """extract_drivers(repos=[...]) extracts all drivers from named repos"""
-        with mock.patch("driver_updates.dd_list", side_effect=[
-            [fake_enhancement],
-            [fake_enhancement, fake_module]]):
-            extract_drivers(repos=['enh_repo', 'mod_repo'])
-        mock_extract.assert_has_calls([
-            mock.call(fake_enhancement.source, "/updates"),
-            mock.call(fake_enhancement.source, "/updates"),
-            mock.call(fake_module.source, "/updates")
-        ])
-        pkglist = "/run/install/dd_packages"
-        mock_append.assert_called_once_with(pkglist, fake_module.name)
-        mock_save.assert_called_once_with(fake_module.repo)
-
-class GrabDriverFilesTestCase(FileTestCaseBase):
-    def test_basic(self):
-        """grab_driver_files: copy drivers into place, return module list"""
-        # create a bunch of fake extracted files
-        outdir = self.tmpdir + '/extract-outdir'
-        moddir = outdir + "/lib/modules/%s/kernel/" % os.uname()[2]
-        fwdir = outdir + "/lib/firmware/"
-        modules = makefiles(moddir+"net/funk.ko", moddir+"fs/lolfs.ko.xz")
-        firmware = makefiles(fwdir+"funk.fw")
-        makefiles(outdir+"/usr/bin/monkey", outdir+"/other/dir/blah.ko")
-        mod_upd_dir = self.tmpdir+'/module-updates'
-        fw_upd_dir = self.tmpdir+'/fw-updates'
-        # use our updates dirs instead of the default updates dirs
-        with mock.patch.multiple("driver_updates",
-                                 module_updates_dir=mod_upd_dir,
-                                 firmware_updates_dir=fw_upd_dir):
-            modnames = grab_driver_files(outdir)
-        self.assertEqual(set(modnames), set(["funk", "lolfs"]))
-        modfiles = set(['funk.ko', 'lolfs.ko.xz'])
-        fwfiles = set(['funk.fw'])
-        # modules/firmware are *not* in their old locations
-        self.assertEqual([f for f in modules+firmware if os.path.exists(f)], [])
-        # modules are in the system's updates dir
-        self.assertEqual(set(os.listdir(mod_upd_dir)), modfiles)
-        # modules are also in outdir's updates dir
-        self.assertEqual(set(os.listdir(outdir+'/'+mod_upd_dir)), modfiles)
-        # repeat for firmware
-        self.assertEqual(set(os.listdir(fw_upd_dir)), fwfiles)
-        self.assertEqual(set(os.listdir(outdir+'/'+fw_upd_dir)), fwfiles)
-
-class LoadDriversTestCase(unittest.TestCase):
-    @mock.patch("driver_updates.subprocess.call")
-    def test_basic(self, call):
-        """load_drivers: runs depmod and modprobes all named modules"""
-        modnames = ['mod1', 'mod2']
-        load_drivers(modnames)
-        call.assert_has_calls([
-            mock.call(["depmod", "-a"]),
-            mock.call(["modprobe", "-a"] + modnames)
-        ])
-
-from driver_updates import process_driver_disk
-class ProcessDriverDiskTestCase(unittest.TestCase):
-    def setUp(self):
-        # an iterable that returns fake mountpoints, for mocking mount()
-        self.fakemount = ["/mnt/DD-%i" % n for n in range(1,10)]
-        # an iterable that returns fake repos, for mocking find_repos()
-        self.frepo = {
-            '/mnt/DD-1': ['/mnt/DD-1/repo1'],
-            '/mnt/DD-2': ['/mnt/DD-2/repo1', '/mnt/DD-2/repo2'],
-        }
-        # fake iso listings for iso_dir
-        self.fiso = {
-            '/mnt/DD-1': [],
-            '/mnt/DD-2': [],
-            '/mnt/DD-3': [],
-        }
-        # a context-manager object to be returned by the mock mounted()
-        mounted_ctx = mock.MagicMock(
-            __enter__=mock.MagicMock(side_effect=self.fakemount), # mount
-            __exit__=mock.MagicMock(return_value=None),           # umount
-        )
-        self.modlist = []
-        # set up our patches
-        patches = (
-            mock.patch("driver_updates.mounted", return_value=mounted_ctx),
-            mock.patch("driver_updates.find_repos", side_effect=self.frepo.get),
-            mock.patch("driver_updates.find_isos", side_effect=self.fiso.get),
-            mock.patch("driver_updates.extract_drivers", return_value=True),
-            mock.patch("driver_updates.load_drivers"),
-            mock.patch('driver_updates.grab_driver_files',
-                                side_effect=lambda: self.modlist),
-        )
-        self.mocks = {p.attribute:p.start() for p in patches}
-        for p in patches: self.addCleanup(p.stop)
-
-    def test_basic(self):
-        """process_driver_disk: mount disk, extract RPMs, grab + load drivers"""
-        dev = '/dev/fake'
-        process_driver_disk(dev)
-        # did we mount the initial device, and then the .iso we find therein?
-        self.mocks['mounted'].assert_called_once_with(dev)
-        self.mocks['extract_drivers'].assert_called_once_with(repos=self.frepo['/mnt/DD-1'])
-        self.mocks['grab_driver_files'].assert_called_once_with()
-        self.mocks['load_drivers'].assert_called_once_with(self.modlist)
-
-    def test_recursive(self):
-        """process_driver_disk: recursively process .isos at toplevel"""
-        dev = '/dev/fake'
-        # first mount has no repos, but an iso
-        self.frepo['/mnt/DD-1'] = []
-        self.fiso['/mnt/DD-1'].append('magic.iso')
-        self.fiso['/mnt/DD-2'].append('ignored.iso')
-        process_driver_disk(dev)
-        # did we mount the initial device, and the iso therein?
-        # also: we ignore ignored.iso because magic.iso is a proper DD
-        self.mocks['mounted'].assert_has_calls([
-            mock.call(dev), mock.call('magic.iso')
-        ])
-        # we extracted drivers from the repo(s) in magic.iso
-        self.mocks['extract_drivers'].assert_called_once_with(repos=self.frepo['/mnt/DD-2'])
-        self.mocks['grab_driver_files'].assert_called_once_with()
-        self.mocks['load_drivers'].assert_called_once_with(self.modlist)
-
-    def test_no_drivers(self):
-        """process_driver_disk: don't run depmod etc. if no new drivers"""
-        dev = '/dev/fake'
-        self.mocks['extract_drivers'].return_value = False
-        process_driver_disk(dev)
-        self.assertFalse(self.mocks['grab_driver_files'].called)
-        self.assertFalse(self.mocks['load_drivers'].called)
-
-from driver_updates import finish, mark_finished, all_finished
-
-class FinishedTestCase(FileTestCaseBase):
-    def test_mark_finished(self):
-        """mark_finished: appends a line to /tmp/dd_finished"""
-        requeststr = "WOW SOMETHING OR OTHER"
-        mark_finished(requeststr, topdir=self.tmpdir)
-        finished = self.tmpdir+'/dd_finished'
-        self.assertTrue(os.path.exists(finished))
-        self.assertEqual(read_lines(finished), [requeststr])
-
-    def test_all_finished(self):
-        """all_finished: True if all lines from dd_todo are in dd_finished"""
-        todo = self.tmpdir+'/dd_todo'
-        requests = ['one', 'two', 'final thingy']
-        with open(todo, 'w') as outf:
-            outf.write(''.join(r+'\n' for r in requests))
-        self.assertEqual(set(read_lines(todo)), set(requests))
-        for r in reversed(requests):
-            self.assertFalse(all_finished(topdir=self.tmpdir))
-            mark_finished(r, topdir=self.tmpdir)
-        self.assertTrue(all_finished(topdir=self.tmpdir))
-
-    def test_extra_finished(self):
-        """all_finished: True if dd_finished has more items than dd_todo"""
-        self.test_all_finished()
-        mark_finished("BONUS", topdir=self.tmpdir)
-        self.assertTrue(all_finished(topdir=self.tmpdir))
-
-    def test_finish(self):
-        """finish: mark request finished, and write dd.done if all complete"""
-        todo = self.tmpdir+'/dd_todo'
-        done = self.tmpdir+'/dd.done'
-        requests = ['one', 'two', 'final thingy']
-        with open(todo, 'w') as outf:
-            outf.write(''.join(r+'\n' for r in requests))
-        for r in reversed(requests):
-            print("marking %s" % r)
-            self.assertFalse(os.path.exists(done))
-            finish(r, topdir=self.tmpdir)
-        self.assertTrue(os.path.exists(done))
-
-from driver_updates import get_deviceinfo, DeviceInfo
-blkid_output = b'''\
-DEVNAME=/dev/sda2
-UUID=0f21a3d1-dcd3-4ab4-a292-c5556850d561
-TYPE=ext4
-
-DEVNAME=/dev/sda1
-UUID=C53C-EE46
-TYPE=vfat
-
-DEVNAME=/dev/sda3
-UUID=4126dbb6-c7d3-47b4-b1fc-9bb461df0067
-TYPE=btrfs
-
-DEVNAME=/dev/loop0
-UUID=6f16967e-0388-4276-bd8d-b88e5b217a55
-TYPE=ext4
-'''
-disk_labels = {
-    '/dev/sdb1': 'metroid_srv',
-    '/dev/loop0': 'I\\x20\u262d\\x20COMMUNISM',
-    '/dev/sda3': 'metroid_root'
-}
-devicelist = [
-    DeviceInfo(DEVNAME='/dev/sda2', TYPE='ext4',
-               UUID='0f21a3d1-dcd3-4ab4-a292-c5556850d561'),
-    DeviceInfo(DEVNAME='/dev/sda1', TYPE='vfat',
-               UUID='C53C-EE46'),
-    DeviceInfo(DEVNAME='/dev/sda3', TYPE='btrfs', LABEL='metroid_root',
-               UUID='4126dbb6-c7d3-47b4-b1fc-9bb461df0067'),
-    DeviceInfo(DEVNAME='/dev/loop0', TYPE='ext4',
-               LABEL='I\\x20\u262d\\x20COMMUNISM',
-               UUID='6f16967e-0388-4276-bd8d-b88e5b217a55'),
-]
-# also covers blkid, get_disk_labels, DeviceInfo
-class DeviceInfoTestCase(unittest.TestCase):
-    @mock.patch('driver_updates.subprocess.check_output')
-    @mock.patch('driver_updates.get_disk_labels')
-    def test_basic(self, get_disk_labels, check_output):
-        """get_deviceinfo: parses DeviceInfo from blkid etc."""
-        # configure mock objects
-        check_output.return_value = blkid_output
-        get_disk_labels.return_value = disk_labels
-        # now we're getting mock deviceinfo, whee
-        disks = get_deviceinfo()
-        self.assertEqual(len(disks), 4)
-        disks.sort(key=lambda d: d.device)
-        loop, efi, boot, root = disks
-        self.assertEqual(vars(boot), vars(devicelist[0]))
-        self.assertEqual(vars(efi), vars(devicelist[1]))
-        self.assertEqual(vars(root), vars(devicelist[2]))
-        self.assertEqual(vars(loop), vars(devicelist[3]))
-
-# TODO: test TextMenu itself
-
-# py2/3 compat
-import sys
-if sys.version_info.major == 3:
-    from io import StringIO
-else:
-    from io import BytesIO as StringIO
-
-from driver_updates import device_menu
-class DeviceMenuTestCase(unittest.TestCase):
-    def setUp(self):
-        patches = (
-            mock.patch('driver_updates.get_deviceinfo',return_value=devicelist),
-        )
-        self.mocks = {p.attribute:p.start() for p in patches}
-        for p in patches: self.addCleanup(p.stop)
-
-    def test_device_menu_exit(self):
-        """device_menu: 'c' exits the menu"""
-        with mock.patch('driver_updates._input', side_effect=['c']):
-            dev = device_menu()
-        self.assertEqual(dev, [])
-        self.assertEqual(self.mocks['get_deviceinfo'].call_count, 1)
-
-    def test_device_menu_refresh(self):
-        """device_menu: 'r' makes the menu refresh"""
-        with mock.patch('driver_updates._input', side_effect=['r','c']):
-            device_menu()
-        self.assertEqual(self.mocks['get_deviceinfo'].call_count, 2)
-
-    @mock.patch("sys.stdout", new_callable=StringIO)
-    def test_device_menu(self, stdout):
-        """device_menu: choosing a number returns that Device"""
-        choose_num='2'
-        with mock.patch('driver_updates._input', return_value=choose_num):
-            result = device_menu()
-        # if you hit '2' you should get the corresponding device from the list
-        self.assertEqual(len(result), 1)
-        dev = result[0]
-        self.assertEqual(vars(dev), vars(devicelist[int(choose_num)-1]))
-        # find the corresponding line on-screen
-        screen = [l.strip() for l in stdout.getvalue().splitlines()]
-        match = [l for l in screen if l.startswith(choose_num+')')]
-        self.assertEqual(len(match), 1)
-        line = match.pop(0)
-        # the device name (at least) should be on this line
-        self.assertIn(os.path.basename(dev.device), line)
diff --git a/tests/dracut_tests/test_driver_updates.py b/tests/dracut_tests/test_driver_updates.py
new file mode 100644
index 0000000..9b87c73
--- /dev/null
+++ b/tests/dracut_tests/test_driver_updates.py
@@ -0,0 +1,631 @@
+# test_driver_updates.py - unittests for driver_updates.py
+
+import unittest
+try:
+    import unittest.mock as mock
+except ImportError:
+    import mock
+
+import os
+import tempfile
+import shutil
+
+import sys
+sys.path.append(os.path.normpath(os.path.dirname(__file__)+'/../../dracut'))
+
+from driver_updates import copy_files, move_files, iter_files, ensure_dir
+from driver_updates import append_line, mkdir_seq
+
+def touch(path):
+    try:
+        open(path, 'a')
+    except IOError as e:
+        if e.errno != 17: raise
+
+def makedir(path):
+    ensure_dir(path)
+    return path
+
+def makefile(path):
+    makedir(os.path.dirname(path))
+    touch(path)
+    return path
+
+def makefiles(*paths):
+    return [makefile(p) for p in paths]
+
+class FileTestCaseBase(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix="test_driver_updates.")
+        self.srcdir = self.tmpdir+'/src/'
+        self.destdir = self.tmpdir+'/dest/'
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def makefiles(self, *paths):
+        return [makefile(os.path.normpath(self.tmpdir+'/'+p)) for p in paths]
+
+class SelfTestCase(FileTestCaseBase):
+    def test_makefiles(self):
+        """check test helpers"""
+        filepaths = ["sub/dir/test.file", "testfile"]
+        self.makefiles(*filepaths)
+        for f in filepaths:
+            self.assertTrue(os.path.exists(self.tmpdir+'/'+f))
+
+class TestCopyFiles(FileTestCaseBase):
+    def test_basic(self):
+        """copy_file: copy files into destdir, leaving existing contents"""
+        files = self.makefiles("src/file1", "src/subdir/file2")
+        self.makefiles("dest/file3")
+        copy_files(files, self.destdir)
+        result = set(os.listdir(self.destdir))
+        self.assertEqual(result, set(["file1", "file2", "file3"]))
+
+    def test_overwrite(self):
+        """copy_file: overwrite files in destdir if they have the same name"""
+        src, dest = self.makefiles("src/file1", "dest/file1")
+        with open(src, 'w') as outf:
+            outf.write("srcfile")
+        with open(dest, 'w') as outf:
+            outf.write("destfile")
+        copy_files([src], self.destdir)
+        self.assertEqual(os.listdir(self.destdir), ["file1"])
+        self.assertEqual(open(dest).read(), "srcfile")
+
+    def test_samefile(self):
+        """copy_file: skip files already in destdir"""
+        (dest,) = self.makefiles("dest/file1")
+        with open(dest, 'w') as outf:
+            outf.write("destfile")
+        copy_files([dest], self.destdir)
+        self.assertEqual(os.listdir(self.destdir), ["file1"])
+        self.assertEqual(open(dest).read(), "destfile")
+
+    def test_copy_to_parent(self):
+        """copy_file: skip files in subdirs of destdir"""
+        files = self.makefiles("dest/subdir/file1")
+        copy_files(files, self.destdir)
+        self.assertEqual(list(iter_files(self.destdir)), files)
+
+class TestIterFiles(FileTestCaseBase):
+    def test_basic(self):
+        """iter_files: iterates over full paths to files under topdir"""
+        files = set(self.makefiles("src/file1", "dest/file2", "src/sub/file3"))
+        makedir(self.tmpdir+'/empty/dir')
+        result = set(iter_files(self.tmpdir))
+        self.assertEqual(files, result)
+
+    def test_pattern(self):
+        """iter_files: match filename against glob pattern"""
+        self.makefiles("src/file1.so", "src/sub.ko/file2")
+        goodfiles = set(self.makefiles("src/sub/file1.ko", "src/file2.ko.xz"))
+        result = set(iter_files(self.tmpdir, pattern="*.ko*"))
+        self.assertEqual(result, goodfiles)
+
+class TestMoveFiles(FileTestCaseBase):
+    def test_basic(self):
+        """move_files: move files to destdir"""
+        files = self.makefiles("src/file1", "src/subdir/file2")
+        move_files(files, self.destdir)
+        self.assertEqual(set(os.listdir(self.destdir)), set(["file1", "file2"]))
+        self.assertEqual(list(iter_files(self.srcdir)), [])
+
+    def test_overwrite(self):
+        """move_files: overwrite files with the same name"""
+        src, dest = self.makefiles("src/file1", "dest/file1")
+        with open(src, 'w') as outf:
+            outf.write("srcfile")
+        with open(dest, 'w') as outf:
+            outf.write("destfile")
+        move_files([src], self.destdir)
+        self.assertEqual(os.listdir(self.destdir), ["file1"])
+        self.assertEqual(open(dest).read(), "srcfile")
+        self.assertEqual(list(iter_files(self.srcdir)), [])
+
+    def test_samefile(self):
+        """move_files: leave files alone if they're already in destdir"""
+        (dest,) = self.makefiles("dest/file1")
+        with open(dest, 'w') as outf:
+            outf.write("destfile")
+        move_files([dest], self.destdir)
+        self.assertEqual(os.listdir(self.destdir), ["file1"])
+        self.assertEqual(open(dest).read(), "destfile")
+
+    def test_move_to_parent(self):
+        """move_files: leave files alone if they're in a subdir of destdir"""
+        files = set(self.makefiles("dest/subdir/file1", "dest/file2"))
+        move_files(files, self.destdir)
+        self.assertEqual(set(iter_files(self.destdir)), files)
+
+class TestAppendLine(FileTestCaseBase):
+    def test_empty(self):
+        """append_line: create file + append \\n when needed"""
+        line = "this is a line of text with no newline"
+        outfile = self.tmpdir+'/outfile'
+        append_line(outfile, line)
+        self.assertEqual(open(outfile).read(), line+'\n')
+
+    def test_append(self):
+        """append_line: adds a line to the end of an existing file"""
+        oldlines = ["line one", "line two", "and I'm line three"]
+        outfile = self.tmpdir+'/outfile'
+        with open(outfile, 'w') as outf:
+            for line in oldlines:
+                outf.write(line+'\n')
+        line = "this line contains a newline already\n"
+        append_line(outfile, line)
+        self.assertEqual(open(outfile).read(), '\n'.join(oldlines+[line]))
+
+from driver_updates import read_lines
+class TestReadLine(FileTestCaseBase):
+    def test_empty(self):
+        """read_lines: return [] for empty file"""
+        [empty] = self.makefiles("emptyfile")
+        self.assertEqual(read_lines(empty), [])
+
+    def test_missing(self):
+        """read_lines: return [] for missing file"""
+        self.assertEqual(read_lines(self.tmpdir+'/no-such-file'),[])
+
+    def test_readlines(self):
+        """read_lines: returns a list of lines without trailing newlines"""
+        filedata = 'line one\nline two\n\nline four\n'
+        outfile = self.tmpdir+'/outfile'
+        with open(outfile, 'w') as outf:
+            outf.write(filedata)
+        lines = read_lines(outfile)
+        self.assertEqual(lines, ['line one', 'line two','','line four'])
+
+    def test_readline_and_append_line(self):
+        """read_lines: returns items as passed to append_line"""
+        filename = self.tmpdir+'/outfile'
+        items = ["one", "two", "five"]
+        for i in items:
+            append_line(filename, i)
+        self.assertEqual(items, read_lines(filename))
+
+class TestMkdirSeq(FileTestCaseBase):
+    def test_basic(self):
+        """mkdir_seq: first dir ends with 1"""
+        newdir = mkdir_seq(self.srcdir+'/DD-')
+        self.assertEqual(newdir, self.srcdir+'/DD-1')
+        self.assertTrue(os.path.isdir(newdir))
+
+    def test_one_exists(self):
+        """mkdir_seq: increment number if file exists"""
+        firstdir = mkdir_seq(self.srcdir+'/DD-')
+        newdir = mkdir_seq(self.srcdir+'/DD-')
+        self.assertEqual(newdir, self.srcdir+'/DD-2')
+        self.assertTrue(os.path.isdir(newdir))
+        self.assertTrue(os.path.isdir(firstdir))
+
+from driver_updates import find_repos, save_repo, ARCH
+# As far as we know, this is what makes a valid repo: rhdd3 + rpms/`uname -m`/
+def makerepo(topdir, desc=None):
+    descfile = makefile(topdir+'/rhdd3')
+    if not desc:
+        desc = os.path.basename(topdir)
+    with open(descfile, "w") as outf:
+        outf.write(desc+"\n")
+    makedir(topdir+'/rpms/'+ARCH)
+
+class TestFindRepos(FileTestCaseBase):
+    def test_basic(self):
+        """find_repos: return RPM dir if a valid repo is found"""
+        makerepo(self.tmpdir)
+        repos = find_repos(self.tmpdir)
+        self.assertEqual(repos, [self.tmpdir+'/rpms/'+ARCH])
+        self.assertTrue(os.path.isdir(repos[0]))
+
+    def test_multiple_subdirs(self):
+        """find_repos: descend multiple subdirs if needed"""
+        makerepo(self.tmpdir+'/driver1')
+        makerepo(self.tmpdir+'/sub/driver1')
+        makerepo(self.tmpdir+'/sub/driver2')
+        repos = find_repos(self.tmpdir)
+        self.assertEqual(len(repos),3)
+
+class TestSaveRepo(FileTestCaseBase):
+    def test_basic(self):
+        """save_repo: copies a directory to /run/install/DD-X"""
+        makerepo(self.srcdir)
+        [repo] = find_repos(self.srcdir)
+        makefile(repo+'/fake-something.rpm')
+        saved = save_repo(repo, target=self.destdir)
+        self.assertEqual(set(os.listdir(saved)), set(["fake-something.rpm"]))
+        self.assertEqual(saved, os.path.join(self.destdir, "DD-1"))
+
+from driver_updates import mount, umount, mounted
+class MountTestCase(unittest.TestCase):
+    @mock.patch('driver_updates.mkdir_seq')
+    @mock.patch('driver_updates.subprocess.check_call')
+    def test_mkdir(self, check_call, mkdir):
+        """mount: makes mountpoint if needed"""
+        dev, mnt = '/dev/fake', '/media/DD-1'
+        mkdir.return_value = mnt
+        mountpoint = mount(dev)
+        mkdir.assert_called_once_with('/media/DD-')
+        check_call.assert_called_once_with(["mount", dev, mnt])
+        self.assertEqual(mnt, mountpoint)
+
+    @mock.patch('driver_updates.mkdir_seq')
+    @mock.patch('driver_updates.subprocess.check_call')
+    def test_basic(self, check_call, mkdir):
+        """mount: calls mount(8) to mount a device/image"""
+        dev, mnt = '/dev/fake', '/media/fake'
+        mount(dev, mnt)
+        check_call.assert_called_once_with(["mount", dev, mnt])
+        self.assertFalse(mkdir.called)
+
+    @mock.patch('driver_updates.subprocess.call')
+    def test_umount(self, call):
+        """umount: calls umount(8)"""
+        mnt = '/mnt/fake'
+        umount(mnt)
+        call.assert_called_once_with(["umount", mnt])
+
+    @mock.patch('driver_updates.mount')
+    @mock.patch('driver_updates.umount')
+    def test_mount_manager(self, mock_umount, mock_mount):
+        """mounted: context manager mounts/umounts as expected"""
+        dev, mnt = '/dev/fake', '/media/fake'
+        mock_mount.return_value = mnt
+        with mounted(dev, mnt) as mountpoint:
+            mock_mount.assert_called_once_with(dev, mnt)
+            self.assertFalse(mock_umount.called)
+            self.assertEqual(mountpoint, mnt)
+        mock_umount.assert_called_once_with(mnt)
+
+# NOTE: dd_list and dd_extract get tested pretty thoroughly in tests/dd_tests,
+# so this is a slightly higher-level test case
+from driver_updates import dd_list, dd_extract, Driver
+fake_module = Driver(
+    source='/repo/path/to/fake-driver-1.0-1.rpm',
+    name='fake-driver',
+    flags='modules firmwares',
+    description='Wow this is totally a fake driver.\nHooray for this',
+    repo='/repo/path/to'
+)
+fake_enhancement = Driver(
+    source='/repo/path/to/fake-enhancement-1.0-1.rpm',
+    name='fake-enhancement',
+    flags='binaries libraries',
+    description='This is enhancing the crap out of the installer.\n\nYeah.',
+    repo=fake_module.repo
+)
+def dd_list_output(driver):
+    out='{0.source}\n{0.name}\n{0.flags}\n{0.description}\n---\n'.format(driver)
+    return out.encode('utf-8')
+
+class DDUtilsTestCase(unittest.TestCase):
+    @mock.patch("driver_updates.subprocess.check_output")
+    def test_dd_list(self, check_output):
+        """dd_list: returns a list of Driver objects parsed from output"""
+        output = dd_list_output(fake_module)+dd_list_output(fake_enhancement)
+        check_output.return_value = output
+        anaconda, kernel = '19.0', os.uname()[2]
+        result = dd_list(fake_module.repo)
+        cmd = check_output.call_args[0][0]
+        self.assertIn(kernel, cmd)
+        self.assertIn(anaconda, cmd)
+        self.assertIn(fake_module.repo, cmd)
+        self.assertTrue(cmd[0].endswith("dd_list"))
+        self.assertEqual(len(result), 2)
+        mod, enh = sorted(result, key=lambda d: d.name)
+        self.assertEqual(mod.__dict__, fake_module.__dict__)
+        self.assertEqual(enh.__dict__, fake_enhancement.__dict__)
+
+    @mock.patch("driver_updates.subprocess.check_output")
+    def test_dd_extract(self, check_output):
+        """dd_extract: call binary with expected arguments"""
+        rpm = "/some/kind/of/path.rpm"
+        outdir = "/output/dir"
+        dd_extract(rpm, outdir)
+        cmd = check_output.call_args[0][0]
+        self.assertIn(os.uname()[2], cmd)
+        self.assertIn(rpm, cmd)
+        self.assertIn(outdir, cmd)
+        self.assertIn("-blmf", cmd)
+        self.assertTrue(cmd[0].endswith("dd_extract"))
+
+from driver_updates import extract_drivers, grab_driver_files, load_drivers
+
+ at mock.patch("driver_updates.ensure_dir")
+ at mock.patch("driver_updates.save_repo")
+ at mock.patch("driver_updates.append_line")
+ at mock.patch("driver_updates.dd_extract")
+class ExtractDriversTestCase(unittest.TestCase):
+    def test_drivers(self, mock_extract, mock_append, mock_save, *args):
+        """extract_drivers: save repo, write pkglist"""
+        extract_drivers(drivers=[fake_enhancement, fake_module])
+        # extracts all listed modules
+        mock_extract.assert_has_calls([
+            mock.call(fake_enhancement.source, "/updates"),
+            mock.call(fake_module.source, "/updates")
+        ], any_order=True)
+        pkglist = "/run/install/dd_packages"
+        mock_append.assert_called_once_with(pkglist, fake_module.name)
+        mock_save.assert_called_once_with(fake_module.repo)
+
+    def test_enhancements(self, mock_extract, mock_append, mock_save, *args):
+        """extract_drivers: extract selected drivers, don't save enhancements"""
+        extract_drivers(drivers=[fake_enhancement])
+        mock_extract.assert_called_once_with(
+            fake_enhancement.source, "/updates"
+        )
+        self.assertFalse(mock_append.called)
+        self.assertFalse(mock_save.called)
+
+    def test_repo(self, mock_extract, mock_append, mock_save, *args):
+        """extract_drivers(repos=[...]) extracts all drivers from named repos"""
+        with mock.patch("driver_updates.dd_list", side_effect=[
+            [fake_enhancement],
+            [fake_enhancement, fake_module]]):
+            extract_drivers(repos=['enh_repo', 'mod_repo'])
+        mock_extract.assert_has_calls([
+            mock.call(fake_enhancement.source, "/updates"),
+            mock.call(fake_enhancement.source, "/updates"),
+            mock.call(fake_module.source, "/updates")
+        ])
+        pkglist = "/run/install/dd_packages"
+        mock_append.assert_called_once_with(pkglist, fake_module.name)
+        mock_save.assert_called_once_with(fake_module.repo)
+
+class GrabDriverFilesTestCase(FileTestCaseBase):
+    def test_basic(self):
+        """grab_driver_files: copy drivers into place, return module list"""
+        # create a bunch of fake extracted files
+        outdir = self.tmpdir + '/extract-outdir'
+        moddir = outdir + "/lib/modules/%s/kernel/" % os.uname()[2]
+        fwdir = outdir + "/lib/firmware/"
+        modules = makefiles(moddir+"net/funk.ko", moddir+"fs/lolfs.ko.xz")
+        firmware = makefiles(fwdir+"funk.fw")
+        makefiles(outdir+"/usr/bin/monkey", outdir+"/other/dir/blah.ko")
+        mod_upd_dir = self.tmpdir+'/module-updates'
+        fw_upd_dir = self.tmpdir+'/fw-updates'
+        # use our updates dirs instead of the default updates dirs
+        with mock.patch.multiple("driver_updates",
+                                 MODULE_UPDATES_DIR=mod_upd_dir,
+                                 FIRMWARE_UPDATES_DIR=fw_upd_dir):
+            modnames = grab_driver_files(outdir)
+        self.assertEqual(set(modnames), set(["funk", "lolfs"]))
+        modfiles = set(['funk.ko', 'lolfs.ko.xz'])
+        fwfiles = set(['funk.fw'])
+        # modules/firmware are *not* in their old locations
+        self.assertEqual([f for f in modules+firmware if os.path.exists(f)], [])
+        # modules are in the system's updates dir
+        self.assertEqual(set(os.listdir(mod_upd_dir)), modfiles)
+        # modules are also in outdir's updates dir
+        self.assertEqual(set(os.listdir(outdir+'/'+mod_upd_dir)), modfiles)
+        # repeat for firmware
+        self.assertEqual(set(os.listdir(fw_upd_dir)), fwfiles)
+        self.assertEqual(set(os.listdir(outdir+'/'+fw_upd_dir)), fwfiles)
+
+class LoadDriversTestCase(unittest.TestCase):
+    @mock.patch("driver_updates.subprocess.call")
+    def test_basic(self, call):
+        """load_drivers: runs depmod and modprobes all named modules"""
+        modnames = ['mod1', 'mod2']
+        load_drivers(modnames)
+        call.assert_has_calls([
+            mock.call(["depmod", "-a"]),
+            mock.call(["modprobe", "-a"] + modnames)
+        ])
+
+from driver_updates import process_driver_disk
+class ProcessDriverDiskTestCase(unittest.TestCase):
+    def setUp(self):
+        # an iterable that returns fake mountpoints, for mocking mount()
+        self.fakemount = ["/mnt/DD-%i" % n for n in range(1,10)]
+        # an iterable that returns fake repos, for mocking find_repos()
+        self.frepo = {
+            '/mnt/DD-1': ['/mnt/DD-1/repo1'],
+            '/mnt/DD-2': ['/mnt/DD-2/repo1', '/mnt/DD-2/repo2'],
+        }
+        # fake iso listings for iso_dir
+        self.fiso = {
+            '/mnt/DD-1': [],
+            '/mnt/DD-2': [],
+            '/mnt/DD-3': [],
+        }
+        # a context-manager object to be returned by the mock mounted()
+        mounted_ctx = mock.MagicMock(
+            __enter__=mock.MagicMock(side_effect=self.fakemount), # mount
+            __exit__=mock.MagicMock(return_value=None),           # umount
+        )
+        self.modlist = []
+        # set up our patches
+        patches = (
+            mock.patch("driver_updates.mounted", return_value=mounted_ctx),
+            mock.patch("driver_updates.find_repos", side_effect=self.frepo.get),
+            mock.patch("driver_updates.find_isos", side_effect=self.fiso.get),
+            mock.patch("driver_updates.extract_drivers", return_value=True),
+            mock.patch("driver_updates.load_drivers"),
+            mock.patch('driver_updates.grab_driver_files',
+                                side_effect=lambda: self.modlist),
+        )
+        self.mocks = {p.attribute:p.start() for p in patches}
+        for p in patches: self.addCleanup(p.stop)
+
+    def test_basic(self):
+        """process_driver_disk: mount disk, extract RPMs, grab + load drivers"""
+        dev = '/dev/fake'
+        process_driver_disk(dev)
+        # did we mount the initial device, and then the .iso we find therein?
+        self.mocks['mounted'].assert_called_once_with(dev)
+        self.mocks['extract_drivers'].assert_called_once_with(repos=self.frepo['/mnt/DD-1'])
+        self.mocks['grab_driver_files'].assert_called_once_with()
+        self.mocks['load_drivers'].assert_called_once_with(self.modlist)
+
+    def test_recursive(self):
+        """process_driver_disk: recursively process .isos at toplevel"""
+        dev = '/dev/fake'
+        # first mount has no repos, but an iso
+        self.frepo['/mnt/DD-1'] = []
+        self.fiso['/mnt/DD-1'].append('magic.iso')
+        self.fiso['/mnt/DD-2'].append('ignored.iso')
+        process_driver_disk(dev)
+        # did we mount the initial device, and the iso therein?
+        # also: we ignore ignored.iso because magic.iso is a proper DD
+        self.mocks['mounted'].assert_has_calls([
+            mock.call(dev), mock.call('magic.iso')
+        ])
+        # we extracted drivers from the repo(s) in magic.iso
+        self.mocks['extract_drivers'].assert_called_once_with(repos=self.frepo['/mnt/DD-2'])
+        self.mocks['grab_driver_files'].assert_called_once_with()
+        self.mocks['load_drivers'].assert_called_once_with(self.modlist)
+
+    def test_no_drivers(self):
+        """process_driver_disk: don't run depmod etc. if no new drivers"""
+        dev = '/dev/fake'
+        self.mocks['extract_drivers'].return_value = False
+        process_driver_disk(dev)
+        self.assertFalse(self.mocks['grab_driver_files'].called)
+        self.assertFalse(self.mocks['load_drivers'].called)
+
+from driver_updates import finish, mark_finished, all_finished
+
+class FinishedTestCase(FileTestCaseBase):
+    def test_mark_finished(self):
+        """mark_finished: appends a line to /tmp/dd_finished"""
+        requeststr = "WOW SOMETHING OR OTHER"
+        mark_finished(requeststr, topdir=self.tmpdir)
+        finished = self.tmpdir+'/dd_finished'
+        self.assertTrue(os.path.exists(finished))
+        self.assertEqual(read_lines(finished), [requeststr])
+
+    def test_all_finished(self):
+        """all_finished: True if all lines from dd_todo are in dd_finished"""
+        todo = self.tmpdir+'/dd_todo'
+        requests = ['one', 'two', 'final thingy']
+        with open(todo, 'w') as outf:
+            outf.write(''.join(r+'\n' for r in requests))
+        self.assertEqual(set(read_lines(todo)), set(requests))
+        for r in reversed(requests):
+            self.assertFalse(all_finished(topdir=self.tmpdir))
+            mark_finished(r, topdir=self.tmpdir)
+        self.assertTrue(all_finished(topdir=self.tmpdir))
+
+    def test_extra_finished(self):
+        """all_finished: True if dd_finished has more items than dd_todo"""
+        self.test_all_finished()
+        mark_finished("BONUS", topdir=self.tmpdir)
+        self.assertTrue(all_finished(topdir=self.tmpdir))
+
+    def test_finish(self):
+        """finish: mark request finished, and write dd.done if all complete"""
+        todo = self.tmpdir+'/dd_todo'
+        done = self.tmpdir+'/dd.done'
+        requests = ['one', 'two', 'final thingy']
+        with open(todo, 'w') as outf:
+            outf.write(''.join(r+'\n' for r in requests))
+        for r in reversed(requests):
+            print("marking %s" % r)
+            self.assertFalse(os.path.exists(done))
+            finish(r, topdir=self.tmpdir)
+        self.assertTrue(os.path.exists(done))
+
+from driver_updates import get_deviceinfo, DeviceInfo
+blkid_out = b'''\
+DEVNAME=/dev/sda2
+UUID=0f21a3d1-dcd3-4ab4-a292-c5556850d561
+TYPE=ext4
+
+DEVNAME=/dev/sda1
+UUID=C53C-EE46
+TYPE=vfat
+
+DEVNAME=/dev/sda3
+UUID=4126dbb6-c7d3-47b4-b1fc-9bb461df0067
+TYPE=btrfs
+
+DEVNAME=/dev/loop0
+UUID=6f16967e-0388-4276-bd8d-b88e5b217a55
+TYPE=ext4
+'''
+disk_labels = {
+    '/dev/sdb1': 'metroid_srv',
+    '/dev/loop0': 'I\\x20\u262d\\x20COMMUNISM',
+    '/dev/sda3': 'metroid_root'
+}
+devicelist = [
+    DeviceInfo(DEVNAME='/dev/sda2', TYPE='ext4',
+               UUID='0f21a3d1-dcd3-4ab4-a292-c5556850d561'),
+    DeviceInfo(DEVNAME='/dev/sda1', TYPE='vfat',
+               UUID='C53C-EE46'),
+    DeviceInfo(DEVNAME='/dev/sda3', TYPE='btrfs', LABEL='metroid_root',
+               UUID='4126dbb6-c7d3-47b4-b1fc-9bb461df0067'),
+    DeviceInfo(DEVNAME='/dev/loop0', TYPE='ext4',
+               LABEL='I\\x20\u262d\\x20COMMUNISM',
+               UUID='6f16967e-0388-4276-bd8d-b88e5b217a55'),
+]
+# also covers blkid, get_disk_labels, DeviceInfo
+class DeviceInfoTestCase(unittest.TestCase):
+    @mock.patch('driver_updates.subprocess.check_output',return_value=blkid_out)
+    @mock.patch('driver_updates.get_disk_labels',return_value=disk_labels)
+    def test_basic(self, get_disk_labels, check_output):
+        """get_deviceinfo: parses DeviceInfo from blkid etc."""
+        disks = get_deviceinfo()
+        self.assertEqual(len(disks), 4)
+        disks.sort(key=lambda d: d.device)
+        loop, efi, boot, root = disks
+        self.assertEqual(vars(boot), vars(devicelist[0]))
+        self.assertEqual(vars(efi), vars(devicelist[1]))
+        self.assertEqual(vars(root), vars(devicelist[2]))
+        self.assertEqual(vars(loop), vars(devicelist[3]))
+
+    def test_shortdev(self):
+        d = DeviceInfo(DEVNAME="/dev/disk/by-label/OEMDRV")
+        with mock.patch("os.path.realpath", return_value="/dev/i2o/hdb"):
+            self.assertEqual(d.shortdev, "i2o/hdb")
+
+# TODO: test TextMenu itself
+
+# py2/3 compat
+import sys
+if sys.version_info.major == 3:
+    from io import StringIO
+else:
+    from io import BytesIO as StringIO
+
+from driver_updates import device_menu
+class DeviceMenuTestCase(unittest.TestCase):
+    def setUp(self):
+        patches = (
+            mock.patch('driver_updates.get_deviceinfo',return_value=devicelist),
+        )
+        self.mocks = {p.attribute:p.start() for p in patches}
+        for p in patches: self.addCleanup(p.stop)
+
+    def test_device_menu_exit(self):
+        """device_menu: 'c' exits the menu"""
+        with mock.patch('driver_updates._input', side_effect=['c']):
+            dev = device_menu()
+        self.assertEqual(dev, [])
+        self.assertEqual(self.mocks['get_deviceinfo'].call_count, 1)
+
+    def test_device_menu_refresh(self):
+        """device_menu: 'r' makes the menu refresh"""
+        with mock.patch('driver_updates._input', side_effect=['r','c']):
+            device_menu()
+        self.assertEqual(self.mocks['get_deviceinfo'].call_count, 2)
+
+    @mock.patch("sys.stdout", new_callable=StringIO)
+    def test_device_menu(self, stdout):
+        """device_menu: choosing a number returns that Device"""
+        choose_num='2'
+        with mock.patch('driver_updates._input', return_value=choose_num):
+            result = device_menu()
+        # if you hit '2' you should get the corresponding device from the list
+        self.assertEqual(len(result), 1)
+        dev = result[0]
+        self.assertEqual(vars(dev), vars(devicelist[int(choose_num)-1]))
+        # find the corresponding line on-screen
+        screen = [l.strip() for l in stdout.getvalue().splitlines()]
+        match = [l for l in screen if l.startswith(choose_num+')')]
+        self.assertEqual(len(match), 1)
+        line = match.pop(0)
+        # the device name (at least) should be on this line
+        self.assertIn(os.path.basename(dev.device), line)
diff --git a/tests/nosetests.sh b/tests/nosetests.sh
index 6db3c3d..e0b1c1e 100755
--- a/tests/nosetests.sh
+++ b/tests/nosetests.sh
@@ -8,7 +8,7 @@ fi
 
 # If no tests were selected, select all of them
 if [ $# -eq 0 ]; then
-    set -- "${top_srcdir}"/tests/*_tests "${top_srcdir}"/dracut/test_*.py
+    set -- "${top_srcdir}"/tests/*_tests
 fi
 
 exec nosetests -v --exclude=logpicker -a \!acceptance,\!slow "$@"


-- 
To view this commit on github, visit https://github.com/rhinstaller/anaconda/commit/e375b65d8981e40f7e2cf75d088b0b6ddbebd271


More information about the anaconda-patches mailing list