test.py revision 6231cd6de184639494528ef60a9aca482283556b
1# Shell class for a test, inherited by all individual tests 2# 3# Methods: 4# __init__ initialise 5# initialize run once for each job 6# setup run once for each new version of the test installed 7# run run the test (wrapped by job.run_test()) 8# 9# Data: 10# job backreference to the job this test instance is part of 11# outputdir eg. results/<job>/<testname.tag> 12# resultsdir eg. results/<job>/<testname.tag>/results 13# profdir eg. results/<job>/<testname.tag>/profiling 14# debugdir eg. results/<job>/<testname.tag>/debug 15# bindir eg. tests/<test> 16# src eg. tests/<test>/src 17# tmpdir eg. tmp/<testname.tag> 18 19import os, re, fcntl, shutil, tarfile 20 21from error import * 22from utils import write_keyval, is_url, update_version 23 24 25class base_test: 26 preserve_srcdir = False 27 28 def __init__(self, job, bindir, outputdir): 29 self.job = job 30 self.autodir = job.autodir 31 32 self.outputdir = outputdir 33 tagged_testname = os.path.basename(self.outputdir) 34 # check if the outputdir already exists, because if it does 35 # then this test has already been run with the same tag earlier 36 # in this job 37 if os.path.exists(self.outputdir): 38 testname, tag = (tagged_testname + '.').split('.', 1) 39 msg = ("%s already exists, test <%s> may have already " 40 + "run with tag <%s>") % (tagged_testname, 41 testname, tag) 42 raise TestError(msg) 43 else: 44 os.mkdir(self.outputdir) 45 46 self.resultsdir = os.path.join(self.outputdir, 'results') 47 os.mkdir(self.resultsdir) 48 self.profdir = os.path.join(self.outputdir, 'profiling') 49 os.mkdir(self.profdir) 50 self.debugdir = os.path.join(self.outputdir, 'debug') 51 os.mkdir(self.debugdir) 52 self.bindir = bindir 53 if hasattr(job, 'libdir'): 54 self.libdir = job.libdir 55 self.srcdir = os.path.join(self.bindir, 'src') 56 57 self.tmpdir = os.path.join(job.tmpdir, tagged_testname) 58 59 if os.path.exists(self.tmpdir): 60 shutil.rmtree(self.tmpdir) 61 os.mkdir(self.tmpdir) 62 63 self.job.stdout.tee_redirect( 64 os.path.join(self.debugdir, 'stdout')) 65 self.job.stderr.tee_redirect( 66 os.path.join(self.debugdir, 'stderr')) 67 try: 68 self.initialize() 69 # compile and install the test, if needed. 70 update_version(self.srcdir, self.preserve_srcdir, 71 self.version, self.setup) 72 finally: 73 self.job.stderr.restore() 74 self.job.stdout.restore() 75 76 77 def assert_(self, expr, msg='Assertion failed.'): 78 if not expr: 79 raise TestError(msg) 80 81 82 def write_keyval(self, dictionary): 83 write_keyval(self.resultsdir, dictionary) 84 85 86 def initialize(self): 87 pass 88 89 90 def setup(self): 91 pass 92 93 94 def cleanup(self): 95 pass 96 97 98 def _exec(self, args, dargs): 99 try: 100 self.job.stdout.tee_redirect( 101 os.path.join(self.debugdir, 'stdout')) 102 self.job.stderr.tee_redirect( 103 os.path.join(self.debugdir, 'stderr')) 104 105 try: 106 os.chdir(self.outputdir) 107 write_keyval(self.outputdir, 108 { 'version' : self.version }) 109 self.execute(*args, **dargs) 110 finally: 111 self.cleanup() 112 self.job.stderr.restore() 113 self.job.stdout.restore() 114 except AutotestError: 115 raise 116 except: 117 raise UnhandledError('running test ' + \ 118 self.__class__.__name__ + "\n") 119 120 121def testname(url): 122 # Extract the testname from the test url. 123 match = re.match('[^:]+://(.*)/([^/]*)$', url) 124 if not match: 125 return ('', url) 126 (group, filename) = match.groups() 127 128 # Generate the group prefix. 129 group = re.sub(r'\W', '_', group) 130 131 # Drop the extension to get the raw test name. 132 testname = re.sub(r'\.tgz', '', filename) 133 134 return (group, testname) 135 136 137def _installtest(job, url): 138 (group, name) = testname(url) 139 140 # Bail if the test is already installed 141 group_dir = os.path.join(job.testdir, "download", group) 142 if os.path.exists(os.path.join(group_dir, name)): 143 return (group, name) 144 145 # If the group directory is missing create it and add 146 # an empty __init__.py so that sub-directories are 147 # considered for import. 148 if not os.path.exists(group_dir): 149 os.mkdir(group_dir) 150 f = file(os.path.join(group_dir, '__init__.py'), 'w+') 151 f.close() 152 153 print name + ": installing test url=" + url 154 get_file(url, os.path.join(group_dir, 'test.tgz')) 155 old_wd = os.getcwd() 156 os.chdir(group_dir) 157 tar = tarfile.TarFile('test.tgz') 158 for member in tar.getmembers(): 159 tar.extract(member) 160 tar.close() 161 os.chdir(old_wd) 162 os.remove(os.path.join(group_dir, 'test.tgz')) 163 164 # For this 'sub-object' to be importable via the name 165 # 'group.name' we need to provide an __init__.py, 166 # so link the main entry point to this. 167 os.symlink(name + '.py', os.path.join(group_dir, name, 168 '__init__.py')) 169 170 # The test is now installed. 171 return (group, name) 172 173 174def runtest(job, url, tag, args, dargs, 175 local_namespace={}, global_namespace={}, after_test_hook=None): 176 local_namespace = local_namespace.copy() 177 global_namespace = global_namespace.copy() 178 179 # if this is not a plain test name then download and install the 180 # specified test 181 if is_url(url): 182 (group, testname) = _installtest(job, url) 183 bindir = os.path.join(job.testdir, 'download', group, testname) 184 site_bindir = None 185 else: 186 # if the test is local, it can be found in either testdir 187 # or site_testdir. tests in site_testdir override tests 188 # defined in testdir 189 (group, testname) = ('', url) 190 bindir = os.path.join(job.testdir, group, testname) 191 if hasattr(job, 'site_testdir'): 192 site_bindir = os.path.join(job.site_testdir, 193 group, testname) 194 else: 195 site_bindir = None 196 197 outputdir = os.path.join(job.resultdir, testname) 198 if tag: 199 outputdir += '.' + tag 200 201 # if we can find the test in site_bindir, use this version 202 if site_bindir and os.path.exists(site_bindir): 203 bindir = site_bindir 204 testdir = job.site_testdir 205 elif os.path.exists(bindir): 206 testdir = job.testdir 207 elif not os.path.exists(bindir): 208 raise TestError(testname + ': test does not exist') 209 210 if group: 211 sys.path.insert(0, os.path.join(testdir, 'download')) 212 group += '.' 213 else: 214 sys.path.insert(0, os.path.join(testdir, testname)) 215 216 local_namespace['job'] = job 217 local_namespace['bindir'] = bindir 218 local_namespace['outputdir'] = outputdir 219 220 try: 221 lockfile = open(os.path.join(job.tmpdir, '.testlock'), 'w') 222 fcntl.flock(lockfile, fcntl.LOCK_EX) 223 exec ("import %s%s" % (group, testname), 224 local_namespace, global_namespace) 225 exec ("mytest = %s%s.%s(job, bindir, outputdir)" % 226 (group, testname, testname), 227 local_namespace, global_namespace) 228 finally: 229 fcntl.flock(lockfile, fcntl.LOCK_UN) 230 lockfile.close() 231 sys.path.pop(0) 232 233 pwd = os.getcwd() 234 os.chdir(outputdir) 235 try: 236 mytest = global_namespace['mytest'] 237 mytest._exec(args, dargs) 238 finally: 239 if after_test_hook: 240 after_test_hook(mytest) 241