test.py revision 75f161ac96565cb1ae13dd3f7bf0002c003da66f 
1# Copyright Martin J. Bligh, Andy Whitcroft, 2006
2#
3# Shell class for a test, inherited by all individual tests
4#
5# Methods:
6#	__init__	initialise
7#	initialize	run once for each job
8#	setup		run once for each new version of the test installed
9#	run		run the test (wrapped by job.run_test())
10#
11# Data:
12#	job		backreference to the job this test instance is part of
13#	outputdir	eg. results/<job>/<testname.tag>
14#	resultsdir	eg. results/<job>/<testname.tag>/results
15#	profdir		eg. results/<job>/<testname.tag>/profiling
16#	debugdir	eg. results/<job>/<testname.tag>/debug
17#	bindir		eg. tests/<test>
18#	src		eg. tests/<test>/src
19#	tmpdir		eg. tmp/<testname.tag>
20
21import os, pickle, tempfile, fcntl, traceback
22from autotest_utils import *
23from common.error import *
24import sysinfo
25
26class test:
27	preserve_srcdir = False
28
29	def __init__(self, job, bindir, outputdir):
30		self.job = job
31		self.autodir = job.autodir
32		self.outputdir = outputdir
33		os.mkdir(self.outputdir)
34		self.resultsdir = os.path.join(self.outputdir, 'results')
35		os.mkdir(self.resultsdir)
36		self.profdir = os.path.join(self.outputdir, 'profiling')
37		os.mkdir(self.profdir)
38		self.debugdir = os.path.join(self.outputdir, 'debug')
39		os.mkdir(self.debugdir)
40
41		self.bindir = bindir
42		self.srcdir = os.path.join(self.bindir, 'src')
43
44		tagged_testname = os.path.basename(self.outputdir)
45		self.tmpdir = os.path.join(job.tmpdir, tagged_testname)
46
47		if os.path.exists(self.tmpdir):
48			system('rm -rf ' + self.tmpdir)
49		os.mkdir(self.tmpdir)
50
51		self.job.stdout.tee_redirect(
52			os.path.join(self.debugdir, 'stdout'))
53		self.job.stderr.tee_redirect(
54			os.path.join(self.debugdir, 'stderr'))
55		try:
56			self.initialize()
57			# compile and install the test, if needed.
58			update_version(self.srcdir, self.preserve_srcdir,
59						self.version, self.setup)
60		finally:
61			self.job.stderr.restore()
62			self.job.stdout.restore()
63
64
65	def initialize(self):
66		pass
67
68
69	def setup(self):
70		pass
71
72
73	def cleanup(self):
74		pass
75
76
77	def _exec(self, args, dargs):
78		try:
79			self.job.stdout.tee_redirect(
80				os.path.join(self.debugdir, 'stdout'))
81			self.job.stderr.tee_redirect(
82				os.path.join(self.debugdir, 'stderr'))
83
84			try:
85				os.chdir(self.outputdir)
86				write_keyval(self.outputdir,
87						{ 'version' : self.version })
88				self.execute(*args, **dargs)
89			finally:
90				self.cleanup()
91				self.job.stderr.restore()
92				self.job.stdout.restore()
93		except AutotestError:
94			raise
95		except:
96			raise UnhandledError('running test ' + \
97				self.__class__.__name__ + "\n")
98
99
100def testname(url):
101	# Extract the testname from the test url.
102	match = re.match('[^:]+://(.*)/([^/]*)$', url)
103	if not match:
104		return ('', url)
105	(group, filename) = match.groups()
106
107	# Generate the group prefix.
108	gfix = re.compile('\W')
109	group = gfix.sub('_', group)
110
111	# Drop the extension to get the raw test name.
112	tfix = re.compile('\.tgz')
113	testname = tfix.sub('', filename)
114
115	return (group, testname)
116
117
118def __installtest(job, url):
119	(group, name) = testname(url)
120
121	##print "group=%s name=%s" % (group, name)
122
123	# Bail if the test is already installed
124	group_dir = os.path.join(job.testdir, "download", group)
125	if os.path.exists(os.path.join(group_dir, name)):
126		return (group, name)
127
128	# If the group directory is missing create it and add
129	# an empty  __init__.py so that sub-directories are
130	# considered for import.
131	if not os.path.exists(group_dir):
132		os.mkdir(group_dir)
133		f = file(os.path.join(group_dir, '__init__.py'), 'w+')
134		f.close()
135
136	print name + ": installing test url=" + url
137	system("wget %s -O %s" % (url, os.path.join(group_dir, 'test.tgz')))
138	system("cd %s; tar zxf %s" % (group_dir, 'test.tgz'))
139	os.unlink(os.path.join(group_dir, 'test.tgz'))
140
141	# For this 'sub-object' to be importable via the name
142	# 'group.name' we need to provide an __init__.py,
143	# so link the main entry point to this.
144	os.symlink(name + '.py', os.path.join(group_dir, name,
145				'__init__.py'))
146
147	# The test is now installed.
148	return (group, name)
149
150
151# runtest: main interface for importing and instantiating new tests.
152def runtest(job, url, tag, args, dargs):
153	# If this is not a plain test name then download and install
154	# the specified test.
155	if is_url(url):
156		(group, testname) = __installtest(job, url)
157		bindir = os.path.join(job.testdir, "download", group, testname)
158	else:
159		(group, testname) = ('', url)
160		bindir = os.path.join(job.testdir, group, testname)
161
162	outputdir = os.path.join(job.resultdir, testname)
163
164	if (tag):
165		outputdir += '.' + tag
166	if not os.path.exists(bindir):
167		raise TestError(testname + ": test does not exist")
168
169	if group:
170		sys.path.insert(0, os.path.join(job.testdir, "download"))
171		group += '.'
172	else:
173		sys.path.insert(0, os.path.join(job.testdir, testname))
174
175	try:
176		lockfile = open(os.path.join(job.tmpdir, ".testlock"), "w")
177		fcntl.flock(lockfile, fcntl.LOCK_EX)
178		exec "import %s%s" % (group, testname)
179		exec "mytest = %s%s.%s(job, bindir, outputdir)" % \
180			(group, testname, testname)
181	finally:
182		fcntl.flock(lockfile, fcntl.LOCK_UN)
183		lockfile.close()
184		sys.path.pop(0)
185
186	pwd = os.getcwd()
187	os.chdir(outputdir)
188	dmesg = os.path.join(mytest.debugdir, 'dmesg')
189	try:
190		mytest._exec(args, dargs)
191	finally:
192		try:
193			# log "after each test" sysinfo
194			sysinfo_dir = os.path.join(mytest.outputdir, 'sysinfo')
195			os.makedirs(sysinfo_dir)
196			try:
197				os.chdir(sysinfo_dir)
198				sysinfo.after_each_test()
199			finally:
200				os.chdir(pwd)
201
202			if os.path.exists(mytest.tmpdir):
203				system('rm -rf ' + mytest.tmpdir)
204		except:
205			print 'post-test error:'
206			traceback.print_exc(file=sys.stdout)
207