[ADD] allure-pytest-log plugin

Author: 左武辉51085

allure-pytest-log/README.rst

@@ -0,0 +1,18 @@
+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
+
+    $ pip install allure-pytest-log
+    $ py.test --allure-capture [--alluredir=%allure_result_folder%] ./tests
+    $ allure serve %allure_result_folder%

allure-pytest-log/__init__.py

@@ -0,0 +1 @@
+# -*- coding: UTF-8 -*-

allure-pytest-log/setup.py

@@ -0,0 +1,58 @@
+import os, sys
+from setuptools import setup
+from pkg_resources import require, DistributionNotFound, VersionConflict
+
+try:
+    require('pytest-allure-adaptor')
+    print("""
+    You have pytest-allure-adaptor installed.
+    You need to remove pytest-allure-adaptor from your site-packages
+    before installing allure-pytest, or conflicts may result.
+    """)
+    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="[email protected]",
+            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()

allure-pytest-log/src/__init__.py

@@ -0,0 +1,79 @@
+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())

allure-pytest-log/src/listener.py

@@ -0,0 +1,48 @@
+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)

allure-pytest-log/src/plugin.py

@@ -0,0 +1,35 @@
+# -*- coding: UTF-8 -*-
+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()

allure-pytest-log/src/utils.py

@@ -0,0 +1 @@
+# -*- coding: UTF-8 -*-

allure-pytest-log/test/__init__.py

@@ -0,0 +1,89 @@
+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()
+
+
+@pytest.fixture(scope='function', autouse=True)
+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)
+
+
+@pytest.fixture(scope='module')
+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
+
+
+@pytest.fixture(scope='module')
+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