nose2 first version (via #513)

Author: sseliverstov

.github/workflows/build.yaml

@@ -10,6 +10,7 @@ jobs:
         package: [
           allure-python-commons-test,
           allure-python-commons,
+          allure-nose2,
           allure-behave,
           allure-pytest,
           allure-pytest-bdd,

.github/workflows/release.yaml

@@ -40,6 +40,11 @@ jobs:
         twine upload dist/*
         popd
 
+        pushd allure-nose2
+        python setup.py sdist bdist_wheel
+        twine upload dist/*
+        popd
+
         pushd allure-pytest
         python setup.py sdist bdist_wheel
         twine upload dist/*

allure-nose2/README.rst

@@ -0,0 +1,58 @@
+import os
+from setuptools import setup
+
+PACKAGE = "allure-nose2"
+
+classifiers = [
+    'Development Status :: 5 - Production/Stable',
+    'Intended Audience :: Developers',
+    'License :: OSI Approved :: Apache Software License',
+    'Topic :: Software Development :: Quality Assurance',
+    'Topic :: Software Development :: Testing',
+    'Programming Language :: Python :: 3',
+    'Programming Language :: Python :: 3.6',
+    'Programming Language :: Python :: 3.7',
+]
+
+setup_requires = [
+    "setuptools_scm"
+]
+
+install_requires = [
+    "nose2"
+]
+
+
+def prepare_version():
+    from setuptools_scm import get_version
+    configuration = {"root": "..", "relative_to": __file__}
+    version = get_version(**configuration)
+    install_requires.append("allure-python-commons=={version}".format(version=version))
+    return configuration
+
+
+def get_readme(fname):
+    return open(os.path.join(os.path.dirname(__file__), fname)).read()
+
+
+def main():
+    setup(
+        name=PACKAGE,
+        use_scm_version=prepare_version,
+        description="Allure nose2 integration",
+        url="https://github.com/allure-framework/allure-python",
+        author="QAMetaSoftware, Stanislav Seliverstov",
+        author_email="[email protected]",
+        license="Apache-2.0",
+        classifiers=classifiers,
+        keywords="allure reporting nose2",
+        long_description=get_readme('README.rst'),
+        packages=["allure_nose2"],
+        package_dir={"allure_nose2": "src"},
+        setup_requires=setup_requires,
+        install_requires=install_requires
+    )
+
+if __name__ == '__main__':
+    main()
+

allure-nose2/setup.py

@@ -0,0 +1,5 @@
+
+# ToDo attaches
+class AllureListener(object):
+    def __init__(self, lifecycle):
+        self.lifecycle = lifecycle

allure-nose2/src/__init__.py

@@ -0,0 +1,129 @@
+from nose2.events import Plugin
+from allure_commons import plugin_manager
+from allure_commons.logger import AllureFileLogger
+from allure_nose2.listener import AllureListener
+from allure_commons.lifecycle import AllureLifecycle
+from nose2 import result
+from allure_commons.model2 import Status
+from allure_commons.model2 import StatusDetails
+from allure_commons.model2 import Label
+from allure_commons.types import LabelType
+from allure_commons.utils import host_tag, thread_tag
+
+from allure_commons.utils import platform_label, md5
+
+
+from .utils import timestamp_millis, status_details, update_attrs, labels, name, fullname, params
+import allure_commons
+
+
+class DecoratorsHelper(object):
+    @classmethod
+    @allure_commons.hookimpl
+    def decorate_as_label(cls, label_type, labels):
+        # ToDo functools.update_wrapper
+        def wrapper(test):
+            update_attrs(test, label_type, labels)
+            return test
+
+        return wrapper
+
+    @classmethod
+    def register(cls):
+        if cls not in plugin_manager.get_plugins():
+            plugin_manager.register(cls)
+
+    @classmethod
+    def unregister(cls):
+        if cls in plugin_manager.get_plugins():
+            plugin_manager.unregister(plugin=cls)
+
+
+DecoratorsHelper.register()
+
+
+class Allure(Plugin):
+    configSection = 'allure'
+    commandLineSwitch = (None, "allure", "Generate an Allure report")
+
+    def __init__(self, *args, **kwargs):
+        super(Allure, self).__init__(*args, **kwargs)
+        self._host = host_tag()
+        self._thread = thread_tag()
+        self.lifecycle = AllureLifecycle()
+        self.logger = AllureFileLogger("allure-result")
+        self.listener = AllureListener(self.lifecycle)
+
+    def registerInSubprocess(self, event):
+        self.unregister_allure_plugins()
+        event.pluginClasses.append(self.__class__)
+
+    def startSubprocess(self, event):
+        self.register_allure_plugins()
+
+    def stopSubprocess(self, event):
+        self.unregister_allure_plugins()
+
+    def register_allure_plugins(self):
+        plugin_manager.register(self.listener)
+        plugin_manager.register(self.logger)
+
+    def unregister_allure_plugins(self):
+        plugin_manager.unregister(plugin=self.listener)
+        plugin_manager.unregister(plugin=self.logger)
+
+    def is_registered(self):
+        return all([plugin_manager.is_registered(self.listener),
+                    plugin_manager.is_registered(self.logger)])
+
+    def startTestRun(self, event):
+        self.register_allure_plugins()
+
+    def afterTestRun(self, event):
+        self.unregister_allure_plugins()
+
+    def startTest(self, event):
+        if self.is_registered():
+            with self.lifecycle.schedule_test_case() as test_result:
+                test_result.name = name(event)
+                test_result.start = timestamp_millis(event.startTime)
+                test_result.fullName = fullname(event)
+                test_result.testCaseId = md5(test_result.fullName)
+                test_result.historyId = md5(event.test.id())
+                test_result.labels.extend(labels(event.test))
+                test_result.labels.append(Label(name=LabelType.HOST, value=self._host))
+                test_result.labels.append(Label(name=LabelType.THREAD, value=self._thread))
+                test_result.labels.append(Label(name=LabelType.FRAMEWORK, value='nose2'))
+                test_result.labels.append(Label(name=LabelType.LANGUAGE, value=platform_label()))
+                test_result.parameters = params(event)
+
+    def stopTest(self, event):
+        if self.is_registered():
+            with self.lifecycle.update_test_case() as test_result:
+                test_result.stop = timestamp_millis(event.stopTime)
+            self.lifecycle.write_test_case()
+
+    def testOutcome(self, event):
+        if self.is_registered():
+            with self.lifecycle.update_test_case() as test_result:
+                if event.outcome == result.PASS and event.expected:
+                    test_result.status = Status.PASSED
+                elif event.outcome == result.PASS and not event.expected:
+                    test_result.status = Status.PASSED
+                    test_result.statusDetails = StatusDetails(message="test passes unexpectedly")
+                elif event.outcome == result.FAIL and not event.expected:
+                    test_result.status = Status.FAILED
+                    test_result.statusDetails = status_details(event)
+                elif event.outcome == result.ERROR:
+                    test_result.status = Status.BROKEN
+                    test_result.statusDetails = status_details(event)
+                elif event.outcome == result.SKIP:
+                    test_result.status = Status.SKIPPED
+                    test_result.statusDetails = status_details(event)
+                # Todo default status and other cases
+                # elif event.outcome == result.FAIL and event.expected:
+                #     pass
+                    # self.skipped += 1
+                    # skipped = ET.SubElement(testcase, 'skipped')
+                    # skipped.set('message', 'expected test failure')
+                    # skipped.text = msg

allure-nose2/src/listener.py

@@ -0,0 +1,98 @@
+from traceback import format_exception_only
+from allure_commons.model2 import StatusDetails, Label
+from allure_commons.model2 import Parameter
+from allure_commons.utils import represent
+from nose2 import util
+import inspect
+
+# ToDo move to commons
+ALLURE_LABELS = [
+    'epic',
+    'feature',
+    'story',
+]
+
+
+def timestamp_millis(timestamp):
+    return int(timestamp * 1000)
+
+
+def status_details(event):
+    message, trace = None, None
+    if event.exc_info:
+        exc_type, value, _ = event.exc_info
+        message = '\n'.join(format_exception_only(exc_type, value)) if exc_type or value else None
+        trace = ''.join(util.exc_info_to_string(event.exc_info, event.test))
+    elif event.reason:
+        message = event.reason
+
+    if message or trace:
+        return StatusDetails(message=message, trace=trace)
+
+
+def update_attrs(test, name, values):
+    if type(values) in (list, tuple, str) and name.isidentifier():
+        attrib = getattr(test, name, values)
+        if attrib and attrib != values:
+            attrib = sum(
+                [tuple(i) if type(i) in (tuple, list) else (i,) for i in (attrib, values)],
+                ()
+            )
+        setattr(test, name, attrib)
+
+
+def labels(test):
+
+    def _get_attrs(obj, keys):
+        pairs = set()
+        for key in keys:
+            values = getattr(obj, key, ())
+            for value in (values,) if type(values) == str else values:
+                pairs.add((key, value))
+        return pairs
+
+    keys = ALLURE_LABELS
+    pairs = _get_attrs(test, keys)
+
+    if hasattr(test, "_testFunc"):
+        pairs.update(_get_attrs(test._testFunc, keys))
+    elif hasattr(test, "_testMethodName"):
+        test_method = getattr(test, test._testMethodName)
+        pairs.update(_get_attrs(test_method, keys))
+    return [Label(name=name, value=value) for name, value in pairs]
+
+
+def name(event):
+    full_name = fullname(event)
+    return full_name.split(".")[-1]
+
+
+def fullname(event):
+    if hasattr(event.test, "_testFunc"):
+        test_module = event.test._testFunc.__module__
+        test_name = event.test._testFunc.__name__
+        return "{module}.{name}".format(module=test_module, name=test_name)
+    test_id = event.test.id()
+    return test_id.split(":")[0]
+
+
+def params(event):
+    def _params(names, values):
+        return [Parameter(name=name, value=represent(value)) for name, value in zip(names, values)]
+
+    test_id = event.test.id()
+
+    if len(test_id.split("\n")) > 1:
+        if hasattr(event.test, "_testFunc"):
+            wrapper_arg_spec = inspect.getfullargspec(event.test._testFunc)
+            arg_set, obj = wrapper_arg_spec.defaults
+            test_arg_spec = inspect.getfullargspec(obj)
+            args = test_arg_spec.args
+            return _params(args, arg_set)
+        elif hasattr(event.test, "_testMethodName"):
+            method = getattr(event.test, event.test._testMethodName)
+            wrapper_arg_spec = inspect.getfullargspec(method)
+            obj, arg_set = wrapper_arg_spec.defaults
+            test_arg_spec = inspect.getfullargspec(obj)
+            args = test_arg_spec.args
+            return _params(args[1:], arg_set)

allure-nose2/src/plugin.py

@@ -0,0 +1,29 @@
+import sys
+import types
+from importlib import util
+from doctest import script_from_examples
+from nose2 import events
+
+
+class CurrentExample(events.Plugin):
+    commandLineSwitch = (None, "current-example", "Method docstring to module")
+
+    def __init__(self, *args, **kwargs):
+        super(CurrentExample, self).__init__(*args, **kwargs)
+        self._current_docstring = ""
+
+    def startTest(self, event):
+        if hasattr(event.test, "_testFunc"):
+            self._current_docstring = event.test._testFunc.__doc__
+        else:
+            self._current_docstring = event.test._testMethodDoc
+
+    def get_example_module(self):
+        module = types.ModuleType("stub")
+        if self._current_docstring:
+            code = script_from_examples(self._current_docstring)
+            spec = util.spec_from_loader("example_module", origin="example_module", loader=None)
+            module = util.module_from_spec(spec)
+            exec(code, module.__dict__)
+        sys.modules['example_module'] = module
+        return module