diff options
Diffstat (limited to 'ipatests')
| -rwxr-xr-x | ipatests/ipa-test-task | 13 | ||||
| -rw-r--r-- | ipatests/pytest_plugins/beakerlib.py | 205 |
2 files changed, 24 insertions, 194 deletions
diff --git a/ipatests/ipa-test-task b/ipatests/ipa-test-task index d89af841d..8c9ab082f 100755 --- a/ipatests/ipa-test-task +++ b/ipatests/ipa-test-task @@ -28,9 +28,13 @@ from ipapython.ipa_log_manager import log_mgr, standard_logging_setup from ipatests.test_integration import config from ipatests.test_integration import tasks from ipatests.test_integration.host import Host -from ipatests.pytest_plugins.beakerlib import BeakerLibProcess from ipatests.pytest_plugins.integration import collect_logs +try: + from pytest_beakerlib import BeakerLibProcess +except ImportError: + BeakerLibProcess = None + log = log_mgr.get_logger(__name__) @@ -245,8 +249,8 @@ class TaskRunner(object): return parser def main(self, argv): - - args = self.get_parser().parse_args(argv) + parser = self.get_parser() + args = parser.parse_args(argv) self.config = config.Config.from_env(os.environ) if not self.config: raise EnvironmentError('Multihost environment not configured') @@ -259,6 +263,9 @@ class TaskRunner(object): self.collect_log = collect_log if args.with_beakerlib: + if BeakerLibProcess is None: + parser.error( + 'pytest_beakerlib not installed, cannot use BeakerLib') beakerlib_process = BeakerLibProcess() args.verbose = True diff --git a/ipatests/pytest_plugins/beakerlib.py b/ipatests/pytest_plugins/beakerlib.py index 45bbb0539..c94f7f417 100644 --- a/ipatests/pytest_plugins/beakerlib.py +++ b/ipatests/pytest_plugins/beakerlib.py @@ -16,52 +16,29 @@ # along with this program. If not, see <http://www.gnu.org/licenses/>. # -"""pytest integration with BeakerLib +"""Test integration with BeakerLib -Runs a Bash process on the side, and feeds BeakerLib commands to it -(rlPhaseStart, rlPhaseEnd, rlPass, rlFail, ...) +IPA-specific configuration for the BeakerLib plugin (from pytest-beakerlib). +If the plugin is active, sets up IPA logging to also log to Beaker. -Other plugins may integrate with this using pytest's -config.pluginmanager.getplugin('BeakerLibPlugin'). If this is None, -BeakerLib integration is not active, otherwise the result's -run_beakerlib_command method can be used to run additional commands. - -IPA logging is also redirected to the Bash process. """ -import os -import re import logging -import subprocess - -import pytest -from ipapython import ipautil from ipapython.ipa_log_manager import log_mgr -@pytest.fixture -def log_files_to_collect(): - return [] - - -def pytest_addoption(parser): - parser.addoption( - '--with-beakerlib', action="store_true", - dest="with_beakerlib", default=None, - help="Report test results via beakerlib") - - -@pytest.mark.tryfirst -def pytest_load_initial_conftests(args, early_config, parser): - ns = early_config.known_args_namespace - if ns.with_beakerlib: - if 'BEAKERLIB' not in os.environ: - raise exit('$BEAKERLIB not set, cannot use --with-beakerlib') - - plugin = BeakerLibPlugin() - pluginmanager = early_config.pluginmanager.register( - plugin, 'BeakerLibPlugin') +def pytest_configure(config): + plugin = config.pluginmanager.getplugin('BeakerLibPlugin') + if plugin: + handler = BeakerLibLogHandler(plugin.run_beakerlib_command) + log_mgr.configure( + { + 'default_level': 'DEBUG', + 'handlers': [{'log_handler': handler, + 'format': '[%(name)s] %(message)s', + 'level': 'info'}]}, + configure_state='beakerlib_plugin') class BeakerLibLogHandler(logging.Handler): @@ -78,157 +55,3 @@ class BeakerLibLogHandler(logging.Handler): 'CRITICAL': 'rlLogFatal', }.get(record.levelname, 'rlLog') self.beakerlib_command([command, self.format(record)]) - - -class BeakerLibProcess(object): - """Manager of a Bash process that is being fed beakerlib commands - """ - def __init__(self, env=os.environ): - self.log = log_mgr.get_logger(self) - - if 'BEAKERLIB' not in env: - raise RuntimeError('$BEAKERLIB not set, cannot use BeakerLib') - - self.env = env - # Set up the Bash process - self.bash = subprocess.Popen(['bash'], - stdin=subprocess.PIPE, - stdout=open(os.devnull, 'w'), - stderr=open(os.devnull, 'w')) - source_path = os.path.join(self.env['BEAKERLIB'], 'beakerlib.sh') - self.run_beakerlib_command(['.', source_path]) - - # 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': 'info'}]}, - configure_state='beakerlib_plugin') - - def run_beakerlib_command(self, cmd): - """Given a command as a Popen-style list, run it in the Bash process""" - if not self.bash: - return - for word in cmd: - self.bash.stdin.write(ipautil.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 log_links(self, docstring): - for match in LINK_RE.finditer(docstring or ''): - self.log.info('Link: %s', match.group()) - - def end(self): - """End the Bash process""" - self.run_beakerlib_command(['exit']) - bash = self.bash - self.bash = None - bash.communicate() - - def log_exception(self, err=None): - """Log an exception - - err is a 3-tuple as returned from sys.exc_info(); if not given, - sys.exc_info() is used. - """ - if err is None: - err = sys.exc_info() - message = ''.join(traceback.format_exception(*err)).rstrip() - self.run_beakerlib_command(['rlLogError', message]) - - -class BeakerLibPlugin(object): - def __init__(self): - self.log = log_mgr.get_logger(self) - - self.process = BeakerLibProcess(env=os.environ) - - self._current_item = None - - def run_beakerlib_command(self, cmd): - """Given a command as a Popen-style list, run it in the Bash process""" - self.process.run_beakerlib_command(cmd) - - def get_item_name(self, item): - """Return a "identifier-style" name for the given item - - The name only contains the characters [^a-zA-Z0-9_]. - """ - bad_char_re = re.compile('[^a-zA-Z0-9_]') - parts = [] - current = item - while current: - if isinstance(current, pytest.Module): - name = current.name - if name.endswith('.py'): - name = name[:-3] - name = bad_char_re.sub('-', name) - parts.append(name) - break - if isinstance(current, pytest.Instance): - pass - else: - name = current.name - name = bad_char_re.sub('-', name) - parts.append(name) - current = current.parent - return '-'.join(reversed(parts)) - - def set_current_item(self, item): - """Set the item that is currently being processed - - No-op if the same item is already being processed. - Ends the phase for the previous item, if any. - """ - if item != self._current_item: - item_name = self.get_item_name(item) - if self._current_item: - self.run_beakerlib_command(['rlPhaseEnd']) - if item: - self.run_beakerlib_command(['rlPhaseStart', 'FAIL', item_name]) - self._current_item = item - - def pytest_collection_modifyitems(self, session, config, items): - """Log all collected items at start of test""" - self.run_beakerlib_command(['rlLogInfo', 'Collected pytest tests:']) - for item in items: - self.run_beakerlib_command(['rlLogInfo', - ' - ' + self.get_item_name(item)]) - - def pytest_runtest_setup(self, item): - """Log item before running it""" - self.set_current_item(item) - - def pytest_runtest_makereport(self, item, call): - """Report pass/fail for setup/call/teardown of an item""" - self.set_current_item(item) - desc = '%s: %s' % (call.when, item) - - if not call.excinfo: - self.run_beakerlib_command(['rlPass', 'PASS %s' % desc]) - else: - self.run_beakerlib_command(['rlLogError', call.excinfo.exconly()]) - short_repr = str(call.excinfo.getrepr(style='short')) - self.run_beakerlib_command(['rlLogInfo', short_repr]) - - # Give super-detailed traceback for DEBUG=1 - long_repr = str(call.excinfo.getrepr( - showlocals=True, funcargs=True)) - self.run_beakerlib_command(['rlLogDebug', long_repr]) - - if call.excinfo.errisinstance(pytest.skip.Exception): - self.run_beakerlib_command(['rlPass', 'SKIP %s' % desc]) - else: - self.run_beakerlib_command(['rlFail', 'FAIL %s' % desc]) - - def pytest_unconfigure(self, config): - """Clean up and exit""" - self.set_current_item(None) - self.process.end() |