diff options
Diffstat (limited to 'ipatests/beakerlib_plugin.py')
| -rw-r--r-- | ipatests/beakerlib_plugin.py | 168 |
1 files changed, 168 insertions, 0 deletions
diff --git a/ipatests/beakerlib_plugin.py b/ipatests/beakerlib_plugin.py new file mode 100644 index 000000000..2ad2674eb --- /dev/null +++ b/ipatests/beakerlib_plugin.py @@ -0,0 +1,168 @@ +# Authors: +# Petr Viktorin <pviktori@redhat.com> +# +# Copyright (C) 2013 Red Hat +# see file 'COPYING' for use and warranty information +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# 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, see <http://www.gnu.org/licenses/>. + +"""A Nose plugin that integrates with BeakerLib""" + +import os +import sys +import subprocess +import traceback +import logging + +import nose +from nose.plugins import Plugin + +from ipapython import ipautil +from ipapython.ipa_log_manager import log_mgr + + +def shell_quote(string): + """Quote a string for the shell + + Adapted from Python3's shlex.quote + """ + return "'" + str(string).replace("'", "'\"'\"'") + "'" + + +class BeakerLibLogHandler(logging.Handler): + def __init__(self, beakerlib_command): + super(BeakerLibLogHandler, self).__init__() + self.beakerlib_command = beakerlib_command + + def emit(self, record): + command = { + 'DEBUG': 'rlLogDebug', + 'INFO': 'rlLogInfo', + 'WARNING': 'rlLogWarning', + 'ERROR': 'rlLogError', + 'CRITICAL': 'rlLogFatal', + }.get(record.levelname, 'rlLog') + self.beakerlib_command([command, self.format(record)]) + + +class BeakerLibPlugin(Plugin): + """A Nose plugin that integrates with BeakerLib""" + # Since BeakerLib is a Bash library, we need to run it in Bash. + # The plugin maintains a Bash process and feeds it with commands + # on events like test start/end, logging, etc. + # See nose.plugins.base.IPluginInterface for Nose plugin interface docs + name = 'beakerlib' + + def options(self, parser, env=os.environ): + super(BeakerLibPlugin, self).options(parser, env=env) + self.env = env + self.parser = parser + + def configure(self, options, conf): + super(BeakerLibPlugin, self).configure(options, conf) + if not self.enabled: + return + + if 'BEAKERLIB' not in self.env: + self.parser.error( + 'BeakerLib not active, cannot use --with-beakerlib') + + # Set up the Bash process + self.bash = subprocess.Popen(['bash'], + stdin=subprocess.PIPE) + source_path = os.path.join(self.env['BEAKERLIB'], 'beakerlib.sh') + self.run_beakerlib_command(['.', source_path]) + + # _in_class is set when we are in setup_class, so its rlPhaseEnd can + # be called when the first test starts + self._in_class = False + + # Redirect logging to our own handlers + self.setup_log_handler(BeakerLibLogHandler(self.run_beakerlib_command)) + + def setup_log_handler(self, handler): + log_mgr.configure( + { + 'default_level': 'DEBUG', + 'handlers': [{'log_handler': handler, + 'format': '[%(name)s] %(message)s', + 'level': 'debug'}]}, + configure_state='beakerlib_plugin') + + def run_beakerlib_command(self, cmd): + """Given a command as a Popen-style list, run it in the Bash process""" + for word in cmd: + self.bash.stdin.write(shell_quote(word)) + self.bash.stdin.write(' ') + self.bash.stdin.write('\n') + self.bash.stdin.flush() + assert self.bash.returncode is None, "BeakerLib Bash process exited" + + def report(self, stream): + """End the Bash process""" + self.run_beakerlib_command(['exit']) + self.bash.communicate() + + def startContext(self, context): + """Start a test context (module, class) + + For test classes, this starts a BeakerLib phase + """ + if not isinstance(context, type): + return + message = 'Class setup: %s' % context.__name__ + self.run_beakerlib_command(['rlPhaseStart', 'FAIL', message]) + self._in_class = True + + def stopContext(self, context): + """End a test context""" + if self._in_class: + self.run_beakerlib_command(['rlPhaseEnd']) + + def startTest(self, test): + """Start a test phase""" + if self._in_class: + self.run_beakerlib_command(['rlPhaseEnd']) + self.run_beakerlib_command(['rlPhaseStart', 'FAIL', + 'Nose test: %s' % test]) + + def stopTest(self, test): + """End a test phase""" + self.run_beakerlib_command(['rlPhaseEnd']) + + def addSuccess(self, test): + self.run_beakerlib_command(['rlPass', 'Test succeeded']) + + def log_exception(self, err): + """Log an exception + + err is a 3-tuple as returned from sys.exc_info() + """ + message = ''.join(traceback.format_exception(*err)).rstrip() + self.run_beakerlib_command(['rlLogError', message]) + + def addError(self, test, err): + if issubclass(err[0], nose.SkipTest): + # Log skipped test. + # Unfortunately we only get to see this if the built-in skip + # plugin is disabled (--no-skip) + self.run_beakerlib_command(['rlPass', 'Test skipped: %s' % err[1]]) + else: + self.log_exception(err) + self.run_beakerlib_command( + ['rlFail', 'Test failed: unhandled exception']) + + def addFailure(self, test, err): + self.log_exception(err) + self.run_beakerlib_command(['rlFail', 'Test failed']) |