RPPT-001 | Rewrite listener, fix rp_logger. Make code more pythonic.

Author: alexandr

.gitignore

@@ -0,0 +1,2 @@
+# Created by .ignore support plugin (hsz.mobi)
+.idea

pytest_reportportal/__init__.py

@@ -1,71 +1 @@
-import sys
-import logging
-
-from .service import PyTestService
-
-
-class RPLogger(logging.getLoggerClass()):
-
-    def __init__(self, name, level=0):
-        super(RPLogger, self).__init__(name, level=level)
-
-    def _log(self, level, msg, args, exc_info=None, extra=None,
-             attachment=None):
-        """
-        Low-level logging routine which creates a LogRecord and then calls
-        all the handlers of this logger to handle the record.
-        """
-        if logging._srcfile:
-            # IronPython doesn't track Python frames, so findCaller raises an
-            # exception on some versions of IronPython. We trap it here so that
-            # IronPython can use logging.
-            try:
-                fn, lno, func = self.findCaller()
-            except ValueError:
-                fn, lno, func = "(unknown file)", 0, "(unknown function)"
-        else:
-            fn, lno, func = "(unknown file)", 0, "(unknown function)"
-
-        if exc_info and not isinstance(exc_info, tuple):
-            exc_info = sys.exc_info()
-
-        record = self.makeRecord(
-            self.name, level, fn, lno, msg, args, exc_info, func, extra
-        )
-        record.attachment = attachment
-        self.handle(record)
-
-
-class RPLogHandler(logging.Handler):
-
-    # Map loglevel codes from `logging` module to ReportPortal text names:
-    _loglevel_map = {
-        logging.NOTSET: "TRACE",
-        logging.DEBUG: "DEBUG",
-        logging.INFO: "INFO",
-        logging.WARNING: "WARN",
-        logging.ERROR: "ERROR",
-        logging.CRITICAL: "ERROR",
-    }
-    _sorted_levelnos = sorted(_loglevel_map.keys(), reverse=True)
-
-    def __init__(self, level=logging.NOTSET):
-        super(RPLogHandler, self).__init__(level)
-
-    def emit(self, record):
-        try:
-            msg = self.format(record)
-        except (KeyboardInterrupt, SystemExit):
-            raise
-        except:
-            self.handleError(record)
-
-        for level in self._sorted_levelnos:
-            if level <= record.levelno:
-                break
-
-        return PyTestService.post_log(
-            msg,
-            loglevel=self._loglevel_map[level],
-            attachment=record.__dict__.get("attachment", None),
-        )
+from .rp_logging import RPLogger, RPLogHandler

pytest_reportportal/listener.py

@@ -0,0 +1,45 @@
+import html
+import pytest
+
+from .service import PyTestService
+
+
+class RPReportListener(object):
+    def __init__(self):
+        # Identifier if TestItem is called:
+        # if setup is failed, pytest will NOT call
+        # TestItem and Result will not reported!
+        self.called = False
+
+        # Test Item result
+        self.result = None
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_protocol(self, item):
+        PyTestService.start_pytest_item(item)
+        yield
+        item_result = self.result if self.called else 'SKIPPED'
+        PyTestService.finish_pytest_item(item_result)
+        self.called = False
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_makereport(self):
+        report = (yield).get_result()
+
+        if report.longrepr:
+            PyTestService.post_log(
+                html.escape(report.longreprtext),
+                loglevel='ERROR',
+            )
+
+        if report.when == 'call':
+            self.called = True
+
+            if report.passed:
+                item_result = 'PASSED'
+            elif report.failed:
+                item_result = 'FAILED'
+            else:
+                item_result = 'SKIPPED'
+
+            self.result = item_result

pytest_reportportal/plugin.py

@@ -2,98 +2,49 @@
 # and/or modify it under the terms of the GPL licence
 
 import logging
-import cgi
-
-import pytest
 
 from .service import PyTestService
-
-
-class RP_Report_Listener(object):
-
-    # Identifier if TestItem is called:
-    # if setup is failed, pytest will NOT call
-    #  TestItem and Result will not reported!
-    called = None
-    # Test Item result
-    result = None
-
-    @pytest.mark.hookwrapper
-    def pytest_runtest_makereport(self, item, call):
-        report = (yield).get_result()
-
-        if report.when == "setup":
-            # when function pytest_setup is called,
-            # test item session will be started in RP
-            PyTestService.start_pytest_item(item)
-
-        if report.longrepr:
-            PyTestService.post_log(
-                cgi.escape(report.longreprtext),
-                loglevel='ERROR',
-            )
-
-        if report.when == "call":
-            self.called = True
-            if report.passed:
-                item_result = "PASSED"
-            elif report.failed:
-                item_result = "FAILED"
-            else:
-                item_result = "SKIPPED"
-
-            self.result = item_result
-
-        if report.when == "teardown":
-            # If item is called, result of TestItem is reported
-            if self.called is True:
-                item_result = self.result
-            else:
-                # If setup - failed or skipped,
-                # the TestItem will reported as SKIPPED
-                item_result = "SKIPPED"
-            PyTestService.finish_pytest_item(item_result)
-            self.called = None
+from .listener import RPReportListener
 
 
 def pytest_sessionstart(session):
     PyTestService.init_service(
-        project=session.config.getini("rp_project"),
-        endpoint=session.config.getini("rp_endpoint"),
-        uuid=session.config.getini("rp_uuid"),
-        log_batch_size=int(session.config.getini("rp_log_batch_size")),
-        ignore_errors=bool(session.config.getini("rp_ignore_errors")),
-        ignored_tags=session.config.getini("rp_ignore_tags"),
+        project=session.config.getini('rp_project'),
+        endpoint=session.config.getini('rp_endpoint'),
+        uuid=session.config.getini('rp_uuid'),
+        log_batch_size=int(session.config.getini('rp_log_batch_size')),
+        ignore_errors=bool(session.config.getini('rp_ignore_errors')),
+        ignored_tags=session.config.getini('rp_ignore_tags'),
     )
 
     PyTestService.start_launch(
         session.config.option.rp_launch,
-        tags=session.config.getini("rp_launch_tags"),
-        description=session.config.getini("rp_launch_description"),
+        tags=session.config.getini('rp_launch_tags'),
+        description=session.config.getini('rp_launch_description'),
     )
 
 
-def pytest_sessionfinish(session):
+def pytest_sessionfinish():
     # FixMe: currently method of RP api takes the string parameter
     # so it is hardcoded
-    PyTestService.finish_launch(status="RP_Launch")
+    PyTestService.finish_launch(status='RP_Launch')
 
 
 def pytest_configure(config):
     if not config.option.rp_launch:
-        config.option.rp_launch = config.getini("rp_launch")
+        config.option.rp_launch = config.getini('rp_launch')
 
     # set Pytest_Reporter and configure it
-    config._reporter = RP_Report_Listener()
+    config._reporter = RPReportListener()
 
-    if hasattr(config, "_reporter"):
+    if hasattr(config, '_reporter'):
         config.pluginmanager.register(config._reporter)
 
 
 def pytest_unconfigure(config):
     PyTestService.terminate_service()
 
-    if hasattr(config, "_reporter"):
+    if hasattr(config, '_reporter'):
         reporter = config._reporter
         del config._reporter
         config.pluginmanager.unregister(reporter)
@@ -103,7 +54,7 @@ def pytest_unconfigure(config):
 def pytest_addoption(parser):
     group = parser.getgroup("reporting")
     group.addoption(
-        "--rp-launch",
+        '--rp-launch',
         action="store",
         dest="rp_launch",
         help="Launch name (overrides rp_launch config option)")

pytest_reportportal/rp_logging.py

@@ -0,0 +1,69 @@
+import sys
+import logging
+
+from .service import PyTestService
+
+
+class RPLogger(logging.getLoggerClass()):
+    def __init__(self, name, level=0):
+        super(RPLogger, self).__init__(name, level=level)
+
+    def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False, attachment=None):
+        """
+        Low-level logging routine which creates a LogRecord and then calls
+        all the handlers of this logger to handle the record.
+        """
+        sinfo = None
+        if logging._srcfile:
+            # IronPython doesn't track Python frames, so findCaller raises an
+            # exception on some versions of IronPython. We trap it here so that
+            # IronPython can use logging.
+            try:
+                fn, lno, func, sinfo = self.findCaller(stack_info)
+            except ValueError:  # pragma: no cover
+                fn, lno, func = '(unknown file)', 0, '(unknown function)'
+        else:
+            fn, lno, func = '(unknown file)', 0, '(unknown function)'
+
+        if exc_info and not isinstance(exc_info, tuple):
+            exc_info = sys.exc_info()
+
+        record = self.makeRecord(self.name, level, fn, lno, msg, args, exc_info, func, extra, sinfo)
+        record.attachment = attachment
+        self.handle(record)
+
+
+class RPLogHandler(logging.Handler):
+    def __init__(self, level=logging.NOTSET):
+        super(RPLogHandler, self).__init__(level)
+
+        # Map loglevel codes from `logging` module to ReportPortal text names:
+        self._loglevel_map = {
+            logging.NOTSET: 'TRACE',
+            logging.DEBUG: 'DEBUG',
+            logging.INFO: 'INFO',
+            logging.WARNING: 'WARN',
+            logging.ERROR: 'ERROR',
+            logging.CRITICAL: 'ERROR',
+        }
+        self._sorted_levelnos = sorted(self._loglevel_map.keys(), reverse=True)
+
+    def emit(self, record):
+        msg = ''
+
+        try:
+            msg = self.format(record)
+        except (KeyboardInterrupt, SystemExit):
+            raise
+        except:
+            self.handleError(record)
+
+        for level in self._sorted_levelnos:
+            if level <= record.levelno:
+                break
+
+        return PyTestService.post_log(
+            msg,
+            loglevel=self._loglevel_map[level],
+            attachment=record.__dict__.get('attachment', None),
+        )