1
# TargetNonInteractive.py
3
# abstraction for non-interactive backends (like chroot, qemu)
6
from DistUpgrade.DistUpgradeConfigParser import DistUpgradeConfig
11
import ConfigParser as configparser
14
from shutil import rmtree
16
# refactor the code so that we have
17
# UpgradeTest - the controler object
18
# UpgradeTestImage - abstraction for chroot/qemu/xen
20
class UpgradeTestImage(object):
21
def runInTarget(self, command):
23
def copyToImage(self, fromFile, toFile):
25
def copyFromImage(self, fromFile, toFile):
27
def bootstrap(self, force=False):
34
class UpgradeTestBackend(object):
35
""" This is a abstrace interface that all backends (chroot, qemu)
36
should implement - very basic currently :)
39
apt_options = ["-y","--allow-unauthenticated"]
41
def __init__(self, profiledir, resultdir=""):
42
" init the backend with the given profile "
44
assert(profiledir != None)
45
profiledir = os.path.normpath(profiledir)
46
profile = os.path.join(os.path.abspath(profiledir), "DistUpgrade.cfg")
47
self.upgradefilesdir = "./DistUpgrade"
49
if os.path.exists("./post_upgrade_tests/"):
50
self.post_upgrade_tests_dir = "./post_upgrade_tests/"
52
self.post_upgrade_tests_dir = "/usr/share/auto-upgrade-tester/post_upgrade_tests/"
54
if os.path.exists(profile):
55
override_cfg_d = os.path.join(profiledir, "..", "override.cfg.d")
56
defaults_cfg_d = os.path.join(profiledir, "..", "defaults.cfg.d")
57
self.profile = os.path.abspath(profile)
58
self.config = DistUpgradeConfig(datadir=os.path.dirname(profile),
59
name=os.path.basename(profile),
60
override_dir=override_cfg_d,
61
defaults_dir=defaults_cfg_d)
63
raise IOError("Can't find profile '%s' (%s) " % (profile, os.getcwd()))
65
base_resultdir = resultdir
67
base_resultdir = self.config.getWithDefault(
68
"NonInteractive", "ResultDir", "results-upgrade-tester")
69
self.resultdir = os.path.abspath(
70
os.path.join(base_resultdir, profiledir.split("/")[-1]))
72
# Cleanup result directory before new run
73
if os.path.exists(self.resultdir):
74
rmtree(self.resultdir)
75
os.makedirs(self.resultdir)
77
self.fromDist = self.config.get("Sources","From")
78
if "http_proxy" in os.environ and not self.config.has_option("NonInteractive","Proxy"):
79
self.config.set("NonInteractive","Proxy", os.environ["http_proxy"])
80
elif self.config.has_option("NonInteractive","Proxy"):
81
proxy=self.config.get("NonInteractive","Proxy")
82
os.putenv("http_proxy",proxy)
83
os.putenv("DEBIAN_FRONTEND","noninteractive")
86
self.cachedir = self.config.get("NonInteractive","CacheDebs")
87
except configparser.NoOptionError:
89
# init a sensible environment (to ensure proper operation if
91
os.environ["PATH"] = "/usr/sbin:/usr/bin:/sbin:/bin"
93
def installPackages(self, pkgs):
95
install packages in the image
99
def getSourcesListFile(self):
101
creates a temporary sources.list file and returns it to
104
# write new sources.list
105
sourceslist = tempfile.NamedTemporaryFile()
106
comps = self.config.getlist("NonInteractive","Components")
107
pockets = self.config.getlist("NonInteractive","Pockets")
108
mirror = self.config.get("NonInteractive","Mirror")
109
sourceslist.write("deb %s %s %s\n" % (mirror, self.fromDist, " ".join(comps)))
110
for pocket in pockets:
111
sourceslist.write("deb %s %s-%s %s\n" % (mirror, self.fromDist,pocket, " ".join(comps)))
116
" bootstaps a pristine install"
120
" upgrade a given install "
124
" test if the upgrade was successful "
127
def resultsToJunitXML(self, results, outputfile = None):
129
Filter results to get Junit XML output
131
:param results: list of results. Each result is a dictionary of the form
132
name: name of the test
133
result: (pass, fail, error)
134
time: execution time of the test in seconds
135
message: optional message in case of failure or error
136
:param output: Output XML to this file instead of returning the value
138
from xml.sax.saxutils import escape
142
res = [x['result'] for x in results]
143
fail_count = res.count('fail')
144
error_count = res.count('error')
145
total_count = len(res)
146
total_time = sum([x['time'] for x in results])
148
output = """<testsuite errors="%d" failures="%d" name="%s" tests="%d" time="%.3f">\n""" % (
149
error_count, fail_count, testsuite_name, total_count,total_time)
151
for result in results:
152
output += """<testcase classname="%s" name="%s" time="%.3f">\n""" % (
153
self.profilename + '.PostUpgradeTest',
154
result['name'][:-3], result['time'])
155
if 'fail' in result['result']:
156
output += """<failure type="%s">%s\n</failure>\n""" % (
157
'exception', escape(result['message']))
158
elif 'error' in result['result']:
159
output += """<error type="%s">%s\n</error>\n""" % (
160
'exception', escape(result['message']))
162
output += "</testcase>\n"
163
output += "</testsuite>\n"
166
with open(outputfile, 'w') as f: