virt-manager/tests/test_inject.py

# Copyright (C) 2013, 2014 Red Hat, Inc.
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.


import atexit
import os
import logging
import sys
import unittest

from tests import utils

from virtinst import Guest
from virtinst import urldetect
from virtinst import urlfetcher
from virtinst import util
from virtinst.initrdinject import perform_initrd_injections

cleanup = []
_alldistros = {}

testconn = utils.URIs.open_testdefault_cached()
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/Server/%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)")
    sys.stdout.flush()
    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.11", "http://vault.centos.org/5.11/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-27", FEDORA_URL % ("27", "x86_64"), ks2=True)


def exit_cleanup():
    for f in cleanup or []:
        try:
            os.unlink(f)
        except Exception:
            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 = urldetect.getDistroStore(guest, fetcher)
        kernel, initrd, ignore = store.acquireKernel()
        cleanup.append(kernel)
        cleanup.append(initrd)
        distro.kernel = kernel
        distro.initrd = initrd
    except Exception:
        logging.error("Fetching distro=%s failed", distro.name, exc_info=True)
    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 -display gtk "
           "-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):
        self.failfast = True
        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)

    idx = 0
    for dname, dobj in _alldistros.items():
        idx += 1
        setattr(FetchTests, "testFetch%.3d_%s" %
                (idx, dname.replace("-", "_")), _make_fetch_cb(dobj))
        setattr(InjectTests, "testInitrd%.3d_%s" %
                (idx, dname.replace("-", "_")), _make_check_cb(dobj))

_make_tests()