/
test_urls.py
265 lines (227 loc) · 8.57 KB
/
test_urls.py
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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
# Copyright (C) 2013 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA 02110-1301 USA.
import logging
import os
import sys
import time
import traceback
import unittest
from tests import utils
from virtinst import Guest
from virtinst import urlfetcher
from virtinst import util
from virtinst.urlfetcher import ALTLinuxDistro
from virtinst.urlfetcher import CentOSDistro
from virtinst.urlfetcher import DebianDistro
from virtinst.urlfetcher import FedoraDistro
from virtinst.urlfetcher import GenericDistro
from virtinst.urlfetcher import MandrivaDistro
from virtinst.urlfetcher import RHELDistro
from virtinst.urlfetcher import SLDistro
from virtinst.urlfetcher import SuseDistro
from virtinst.urlfetcher import UbuntuDistro
class _DistroURL(object):
def __init__(self, name, url, detectdistro,
testxen, testbootiso, testshortcircuit):
self.name = name
self.url = url
self.detectdistro = detectdistro
self.arch = self._find_arch()
self.distroclass = self._distroclass_for_name(self.name)
self.testxen = testxen
self.testbootiso = testbootiso
# If True, pass in the expected distro value to getDistroStore
# so it can short circuit the lookup checks. Speeds up the tests
# and exercises the shortcircuit infrastructure
self.testshortcircuit = testshortcircuit
def _distroclass_for_name(self, name):
# Map the test case name to the expected urlfetcher distro
# class we should be detecting
if "fedora" in name:
return FedoraDistro
if "centos" in name:
return CentOSDistro
if "rhel" in name:
return RHELDistro
if "suse" in name:
return SuseDistro
if "debian" in name:
return DebianDistro
if name.startswith("sl-"):
return SLDistro
if "ubuntu" in name:
return UbuntuDistro
if "mageia" in name:
return MandrivaDistro
if "altlinux" in name:
return ALTLinuxDistro
if "generic" in name:
return GenericDistro
raise RuntimeError("name=%s didn't map to any distro class. Extend "
"_distroclass_for_name" % name)
def _find_arch(self):
if ("i686" in self.url or
"i386" in self.url or
"i586" in self.url):
return "i686"
if ("arm64" in self.url or
"aarch64" in self.url):
return "aarch64"
if ("ppc64el" in self.url or
"ppc64le" in self.url):
return "ppc64le"
if "s390" in self.url:
return "s390x"
if ("x86_64" in self.url or
"amd64" in self.url):
return "x86_64"
return "x86_64"
testconn = utils.open_testdefault()
hvmguest = Guest(testconn)
hvmguest.os.os_type = "hvm"
xenguest = Guest(testconn)
xenguest.os.os_type = "xen"
meter = util.make_meter(quiet=not utils.get_debug())
def _storeForDistro(fetcher, guest):
"""
Helper to lookup the Distro store object, basically detecting the
URL. Handle occasional proxy errors
"""
for ignore in range(0, 10):
try:
return urlfetcher.getDistroStore(guest, fetcher)
except Exception as e:
if str(e).count("502"):
logging.debug("Caught proxy error: %s", str(e))
time.sleep(.5)
continue
raise
raise # pylint: disable=misplaced-bare-raise
def _testURL(fetcher, distroobj):
"""
Test that our URL detection logic works for grabbing kernel, xen
kernel, and boot.iso
"""
distname = distroobj.name
arch = distroobj.arch
hvmguest.os.arch = arch
xenguest.os.arch = arch
if distroobj.testshortcircuit:
hvmguest.os_variant = distroobj.detectdistro
xenguest.os_variant = distroobj.detectdistro
else:
hvmguest.os_variant = None
xenguest.os_variant = None
try:
hvmstore = _storeForDistro(fetcher, hvmguest)
xenstore = None
if distroobj.testxen:
xenstore = _storeForDistro(fetcher, xenguest)
except Exception:
raise AssertionError("\nFailed to detect URLDistro class:\n"
"name = %s\n"
"url = %s\n\n%s" %
(distname, fetcher.location, "".join(traceback.format_exc())))
for s in [hvmstore, xenstore]:
if (s and distroobj.distroclass and
not isinstance(s, distroobj.distroclass)):
raise AssertionError("Unexpected URLDistro class:\n"
"found = %s\n"
"expect = %s\n"
"name = %s\n"
"url = %s" %
(s.__class__, distroobj.distroclass, distname,
fetcher.location))
# Make sure the stores are reporting correct distro name/variant
if (s and distroobj.detectdistro and
distroobj.detectdistro != s.get_osdict_info()):
raise AssertionError(
"Detected OS did not match expected values:\n"
"found = %s\n"
"expect = %s\n"
"name = %s\n"
"url = %s\n"
"store = %s" %
(s.os_variant, distroobj.detectdistro,
distname, fetcher.location, distroobj.distroclass))
# Do this only after the distro detection, since we actually need
# to fetch files for that part
def fakeAcquireFile(filename):
logging.debug("Fake acquiring %s", filename)
return fetcher.hasFile(filename)
fetcher.acquireFile = fakeAcquireFile
# Fetch boot iso
if distroobj.testbootiso:
boot = hvmstore.acquireBootDisk(hvmguest)
logging.debug("acquireBootDisk: %s", str(boot))
if boot is not True:
raise AssertionError("%s-%s: bootiso fetching failed" %
(distname, arch))
# Fetch regular kernel
kern = hvmstore.acquireKernel(hvmguest)
logging.debug("acquireKernel (hvm): %s", str(kern))
if kern[0] is not True or kern[1] is not True:
AssertionError("%s-%s: hvm kernel fetching failed" %
(distname, arch))
# Fetch xen kernel
if xenstore:
kern = xenstore.acquireKernel(xenguest)
logging.debug("acquireKernel (xen): %s", str(kern))
if kern[0] is not True or kern[1] is not True:
raise AssertionError("%s-%s: xen kernel fetching" %
(distname, arch))
def _testURLWrapper(distroobj):
os.environ.pop("VIRTINST_TEST_SUITE", None)
logging.debug("Testing for media arch=%s distroclass=%s",
distroobj.arch, distroobj.distroclass)
sys.stdout.write("\nTesting %-25s " % distroobj.name)
sys.stdout.flush()
fetcher = urlfetcher.fetcherForURI(distroobj.url, "/tmp", meter)
try:
fetcher.prepareLocation()
return _testURL(fetcher, distroobj)
finally:
fetcher.cleanupLocation()
# Register tests to be picked up by unittest
class URLTests(unittest.TestCase):
pass
def _make_tests():
import configparser
cfg = configparser.ConfigParser()
cfg.read("tests/test_urls.ini")
manualpath = "tests/test_urls_manual.ini"
cfg.read(manualpath)
if not os.path.exists(manualpath):
print("NOTE: Pass in manual data with %s" % manualpath)
urls = {}
for name in cfg.sections():
vals = dict(cfg.items(name))
d = _DistroURL(name, vals["url"],
vals.get("distro", None),
vals.get("testxen", "0") == "1",
vals.get("testbootiso", "0") == "1",
vals.get("testshortcircuit", "0") == "1")
urls[d.name] = d
keys = urls.keys()
keys.sort()
for key in keys:
distroobj = urls[key]
def _make_wrapper(d):
return lambda _self: _testURLWrapper(d)
setattr(URLTests, "testURL%s" % key.replace("-", "_"),
_make_wrapper(distroobj))
_make_tests()