Refactor the server and client side test classes into a the common lib
Big change here. Pull all of the common code in the server and client
versions of test into the common lib. This also requires pulling
chunks of the client and server utility functions into the common lib
as well.
Signed-off-by: John Admanski <jadmanski@google.com>
git-svn-id: http://test.kernel.org/svn/autotest/trunk@1215 592f7852-d20e-0410-864c-8624ca9c26a4
diff --git a/server/test.py b/server/test.py
index 159aba7..2be1734 100755
--- a/server/test.py
+++ b/server/test.py
@@ -1,195 +1,26 @@
# Copyright Martin J. Bligh, Andy Whitcroft, 2007
#
-# Shell class for a test, inherited by all individual tests
+# Define the server-side test class
#
-# Methods:
-# __init__ initialise
-# initialize run once for each job
-# setup run once for each new version of the test installed
-# run run the test (wrapped by job.run_test())
-#
-# Data:
-# job backreference to the job this test instance is part of
-# outputdir eg. results/<job>/<testname.tag>
-# resultsdir eg. results/<job>/<testname.tag>/results
-# profdir eg. results/<job>/<testname.tag>/profiling
-# debugdir eg. results/<job>/<testname.tag>/debug
-# bindir eg. tests/<test>
-# src eg. tests/<test>/src
-# tmpdir eg. tmp/<test>
-import os, pickle, tempfile, re
+import os
+
from subcommand import *
from common.error import *
from utils import *
-class test:
- preserve_srcdir = False
-
- def __init__(self, job, bindir, outputdir):
- testname = self.__class__.__name__
-
- self.job = job
-
- self.outputdir = outputdir
- tagged_testname = os.path.basename(self.outputdir)
- # check if the outputdir already exists, because if it does
- # then this test has already been run with the same tag earlier
- # in this job
- if os.path.exists(self.outputdir):
- testname, tag = (tagged_testname + '.').split('.', 1)
- msg = ("%s already exists, test <%s> may have already "
- + "run with tag <%s>") % (tagged_testname,
- testname, tag)
- raise TestError(msg)
- else:
- os.mkdir(self.outputdir)
-
- self.resultsdir = self.outputdir + "/results"
- os.mkdir(self.resultsdir)
- self.profdir = self.outputdir + "/profiling"
- os.mkdir(self.profdir)
- self.debugdir = self.outputdir + "/debug"
- os.mkdir(self.debugdir)
-
- self.bindir = bindir
- self.srcdir = bindir + '/src'
-
- self.tmpdir = job.tmpdir + '/' + testname
- if os.path.exists(self.tmpdir):
- system('rm -rf ' + self.tmpdir)
- os.mkdir(self.tmpdir)
-
- self.initialize()
- # compile and install the test, if needed.
- update_version(self.srcdir, self.preserve_srcdir, self.version, self.setup)
+import common.test
- def assert_(self, expr, msg='Assertion failed.'):
- if not expr:
- raise TestError(msg)
+class test(common.test.base_test):
+ pass
- def initialize(self):
- pass
-
-
- def setup(self):
- pass
-
-
- def cleanup(self):
- pass
-
-
- def run(self, args, dargs):
- try:
- # self.job.stdout.tee_redirect(
- # os.path.join(self.debugdir, 'stdout'))
- # self.job.stderr.tee_redirect(
- # os.path.join(self.debugdir, 'stderr'))
-
- try:
- os.chdir(self.outputdir)
- write_keyval(self.outputdir, { 'version':self.version })
- self.execute(*args, **dargs)
- finally:
- self.cleanup()
- # self.job.stderr.restore()
- # self.job.stdout.restore()
- except AutotestError:
- raise
- except:
- raise UnhandledError('running test ' + \
- self.__class__.__name__ + "\n")
-
-
-def testname(url):
- # Extract the testname from the test url.
- match = re.match('[^:]+://(.*)/([^/]*)$', url)
- if not match:
- return ('', url)
- (group, filename) = match.groups()
-
- # Generate the group prefix.
- gfix = re.compile('\W')
- group = gfix.sub('_', group)
-
- # Drop the extension to get the raw test name.
- tfix = re.compile('\.tgz')
- testname = tfix.sub('', filename)
-
- return (group, testname)
-
-
-def __installtest(job, url):
- (group, name) = testname(url)
-
- ##print "group=%s name=%s" % (group, name)
-
- # Bail if the test is already installed
- group_dir = os.path.join(job.testdir, "download", group)
- if os.path.exists(os.path.join(group_dir, name)):
- return (group, name)
-
- # If the group directory is missing create it and add
- # an empty __init__.py so that sub-directories are
- # considered for import.
- if not os.path.exists(group_dir):
- os.mkdir(group_dir)
- f = file(os.path.join(group_dir, '__init__.py'), 'w+')
- f.close()
-
- print name + ": installing test url=" + url
- system("wget %s -O %s" % (url, os.path.join(group_dir, 'test.tgz')))
- system("cd %s; tar zxf %s" % (group_dir, 'test.tgz'))
- os.unlink(os.path.join(group_dir, 'test.tgz'))
-
- # For this 'sub-object' to be importable via the name
- # 'group.name' we need to provide an __init__.py,
- # so link the main entry point to this.
- os.symlink(name + '.py', os.path.join(group_dir, name,
- '__init__.py'))
-
- # The test is now installed.
- return (group, name)
-
-
-# runtest: main interface for importing and instantiating new tests.
-def __runtest(job, url, tag, args, dargs):
- (group, testname) = ('', url)
- bindir = os.path.join(job.testdir, group, testname)
-
- outputdir = os.path.join(job.resultdir, testname)
- if (tag):
- outputdir += '.' + tag
- if os.path.exists(outputdir):
- system('rm -rf ' + outputdir)
-
- if not os.path.exists(bindir):
- raise TestError(testname + ": test does not exist")
-
- try:
- if group:
- sys.path.insert(0, os.path.join(job.testdir,
- "download"))
- group += '.'
- else:
- sys.path.insert(0, os.path.join(job.testdir, testname))
-
- exec "import %s%s" % (group, testname)
- exec "mytest = %s%s.%s(job, bindir, outputdir)" % \
- (group, testname, testname)
- finally:
- sys.path.pop(0)
-
- pwd = os.getcwd()
- os.chdir(outputdir)
- mytest.run(args, dargs)
- os.chdir(pwd)
+testname = common.test.testname
def runtest(job, url, tag, args, dargs):
- t = subcommand(__runtest, [job, url, tag, args, dargs])
+ t = subcommand(common.test.runtest,
+ [job, url, tag, args, dargs, locals(), globals()])
t.fork_start()
t.fork_waitfor()
diff --git a/server/utils.py b/server/utils.py
index 9c534fb..75c4725 100644
--- a/server/utils.py
+++ b/server/utils.py
@@ -17,6 +17,9 @@
import hosts
from common.error import *
+from common.utils import *
+
+
# A dictionary of pid and a list of tmpdirs for that pid
__tmp_dirs = {}
@@ -365,42 +368,6 @@
return source_material
-def write_keyval(dirname, dictionary):
- keyval = open(os.path.join(dirname, 'keyval'), 'w')
- for key in dictionary.keys():
- value = '%s' % dictionary[key] # convert numbers to strings
- if re.search(r'\W', key):
- raise ValueError('Invalid key: %s' % key)
- keyval.write('%s=%s\n' % (key, str(value)))
- keyval.close()
-
-
-def update_version(srcdir, preserve_srcdir, new_version, install, *args, **dargs):
- """
- Make sure srcdir is version new_version
-
- If not, delete it and install() the new version.
-
- In the preserve_srcdir case, we just check it's up to date,
- and if not, we rerun install, without removing srcdir
- """
- versionfile = srcdir + '/.version'
- install_needed = True
-
- if os.path.exists(srcdir):
- if os.path.exists(versionfile):
- old_version = pickle.load(open(versionfile, 'r'))
- if (old_version == new_version):
- install_needed = False
-
- if install_needed:
- if not preserve_srcdir:
- system('rm -rf ' + srcdir)
- install(*args, **dargs)
- if os.path.exists(srcdir):
- pickle.dump(new_version, open(versionfile, 'w'))
-
-
def get_server_dir():
path = os.path.dirname(sys.modules['utils'].__file__)
return os.path.abspath(path)