yocto-poky/bitbake/lib/toaster/contrib/tts/runner.py - stefanberger/openbmc - Gitiles

 #!/usr/bin/python

 # ex:ts=4:sw=4:sts=4:et
 # -*- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*-
 #
 # Copyright (C) 2015 Alexandru Damian for Intel Corp.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 2 as
 # published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License along
 # with this program; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.


 # This is the main test execution controller. It is designed to be run
 # manually from the command line, or to be called from a different program
 # that schedules test execution.
 #
 # Execute   runner.py -h   for help.


 from __future__ import print_function
 import sys, os
 import unittest, importlib
 import logging, pprint, json
 import re
 from shellutils import ShellCmdException, mkdirhier, run_shell_cmd

 import config

 # we also log to a file, in addition to console, because our output is important
 __log_file_name__ = os.path.join(os.path.dirname(__file__), "log/tts_%d.log" % config.OWN_PID)
 mkdirhier(os.path.dirname(__log_file_name__))
 __log_file__ = open(__log_file_name__, "w")
 __file_handler__ = logging.StreamHandler(__log_file__)
 __file_handler__.setFormatter(logging.Formatter("%(asctime)s %(levelname)s: %(message)s"))

 config.logger.addHandler(__file_handler__)

 # set up log directory
 try:
     if not os.path.exists(config.LOGDIR):
         os.mkdir(config.LOGDIR)
     else:
         if not os.path.isdir(config.LOGDIR):
             raise Exception("Expected log dir '%s' is not actually a directory." % config.LOGDIR)
 except OSError as exc:
     raise exc

 # creates the under-test-branch as a separate directory
 def set_up_test_branch(settings, branch_name):
     testdir = "%s/%s.%d" % (settings['workdir'], config.TEST_DIR_NAME, config.OWN_PID)

     # creates the host dir
     if os.path.exists(testdir):
         raise Exception("Test dir '%s'is already there, aborting" % testdir)

     # may raise OSError, is to be handled by the caller
     os.makedirs(testdir)


     # copies over the .git from the localclone
     run_shell_cmd("cp -a '%s'/.git '%s'" % (settings['localclone'], testdir))

     # add the remote if it doesn't exist
     crt_remotes = run_shell_cmd("git remote -v", cwd=testdir)
     remotes = [word for line in crt_remotes.split("\n") for word in line.split()]
     if not config.CONTRIB_REPO in remotes:
         remote_name = "tts_contrib"
         run_shell_cmd("git remote add %s %s" % (remote_name, config.CONTRIB_REPO), cwd=testdir)
     else:
         remote_name = remotes[remotes.index(config.CONTRIB_REPO) - 1]

     # do the fetch
     run_shell_cmd("git fetch %s -p" % remote_name, cwd=testdir)

     # do the checkout
     run_shell_cmd("git checkout origin/master && git branch -D %s; git checkout %s/%s -b %s && git reset --hard" % (branch_name, remote_name, branch_name, branch_name), cwd=testdir)

     return testdir


 def __search_for_tests():
     # we find all classes that can run, and run them
     tests = []
     for _, _, files_list in os.walk(os.path.dirname(os.path.abspath(__file__))):
         for module_file in [f[:-3] for f in files_list if f.endswith(".py") and not f.startswith("__init__")]:
             config.logger.debug("Inspecting module %s", module_file)
             current_module = importlib.import_module(module_file)
             crtclass_names = vars(current_module)
             for name in crtclass_names:
                 tested_value = crtclass_names[name]
                 if isinstance(tested_value, type(unittest.TestCase)) and issubclass(tested_value, unittest.TestCase):
                     tests.append((module_file, name))
         break
     return tests


 # boilerplate to self discover tests and run them
 def execute_tests(dir_under_test, testname):

     if testname is not None and "." in testname:
         tests = []
         tests.append(tuple(testname.split(".", 2)))
     else:
         tests = __search_for_tests()

     # let's move to the directory under test
     crt_dir = os.getcwd()
     os.chdir(dir_under_test)

     # execute each module
     # pylint: disable=broad-except
     # we disable the broad-except because we want to actually catch all possible exceptions
     try:
         # sorting the tests by the numeric order in the class name
         tests = sorted(tests, key=lambda x: int(re.search(r"[0-9]+", x[1]).group(0)))
         config.logger.debug("Discovered test clases: %s", pprint.pformat(tests))
         unittest.installHandler()
         suite = unittest.TestSuite()
         loader = unittest.TestLoader()
         result = unittest.TestResult()
         result.failfast = True
         for module_file, test_name in tests:
             suite.addTest(loader.loadTestsFromName("%s.%s" % (module_file, test_name)))
         config.logger.info("Running %d test(s)", suite.countTestCases())
         suite.run(result)

         for error in result.errors:
             config.logger.error("Exception on test: %s\n%s", error[0],
                                 "\n".join(["-- %s" % x for x in error[1].split("\n")]))

         for failure in result.failures:
             config.logger.error("Failed test: %s:\n%s\n", failure[0],
                                 "\n".join(["--  %s" % x for x in failure[1].split("\n")]))

         config.logger.info("Test results: %d ran, %d errors, %d failures", result.testsRun, len(result.errors), len(result.failures))

     except Exception as exc:
         import traceback
         config.logger.error("Exception while running test. Tracedump: \n%s", traceback.format_exc(exc))
     finally:
         os.chdir(crt_dir)
     return len(result.failures)

 # verify that we had a branch-under-test name as parameter
 def validate_args():
     from optparse import OptionParser
     parser = OptionParser(usage="usage: %prog [options] branch_under_test")

     parser.add_option("-t", "--test-dir", dest="testdir", default=None, help="Use specified directory to run tests, inhibits the checkout.")
     parser.add_option("-s", "--single", dest="singletest", default=None, help="Run only the specified test")

     (options, args) = parser.parse_args()
     if len(args) < 1:
         raise Exception("Please specify the branch to run on. Use option '-h' when in doubt.")
     return (options, args)


 # load the configuration options
 def read_settings():
     if not os.path.exists(config.SETTINGS_FILE) or not os.path.isfile(config.SETTINGS_FILE):
         raise Exception("Config file '%s' cannot be openend" % config.SETTINGS_FILE)
     return json.loads(open(config.SETTINGS_FILE, "r").read())


 # cleanup !
 def clean_up(testdir):
     run_shell_cmd("rm -rf -- '%s'" % testdir)

 def dump_info(settings, options, args):
     """ detailed information about current run configuration, for debugging purposes.
     """
     config.logger.debug("Settings:\n%s\nOptions:\n%s\nArguments:\n%s\n", settings, options, args)

 def main():
     (options, args) = validate_args()

     settings = read_settings()
     need_cleanup = False

     # dump debug info
     dump_info(settings, options, args)

     testdir = None
     no_failures = 1
     try:
         if options.testdir is not None and os.path.exists(options.testdir):
             testdir = os.path.abspath(options.testdir)
             config.logger.info("No checkout, using %s", testdir)
         else:
             need_cleanup = True
             testdir = set_up_test_branch(settings, args[0]) # we expect a branch name as first argument

         config.TESTDIR = testdir    # we let tests know where to run

         # ensure that the test dir only contains no *.pyc leftovers
         run_shell_cmd("find '%s' -type f -name *.pyc -exec rm {} \\;" % testdir)

         no_failures = execute_tests(testdir, options.singletest)

     except ShellCmdException as exc:
         import traceback
         config.logger.error("Error while setting up testing. Traceback: \n%s", traceback.format_exc(exc))
     finally:
         if need_cleanup and testdir is not None:
             clean_up(testdir)

     sys.exit(no_failures)

 if __name__ == "__main__":
     main()
	#!/usr/bin/python

	# ex:ts=4:sw=4:sts=4:et
	# -- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil --
	#
	# Copyright (C) 2015 Alexandru Damian for Intel Corp.
	#
	# This program is free software; you can redistribute it and/or modify
	# it under the terms of the GNU General Public License version 2 as
	# published by the Free Software Foundation.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.
	#
	# You should have received a copy of the GNU General Public License along
	# with this program; if not, write to the Free Software Foundation, Inc.,
	# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.


	# This is the main test execution controller. It is designed to be run
	# manually from the command line, or to be called from a different program
	# that schedules test execution.
	#
	# Execute runner.py -h for help.



	from __future__ import print_function
	import sys, os
	import unittest, importlib
	import logging, pprint, json
	import re
	from shellutils import ShellCmdException, mkdirhier, run_shell_cmd

	import config

	# we also log to a file, in addition to console, because our output is important
	__log_file_name__ = os.path.join(os.path.dirname(__file__), "log/tts_%d.log" % config.OWN_PID)
	mkdirhier(os.path.dirname(__log_file_name__))
	__log_file__ = open(__log_file_name__, "w")
	__file_handler__ = logging.StreamHandler(__log_file__)
	__file_handler__.setFormatter(logging.Formatter("%(asctime)s %(levelname)s: %(message)s"))

	config.logger.addHandler(__file_handler__)

	# set up log directory
	try:
	if not os.path.exists(config.LOGDIR):
	os.mkdir(config.LOGDIR)
	else:
	if not os.path.isdir(config.LOGDIR):
	raise Exception("Expected log dir '%s' is not actually a directory." % config.LOGDIR)
	except OSError as exc:
	raise exc

	# creates the under-test-branch as a separate directory
	def set_up_test_branch(settings, branch_name):
	testdir = "%s/%s.%d" % (settings['workdir'], config.TEST_DIR_NAME, config.OWN_PID)

	# creates the host dir
	if os.path.exists(testdir):
	raise Exception("Test dir '%s'is already there, aborting" % testdir)

	# may raise OSError, is to be handled by the caller
	os.makedirs(testdir)


	# copies over the .git from the localclone
	run_shell_cmd("cp -a '%s'/.git '%s'" % (settings['localclone'], testdir))

	# add the remote if it doesn't exist
	crt_remotes = run_shell_cmd("git remote -v", cwd=testdir)
	remotes = [word for line in crt_remotes.split("\n") for word in line.split()]
	if not config.CONTRIB_REPO in remotes:
	remote_name = "tts_contrib"
	run_shell_cmd("git remote add %s %s" % (remote_name, config.CONTRIB_REPO), cwd=testdir)
	else:
	remote_name = remotes[remotes.index(config.CONTRIB_REPO) - 1]

	# do the fetch
	run_shell_cmd("git fetch %s -p" % remote_name, cwd=testdir)

	# do the checkout
	run_shell_cmd("git checkout origin/master && git branch -D %s; git checkout %s/%s -b %s && git reset --hard" % (branch_name, remote_name, branch_name, branch_name), cwd=testdir)

	return testdir


	def __search_for_tests():
	# we find all classes that can run, and run them
	tests = []
	for _, _, files_list in os.walk(os.path.dirname(os.path.abspath(__file__))):
	for module_file in [f[:-3] for f in files_list if f.endswith(".py") and not f.startswith("__init__")]:
	config.logger.debug("Inspecting module %s", module_file)
	current_module = importlib.import_module(module_file)
	crtclass_names = vars(current_module)
	for name in crtclass_names:
	tested_value = crtclass_names[name]
	if isinstance(tested_value, type(unittest.TestCase)) and issubclass(tested_value, unittest.TestCase):
	tests.append((module_file, name))
	break
	return tests


	# boilerplate to self discover tests and run them
	def execute_tests(dir_under_test, testname):

	if testname is not None and "." in testname:
	tests = []
	tests.append(tuple(testname.split(".", 2)))
	else:
	tests = __search_for_tests()

	# let's move to the directory under test
	crt_dir = os.getcwd()
	os.chdir(dir_under_test)

	# execute each module
	# pylint: disable=broad-except
	# we disable the broad-except because we want to actually catch all possible exceptions
	try:
	# sorting the tests by the numeric order in the class name
	tests = sorted(tests, key=lambda x: int(re.search(r"[0-9]+", x[1]).group(0)))
	config.logger.debug("Discovered test clases: %s", pprint.pformat(tests))
	unittest.installHandler()
	suite = unittest.TestSuite()
	loader = unittest.TestLoader()
	result = unittest.TestResult()
	result.failfast = True
	for module_file, test_name in tests:
	suite.addTest(loader.loadTestsFromName("%s.%s" % (module_file, test_name)))
	config.logger.info("Running %d test(s)", suite.countTestCases())
	suite.run(result)

	for error in result.errors:
	config.logger.error("Exception on test: %s\n%s", error[0],
	"\n".join(["-- %s" % x for x in error[1].split("\n")]))

	for failure in result.failures:
	config.logger.error("Failed test: %s:\n%s\n", failure[0],
	"\n".join(["-- %s" % x for x in failure[1].split("\n")]))

	config.logger.info("Test results: %d ran, %d errors, %d failures", result.testsRun, len(result.errors), len(result.failures))

	except Exception as exc:
	import traceback
	config.logger.error("Exception while running test. Tracedump: \n%s", traceback.format_exc(exc))
	finally:
	os.chdir(crt_dir)
	return len(result.failures)

	# verify that we had a branch-under-test name as parameter
	def validate_args():
	from optparse import OptionParser
	parser = OptionParser(usage="usage: %prog [options] branch_under_test")

	parser.add_option("-t", "--test-dir", dest="testdir", default=None, help="Use specified directory to run tests, inhibits the checkout.")
	parser.add_option("-s", "--single", dest="singletest", default=None, help="Run only the specified test")

	(options, args) = parser.parse_args()
	if len(args) < 1:
	raise Exception("Please specify the branch to run on. Use option '-h' when in doubt.")
	return (options, args)




	# load the configuration options
	def read_settings():
	if not os.path.exists(config.SETTINGS_FILE) or not os.path.isfile(config.SETTINGS_FILE):
	raise Exception("Config file '%s' cannot be openend" % config.SETTINGS_FILE)
	return json.loads(open(config.SETTINGS_FILE, "r").read())


	# cleanup !
	def clean_up(testdir):
	run_shell_cmd("rm -rf -- '%s'" % testdir)

	def dump_info(settings, options, args):
	""" detailed information about current run configuration, for debugging purposes.
	"""
	config.logger.debug("Settings:\n%s\nOptions:\n%s\nArguments:\n%s\n", settings, options, args)

	def main():
	(options, args) = validate_args()

	settings = read_settings()
	need_cleanup = False

	# dump debug info
	dump_info(settings, options, args)

	testdir = None
	no_failures = 1
	try:
	if options.testdir is not None and os.path.exists(options.testdir):
	testdir = os.path.abspath(options.testdir)
	config.logger.info("No checkout, using %s", testdir)
	else:
	need_cleanup = True
	testdir = set_up_test_branch(settings, args[0]) # we expect a branch name as first argument

	config.TESTDIR = testdir # we let tests know where to run

	# ensure that the test dir only contains no *.pyc leftovers
	run_shell_cmd("find '%s' -type f -name *.pyc -exec rm {} \\;" % testdir)

	no_failures = execute_tests(testdir, options.singletest)

	except ShellCmdException as exc:
	import traceback
	config.logger.error("Error while setting up testing. Traceback: \n%s", traceback.format_exc(exc))
	finally:
	if need_cleanup and testdir is not None:
	clean_up(testdir)

	sys.exit(no_failures)

	if __name__ == "__main__":
	main()