test.py revision b53a3474f65f3f631bddef6f743c2944fed186dc 
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 execute(self):
118        raise NotImplementedError("This function must be overriden by "
119                                  "the test class")
120
121
122    def _exec(self, args, dargs):
123        try:
124            self.job.stdout.tee_redirect(
125                os.path.join(self.debugdir, 'stdout'))
126            self.job.stderr.tee_redirect(
127                os.path.join(self.debugdir, 'stderr'))
128
129            try:
130                os.chdir(self.outputdir)
131                dargs   = dargs.copy()
132                keyvals = dargs.pop('test_attributes', dict())
133                keyvals = keyvals.copy()
134                keyvals['version'] = self.version
135                self.write_test_keyval(keyvals)
136
137                self.execute(*args, **dargs)
138            finally:
139                self.cleanup()
140                self.job.stderr.restore()
141                self.job.stdout.restore()
142        except error.AutotestError:
143            raise
144        except Exception, e:
145            raise error.UnhandledError(e)
146
147
148def testname(url):
149    # Extract the testname from the test url.
150    match = re.match('[^:]+://(.*)/([^/]*)$', url)
151    if not match:
152        return ('', url)
153    (group, filename) = match.groups()
154
155    # Generate the group prefix.
156    group = re.sub(r'\W', '_', group)
157
158    # Drop the extension to get the raw test name.
159    testname = re.sub(r'\.tgz', '', filename)
160
161    return (group, testname)
162
163
164def _installtest(job, url):
165    (group, name) = testname(url)
166
167    # Bail if the test is already installed
168    group_dir = os.path.join(job.testdir, "download", group)
169    if os.path.exists(os.path.join(group_dir, name)):
170        return (group, name)
171
172    # If the group directory is missing create it and add
173    # an empty  __init__.py so that sub-directories are
174    # considered for import.
175    if not os.path.exists(group_dir):
176        os.mkdir(group_dir)
177        f = file(os.path.join(group_dir, '__init__.py'), 'w+')
178        f.close()
179
180    print name + ": installing test url=" + url
181    utils.get_file(url, os.path.join(group_dir, 'test.tgz'))
182    old_wd = os.getcwd()
183    os.chdir(group_dir)
184    tar = tarfile.open('test.tgz')
185    for member in tar.getmembers():
186        tar.extract(member)
187    tar.close()
188    os.chdir(old_wd)
189    os.remove(os.path.join(group_dir, 'test.tgz'))
190
191    # For this 'sub-object' to be importable via the name
192    # 'group.name' we need to provide an __init__.py,
193    # so link the main entry point to this.
194    os.symlink(name + '.py', os.path.join(group_dir, name,
195                            '__init__.py'))
196
197    # The test is now installed.
198    return (group, name)
199
200
201def runtest(job, url, tag, args, dargs,
202            local_namespace={}, global_namespace={}, after_test_hook=None):
203    local_namespace = local_namespace.copy()
204    global_namespace = global_namespace.copy()
205
206    # if this is not a plain test name then download and install the
207    # specified test
208    if utils.is_url(url):
209        (group, testname) = _installtest(job, url)
210        bindir = os.path.join(job.testdir, 'download', group, testname)
211        site_bindir = None
212    else:
213        # if the test is local, it can be found in either testdir
214        # or site_testdir. tests in site_testdir override tests
215        # defined in testdir
216        (group, testname) = ('', url)
217        bindir = os.path.join(job.testdir, group, testname)
218        if hasattr(job, 'site_testdir'):
219            site_bindir = os.path.join(job.site_testdir,
220                                       group, testname)
221        else:
222            site_bindir = None
223
224    outputdir = os.path.join(job.resultdir, testname)
225    if tag:
226        outputdir += '.' + tag
227
228    # if we can find the test in site_bindir, use this version
229    if site_bindir and os.path.exists(site_bindir):
230        bindir = site_bindir
231        testdir = job.site_testdir
232    elif os.path.exists(bindir):
233        testdir = job.testdir
234    elif not os.path.exists(bindir):
235        raise error.TestError(testname + ': test does not exist')
236
237    if group:
238        sys.path.insert(0, os.path.join(testdir, 'download'))
239        group += '.'
240    else:
241        sys.path.insert(0, os.path.join(testdir, testname))
242
243    local_namespace['job'] = job
244    local_namespace['bindir'] = bindir
245    local_namespace['outputdir'] = outputdir
246
247    lockfile = open(os.path.join(job.tmpdir, '.testlock'), 'w')
248    try:
249        fcntl.flock(lockfile, fcntl.LOCK_EX)
250        exec ("import %s%s" % (group, testname),
251              local_namespace, global_namespace)
252        exec ("mytest = %s%s.%s(job, bindir, outputdir)" %
253              (group, testname, testname),
254              local_namespace, global_namespace)
255    finally:
256        fcntl.flock(lockfile, fcntl.LOCK_UN)
257        lockfile.close()
258        sys.path.pop(0)
259
260    pwd = os.getcwd()
261    os.chdir(outputdir)
262    try:
263        mytest = global_namespace['mytest']
264        mytest._exec(args, dargs)
265    finally:
266        if after_test_hook:
267            after_test_hook(mytest)
268