test.py revision 0afbb6369aa5aa9a75ea67dd9e95ec4b21c0c181 
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, warnings
20
21from autotest_lib.client.common_lib import error, utils
22
23
24class base_test:
25    preserve_srcdir = False
26
27    def __init__(self, job, bindir, outputdir):
28        self.job = job
29        self.autodir = job.autodir
30
31        self.outputdir = outputdir
32        tagged_testname = os.path.basename(self.outputdir)
33        self.resultsdir = os.path.join(self.outputdir, 'results')
34        os.mkdir(self.resultsdir)
35        self.profdir = os.path.join(self.outputdir, 'profiling')
36        os.mkdir(self.profdir)
37        self.debugdir = os.path.join(self.outputdir, 'debug')
38        os.mkdir(self.debugdir)
39        self.bindir = bindir
40        if hasattr(job, 'libdir'):
41            self.libdir = job.libdir
42        self.srcdir = os.path.join(self.bindir, 'src')
43
44        self.tmpdir = os.path.join(job.tmpdir, tagged_testname)
45
46        if os.path.exists(self.tmpdir):
47            shutil.rmtree(self.tmpdir)
48        os.mkdir(self.tmpdir)
49
50        self.job.stdout.tee_redirect(
51                os.path.join(self.debugdir, 'stdout'))
52        self.job.stderr.tee_redirect(
53                os.path.join(self.debugdir, 'stderr'))
54        try:
55            self.initialize()
56            # compile and install the test, if needed.
57            utils.update_version(self.srcdir, self.preserve_srcdir,
58                                 self.version, self.setup)
59        finally:
60            self.job.stderr.restore()
61            self.job.stdout.restore()
62
63
64    def assert_(self, expr, msg='Assertion failed.'):
65        if not expr:
66            raise error.TestError(msg)
67
68
69    def write_test_keyval(self, attr_dict):
70        utils.write_keyval(self.outputdir, attr_dict)
71
72
73    @staticmethod
74    def _append_type_to_keys(dictionary, typename):
75        new_dict = {}
76        for key, value in dictionary.iteritems():
77            new_key = "%s{%s}" % (key, typename)
78            new_dict[new_key] = value
79        return new_dict
80
81
82    def write_iteration_keyval(self, attr_dict, perf_dict):
83        attr_dict = self._append_type_to_keys(attr_dict, "attr")
84        perf_dict = self._append_type_to_keys(perf_dict, "perf")
85
86        utils.write_keyval(self.resultsdir, attr_dict,
87                           type_tag="attr")
88        utils.write_keyval(self.resultsdir, perf_dict,
89                           type_tag="perf")
90
91        keyval_path = os.path.join(self.resultsdir, "keyval")
92        print >> open(keyval_path, "a"), ""
93
94
95    # TODO: deprecate, remove from code in favour of
96    # the write_*_keyval methods
97    def write_keyval(self, dictionary):
98        warnings.warn("test.write_keyval is deprecated, use "
99                      "test.write_test_keyval or "
100                      "test.write_iteration_keyval instead",
101                      DeprecationWarning)
102        self.write_iteration_keyval({}, dictionary)
103
104
105    def initialize(self):
106        pass
107
108
109    def setup(self):
110        pass
111
112
113    def cleanup(self):
114        pass
115
116
117    def _exec(self, args, dargs):
118        try:
119            self.job.stdout.tee_redirect(
120                os.path.join(self.debugdir, 'stdout'))
121            self.job.stderr.tee_redirect(
122                os.path.join(self.debugdir, 'stderr'))
123
124            try:
125                os.chdir(self.outputdir)
126                dargs   = dargs.copy()
127                keyvals = dargs.pop('test_attributes', dict())
128                keyvals = keyvals.copy()
129                keyvals['version'] = self.version
130                self.write_test_keyval(keyvals)
131
132                self.execute(*args, **dargs)
133            finally:
134                self.cleanup()
135                self.job.stderr.restore()
136                self.job.stdout.restore()
137        except error.AutotestError:
138            raise
139        except Exception, e:
140            msg = "Unhandled %s error occured during test\n"
141            msg %= str(e.__class__.__name__)
142            raise error.UnhandledError(msg)
143
144
145def testname(url):
146    # Extract the testname from the test url.
147    match = re.match('[^:]+://(.*)/([^/]*)$', url)
148    if not match:
149        return ('', url)
150    (group, filename) = match.groups()
151
152    # Generate the group prefix.
153    group = re.sub(r'\W', '_', group)
154
155    # Drop the extension to get the raw test name.
156    testname = re.sub(r'\.tgz', '', filename)
157
158    return (group, testname)
159
160
161def _installtest(job, url):
162    (group, name) = testname(url)
163
164    # Bail if the test is already installed
165    group_dir = os.path.join(job.testdir, "download", group)
166    if os.path.exists(os.path.join(group_dir, name)):
167        return (group, name)
168
169    # If the group directory is missing create it and add
170    # an empty  __init__.py so that sub-directories are
171    # considered for import.
172    if not os.path.exists(group_dir):
173        os.mkdir(group_dir)
174        f = file(os.path.join(group_dir, '__init__.py'), 'w+')
175        f.close()
176
177    print name + ": installing test url=" + url
178    get_file(url, os.path.join(group_dir, 'test.tgz'))
179    old_wd = os.getcwd()
180    os.chdir(group_dir)
181    tar = tarfile.open('test.tgz')
182    for member in tar.getmembers():
183        tar.extract(member)
184    tar.close()
185    os.chdir(old_wd)
186    os.remove(os.path.join(group_dir, 'test.tgz'))
187
188    # For this 'sub-object' to be importable via the name
189    # 'group.name' we need to provide an __init__.py,
190    # so link the main entry point to this.
191    os.symlink(name + '.py', os.path.join(group_dir, name,
192                            '__init__.py'))
193
194    # The test is now installed.
195    return (group, name)
196
197
198def runtest(job, url, tag, args, dargs,
199            local_namespace={}, global_namespace={}, after_test_hook=None):
200    local_namespace = local_namespace.copy()
201    global_namespace = global_namespace.copy()
202
203    # if this is not a plain test name then download and install the
204    # specified test
205    if utils.is_url(url):
206        (group, testname) = _installtest(job, url)
207        bindir = os.path.join(job.testdir, 'download', group, testname)
208        site_bindir = None
209    else:
210        # if the test is local, it can be found in either testdir
211        # or site_testdir. tests in site_testdir override tests
212        # defined in testdir
213        (group, testname) = ('', url)
214        bindir = os.path.join(job.testdir, group, testname)
215        if hasattr(job, 'site_testdir'):
216            site_bindir = os.path.join(job.site_testdir,
217                                       group, testname)
218        else:
219            site_bindir = None
220
221    outputdir = os.path.join(job.resultdir, testname)
222    if tag:
223        outputdir += '.' + tag
224
225    # if we can find the test in site_bindir, use this version
226    if site_bindir and os.path.exists(site_bindir):
227        bindir = site_bindir
228        testdir = job.site_testdir
229    elif os.path.exists(bindir):
230        testdir = job.testdir
231    elif not os.path.exists(bindir):
232        raise error.TestError(testname + ': test does not exist')
233
234    if group:
235        sys.path.insert(0, os.path.join(testdir, 'download'))
236        group += '.'
237    else:
238        sys.path.insert(0, os.path.join(testdir, testname))
239
240    local_namespace['job'] = job
241    local_namespace['bindir'] = bindir
242    local_namespace['outputdir'] = outputdir
243
244    lockfile = open(os.path.join(job.tmpdir, '.testlock'), 'w')
245    try:
246        fcntl.flock(lockfile, fcntl.LOCK_EX)
247        exec ("import %s%s" % (group, testname),
248              local_namespace, global_namespace)
249        exec ("mytest = %s%s.%s(job, bindir, outputdir)" %
250              (group, testname, testname),
251              local_namespace, global_namespace)
252    finally:
253        fcntl.flock(lockfile, fcntl.LOCK_UN)
254        lockfile.close()
255        sys.path.pop(0)
256
257    pwd = os.getcwd()
258    os.chdir(outputdir)
259    try:
260        mytest = global_namespace['mytest']
261        mytest._exec(args, dargs)
262    finally:
263        if after_test_hook:
264            after_test_hook(mytest)
265