Allure With Log Capturing Pytest Plugin

- `Source <https://github.com/allure-framework/allure-python>`_

- `Documentation <https://docs.qameta.io/allure/2.0/>`_

- `Gitter <https://gitter.im/allure-framework/allure-core>`_

Installation and Usage

.. code:: bash

import os, sys

from setuptools import setup

from pkg_resources import require, DistributionNotFound, VersionConflict

try:

require('pytest-allure-adaptor')

print("""

""")

sys.exit()

except (DistributionNotFound, VersionConflict):

pass

PACKAGE = "allure-pytest-log"

VERSION = "0.1.0"

classifiers = [

'Development Status :: 5 - Production/Stable',

'Framework :: Pytest',

'Intended Audience :: Developers',

'License :: OSI Approved :: Apache Software License',

'Topic :: Software Development :: Quality Assurance',

'Topic :: Software Development :: Testing',

]

install_requires = [

"allure-pytest>=2.4.1",

"allure-python-commons>=2.4.1"

]

def read(fname):

return open(os.path.join(os.path.dirname(__file__), fname)).read()

def main():

setup(

name=PACKAGE,

version=VERSION,

description="Allure pytest integration with stdout capturing",

url="https://github.com/allure-framework/allure-python-log",

author="WuhuiZuo",

author_email="wuhuizuo@126.com",

license="Apache-2.0",

classifiers=classifiers,

keywords="allure reporting pytest output_capture",

long_description=read('README.rst'),

packages=["allure_pytest_log"],

package_dir={"allure_pytest_log": "src"},

entry_points={"pytest11": ["allure_pytest_log = allure_pytest_log.plugin"]},

install_requires=install_requires

)

if __name__ == '__main__':

main()

import pytest

import allure_commons

from allure_commons.types import AttachmentType

from allure_pytest.listener import ItemCache

from .utils import Tee

class AllureLogListener(object):

def __init__(self, allure_listener=None):

self.allure_listener = allure_listener

self.modify_allure_listener(self.allure_listener)

self._cache = allure_listener._cache

self._tee_cache = TeeCache()

def modify_allure_listener(self, listener):

origin_start_before_fixture = listener.allure_logger.start_before_fixture

origin_stop_before_fixture = listener.allure_logger.stop_before_fixture

origin_start_after_fixture = listener.allure_logger.start_after_fixture

origin_stop_after_fixture = listener.allure_logger.stop_after_fixture

def start_before_fixture(parent_uuid, uuid, fixture):

origin_start_before_fixture(parent_uuid, uuid, fixture)

self.start_tee(uuid)

def stop_before_fixture(uuid, **kwargs):

self.finish_tee(uuid, 'fixture log')

origin_stop_before_fixture(uuid, **kwargs)

def start_after_fixture(parent_uuid, uuid, fixture):

origin_start_after_fixture(parent_uuid, uuid, fixture)

self.start_tee(uuid)

def stop_after_fixture(uuid, **kwargs):

self.finish_tee(uuid, 'fixture[after] log')

origin_stop_after_fixture(uuid, **kwargs)

listener.allure_logger.start_before_fixture = start_before_fixture

listener.allure_logger.stop_before_fixture = stop_before_fixture

listener.allure_logger.start_after_fixture = start_after_fixture

listener.allure_logger.stop_after_fixture = stop_after_fixture

def start_tee(self, uuid):

tee = self._tee_cache.set(uuid)

tee.start()

def finish_tee(self, uuid, attach_name='log'):

tee = self._tee_cache.pop(uuid)

if not tee:

return None

try:

self.allure_listener.allure_logger.attach_data(uuid,

body=tee.getvalue(),

name=attach_name,

attachment_type=AttachmentType.TEXT)

finally:

tee.close()

@allure_commons.hookimpl(hookwrapper=True)

def start_step(self, uuid, title, params):

yield

self.start_tee(uuid)

@allure_commons.hookimpl(hookwrapper=True)

def stop_step(self, uuid, exc_type, exc_val, exc_tb):

self.finish_tee(uuid, 'step log')

yield

@pytest.hookimpl(hookwrapper=True)

def pytest_runtest_call(self, item):

uuid = self._cache.get(item.nodeid)

self.start_tee(uuid)

yield

self.finish_tee(uuid, 'test log')

class TeeCache(ItemCache):

def set(self, _id):

return self._items.setdefault(str(_id), Tee())

import allure_commons

from allure_commons.logger import AllureFileLogger

from allure_pytest.listener import AllureListener

from .listener import AllureLogListener

def _enable_allure_capture(config):

allure_listener = config.pluginmanager.getplugin('AllureListener')

if not allure_listener:

# registry allure-pytest(dependency) plugin first

report_dir = config.option.allure_report_dir or 'reports'

clean = config.option.clean_alluredir

allure_listener = AllureListener(config)

config.pluginmanager.register(allure_listener)

allure_commons.plugin_manager.register(allure_listener)

config.add_cleanup(cleanup_factory(allure_listener))

file_logger = AllureFileLogger(report_dir, clean)

allure_commons.plugin_manager.register(file_logger)

config.add_cleanup(cleanup_factory(file_logger))

allure_log_listener = AllureLogListener(allure_listener)

config.pluginmanager.register(allure_log_listener)

allure_commons.plugin_manager.register(allure_log_listener)

config.add_cleanup(cleanup_factory(allure_log_listener))

def pytest_addoption(parser):

parser.getgroup("reporting").addoption('--allure-capture',

action="store_true",

dest="allure_capture",

help="Capture standard output to Allure report")

def cleanup_factory(plugin):

def clean_up():

name = allure_commons.plugin_manager.get_name(plugin)

allure_commons.plugin_manager.unregister(name=name)

return clean_up

def pytest_configure(config):

allure_capture = config.option.allure_capture

if allure_capture:

_enable_allure_capture(config)

import io

import sys

class Tee(object):

def __init__(self):

self.memory = io.StringIO()

self.origin_stdout = None

def start(self):

self.origin_stdout, sys.stdout = sys.stdout, self

def close(self):

if self.origin_stdout:

sys.stdout = self.origin_stdout

self.flush()

def getvalue(self, *args, **kwargs):

return self.memory.getvalue(*args, **kwargs)

def write(self, data):

self.memory.write(data)

if self.origin_stdout:

self.origin_stdout.write(data)

def flush(self):

self.memory.seek(0)

def __enter__(self):

self.start()

return self

def __exit__(self, exc_type, exc_val, exc_tb):

self.close()

from __future__ import print_function

import pytest

import os

import sys

import subprocess

import shlex

import hashlib

from inspect import getmembers, isfunction

from allure_commons_test.report import AllureReport

from allure_commons.utils import thread_tag

with open("debug-runner", "w") as debugfile:

# overwrite debug-runner file with an empty one

print("New session", file=debugfile)

def _get_hash(input):

if sys.version_info < (3, 0):

data = bytes(input)

else:

data = bytes(input, 'utf8')

return hashlib.md5(data).hexdigest()

def inject_matchers(doctest_namespace):

import hamcrest

for name, function in getmembers(hamcrest, isfunction):

doctest_namespace[name] = function

from allure_commons_test import container, label, report, result

for module in [container, label, report, result]:

for name, function in getmembers(module, isfunction):

doctest_namespace[name] = function

def _runner(allure_dir, module, *extra_params):

extra_params = ' '.join(extra_params)

cmd = shlex.split('%s -m pytest --allure-capture --alluredir=%s %s %s' % (sys.executable, allure_dir, extra_params, module),

posix=False if os.name == "nt" else True)

with open("debug-runner", "a") as debugfile:

try:

subprocess.check_output(cmd, stderr = subprocess.STDOUT)

except subprocess.CalledProcessError as e:

# Save to debug file errors on execution (includes pytest failing tests)

print(e.output, file=debugfile)

def allure_report_with_params(request, tmpdir_factory):

module = request.module.__file__

tmpdir = tmpdir_factory.mktemp('data')

def run_with_params(*params, **kwargs):

cache = kwargs.get("cache", True)

key = _get_hash('{thread}{module}{param}'.format(thread=thread_tag(), module=module, param=''.join(params)))

if not request.config.cache.get(key, False):

_runner(tmpdir.strpath, module, *params)

if cache:

request.config.cache.set(key, True)

def clear_cache():

request.config.cache.set(key, False)

request.addfinalizer(clear_cache)

return AllureReport(tmpdir.strpath)

return run_with_params

def allure_report(request, tmpdir_factory):

module = request.module.__file__

tmpdir = tmpdir_factory.mktemp('data')

_runner(tmpdir.strpath, module)

return AllureReport(tmpdir.strpath)

def pytest_collection_modifyitems(items, config):

if config.option.doctestmodules:

items[:] = [item for item in items if item.__class__.__name__ == 'DoctestItem']

def pytest_ignore_collect(path, config):

if sys.version_info.major < 3 and "py3_only" in path.strpath:

return True

if sys.version_info.major > 2 and "py2_only" in path.strpath:

return True