[master 5/6] dd-refactor: new driver_updates.py + tests

wgwoods installerbot-noreply at redhat.com
Tue Jul 7 18:28:43 UTC 2015


From: Will Woods <wwoods at redhat.com>

This commit adds a new, rewritten `driver_updates.py`, along with some
driverdisk tests.

It also adds `tests/lib/mkdud.py`, which is a helper tool for making
driver disk images.
---
 dracut/driver_updates.py                 | 576 ++++++++++++++++++++++++++++
 dracut/test_driver_updates.py            | 624 +++++++++++++++++++++++++++++++
 tests/kickstart_tests/driverdisk-disk.ks |  41 ++
 tests/kickstart_tests/driverdisk-disk.sh |  33 ++
 tests/lib/mkdud.py                       | 120 ++++++
 5 files changed, 1394 insertions(+)
 create mode 100755 dracut/driver_updates.py
 create mode 100644 dracut/test_driver_updates.py
 create mode 100644 tests/kickstart_tests/driverdisk-disk.ks
 create mode 100755 tests/kickstart_tests/driverdisk-disk.sh
 create mode 100755 tests/lib/mkdud.py

diff --git a/dracut/driver_updates.py b/dracut/driver_updates.py
new file mode 100755
index 0000000..4c331d5
--- /dev/null
+++ b/dracut/driver_updates.py
@@ -0,0 +1,576 @@
+#!/usr/bin/python
+#
+# Copyright (C) 2015 by Red Hat, Inc.  All rights reserved.
+#
+# 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; either version 2 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/>.
+#
+# Author(s):
+#   Brian C. Lane <bcl at brianlane.com>
+#   Will Woods <wwoods at redhat.com>
+#
+"""
+Driver Update Disk handler program.
+
+This will be called once for each requested driverdisk (non-interactive), and
+once for interactive mode (if requested).
+
+Usage is one of:
+
+    driver-updates --disk DISKSTR DEVNODE
+
+        DISKSTR is the string passed by the user ('/dev/sda3', 'LABEL=DD', etc.)
+        DEVNODE is the actual device node (/dev/sda3, /dev/sr0, etc.)
+
+    driver-updates --net URL LOCALFILE
+
+        URL is the string passed by the user ('http://.../something.iso')
+        LOCALFILE is the location of the downloaded file
+
+    driver-updates --interactive
+
+        The user will be presented with a menu where they can choose a disk
+        and pick individual drivers to install.
+
+/tmp/dd_net contains the list of URLs given by the user.
+/tmp/dd_disk contains the list of disk devices given by the user.
+/tmp/dd_interactive contains "menu" if interactive mode was requested.
+
+/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.
+
+Selected driver package names are saved in /run/install/dd_packages.
+
+Anaconda uses the repository and package list to install the same set of drivers
+to the target system.
+"""
+
+import logging
+import sys
+import os
+import subprocess
+import fnmatch
+import readline # pylint:disable=unused-import
+
+from contextlib import contextmanager
+from logging.handlers import SysLogHandler
+
+# py2 compat
+try:
+    from subprocess import DEVNULL
+except ImportError:
+    DEVNULL = open("/dev/null", 'a+')
+try:
+    _input = raw_input # pylint: disable=undefined-variable
+except NameError:
+    _input = input
+
+log = logging.getLogger("DD")
+
+arch = os.uname()[4]
+kernelver = os.uname()[2]
+
+def mkdir_seq(stem):
+    """
+    Create sequentially-numbered directories starting with stem.
+
+    For example, mkdir_seq("/tmp/DD-") would create "/tmp/DD-1";
+    if that already exists, try "/tmp/DD-2", "/tmp/DD-3", and so on,
+    until a directory is created.
+
+    Returns the newly-created directory name.
+    """
+    n = 1
+    while True:
+        dirname = str(stem) + str(n)
+        try:
+            os.makedirs(dirname)
+        except OSError as e:
+            if e.errno != 17: raise
+            n += 1
+        else:
+            return dirname
+
+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
+        if "rhdd3" in files and "rpms" in dirs and os.path.isdir(repo):
+            log.debug("found repo: %s", repo)
+            dd_repos.append(repo)
+    return dd_repos
+
+# NOTE: it's unclear whether or not we're supposed to recurse subdirs looking
+# for .iso files, but that seems like a bad idea if you mount some huge disk..
+# So I've made a judgement call: we only load .iso files from the toplevel.
+def find_isos(mnt):
+    """find files named '.iso' at the top level of mnt."""
+    return [mnt+'/'+f for f in os.listdir(mnt) if f.lower().endswith('.iso')]
+
+class Driver(object):
+    """Represents a single driver (rpm), as listed by dd_list"""
+    def __init__(self, source="", name="", flags="", description="", repo=""):
+        self.source = source
+        self.name = name
+        self.flags = flags
+        self.description = description
+        self.repo = 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'
+    if not kernel_ver:
+        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')
+    drivers = [Driver(*d.split('\n',3)) for d in out.split('\n---\n') if d]
+    log.debug("dd_list: found drivers: %s", ' '.join(d.name for d in drivers))
+    for d in drivers: d.repo = dd_path
+    return drivers
+
+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
+    cmd = ["dd_extract", flags, '-r', rpm_path, '-d', outdir, '-k', kernel_ver]
+    subprocess.check_output(cmd, stderr=DEVNULL) # discard stdout
+
+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
+    if not mnt:
+        mnt = mkdir_seq("/media/DD-")
+    cmd = ["mount", dev, mnt]
+    log.debug("mounting %s at %s", dev, mnt)
+    subprocess.check_call(cmd)
+    return mnt
+
+def umount(mnt):
+    log.debug("unmounting %s", mnt)
+    subprocess.call(["umount", mnt])
+
+ at contextmanager
+def mounted(dev, mnt=None):
+    mnt = mount(dev, mnt)
+    try:
+        yield mnt
+    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):
+        for f in files:
+            if pattern is None or fnmatch.fnmatch(f, pattern):
+                yield os.path.join(head, f)
+
+def ensure_dir(d):
+    """make sure the given directory exists."""
+    subprocess.check_call(["mkdir", "-p", d])
+
+def move_files(files, destdir):
+    """move files into destdir (iff they're not already under destdir)"""
+    ensure_dir(destdir)
+    for f in files:
+        if f.startswith(destdir):
+            continue
+        subprocess.call(["mv", "-f", f, destdir])
+
+def copy_files(files, destdir):
+    """copy files into destdir (iff they're not already under destdir)"""
+    ensure_dir(destdir)
+    for f in files:
+        if f.startswith(destdir):
+            continue
+        subprocess.call(["cp", "-a", f, destdir])
+
+def append_line(filename, line):
+    """simple helper to append a line to a file"""
+    if not line.endswith("\n"):
+        line += "\n"
+    with open(filename, 'a') as outf:
+        outf.write(line)
+
+def read_lines(filename):
+    try:
+        return open(filename).read().splitlines()
+    except IOError:
+        return []
+
+def save_repo(repo, target="/run/install"):
+    """copy a repo to the place where the installer will look for it later."""
+    newdir = mkdir_seq(os.path.join(target, "DD-"))
+    log.debug("save_repo: copying %s to %s", repo, newdir)
+    subprocess.call(["cp", "-arT", repo, newdir])
+    return newdir
+
+def extract_drivers(drivers=None, repos=None, outdir="/updates",
+                    pkglist="/run/install/dd_packages"):
+    """
+    Extract drivers - either a user-selected driver list or full repos.
+
+    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..)
+
+    If any packages containing modules or firmware are extracted, also:
+    * call save_repo for that package's repo
+    * write the package name(s) to pkglist.
+
+    Returns True if any package containing modules was extracted.
+    """
+    if not drivers:
+        drivers = []
+    if repos:
+        drivers += [d for repo in repos for d in dd_list(repo)]
+
+    save_repos = set()
+    new_drivers = False
+
+    ensure_dir(outdir)
+
+    for driver in drivers:
+        log.info("Extracting: %s", driver.name)
+        dd_extract(driver.source, outdir)
+        # Make sure we install modules/firmware into the target system
+        if 'modules' in driver.flags or 'firmwares' in driver.flags:
+            append_line(pkglist, driver.name)
+            save_repos.add(driver.repo)
+            new_drivers = True
+
+    # save the repos containing those packages
+    for repo in save_repos:
+        save_repo(repo)
+
+    return new_drivers
+
+def grab_driver_files(outdir="/updates"):
+    """
+    copy any modules/firmware we just extracted into the running system.
+    return a list of the names of any modules we just copied.
+    """
+    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)
+    return [os.path.basename(m).split('.ko')[0] for m in modules]
+
+def load_drivers(modnames):
+    """run depmod and try to modprobe all the given module names."""
+    log.debug("load_drivers: %s", modnames)
+    subprocess.call(["depmod", "-a"])
+    subprocess.call(["modprobe", "-a"] + modnames)
+
+def process_driver_disk(dev, interactive=False):
+    try:
+        _process_driver_disk(dev, interactive=interactive)
+    except (subprocess.CalledProcessError, IOError) as e:
+        log.error("ERROR: %s", e)
+
+def _process_driver_disk(dev, interactive=False):
+    """
+    Main entry point for processing a single driver disk.
+    Mount the device/image, find repos, and install drivers from those repos.
+
+    If there are no repos, look for .iso files, and (if present) recursively
+    process those.
+
+    If interactive, ask the user which driver(s) to install from the repos,
+    or ask which iso file to process (if no repos).
+    """
+    log.info("Examining %s", dev)
+    with mounted(dev) as mnt:
+        repos = find_repos(mnt)
+        isos = find_isos(mnt)
+
+        if repos:
+            if interactive:
+                new_modules = extract_drivers(drivers=repo_menu(repos))
+            else:
+                new_modules = extract_drivers(repos=repos)
+            if new_modules:
+                modules = grab_driver_files()
+                load_drivers(modules)
+        elif isos:
+            if interactive:
+                isos = iso_menu(isos)
+            for iso in isos:
+                process_driver_disk(iso, interactive=interactive)
+        else:
+            print("=== No driver disks found in %s! ===\n" % dev)
+
+def mark_finished(user_request, topdir="/tmp"):
+    log.debug("marking %s complete in %s", user_request, topdir)
+    append_line(topdir+"/dd_finished", user_request)
+
+def all_finished(topdir="/tmp"):
+    finished = read_lines(topdir+"/dd_finished")
+    todo = read_lines(topdir+"/dd_todo")
+    return all(r in finished for r in todo)
+
+def finish(user_request, topdir="/tmp"):
+    # mark that we've finished processing this request
+    mark_finished(user_request, topdir)
+    # if we're done now, let dracut know
+    if all_finished(topdir):
+        append_line(topdir+"/dd.done", "true")
+
+# --- DEVICE LISTING HELPERS FOR THE MENU -----------------------------------
+
+class DeviceInfo(object):
+    def __init__(self, **kwargs):
+        self.device = kwargs.get("DEVNAME", '')
+        self.uuid = kwargs.get("UUID", '')
+        self.fs_type = kwargs.get("TYPE", '')
+        self.label = kwargs.get("LABEL", '')
+
+    def __repr__(self):
+        return '<DeviceInfo %s>' % self.device
+
+    @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
+
+def blkid():
+    out = subprocess.check_output("blkid -o export -s UUID -s TYPE".split())
+    out = out.decode('ascii')
+    return [dict(kv.split('=',1) for kv in block.splitlines())
+                                 for block in out.split('\n\n')]
+
+def get_disk_labels():
+    return {os.path.realpath(s):os.path.basename(s)
+            for s in iter_files("/dev/disk/by-label")}
+
+def get_deviceinfo():
+    disk_labels = get_disk_labels()
+    deviceinfo = [DeviceInfo(**d) for d in blkid()]
+    for dev in deviceinfo:
+        dev.label = disk_labels.get(dev.device, '')
+    return deviceinfo
+
+# --- INTERACTIVE MENU JUNK ------------------------------------------------
+
+class TextMenu(object):
+    def __init__(self, items, title=None, formatter=None, headeritem=None,
+                 refresher=None, multi=False, page_height=20):
+        self.items = items
+        self.title = title
+        self.formatter = formatter
+        self.headeritem = headeritem
+        self.refresher = refresher
+        self.multi = multi
+        self.page_height = page_height
+        self.pagenum = 1
+        self.selected_items = []
+        self.is_done = False
+        if callable(items):
+            self.refresher = items
+            self.refresh()
+
+    @property
+    def num_pages(self):
+        pages, leftover = divmod(len(self.items), self.page_height)
+        if leftover:
+            return pages+1
+        else:
+            return pages
+
+    def next(self):
+        if self.pagenum < self.num_pages:
+            self.pagenum += 1
+
+    def prev(self):
+        if self.pagenum > 1:
+            self.pagenum -= 1
+
+    def refresh(self):
+        if callable(self.refresher):
+            self.items = self.refresher()
+
+    def done(self):
+        self.is_done = True
+
+    def invalid(self, k):
+        print("Invalid selection %r" % k)
+
+    def toggle_item(self, item):
+        if item in self.selected_items:
+            self.selected_items.remove(item)
+        else:
+            self.selected_items.append(item)
+        if not self.multi:
+            self.done()
+
+    def items_on_page(self):
+        start_idx = (self.pagenum-1) * self.page_height
+        if start_idx > len(self.items):
+            return []
+        else:
+            items = self.items[start_idx:start_idx+self.page_height]
+            return enumerate(items, start=start_idx)
+
+    def format_item(self, item):
+        if callable(self.formatter):
+            return self.formatter(item)
+        else:
+            return str(item)
+
+    def format_items(self):
+        for n, i in self.items_on_page():
+            if self.multi:
+                x = 'x' if i in self.selected_items else ' '
+                yield "%2d) [%s] %s" % (n+1, x, self.format_item(i))
+            else:
+                yield "%2d) %s" % (n+1, self.format_item(i))
+
+    def format_header(self):
+        if self.multi:
+            return '        '+self.format_item(self.headeritem)
+        else:
+            return '    '+self.format_item(self.headeritem)
+
+    def action_dict(self):
+        actions = {
+            'r': self.refresh,
+            'n': self.next,
+            'p': self.prev,
+            'c': self.done,
+        }
+        for n, i in self.items_on_page():
+            actions[str(n+1)] = lambda item=i: self.toggle_item(item)
+        return actions
+
+    def format_page(self):
+        page = '\n(Page {pagenum} of {num_pages}) {title}\n{items}'
+        items = list(self.format_items())
+        if self.headeritem:
+            items.insert(0, self.format_header())
+        return page.format(pagenum=self.pagenum,
+                           num_pages=self.num_pages,
+                           title=self.title or '',
+                           items='\n'.join(items))
+
+    def format_prompt(self):
+        options = [
+            '# to toggle selection' if self.multi else '# to select',
+            "'r'-refresh" if callable(self.refresher) else None,
+            "'n'-next page" if self.pagenum < self.num_pages else None,
+            "'p'-previous page" if self.pagenum > 1 else None,
+            "or 'c'-continue"
+        ]
+        return ', '.join(o for o in options if o is not None) + ': '
+
+    def run(self):
+        while not self.is_done:
+            print(self.format_page())
+            k = _input(self.format_prompt())
+            action = self.action_dict().get(k)
+            if action:
+                action()
+            else:
+                self.invalid(k)
+        return self.selected_items
+
+def repo_menu(repos):
+    drivers = [d for r in repos for d in dd_list(r)]
+    if not drivers:
+        log.info("No suitable drivers found.")
+        return []
+    menu = TextMenu(drivers, title="Select drivers to install",
+                             formatter=lambda d: d.source,
+                             multi=True)
+    result = menu.run()
+    return result
+
+def iso_menu(isos):
+    menu = TextMenu(isos, title="Choose driver disk ISO file")
+    result = menu.run()
+    return result
+
+def device_menu():
+    fmt = '{0.shortdev:<8.8} {0.fs_type:<8.8} {0.label:<20.20} {0.uuid:<.36}'
+    hdr = DeviceInfo(DEVNAME='DEVICE', TYPE='TYPE', LABEL='LABEL', UUID='UUID')
+    menu = TextMenu(get_deviceinfo, title="Driver disk device selection",
+                                    formatter=fmt.format,
+                                    headeritem=hdr)
+    result = menu.run()
+    return result
+
+# --- COMMANDLINE-TYPE STUFF ------------------------------------------------
+
+def setup_log():
+    log.setLevel(logging.DEBUG)
+    handler = SysLogHandler(address="/dev/log")
+    log.addHandler(handler)
+    handler = logging.StreamHandler()
+    handler.setLevel(logging.INFO)
+    formatter = logging.Formatter("DD: %(message)s")
+    handler.setFormatter(formatter)
+    log.addHandler(handler)
+
+def print_usage():
+    print("usage: driver-updates --interactive")
+    print("       driver-updates --disk DISK KERNELDEV")
+    print("       driver-updates --net URL LOCALFILE")
+
+def check_args(args):
+    if args and args[0] == '--interactive':
+        return True
+    elif len(args) == 3 and args[0] in ('--disk', '--net'):
+        return True
+    else:
+        return False
+
+def main(args):
+    if not check_args(args):
+        print_usage()
+        raise SystemExit(2)
+
+    mode = args.pop(0)
+
+    if mode in ('--disk', '--net'):
+        request, dev = args
+        process_driver_disk(dev)
+
+    elif mode == '--interactive':
+        log.info("starting interactive mode")
+        request = 'menu'
+        while True:
+            dev = device_menu()
+            if not dev: break
+            process_driver_disk(dev.pop().device, interactive=True)
+
+    finish(request)
+
+if __name__ == '__main__':
+    setup_log()
+    try:
+        main(sys.argv[1:])
+    except KeyboardInterrupt:
+        log.info("exiting.")
diff --git a/dracut/test_driver_updates.py b/dracut/test_driver_updates.py
new file mode 100644
index 0000000..2aaa993
--- /dev/null
+++ b/dracut/test_driver_updates.py
@@ -0,0 +1,624 @@
+# 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/kickstart_tests/driverdisk-disk.ks b/tests/kickstart_tests/driverdisk-disk.ks
new file mode 100644
index 0000000..e8025cd
--- /dev/null
+++ b/tests/kickstart_tests/driverdisk-disk.ks
@@ -0,0 +1,41 @@
+#version=DEVEL
+url --url="http://dl.fedoraproject.org/pub/fedora/linux/development/$releasever/$basearch/os/"
+install
+network --bootproto=dhcp
+
+keyboard us
+lang en_US.UTF-8
+timezone America/New_York --utc
+rootpw testcase
+shutdown
+
+bootloader --timeout=1
+zerombr
+clearpart --all
+autopart
+
+driverdisk /dev/disk/by-label/TEST_DD
+
+%packages
+ at core
+%end
+
+%post --nochroot
+SYSROOT=${ANA_INSTALL_PATH:-/mnt/sysimage}
+RESULTFILE=$SYSROOT/root/RESULT
+fail() { echo "*** $*" >> $RESULTFILE; }
+
+# check the installer environment
+[ -f /lib/modules/`uname -r`/updates/fake-dd.ko ] || fail "kmod not loaded"
+[ -f /usr/bin/fake-dd-bin ] || fail "installer-enhancement not loaded"
+
+# check the installed system
+[ -f $SYSROOT/root/fake-dd-2.ko ] || fail "kmod rpm not installed"
+[ ! -f $SYSROOT/usr/bin/fake-dd-bin ] || \
+    fail "installer-enhancement package installed to target system"
+
+# write successful result if nothing failed
+if [[ ! -e $RESULTFILE ]]; then
+    echo SUCCESS > $RESULTFILE
+fi
+%end
diff --git a/tests/kickstart_tests/driverdisk-disk.sh b/tests/kickstart_tests/driverdisk-disk.sh
new file mode 100755
index 0000000..c510742
--- /dev/null
+++ b/tests/kickstart_tests/driverdisk-disk.sh
@@ -0,0 +1,33 @@
+# Copyright (c) 2015 Red Hat, Inc.
+#
+# 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; either version 2 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/>.
+#
+# Author: Will Woods <wwoods at redhat.com>
+
+. ${KSTESTDIR}/functions.sh
+
+prepare_disks() {
+    local diskdir="$1/disks"
+    # main disk
+    qemu-img create -q -f qcow2 ${diskdir}/a.img 10G
+    echo "${diskdir}/a.img"
+
+    # driverdisk image
+    ${KSTESTDIR}/../lib/mkdud.py -k -b -L "TEST_DD" ${diskdir}/dd.iso >/dev/null
+    echo "${diskdir}/dd.iso,device=cdrom,readonly=on"
+}
+
+#kernel_args() {
+#    echo inst.dd=/dev/disk/by-label/TEST_DD
+#}
diff --git a/tests/lib/mkdud.py b/tests/lib/mkdud.py
new file mode 100755
index 0000000..1fb2b3c
--- /dev/null
+++ b/tests/lib/mkdud.py
@@ -0,0 +1,120 @@
+#!/usr/bin/python
+# mkdud.py - test helper that makes driverdisk images
+#
+# Copyright (c) 2015 Red Hat, Inc.
+#
+# 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; either version 2 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/>.
+#
+# Author: Will Woods <wwoods at redhat.com>
+
+import os
+import rpmfluff
+import subprocess
+import argparse
+import tempfile
+import shutil
+
+from contextlib import contextmanager
+
+ at contextmanager
+def in_tempdir(prefix='tmp'):
+    oldcwd = os.getcwd()
+    tmpdir = tempfile.mkdtemp(prefix=prefix)
+    os.chdir(tmpdir)
+    yield
+    os.chdir(oldcwd)
+    shutil.rmtree(tmpdir)
+
+def parse_args():
+    p = argparse.ArgumentParser(
+        description="make fake driver disk images for testing",
+        epilog="ex: %(prog)s dd.iso",
+    )
+    p.add_argument("filename",
+        help="image filename to write")
+    p.add_argument("--label", "-L", default="OEMDRV",
+        help="disk image label (default: %(default)s)")
+    p.add_argument("--description", "-d", default="fake driverdisk",
+        help="driverdisk description (default: %(default)r)")
+    p.add_argument("--arch", "-a", default=rpmfluff.expectedArch,
+        help="arch to create RPMs for (default: host arch [%(default)s])")
+    p.add_argument("--kmod", "-k", action="store_true", default=False,
+        help="add a fake kmod to the driverdisk")
+    p.add_argument("--binary", "-b", action="store_true", default=False,
+        help="add a fake binary to the driverdisk")
+    p.add_argument("--createrepo", "-c", action="store_true", default=False,
+        help="run createrepo to add repodata to the driverdisk")
+
+    return p.parse_args()
+
+def write_description(desc):
+    with open("rhdd3",'w') as rhdd3:
+        rhdd3.write(desc+'\n')
+
+def make_rpm(pkg, outdir=".", arch=None):
+    outdir = os.path.abspath(outdir)
+    with in_tempdir(prefix='mkdud.rpmfluff.'):
+        pkg.make()
+        rpmfile = pkg.get_built_rpm(arch or rpmfluff.expectedArch)
+        outfile = os.path.join(outdir, os.path.basename(rpmfile))
+        shutil.move(rpmfile, outfile)
+    return outfile
+
+def write_kmod_rpm(outdir, for_kernel_ver=None, arch=None):
+    pkg = rpmfluff.SimpleRpmBuild('fake_kmod', '1.0', '1')
+    pkg.add_provides('kernel-modules >= %s' % for_kernel_ver)
+    pkg.add_installed_file("/lib/modules/%s/extra/fake-dd.ko" % for_kernel_ver,
+        rpmfluff.SourceFile("fake-dd.ko", "this is a fake kernel module"),
+    )
+    pkg.add_installed_file("/root/fake-dd-2.ko",
+        rpmfluff.SourceFile("fake-dd-2.ko", "another fake kernel module"),
+    )
+    return make_rpm(pkg, outdir, arch)
+
+def write_installer_enhancement_rpm(outdir, for_anaconda_ver=None, arch=None):
+    pkg = rpmfluff.SimpleRpmBuild('fake_bin', '1.0', '1')
+    pkg.add_provides('installer-enhancement = %s' % for_anaconda_ver)
+    pkg.add_installed_file("/usr/bin/fake-dd-bin",
+        rpmfluff.SourceFile("fake-dd-bin", "#!/bin/sh\necho FAKE BINARY OK"),
+        mode='755',
+    )
+    return make_rpm(pkg, outdir, arch)
+
+def createrepo(repodir):
+    return subprocess.check_call(["createrepo", repodir])
+
+def mkisofs(outfile, cd_dir, label=None):
+    cmd = ["mkisofs", "-o", outfile, "-r", "-input-charset", "utf-8"]
+    if label:
+        cmd += ["-V", label]
+    cmd.append(cd_dir)
+    subprocess.check_call(cmd)
+
+def main():
+    opts = parse_args()
+    outfile = os.path.abspath(opts.filename)
+    with in_tempdir(prefix='mkdud.'):
+        write_description(opts.description)
+        rpmdir = os.path.join("rpms", opts.arch)
+        os.makedirs(rpmdir)
+        if opts.kmod:
+            write_kmod_rpm(rpmdir, "3.0.0")
+        if opts.binary:
+            write_installer_enhancement_rpm(rpmdir, "19.0")
+        if opts.createrepo:
+            createrepo(rpmdir)
+        mkisofs(outfile, cd_dir=".", label=opts.label)
+
+if __name__ == '__main__':
+    main()


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


More information about the anaconda-patches mailing list