virt-manager/tests/test_inject.py

#!/usr/bin/python2 -tt
# Copyright (C) 2013, 2014 Red Hat, Inc.

import atexit
import os
import sys
import unittest

from tests import INITRD_TEST_DISTROS
from tests import utils

from virtinst import Guest
from virtinst import urlfetcher
from virtinst import util
from virtinst.distroinstaller import _perform_initrd_injections

cleanup = []
_alldistros = {}

testconn = utils.open_testdefault()
guest = Guest(testconn)
guest.os.os_type = "hvm"
guest.os.arch = "x86_64"
meter = util.make_meter(quiet=False)

DEVFEDORA_URL = "http://dl.fedoraproject.org/pub/fedora/linux/development/%s/%s/os/"
OLD_FEDORA_URL = "http://dl.fedoraproject.org/pub/fedora/linux/releases/%s/Fedora/%s/os/"
FEDORA_URL = "http://dl.fedoraproject.org/pub/fedora/linux/releases/%s/Server/%s/os/"

(WARN_RHEL4,
 WARN_RHEL5,
 WARN_LATEST) = range(1, 4)


def prompt():
    sys.stdout.write("(press enter to continue)")
    return sys.stdin.readline()


class Distro(object):
    def __init__(self, name, url, warntype=WARN_LATEST,
                 ks2=False, virtio=True):
        self.name = name
        self.url = url
        self.virtio = virtio
        self.warntype = warntype

        self.ks = "tests/inject-data/old-kickstart.ks"
        if ks2:
            self.ks = "tests/inject-data/new-kickstart.ks"

        self.kernel = None
        self.initrd = None


def _add(*args, **kwargs):
    _d = Distro(*args, **kwargs)
    _alldistros[_d.name] = _d


_add("centos-4.9", "http://vault.centos.org/4.9/os/x86_64",
     warntype=WARN_RHEL4, ks2=True, virtio=False)
_add("centos-5-latest", "http://ftp.linux.ncsu.edu/pub/CentOS/5/os/x86_64/",
     warntype=WARN_RHEL5)
_add("centos-6-latest", "http://ftp.linux.ncsu.edu/pub/CentOS/6/os/x86_64/",
     warntype=WARN_RHEL5)
_add("centos-7-latest", "http://ftp.linux.ncsu.edu/pub/CentOS/7/os/x86_64/",
     ks2=True)
_add("fedora-20", OLD_FEDORA_URL % ("20", "x86_64"), ks2=True)
_add("fedora-21", FEDORA_URL % ("21", "x86_64"), ks2=True)
_add("fedora-22", DEVFEDORA_URL % ("22", "x86_64"), ks2=True)


def exit_cleanup():
    for f in cleanup or []:
        try:
            os.unlink(f)
        except:
            pass
atexit.register(exit_cleanup)


def _fetch_distro(distro):
    print "Fetching distro=%s" % distro.name

    fetcher = urlfetcher.fetcherForURI(distro.url, "/tmp", meter)
    origenv = os.environ.pop("VIRTINST_TEST_SUITE")
    try:
        fetcher.prepareLocation()
        store = urlfetcher.getDistroStore(guest, fetcher)
        kernel, initrd, ignore = store.acquireKernel(guest)
        cleanup.append(kernel)
        cleanup.append(initrd)
        distro.kernel = kernel
        distro.initrd = initrd
    finally:
        fetcher.cleanupLocation()
        if origenv:
            os.environ["VIRTINST_TEST_SUITE"] = origenv


def _test_distro(distro):
    os.system("clear")
    print "\n"
    if distro.warntype == WARN_RHEL4:
        print "RHEL4: Makes its way to the text installer, then chokes "
        print "on our bogus URI http://HEY-THIS-IS-OUR-BAD-KICKSTART-URL.com/"
    elif distro.warntype == WARN_RHEL5:
        print "RHEL5, RHEL6, Fedora < 17: You'll get an error about a "
        print "bogus bootproto ITREADTHEKICKSTART. This means anaconda "
        print "read our busted kickstart."
    else:
        print "RHEL7, Fedora >= 17: Chokes on the bogus URI in the early "
        print "console screen when fetching the installer squashfs image."

    originitrd = distro.initrd
    kernel = distro.kernel
    newinitrd = originitrd + ".copy"
    injectfile = distro.ks

    os.system("cp -f %s %s" % (originitrd, newinitrd))
    cleanup.append(newinitrd)
    _perform_initrd_injections(newinitrd, [injectfile], ".")

    nic = distro.virtio and "virtio" or "rtl8139"
    append = "-append \"ks=file:/%s\"" % os.path.basename(injectfile)
    cmd = ("sudo qemu-kvm -enable-kvm -name %s "
           "-cpu host -m 1500 -sdl "
           "-net bridge,br=virbr0 -net nic,model=%s "
           "-kernel %s -initrd %s %s" %
           (distro.name, nic, kernel, newinitrd, append))
    print "\n\n" + cmd
    os.system(cmd)


_printinitrd = False
_printfetch = False


class FetchTests(unittest.TestCase):
    def setUp(self):
        global _printfetch
        if _printfetch:
            return
        print """



This is an interactive test.

First step is we need to go and fetch a bunch of distro kernel/initrd
from public trees. This is going to take a while. Let it run then come
back later and we will be waiting to continue.

"""
        prompt()
        _printfetch = True


class InjectTests(unittest.TestCase):
    def setUp(self):
        global _printinitrd
        if _printinitrd:
            return

        print """


Okay, we have all the media. We are going to perform the initrd injection
of some broken kickstarts, then manually launch a qemu instance to verify
the kickstart is detected. How you know it's working depends on the distro.
When each test launches, we will print the manual verification instructions.

"""
        prompt()
        _printinitrd = True


def _make_tests():
    def _make_fetch_cb(_d):
        return lambda s: _fetch_distro(_d)
    def _make_check_cb(_d):
        return lambda s: _test_distro(_d)

    distros = INITRD_TEST_DISTROS or _alldistros.keys()
    idx = 0
    for d in distros:
        dobj = _alldistros[d]
        idx += 1
        setattr(FetchTests, "testFetch%.3d" % idx, _make_fetch_cb(dobj))
        setattr(InjectTests, "testInitrd%.3d" % idx, _make_check_cb(dobj))

_make_tests()