#!/usr/bin/python3 # # Copyright (C) 2014 Red Hat, Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU Lesser General Public License as published # by the Free Software Foundation; either version 2.1 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 Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public License # along with this program. If not, see . # # Author: Chris Lumens # Ignore any interruptible calls # pylint: disable=interruptible-system-call # pylint: disable=ignorable-system-call __all__ = ["Creator", "OutsideMixin"] from blivet.size import MiB from contextlib import contextmanager from nose.plugins.multiprocess import TimedOutException import os import shutil import subprocess import tempfile class Creator(object): """A Creator subclass defines all the parameters for making a VM to run a test against as well as handles creating and running that VM, inspecting results, and managing temporary data. Most Creator subclasses will only need to define the following four attributes: drives -- A list of tuples describing disk images to create. Each tuple is the name of the drive and its size as a blivet.Size. environ -- A dictionary of environment variables that should be added to the environment the test suite will run under. name -- A unique string that names a Creator. This name will be used in creating the results directory (and perhaps other places in the future) so make sure it doesn't conflict with another object. tests -- A list of tuples describing which test cases make up this test. Each tuple is the name of the module containing the test case (minus the leading "inside." and the name of the test case class. Tests will be run in the order provided. """ drives = [] environ = {} name = "Creator" tests = [] def __init__(self): self._drivePaths = {} self._mountpoint = None self._proc = None self._tempdir = None self._reqMemory = 1536 def _call(self, args): subprocess.call(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def archive(self): """Copy all log files and other test results to a subdirectory of the given resultsdir. If logs are no longer available, this method does nothing. It is up to the caller to make sure logs are available beforehand and clean up afterwards. """ from testconfig import config if not os.path.ismount(self.mountpoint): return shutil.copytree(self.mountpoint + "/result", config["resultsdir"] + "/" + self.name) def cleanup(self): """Remove all disk images used during this test case and the temporary directory they were stored in. """ shutil.rmtree(self.tempdir, ignore_errors=True) def die(self): """Kill any running qemu process previously started by this test.""" if self._proc: self._proc.kill() self._proc = None def makeDrives(self): """Create all hard drive images associated with this test. Images must be listed in Creator.drives and will be stored in a temporary directory this method creates. It is up to the caller to remove everything later by calling Creator.cleanup. """ for (drive, size) in self.drives: (fd, diskimage) = tempfile.mkstemp(dir=self.tempdir) os.close(fd) # For now we are using qemu-img to create these files but specifying # sizes in blivet Size objects. Unfortunately, qemu-img wants sizes # as xM or xG, not xMB or xGB. That's what the conversion here is for. self._call(["/usr/bin/qemu-img", "create", "-f", "raw", diskimage, "%sM" % size.convertTo(MiB)]) self._drivePaths[drive] = diskimage @property def template(self): with open("outside/template.py", "r") as f: return f.read() def makeSuite(self): """The suite is a small disk image attached to every test VM automatically by the test framework. It includes all the inside/ stuff, a special suite.py file that will be automatically run by the live CD (and is what actually runs the test), and a directory structure for reporting results. It is mounted under Creator.mountpoint as needed. This method creates the suite image and adds it to the internal list of images associated with this test. Note that because this image is attached to the VM, anaconda will always see two hard drives and thus will never automatically select disks. Note also that this means tests must be careful to not select this disk. """ from testconfig import config self._call(["/usr/bin/qemu-img", "create", "-f", "raw", self.suitepath, "10M"]) self._call(["/sbin/mkfs.ext4", "-F", self.suitepath, "-L", "ANACTEST"]) self._call(["/usr/bin/mount", "-o", "loop", self.suitepath, self.mountpoint]) # Create the directory structure needed for storing results. os.makedirs(self.mountpoint + "/result/anaconda") # Copy all the inside stuff into the mountpoint. shutil.copytree("inside", self.mountpoint + "/inside") # Create the suite file, which contains all the test cases to run and is how # the VM will figure out what to run. with open(self.mountpoint + "/suite.py", "w") as f: imports = map(lambda path_cls: " from inside.%s import %s" % (path_cls[0], path_cls[1]), self.tests) addtests = map(lambda path_cls1: " s.addTest(%s())" % path_cls1[1], self.tests) f.write(self.template % {"environ": " os.environ.update(%s)" % self.environ, "imports": "\n".join(imports), "addtests": "\n".join(addtests), "anacondaArgs": config.get("anacondaArgs", "").strip('"')}) self._call(["/usr/bin/umount", self.mountpoint]) # This ensures it gets passed to qemu-kvm as a disk arg. self._drivePaths[self.suitename] = self.suitepath @contextmanager def suiteMounted(self): """This context manager allows for wrapping code that needs to access the suite. It mounts the disk image beforehand and unmounts it afterwards. """ if self._drivePaths.get(self.suitename, "") == "": return self._call(["/usr/bin/mount", "-o", "loop", self.suitepath, self.mountpoint]) try: yield except: raise finally: self._call(["/usr/bin/umount", self.mountpoint]) def run(self): """Given disk images previously created by Creator.makeDrives and Creator.makeSuite, start qemu and wait for it to terminate. """ from testconfig import config args = ["/usr/bin/qemu-kvm", "-vnc", "none", "-m", str(self._reqMemory), "-boot", "d", "-drive", "file=%s,media=cdrom,readonly" % config["liveImage"]] for drive in self._drivePaths.values(): args += ["-drive", "file=%s,media=disk" % drive] # Save a reference to the running qemu process so we can later kill # it if necessary. For now, the only reason we'd want to kill it is # an expired timer. self._proc = subprocess.Popen(args) try: self._proc.wait() except TimedOutException: self.die() self.cleanup() raise finally: self._proc = None @property def mountpoint(self): """The directory where the suite is mounted. This is a subdirectory of Creator.tempdir, and it is assumed the mountpoint directory (though not the mount itself) exists throughout this test. """ if not self._mountpoint: self._mountpoint = tempfile.mkdtemp(dir=self.tempdir) return self._mountpoint @property def tempdir(self): """The temporary directory used to store disk images and other data this test requires. This directory will be removed by Creator.cleanup. It is up to the caller to call that method, though. """ if not self._tempdir: self._tempdir = tempfile.mkdtemp(prefix="%s-" % self.name, dir="/var/tmp") return self._tempdir @property def suitename(self): return self.name + "_suite" @property def suitepath(self): return self.tempdir + "/" + self.suitename class OutsideMixin(object): """A BaseOutsideTestCase subclass is the interface between the unittest framework and a running VM. It interfaces with an associated Creator object to create devices and fire up a VM, and also handles actually reporting a result that unittest knows how to process. Each subclass will likely only want to define a single attribute: creatorClass -- A Creator subclass that goes with this test. """ creatorClass = None def archive(self): self.creator.archive() def runTest(self): self.creator.run() with self.creator.suiteMounted(): self.assertTrue(os.path.exists(self.creator.mountpoint + "/result"), msg="results directory does not exist") self.archive() self.assertFalse(os.path.exists(self.creator.mountpoint + "/result/unittest-failures"), msg="automated UI test %s failed" % self.creator.name) def setUp(self): # pylint: disable=not-callable self.creator = self.creatorClass() self.creator.makeDrives() self.creator.makeSuite() def tearDown(self): self.creator.cleanup()