mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2025-02-25 18:55:28 -06:00
ipatests: Use pytest-beakerlib
The plugin for BeakerLib integration was split into a separate project. If BeakerLib integration is desired, python-pytest-beakerlib shoule be installed separately. The IPA-specific beakerlib integration only sets up logging to BeakerLib, if the plugin is active. Reviewed-By: Tomas Babej <tbabej@redhat.com>
This commit is contained in:
Petr Viktorin
committed by
Tomas Babej
Tomas Babej
parent
b7e58ce746
commit
1e27fcc3b1
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user