test.py revision 313f12ceca6185ea4890fec8e66baf06fd5d6f40 
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, sys, re, fcntl, shutil, tarfile
20
21import error
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 error.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 error.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 error.AutotestError:
115			raise
116		except:
117			raise error.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.open('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 error.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	lockfile = open(os.path.join(job.tmpdir, '.testlock'), 'w')
221	try:
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