Okay, here it is - the big refactor of driver-updates. `README.dd` still needs to be updated, but the code is all here. It also adds a suite of unit tests for `driver-updates` and a kickstart test for the `driverdisk` command.
Other than some minor text changes the overall behavior should be the same, except for one thing: we no longer attempt to unload running modules. (This is, I think, outside the scope of what problems the driver-updates stuff is designed to solve. `modprobe.blacklist` can handle the rare cases where a driver-updates disk needs to replace an existing module.)
Resolves: RHBZ#1084197 Resolves: RHBZ#1164131
From: Will Woods wwoods@redhat.com
addDriverRepos needs to happen before gatherRepoMetadata, otherwise we don't gather its metadata. --- pyanaconda/packaging/__init__.py | 1 + pyanaconda/packaging/dnfpayload.py | 1 - 2 files changed, 1 insertion(+), 1 deletion(-)
diff --git a/pyanaconda/packaging/__init__.py b/pyanaconda/packaging/__init__.py index 3371c76..7201184 100644 --- a/pyanaconda/packaging/__init__.py +++ b/pyanaconda/packaging/__init__.py @@ -1280,6 +1280,7 @@ def _runThread(self, storage, ksdata, payload, instClass, fallback, checkmount): # Download package metadata try: payload.updateBaseRepo(fallback=fallback, checkmount=checkmount) + payload.addDriverRepos() except (OSError, PayloadError) as e: log.error("PayloadError: %s", e) self._error = self.ERROR_SETUP diff --git a/pyanaconda/packaging/dnfpayload.py b/pyanaconda/packaging/dnfpayload.py index 5c60b4b..906c4c7 100644 --- a/pyanaconda/packaging/dnfpayload.py +++ b/pyanaconda/packaging/dnfpayload.py @@ -699,7 +699,6 @@ def preInstall(self, packages=None, groups=None): if packages: self.requiredPackages += packages self.requiredGroups = groups - self.addDriverRepos()
def reset(self): super(DNFPayload, self).reset()
From: Will Woods wwoods@redhat.com
No substantive code changes here. Just some comments, some pylint directives, and an abstract mixin class so the extra arguments to dracut_args() don't look unused. --- dracut/parse-kickstart | 37 +++++++++++++++++++++---------------- 1 file changed, 21 insertions(+), 16 deletions(-)
diff --git a/dracut/parse-kickstart b/dracut/parse-kickstart index fc4d406..0c75f81 100755 --- a/dracut/parse-kickstart +++ b/dracut/parse-kickstart @@ -34,7 +34,7 @@ from pykickstart.parser import KickstartParser, preprocessKickstart from pykickstart.sections import NullSection from pykickstart.version import returnClassForVersion from pykickstart.errors import KickstartError -# pylint: disable=wildcard-import +# pylint: disable=wildcard-import,unused-wildcard-import from pykickstart.constants import * from pykickstart import commands from collections import OrderedDict @@ -74,7 +74,7 @@ def read_cmdline(f): # pylint: disable=redefined-outer-name for line in lines: for arg in line.split(): - k,_e,v = arg.partition("=") + k,_,v = arg.partition("=") args[k] = v return args
@@ -95,20 +95,25 @@ def setting_only_hostname(net, args):
proc_cmdline = read_cmdline("/proc/cmdline")
+class DracutArgsMixin(object): + """A mixin class to make a Command generate dracut args.""" + def dracut_args(self, args, lineno, obj): + raise NotImplementedError + # Here are the kickstart commands we care about:
-class Cdrom(commands.cdrom.FC3_Cdrom): +class Cdrom(commands.cdrom.FC3_Cdrom, DracutArgsMixin): def dracut_args(self, args, lineno, obj): return "inst.repo=cdrom"
-class HardDrive(commands.harddrive.FC3_HardDrive): +class HardDrive(commands.harddrive.FC3_HardDrive, DracutArgsMixin): def dracut_args(self, args, lineno, obj): if self.biospart: return "inst.repo=bd:%s:%s" % (self.partition, self.dir) else: return "inst.repo=hd:%s:%s" % (self.partition, self.dir)
-class NFS(commands.nfs.FC6_NFS): +class NFS(commands.nfs.FC6_NFS, DracutArgsMixin): def dracut_args(self, args, lineno, obj): if self.opts: method = "nfs:%s:%s:%s" % (self.opts, self.server, self.dir) @@ -119,7 +124,7 @@ class NFS(commands.nfs.FC6_NFS): method = method.replace(" ", "\ ") return "inst.repo=%s" % method
-class URL(commands.url.F18_Url): +class URL(commands.url.F18_Url, DracutArgsMixin): def dracut_args(self, args, lineno, obj): # Spaces in the url need to be %20 if self.url: @@ -136,19 +141,19 @@ class URL(commands.url.F18_Url):
return "\n".join(args)
-class Updates(commands.updates.F7_Updates): +class Updates(commands.updates.F7_Updates, DracutArgsMixin): def dracut_args(self, args, lineno, obj): if self.url == "floppy": return "live.updates=/dev/fd0" elif self.url: return "live.updates=%s" % self.url
-class MediaCheck(commands.mediacheck.FC4_MediaCheck): +class MediaCheck(commands.mediacheck.FC4_MediaCheck, DracutArgsMixin): def dracut_args(self, args, lineno, obj): if self.mediacheck: return "rd.live.check"
-class DriverDisk(commands.driverdisk.F14_DriverDisk): +class DriverDisk(commands.driverdisk.F14_DriverDisk, DracutArgsMixin): def dracut_args(self, args, lineno, obj): dd_net = [] dd_disk = [] @@ -168,7 +173,7 @@ class DriverDisk(commands.driverdisk.F14_DriverDisk): # are processed later. return "\n".join(dd_net)
-class Network(commands.network.F22_Network): +class Network(commands.network.F22_Network, DracutArgsMixin): def dracut_args(self, args, lineno, net): ''' NOTE: The first 'network' line get special treatment: @@ -212,7 +217,7 @@ class Network(commands.network.F22_Network):
return netline
-class DisplayMode(commands.displaymode.FC3_DisplayMode): +class DisplayMode(commands.displaymode.FC3_DisplayMode, DracutArgsMixin): def dracut_args(self, args, lineno, obj): if self.displayMode == DISPLAY_MODE_CMDLINE: return "inst.cmdline" @@ -221,12 +226,12 @@ class DisplayMode(commands.displaymode.FC3_DisplayMode): elif self.displayMode == DISPLAY_MODE_GRAPHICAL: return "inst.graphical"
-class Bootloader(commands.bootloader.F21_Bootloader): +class Bootloader(commands.bootloader.F21_Bootloader, DracutArgsMixin): def dracut_args(self, args, lineno, obj): if self.extlinux: return "extlinux"
-# TODO: keymap, lang... device? selinux? +# FUTURE: keymap, lang... device? selinux?
dracutCmds = { 'cdrom': Cdrom, @@ -366,7 +371,7 @@ def ksnet_to_dracut(args, lineno, net, bootdev=False): open(TMPDIR+"/net.ifaces", "a")
if net.essid or net.wepkey or net.wpakey: - # TODO: make dracut support wireless? (do we care?) + # NOTE: does dracut actually support wireless? (do we care?) log.error("'%s': dracut doesn't support wireless networks", " ".join(args))
@@ -446,7 +451,7 @@ def ksnet_to_ifcfg(net, filename=None): if net.nodefroute: ifcfg['DEFROUTE'] = "no"
- # TODO: dhcpclass, ethtool, essid/wepkey/wpakay, etc. + # FUTURE: dhcpclass, ethtool, essid/wepkey/wpakay, etc.
if net.bootProto == 'dhcp': srcpath = TMPDIR+"/dhclient.%s.lease" % dev @@ -514,7 +519,7 @@ def ksnet_to_ifcfg(net, filename=None):
options = {} for opt in net.bridgeopts.split(","): - key, _sep, value = opt.partition("=") + key, _, value = opt.partition("=") if not value: log.error("Invalid bridge option %s", opt) continue
From: Will Woods wwoods@redhat.com
The fancy handling here is only needed because of the truly messed-up way we were handling "inst.dd" compared to every other boot arg.
Fixing driverdisk handling makes this unnecessary. --- dracut/parse-kickstart | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-)
diff --git a/dracut/parse-kickstart b/dracut/parse-kickstart index 0c75f81..0d21053 100755 --- a/dracut/parse-kickstart +++ b/dracut/parse-kickstart @@ -155,23 +155,16 @@ class MediaCheck(commands.mediacheck.FC4_MediaCheck, DracutArgsMixin):
class DriverDisk(commands.driverdisk.F14_DriverDisk, DracutArgsMixin): def dracut_args(self, args, lineno, obj): - dd_net = [] - dd_disk = [] + dd_args = [] for dd in self.driverdiskList: if dd.partition: - dd_disk.append(dd.partition) + dd_args.append("inst.dd=hd:%s" % dd.partition) elif dd.source: - dd_net.append("inst.dd=%s" % dd.source) + dd_args.append("inst.dd=%s" % dd.source) + elif dd.biospart: + dd_args.append("inst.dd=bd:%s" % dd.biospart)
- # disk sources cannot be added to cmdline because the initial - # driver-update run has already finished. - if dd_disk: - with open(TMPDIR+"/dd_args_ks", "w") as f: - f.write(" ".join(dd_disk)) - - # network sources can be added to the existing cmdline, they - # are processed later. - return "\n".join(dd_net) + return "\n".join(dd_args)
class Network(commands.network.F22_Network, DracutArgsMixin): def dracut_args(self, args, lineno, net):
From: Will Woods wwoods@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@brianlane.com +# Will Woods wwoods@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]) + +@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/$basear..." +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 +@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@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@redhat.com + +import os +import rpmfluff +import subprocess +import argparse +import tempfile +import shutil + +from contextlib import contextmanager + +@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()
I get an error on the calls to extract_drivers when it tries to mkdir /updates. Do the extract_drivers calls in test_driver_updates need to be called with a temporary directory as outdir?
In reply to line 196 of dracut/driver_updates.py:
Mostly: the previous version did it this way. Also the docstring for `shutil.move` says:
A lot more could be done here... A look at a mv.c shows a lot of the issues this implementation glosses over.
Also, `shutil.move` raises IOError if `dst` exists, so we have to wrap it in a try/except.. but what if `dst` is a symlink? Do we have to unlink it first? We may be copying across a filesystem boundary, so `os.rename()` won't work, and... ugh. It gets complicated very quickly.
I'd take a patch that replaced these with python versions, but this runs in a pretty strange environment so I'm not yet sure it's as easy as "just use the shutil versions instead". Need more unit tests here, I guess..
In reply to line 456 of dracut/driver_updates.py:
Done in latest patch!
Heh! Whoops, you're right - it worked here because my system had `/updates` present. Fixed in latest commit.
In reply to line 196 of dracut/driver_updates.py:
Wow, that's even more messy than I expected! Thanks for the exaplanation. I'm absolutely okay with using the ``mv`` utility then :)
From: Will Woods wwoods@redhat.com
Here's where we actually modify stuff to use `driver_updates.py`:
* Parse commandline options to /tmp/dd_{net,disk,interactive,todo} * Generate udev rules etc. in `driver-updates-genrules.sh` * `fetch-driver-net.sh`: fetch .iso file and call driver-updates * Modify `driver-updates@.service`: * Use `dmesg -n1` and `/proc/sys/kernel/printk` to hide console messages * Make sure we run after vconsole is set up * Use `plymouth --hide-splash` rather than `plymouth quit` (this is how dracut handles emergency shells) * Drop `driver-updates.sh` (which ran driver-updates in `pre-trigger`) * Drop `driver-updates-net.sh` (`driver_updates.py` handles saving repos) * Drop old `driver-updates` * Make necessary changes to `Makefile.am` and `module-setup.sh` * Add `dracut/test_*.py` to `nosetests.sh` * `anaconda-lib.sh`: Add `wait_for_dd`, remove `start_driver_update`, fix `run_kickstart` for the new stuff above --- dracut/Makefile.am | 6 +- dracut/anaconda-lib.sh | 50 +-- dracut/driver-updates | 840 -------------------------------------- dracut/driver-updates-genrules.sh | 23 ++ dracut/driver-updates-net.sh | 10 - dracut/driver-updates.sh | 23 -- dracut/driver-updates@.service | 12 +- dracut/fetch-driver-net.sh | 36 +- dracut/module-setup.sh | 7 +- dracut/parse-anaconda-dd.sh | 32 +- tests/nosetests.sh | 2 +- 11 files changed, 85 insertions(+), 956 deletions(-) delete mode 100755 dracut/driver-updates create mode 100644 dracut/driver-updates-genrules.sh delete mode 100755 dracut/driver-updates-net.sh delete mode 100755 dracut/driver-updates.sh
diff --git a/dracut/Makefile.am b/dracut/Makefile.am index ae704e9..44a5b46 100644 --- a/dracut/Makefile.am +++ b/dracut/Makefile.am @@ -44,9 +44,9 @@ dist_dracut_SCRIPTS = module-setup.sh \ parse-anaconda-dd.sh \ fetch-driver-net.sh \ driver-updates@.service \ - driver-updates.sh \ - driver-updates-net.sh \ + driver-updates-genrules.sh \ anaconda-depmod.sh \ - driver-updates + driver_updates.py \ + test_driver_updates.py
MAINTAINERCLEANFILES = Makefile.in diff --git a/dracut/anaconda-lib.sh b/dracut/anaconda-lib.sh index 0b1679c..ab8590c 100755 --- a/dracut/anaconda-lib.sh +++ b/dracut/anaconda-lib.sh @@ -246,39 +246,21 @@ run_kickstart() { . $hookdir/cmdline/*parse-anaconda-repo.sh . $hookdir/cmdline/*parse-livenet.sh . $hookdir/cmdline/*parse-anaconda-dd.sh - case "$repotype" in - http*|ftp|nfs*) do_net=1 ;; - cdrom|hd|bd) do_disk=1 ;; - esac - [ "$root" = "anaconda-auto-cd" ] && do_disk=1 - - # kickstart Driver Disk Handling - # parse-kickstart may have added network inst.dd entries to the cmdline - # Or it may have written devices to /tmp/dd_ks - - # Does network need to be rerun? - dd_args="$(getargs dd= inst.dd=)" - for dd in $dd_args; do - case "${dd%%:*}" in - http|https|ftp|nfs|nfs4) - do_net=1 - rm /tmp/dd_net.done - break - ;; - esac - done
- # Run the driver update UI for disks - if [ -e "/tmp/dd_args_ks" ]; then - # TODO: Seems like this should be a function, a mostly same version is used in 3 places - start_driver_update "Kickstart Driver Update Disk" - rm /tmp/dd_args_ks - fi + # Figure out whether we need to retry disk/net stuff + case "$root" in + anaconda-net:*) do_net=1 ;; + anaconda-disk:*) do_disk=1 ;; + anaconda-auto-cd) do_disk=1 ;; + esac + [ -f /tmp/dd_net ] && do_net=1 + [ -f /tmp/dd_disk ] && do_disk=1
# disk: replay udev events to trigger actions if [ "$do_disk" ]; then # set up new rules . $hookdir/pre-trigger/*repo-genrules.sh + . $hookdir/pre-trigger/*driver-updates-genrules.sh udevadm control --reload # trigger the rules for all the block devices we see udevadm trigger --action=change --subsystem-match=block @@ -304,16 +286,6 @@ wait_for_updates() { echo "[ -e /tmp/liveupdates.done ]" > $hookdir/initqueue/finished/updates.sh }
-start_driver_update() { - local title="$1" - - tty=$(find_tty) - - # save module state - cat /proc/modules > /tmp/dd_modules - - info "Starting $title Service on $tty" - systemctl start driver-updates@$tty.service - status=$(systemctl -p ExecMainStatus show driver-updates@$tty.service) - info "DD status=$status" +wait_for_dd() { + echo "[ -e /tmp/dd.done ]" > $hookdir/initqueue/finished/dd.sh } diff --git a/dracut/driver-updates b/dracut/driver-updates deleted file mode 100755 index e16fe1d..0000000 --- a/dracut/driver-updates +++ /dev/null @@ -1,840 +0,0 @@ -#!/usr/bin/python3 -# -# Copyright (C) 2013 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@brianlane.com -# -""" -Driver Update Disk UI - -/tmp/dd_modules is a copy of /proc/modules at startup time -/tmp/dd_args is a parsed list of the inst.dd= cmdline args, and may include - 'dd' or 'inst.dd' if it was specified without arguments -/tmp/dd_args_ks is the same format, but skips processing existing OEMDRV devices. - -Pass a path and it will install the driver rpms from the path before checking -for new OEMDRV devices. - -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 -from logging.handlers import SysLogHandler -import sys -import os -import subprocess -import time -import glob -import readline # pylint:disable=unused-import - -log = logging.getLogger("DD") - -import functools -def eintr_retry_call(func, *args, **kwargs): - """Retry an interruptible system call if interrupted.""" - while True: - try: - return func(*args, **kwargs) - except InterruptedError: - continue -open = functools.partial(eintr_retry_call, open) # pylint: disable=redefined-builtin - -class RunCmdError(Exception): - """ Raised when run_cmd gets a non-zero returncode - """ - pass - - -def run_cmd(cmd): - """ Run a command, collect stdout and the returncode. stderr is ignored. - - :param cmd: command and arguments to run - :type cmd: list of strings - :returns: exit code and stdout from the command - :rtype: (int, string) - :raises: OSError if the cmd doesn't exist, RunCmdError if the rc != 0 - """ - try: - log.debug(" ".join(cmd)) - proc = subprocess.Popen(cmd, universal_newlines=True, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL) - out = proc.communicate()[0] - if out: - for line in out.splitlines(): - log.debug(line) - except OSError as e: - log.error("Error running %s: %s", cmd[0], e.strerror) - raise - if proc.returncode: - log.debug("%s returned %s", cmd[0], proc.returncode) - raise RunCmdError() - return (proc.returncode, out) - - -def oemdrv_list(): - """ Get a list of devices labeled as OEMDRV - - :returns: list of devices - :rtype: list - """ - try: - outlines = run_cmd(["blkid", "-t", "LABEL=OEMDRV", "-o", "device"])[1] - except (OSError, RunCmdError): - # Nothing with that label - return [] - else: - return outlines.splitlines() - - -def get_dd_args(): - """ Get the dd arguments from /tmp/dd_args or /tmp/dd_args_ks - - :returns: List of arguments - :rtype: list of strings - """ - net_protocols = ["http", "https", "ftp", "nfs", "nfs4"] - args = [] - for dd_args_file in ["/tmp/dd_args", "/tmp/dd_args_ks"]: - if not os.path.exists(dd_args_file): - continue - try: - dd_args = open(dd_args_file, "r").readline().split() - except IOError: - return [] - - # skip dd args that need networking - args.extend(x for x in dd_args if x.split(":")[0].lower() not in net_protocols) - return args - - -def is_interactive(): - """ Determine if the user requested interactive driver selection - - :returns: True if 'dd' or 'inst.dd' included in /tmp/dd_args False if not - :rtype: bool - """ - dd_args = get_dd_args() - if "dd" in dd_args or "inst.dd" in dd_args: - return True - else: - return False - - -def umount(device): - """ Unmount the device - - :param device: Device or mountpoint to unmount - :type device: string - :returns: None - """ - if not device: - return - - try: - run_cmd(["umount", device]) - except (OSError, RunCmdError): - pass - - -def mount_device(device, mnt="/media/DD/"): - """ Mount a device and check to see if it really is a driver disk - - :param device: path to device to mount - :type device: string - :param mnt: path to mount the device on - :type mnt: string - :returns: True if it is a DD, False if not - :rtype: bool - - It is unmounted if it is not a DD and left mounted if it is. - """ - try: - run_cmd(["mount", device, mnt]) - except (OSError, RunCmdError): - return False - return True - - -def copy_repo(dd_path, dest_prefix): - """ Copy the current arch's repository to a unique destination - - :param dd_path: Path to the driver repo directory - :type dd_path: string - :param dest_prefix: Destination directory prefix, a number is added - :type dest_prefix: string - :returns: None - - The destination directory names are in the order that the drivers - were loaded, starting from 1 - """ - suffix = 1 - while os.path.exists(dest_prefix+str(suffix)): - suffix += 1 - dest = dest_prefix+str(suffix) - os.makedirs(dest) - try: - run_cmd(["cp", "-ar", dd_path, dest]) - except (OSError, RunCmdError): - pass - - -def copy_file(src, dest): - """ Copy a file - - :param src: Source file - :type src: string - :param dest: Destination file - :type dest: string - :returns: None - """ - try: - run_cmd(["cp", "-a", src, dest]) - except (OSError, RunCmdError): - pass - - -def move_file(src, dest): - """ Move a file - - :param src: Source file - :type src: string - :param dest: Destination file - :type dest: string - :returns: None - """ - try: - run_cmd(["mv", "-f", src, dest]) - except (OSError, RunCmdError): - pass - - -def find_dd(mnt="/media/DD"): - """ Find all suitable DD repositories under a path - - :param mnt: Top of the directory tree to search - :type mnt: string - :returns: list of DD repositories - :rtype: list - """ - dd_repos = [] - arch = os.uname()[4] - for root, dirs, files in os.walk(mnt, followlinks=True): - if "rhdd3" in files and "rpms" in dirs and \ - os.path.exists(root+"/rpms/"+arch): - dd_repos.append(root+"/rpms/"+arch) - log.debug("Found repos - %s", " ".join(dd_repos)) - return dd_repos - - -def get_module_set(fname): - """ Read a module list and return a set of the names - - :param fname: Full path to filename - :type fname: string - :returns: set of the module names - """ - modules = set() - if os.path.exists(fname): - with open(fname, "r") as f: - for line in f: - mod_args = line.strip().split() - if mod_args: - modules.update([mod_args[0]]) - return modules - -def to_modname(modfile): - return os.path.basename(modfile)[:-3].replace('-','_') - -def reload_modules(newfiles): - """ Reload new module versions from /lib/modules/<kernel>/updates/ - """ - try: - run_cmd(["depmod", "-a"]) - except (OSError, RunCmdError): - pass - - # Make a list of modules added since startup - startup_modules = get_module_set("/tmp/dd_modules") - current_modules = get_module_set("/proc/modules") - new_modules = current_modules.difference(startup_modules) - log.debug("new_modules = %s", " ".join(new_modules)) - - # And a list of modules contained in the disk we just extracted - dd_modules = set(to_modname(f) for f in newfiles if f.endswith(".ko")) - # TODO: what modules do we unload when there's new firmware? - log.debug("dd_modules = %s", " ".join(dd_modules)) - new_modules.update(dd_modules) - - # I think we can just iterate once using modprobe -r to remove unused deps - for module in new_modules: - try: - run_cmd(["modprobe", "-r", module]) - except (OSError, RunCmdError): - pass - - time.sleep(2) - - # Reload the modules, using the new versions from /lib/modules/<kernel>/updates/ - try: - run_cmd(["udevadm", "trigger"]) - except (OSError, RunCmdError): - pass - - -class Driver(object): - def __init__(self): - self.source = "" - self.name = "" - self.flags = "" - self.description = [] - self.selected = False - - @property - def args(self): - return ["--%s" % a for a in self.flags.split()] - - @property - def rpm(self): - return self.source - - -def fake_drivers(num): - """ Generate a number of fake drivers for testing - """ - drivers = [] - for i in range(0, num): - d = Driver() - d.source = "driver-%d" % i - d.flags = "modules" - drivers.append(d) - return drivers - - -def dd_list(dd_path, kernel_ver=None, anaconda_ver=None): - """ Build a list of the drivers in the directory - - :param dd_path: Path to the driver repo - :type dd_path: string - :returns: list of drivers - :rtype: Driver object - - By default none of the drivers are selected - """ - if not kernel_ver: - kernel_ver = os.uname()[2] - if not anaconda_ver: - anaconda_ver = "19.0" - - try: - outlines = run_cmd(["dd_list", "-k", kernel_ver, "-a", anaconda_ver, "-d", dd_path])[1] - except (OSError, RunCmdError): - return [] - - # Output format is: - # source rpm\n - # name\n - # flags\n - # description (multi-line)\n - # ---\n - drivers = [] - new_driver = Driver() - line_idx = 0 - for line in outlines.splitlines(): - log.debug(line) - if line == "---": - drivers.append(new_driver) - new_driver = Driver() - line_idx = 0 - elif line_idx == 0: - new_driver.source = line - line_idx += 1 - elif line_idx == 1: - new_driver.name = line - line_idx += 1 - elif line_idx == 2: - new_driver.flags = line - line_idx += 1 - elif line_idx == 3: - new_driver.description.append(line) - - return drivers - - -def dd_extract(driver, dest_path="/updates/", kernel_ver=None): - """ Extract a driver rpm to a destination path - - :param driver: Driver to extract - :type driver: Driver object - :param dest_path: Top directory of the destination path - :type dest_path: string - :returns: list of paths to extracted firmware and modules - - This extracts the driver's files into 'dest_path' (which defaults - to /updates/ so that the normal live updates handling will overlay - any binary or library updates onto the initrd automatically. - """ - if not kernel_ver: - kernel_ver = os.uname()[2] - - cmd = ["dd_extract", "-k", kernel_ver] - cmd += driver.args - cmd += ["--rpm", driver.rpm, "--directory", dest_path] - log.info("Extracting files from %s", driver.rpm) - - # make sure the to be used directory exists - if not os.path.isdir(dest_path): - os.makedirs(dest_path) - - try: - run_cmd(cmd) - except (OSError, RunCmdError): - log.error("dd_extract failed, skipped %s", driver.rpm) - return - - # Create the destination directories - initrd_updates = "/lib/modules/" + os.uname()[2] + "/updates/" - ko_updates = dest_path + initrd_updates - initrd_firmware = "/lib/firmware/updates/" - firmware_updates = dest_path + initrd_firmware - for d in (initrd_updates, ko_updates, initrd_firmware, firmware_updates): - if not os.path.exists(d): - os.makedirs(d) - - filelist = [] - - # Copy *.ko files over to /updates/lib/modules/<kernel>/updates/ - for root, _dirs, files in os.walk(dest_path+"/lib/modules/"): - if root.endswith("/updates") and os.path.isdir(root): - continue - for f in (f for f in files if f.endswith(".ko")): - src = root+"/"+f - filelist.append(src) - copy_file(src, ko_updates) - move_file(src, initrd_updates) - - # Copy the firmware updates - for root, _dirs, files in os.walk(dest_path+"/lib/firmware/"): - if root.endswith("/updates") and os.path.isdir(root): - continue - for f in (f for f in files): - src = root+"/"+f - filelist.append(src) - copy_file(src, firmware_updates) - move_file(src, initrd_firmware) - - # Tell our caller about the newly-extracted stuff - return filelist - - -# an arbitrary value to signal refreshing the menu contents -DoRefresh = True - -def selection_menu(items, title, info_func, multi_choice=True, refresh=False): - """ Display menu and let user select one or more choices. - - :param items: list of items - :type items: list of objects (with the 'selected' property/attribute if - multi_choice=True is used) - :param title: title for the menu - :type title: str - :param info_func: function providing info about items - :type info_func: item -> str - :param multi_choice: whether it is a multiple choice menu or not - :type multi_choice: bool - :returns: the selected item in case of multi_choice=False and user did - selection, None otherwise - """ - - page_length = 20 - page = 1 - num_pages = len(items) / page_length - if len(items) % page_length > 0: - num_pages += 1 - - if multi_choice: - choice_format = "[%s]" - else: - choice_format = "" - format_str = "%3d) " + choice_format + " %s" - - while True: - # show a page of items - print("\nPage %d of %d" % (page, num_pages)) - print(title) - if page * page_length <= len(items): - num_items = page_length - else: - num_items = len(items) % page_length - for i in range(0, num_items): - item_idx = ((page-1) * page_length) + i - if multi_choice: - if items[item_idx].selected: - selected = "x" - else: - selected = " " - args = (i+1, selected, info_func(items[item_idx])) - else: - args = (i+1, info_func(items[item_idx])) - print(format_str % args) - - # Select an item to toggle, continue or change pages - opts = ["# to select", - "'n'-next page", - "'p'-previous page", - "'c'-continue"] - if multi_choice: - opts[0] = "# to toggle selection" - if refresh: - opts.insert(1,"'r'-refresh") - idx = input(''.join(['\n', - ", ".join(opts[:-1]), - " or ", opts[-1], ": "])) - if idx.isdigit() and not (int(idx) < 1 or int(idx) > num_items): - item_idx = ((page-1) * page_length) + int(idx) - 1 - if multi_choice: - items[item_idx].selected = not items[item_idx].selected - else: - # single choice only, we can return now - return items[item_idx] - elif idx.lower() == 'n': - if page < num_pages: - page += 1 - else: - print("Last page") - elif idx.lower() == 'p': - if page > 1: - page -= 1 - else: - print("First page") - elif idx.lower() == 'r' and refresh: - return DoRefresh - elif idx.lower() == 'c': - return - else: - print("Invalid selection") - -def select_drivers(drivers): - """ Display pages of drivers to be loaded. - - :param drivers: Drivers to be selected by the user - :type drivers: list of Driver objects - :returns: None - """ - if not drivers: - return - - selection_menu(drivers, "Select drivers to install", - lambda driver: driver.source) - -def process_dd(dd_path): - """ Handle installing modules, firmware, enhancements from the dd repo - - :param dd_path: Path to the driver repository - :type dd_path: string - :returns: None - """ - drivers = dd_list(dd_path) - log.debug("drivers = %s", " ".join([d.rpm for d in drivers])) - - # If interactive mode or rhdd3.rules pass flag to deselect by default? - if os.path.exists(dd_path+"/rhdd3.rules") or is_interactive(): - select_drivers(drivers) - if not any((d.selected for d in drivers)): - return - else: - for driver in drivers: - setattr(driver, "selected", True) - - # Copy the repository for Anaconda to use during install - copy_repo(dd_path, "/updates/run/install/DD-") - - extracted = [] - - for driver in (d for d in drivers if d.selected): - extracted += dd_extract(driver, "/updates/") - - # Write the package names for all modules and firmware for Anaconda - if "modules" in driver.flags or "firmwares" in driver.flags: - with open("/run/install/dd_packages", "a") as f: - f.write("%s\n" % driver.name) - - reload_modules(extracted) - - -def select_dd(device): - """ Mount a device and check it for Driver Update repos - - :param device: Path to the device to mount and check - :type device: string - :returns: None - """ - mnt = "/media/DD/" - if not os.path.isdir(mnt): - os.makedirs(mnt) - if not mount_device(device, mnt): - return - - dd_repos = find_dd(mnt) - for repo in dd_repos: - log.info("Processing DD repo %s on %s", repo, device) - process_dd(repo) - - # TODO - does this need to be done before module reload? - umount(device) - - -def network_driver(dd_path): - """ Handle network driver download, then scan for new OEMDRV devices. - - :param dd_path: Path to the downloaded driver rpms - :type dd_path: string - :returns: None - """ - skip_dds = set(oemdrv_list()) - - log.info("Processing Network Drivers from %s", dd_path) - isos = glob.glob(os.path.join(dd_path, "*.iso")) - for iso in isos: - select_dd(iso) - - process_dd(dd_path) - - # TODO: May need to add new drivers to /tmp/dd_modules to prevent them from being unloaded - - # Scan for new OEMDRV devices and ignore dd_args - dd_scan(skip_dds, scan_dd_args=False, skip_device_menu=True) - -class DeviceInfo(object): - def __init__(self, **kwargs): - self.device = kwargs.get("device", None) - self.label = kwargs.get("label", None) - self.uuid = kwargs.get("uuid", None) - self.fs_type = kwargs.get("fs_type", None) - - def __str__(self): - return "%-10s %-20s %-15s %s" % (self.device or "", self.fs_type or "", - self.label or "", self.uuid or "") - -def parse_blkid(line): - """ Parse a line of output from blkid - - :param line: line of output from blkid - :param type: string - :returns: {} or dict of NAME=VALUE pairs including "device" - :rtype: dict - - blkid output cannot be trusted. labels may be missing or in a different - order so we parse what we get and return a dict with their values. - """ - import shlex - - device = {"device":None, "label":None, "uuid":None, "fs_type":None} - fields = shlex.split(line) - if len(fields) < 2 or not fields[0].startswith("/dev/"): - return {} - - # device is in [0] and the remainder are NAME=VALUE with possible spaces - # Use the sda1 part of device "/dev/sda1:" - device['device'] = fields[0][5:-1] - for f in fields[1:]: - if "=" in f: - (key, val) = f.split("=", 1) - if key == "TYPE": - key = "fs_type" - device[key.lower()] = val - return device - -def select_iso(): - """ Let user select device and DD ISO on it. - - :returns: path to the selected ISO file and mountpoint to be unmounted - or (None, None) if no ISO file is selected - :rtype: (str, str) - """ - header = " %-10s %-20s %-15s %s" % ("DEVICE", "TYPE", "LABEL", "UUID") - - iso_dev = DoRefresh - while iso_dev is DoRefresh: - try: - _ret, out = run_cmd(["blkid"]) - except (OSError, RunCmdError): - return (None, None) - - devices = [] - for line in out.splitlines(): - dev = parse_blkid(line) - if dev: - devices.append(DeviceInfo(**dev)) - - iso_dev = selection_menu(devices, - "Driver disk device selection\n" + header, - str, multi_choice=False, refresh=True) - - if not iso_dev: - return (None, None) - - mnt = "/media/DD-search" - if not os.path.isdir(mnt): - os.makedirs(mnt) - if not mount_device("/dev/" + iso_dev.device, mnt): - print("===Cannot mount the chosen device!===\n") - return select_iso() - - # is this device a Driver Update Disc? - if find_dd(mnt): - umount(mnt) # BLUH. unmount it first so select_dd can mount it OK - return ("/dev/" + iso_dev.device, None) - - # maybe it's a device containing multiple DUDs - let the user pick one - isos = list() - for dir_path, _dirs, files in os.walk(mnt): - # trim the mount point path - rel_dir = dir_path[len(mnt):] - - # and the starting "/" (if any) - if rel_dir.startswith("/"): - rel_dir = rel_dir[1:] - - isos += (os.path.join(rel_dir, iso_file) - for iso_file in files if iso_file.endswith(".iso")) - - if not isos: - print("===No ISO files found on %s!===\n" % iso_dev.device) - umount(mnt) - return select_iso() - else: - # mount writes out some mounting information, add blank line - print() - - # let user choose the ISO file - dd_iso = selection_menu(isos, "Choose driver disk ISO file", - lambda iso_file: iso_file, - multi_choice=False) - - if not dd_iso: - return (None, None) - - return (os.path.join(mnt, dd_iso), "/media/DD-search") - -def dd_scan(skip_dds=None, scan_dd_args=True, skip_device_menu=False): - """ Scan the system for OEMDRV devices and and specified by dd=/dev/<device> - - :param skip_dds: devices to skip when checking for OEMDRV label - :type skip_dds: set() - :param scan_dd_args: Scan devices passed in /tmp/dd_args or dd_args_ks - :type scan_dd_args: bool - :returns: None - """ - dd_todo = set(oemdrv_list()) - - if skip_dds is None: - skip_dds = set() - - if skip_dds: - dd_todo.difference_update(skip_dds) - if dd_todo: - log.info("Found new OEMDRV device(s) - %s", ", ".join(dd_todo)) - - if scan_dd_args: - # Add the user specified devices - dd_devs = get_dd_args() - dd_devs = [dev for dev in dd_devs if dev not in ("dd", "inst.dd")] - dd_todo.update(dd_devs) - log.info("Checking devices %s", ", ".join(dd_todo)) - - # Process each Driver Disk, checking for new disks after each one - dd_finished = dd_load(dd_todo, skip_dds=skip_dds) - skip_dds.update(dd_finished) - - # Skip interactive selection of an iso if OEMDRV was found - if skip_dds or skip_device_menu or not is_interactive(): - return - - # Handle interactive driver selection - mount_point = None - while True: - iso, mount_point = select_iso() - if iso: - if iso in skip_dds: - skip_dds.remove(iso) - dd_load(set([iso]), skip_dds=skip_dds) - # NOTE: we intentionally do not add the newly-loaded device to - # skip_dds - the user might (e.g.) swap DVDs and use /dev/sr0 twice - umount(mount_point) - else: - break - -def dd_load(dd_todo, skip_dds=None): - """ Process each Driver Disk, checking for new disks after each one. - Return the set of devices that loaded stuff from. - - :param dd_todo: devices to load drivers from - :type dd_todo: set - :param skip_dds: devices to skip when checking for OEMDRV label - :type skip_dds: set - :returns: set of devices that have been loaded - """ - if skip_dds is None: - skip_dds = set() - - dd_finished = set() - while dd_todo: - device = dd_todo.pop() - if device in skip_dds: - continue - log.info("Checking device %s", device) - select_dd(device) - dd_finished.add(device) - new_oemdrv = set(oemdrv_list()).difference(dd_finished, dd_todo) - if new_oemdrv: - log.info("Found new OEMDRV device(s) - %s", ", ".join(new_oemdrv)) - dd_todo.update(new_oemdrv) - return dd_finished - -if __name__ == '__main__': - 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) - - if len(sys.argv) > 1: - # Network driver source - network_driver(sys.argv[1]) - elif os.path.exists("/tmp/DD-net/"): - network_driver("/tmp/DD-net/") - elif os.path.exists("/tmp/dd_args_ks"): - # Kickstart driverdisk command, skip existing OEMDRV devices and - # process cmdline dd entries. This will process any OEMDRV that - # appear after loading the other drivers. - skip_devices = set(oemdrv_list()) - dd_scan(skip_devices, skip_device_menu=True) - else: - # Process /tmp/dd_args and OEMDRV devices - # Show device selection menu when inst.dd passed and no OEMDRV devices - dd_scan() - - sys.exit(0) - diff --git a/dracut/driver-updates-genrules.sh b/dracut/driver-updates-genrules.sh new file mode 100644 index 0000000..b20dbe2 --- /dev/null +++ b/dracut/driver-updates-genrules.sh @@ -0,0 +1,23 @@ +#!/bin/bash + +command -v wait_for_dd >/dev/null || . /lib/anaconda-lib.sh + +# Don't leave initqueue until we've finished with the requested dd stuff +[ -f /tmp/dd_todo ] && wait_for_dd + +if [ -f /tmp/dd_interactive ]; then + initqueue --onetime --settled --name zz_dd_interactive \ + systemctl start driver-updates@$(find_tty).service +fi + +# Run driver-updates for LABEL=OEMDRV and any other requested disk +for dd in LABEL=OEMDRV $(cat /tmp/dd_disk); do + when_diskdev_appears "$(disk_to_dev_path $dd)" \ + driver-updates --disk $dd $devnode +done + +# force us to wait at least until we've settled at least once +echo '> /tmp/settle.done' > $hookdir/initqueue/settled/settle_done.sh +echo '[ -f /tmp/settle.done ]' > $hookdir/initqueue/finished/wait_for_settle.sh + +# NOTE: dd_net is handled by fetch-driver-net.sh diff --git a/dracut/driver-updates-net.sh b/dracut/driver-updates-net.sh deleted file mode 100755 index f8bed52..0000000 --- a/dracut/driver-updates-net.sh +++ /dev/null @@ -1,10 +0,0 @@ -#!/bin/sh -[ -e /tmp/DD-net ] || return 0 - -command -v getarg >/dev/null || . /lib/dracut-lib.sh -. /lib/anaconda-lib.sh - -if [ -n "$(ls /tmp/DD-net)" ]; then - start_driver_update "Network Driver Update Disk" - rm -rf /tmp/DD-net -fi diff --git a/dracut/driver-updates.sh b/dracut/driver-updates.sh deleted file mode 100755 index fca0b83..0000000 --- a/dracut/driver-updates.sh +++ /dev/null @@ -1,23 +0,0 @@ -#!/bin/bash -# Determine if a Driver Update Disk is present, or inst.dd passed on the cmdline -# and launch the driver update systemd service - -# load all modules -udevadm trigger -udevadm settle - -# Look for devices with the OEMDRV label -blkid -t LABEL=OEMDRV > /dev/null -blkid_rc=$? - -command -v getarg >/dev/null || . /lib/dracut-lib.sh -dd_args="$(getargs dd= inst.dd=)" -if [ -n "$dd_args" -o $blkid_rc -eq 0 ]; then - command -v getarg >/dev/null || . /lib/dracut-lib.sh - . /lib/anaconda-lib.sh - - # kludge to let kernel spit out some extra info w/o stomping on our UI - sleep 5 - echo "$dd_args" > /tmp/dd_args - start_driver_update "Driver Update Disk" -fi diff --git a/dracut/driver-updates@.service b/dracut/driver-updates@.service index 210bcd5..e8c876e 100644 --- a/dracut/driver-updates@.service +++ b/dracut/driver-updates@.service @@ -1,15 +1,19 @@ [Unit] Description=Driver Update Disk UI on %I DefaultDependencies=no +After=systemd-vconsole-setup.service +Wants=systemd-vconsole-setup.service Before=shutdown.target
[Service] Type=oneshot RemainAfterExit=no -WorkingDirectory=/tmp -Environment=LANG=en_US.UTF-8 -ExecStartPre=-/bin/plymouth quit -ExecStart=/bin/driver-updates +WorkingDirectory=/ +ExecStartPre=-/bin/plymouth --hide-splash +ExecStartPre=-/bin/cp -n /proc/sys/kernel/printk /tmp/printk +ExecStartPre=-/bin/dmesg -n 1 +ExecStart=/bin/driver-updates --interactive +ExecStopPost=-/bin/cp /tmp/printk /proc/sys/kernel/printk StandardInput=tty-force StandardOutput=inherit StandardError=inherit diff --git a/dracut/fetch-driver-net.sh b/dracut/fetch-driver-net.sh index 5adc39b..1632d00 100755 --- a/dracut/fetch-driver-net.sh +++ b/dracut/fetch-driver-net.sh @@ -5,26 +5,20 @@ # initqueue/online hook passes interface name as $1 netif="$1"
-# We already processed the dd_args - exit -[ -e /tmp/dd_net.done ] && return 0 - -command -v getarg >/dev/null || . /lib/dracut-lib.sh -dd_args="$(getargs dd= inst.dd=)" -[ -n "$dd_args" ] || return 0 +# No dd_net was requested - exit +[ -f /tmp/dd_net ] || return 0
. /lib/url-lib.sh -dd_repo=/tmp/DD-net/ -for dd in $dd_args; do - case "${dd%%:*}" in - http|https|ftp|nfs|nfs4) - [ -e "$dd_repo" ] || mkdir -p $dd_repo - info "Fetching driver from $dd" - if driver=$(fetch_url "$dd"); then - mv "$driver" $dd_repo - else - warn "Failed to fetch driver from $dd" - fi - ;; - esac -done -echo > /tmp/dd_net.done + +while read dd; do + # If we already fetched this URL, skip it + grep -Fqx "$dd" /tmp/dd_net.done && continue + # Otherwise try to fetch it + info "Fetching driverdisk from $dd" + if driver=$(fetch_url "$dd"); then + echo "$dd" >> /tmp/dd_net.done # mark it done so we don't fetch it again + driver-updates --net "$dd" "$driver" + else + warn "Failed to fetch driver from $dd" + fi +done < /tmp/dd_net diff --git a/dracut/module-setup.sh b/dracut/module-setup.sh index ee92f69..a691365 100755 --- a/dracut/module-setup.sh +++ b/dracut/module-setup.sh @@ -46,16 +46,15 @@ install() { inst "$moddir/parse-kickstart" "/sbin/parse-kickstart" # Driver Update Disks inst_hook cmdline 29 "$moddir/parse-anaconda-dd.sh" + inst_hook pre-trigger 55 "$moddir/driver-updates-genrules.sh" inst_hook initqueue/online 20 "$moddir/fetch-driver-net.sh" - inst_hook pre-trigger 40 "$moddir/driver-updates.sh" - inst_hook pre-pivot 10 "$moddir/driver-updates-net.sh" inst_hook pre-pivot 50 "$moddir/anaconda-depmod.sh" - inst "$moddir/driver-updates" "/bin/driver-updates" + inst "$moddir/driver_updates.py" "/bin/driver-updates" inst_simple "$moddir/driver-updates@.service" "/etc/systemd/system/driver-updates@.service" # rpm configuration file (needed by dd_extract) inst "/usr/lib/rpm/rpmrc" # python deps for parse-kickstart. DOUBLE WOOOO - PYTHONHASHSEED=42 $moddir/python-deps $moddir/parse-kickstart $moddir/driver-updates | while read dep; do + PYTHONHASHSEED=42 $moddir/python-deps $moddir/parse-kickstart $moddir/driver_updates.py | while read dep; do case "$dep" in *.so) inst_library $dep ;; *.py) inst_simple $dep ;; diff --git a/dracut/parse-anaconda-dd.sh b/dracut/parse-anaconda-dd.sh index ebabd07..ee65501 100755 --- a/dracut/parse-anaconda-dd.sh +++ b/dracut/parse-anaconda-dd.sh @@ -1,18 +1,28 @@ #!/bin/bash # parse-anaconda-dd.sh: handle driver update disk settings
-# no need to do this twice -[ -f /tmp/dd_net.done ] && return +# Creates the following files: +# /tmp/dd_net: list of URLs to fetch +# /tmp/dd_disk: list of disk devices to load from +# /tmp/dd_interactive: "menu" if interactive mode requested +# /tmp/dd_todo: concatenation of the above files
-command -v getarg >/dev/null || . /lib/dracut-lib.sh +# clear everything to ensure idempotency +rm -f /tmp/dd_interactive /tmp/dd_net /tmp/dd_disk /tmp/dd_todo
-# inst.dd: May provide a "URI" for the driver rpm (possibly more than one) -dd_args="$(getargs dd= inst.dd=)" -for dd in $dd_args; do - case "${dd%%:*}" in - http|https|ftp|nfs|nfs4) - set_neednet - break - ;; +# parse any dd/inst.dd args found +for dd in $(getargs dd= inst.dd=); do + case "$dd" in + # plain 'dd'/'inst.dd': Engage interactive mode! + dd|inst.dd) echo menu > /tmp/dd_interactive ;; + # network URLs: add to dd_net + http:*|https:*|ftp:*|nfs:*|nfs4:*) echo $dd >> /tmp/dd_net ;; + # disks: strip "cdrom:" or "hd:" and add to dd_disk + cdrom:*|hd:*) echo ${dd#*:} >> /tmp/dd_disk ;; + # anything else is assumed to be a disk + *) echo $dd >> /tmp/dd_disk esac done + +# for convenience's sake, mash 'em all into one list +cat /tmp/dd_net /tmp/dd_disk /tmp/dd_interactive > /tmp/dd_todo diff --git a/tests/nosetests.sh b/tests/nosetests.sh index de2c111..ec7e1df 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 + set -- "${top_srcdir}"/tests/*_tests "${top_srcdir}"/dracut/test_*.py fi
exec nosetests-3.4 -v --exclude=logpicker -a !acceptance,!slow "$@"
From: Will Woods wwoods@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 --- 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@.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 + +@mock.patch("driver_updates.ensure_dir") +@mock.patch("driver_updates.save_repo") +@mock.patch("driver_updates.append_line") +@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 ec7e1df..de2c111 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-3.4 -v --exclude=logpicker -a !acceptance,!slow "$@"
Closed.
anaconda-patches@lists.fedorahosted.org
-
dashea