summaryrefslogtreecommitdiff
path: root/tests/test_inject.py
blob: 21099cb73723d6c6bf14b343372e692adae76b57 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
#!/usr/bin/python2 -tt
# Copyright (C) 2013, 2014 Red Hat, Inc.

import atexit
import os
import sys
import unittest

import urlgrabber

from tests import INITRD_TEST_DISTROS
from tests import utils

from virtinst import Guest
from virtinst import urlfetcher
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 = urlgrabber.progress.TextMeter(fo=sys.stdout)

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()