Merge pull request #2 from ppapou/master

Pytest-Report Portal Logging

Author: avarabyeu

CONTRIBUTING.rst

@@ -1,5 +1,3 @@
-.. highlight:: shell
-
 ============
 Contributing
 ============

README.md

@@ -1,24 +1,44 @@
 # agent-python-pytest
-Framework integration with PyTest
+Pytest plugin for reporting test results of Pytest to the 'Reportal Portal'.
 
-Description:
-Plugin for reporting items results of Pytest to the 'Reportal Portal'.
-
-Install Plugin
+## Usage:
+### Install the plugin
+```bash
 pip install pytest-reportportal
+```
 
-Usage:
-
-Prepare the config file pytest.ini in root dir of tests.
+### Prepare the config file pytest.ini in root dir of tests.
 Content of the pytest.ini file:
-  [pytest]
-  rp_uuid = uid reportportal
-  rp_endpoint = http://ip:port
-  rp_project = Project of ReportPortal
-
-Run tests:
-py.test ./tests --rp_launch AnyLaunchName
+```
+[pytest]
+mandatory fields
+rp_uuid = uid reportportal
+rp_endpoint = http://ip:port
+rp_project = Project of ReportPortal
+```
+also launch tags could be added. But this parapmeter if not mandatory `rp_launch_tags = 'PyTest' 'Report_Portal'`.
 
-Features:
-  - add logging
-  - add Tags parameter to config file
+### For logging of the test item flow please use the python logging util.
+```python
+    # import ReportPortal handler in the test module
+    from pytest_reportportal import RPlogHandler
+    # get logger
+    logger = logging.getLogger()
+    # create hanler, set log level add it to the logger
+    rp_handler = RPlogHandler()
+    rp_handler.setLevel(logging.INFO)
+    logger.addHandler(rp_handler)
+    # In this case only INFO messages will be sent to the RP
+    def test_one(self):
+        logger.info("Case1. Step1")
+        x = "this"
+        logger.info("Case1. Step2")
+        assert 'h' in x
+```
+### Run tests:
+```bash
+py.test ./tests --rp-launch AnyLaunchName
+```
+## Copyright Notice
+Licensed under the [GPLv3](https://www.gnu.org/licenses/quick-guide-gplv3.html)
+license (see the LICENSE file).

pytest_reportportal/__init__.py

@@ -0,0 +1,17 @@
+import logging
+from .service import PyTestService
+
+
+class RPlogHandler(logging.Handler):
+    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)
+
+        return PyTestService.post_log(msg, log_level=self.level)

pytest_reportportal/pytest_rp_plugin.py renamed to pytest_reportportal/plugin.py

@@ -1,7 +1,8 @@
-# This program is free software: you can redistribute it and/or modify iit under the terms of the GPL licence
 import pytest
+# This program is free software: you can redistribute it
+# and/or modify it under the terms of the GPL licence
 import logging
-from service import PyTestService
+from .service import PyTestService
 
 
 class RP_Report_Listener(object):
@@ -17,8 +18,9 @@ class RP_Report_Listener(object):
     def pytest_runtest_makereport(self, item, call):
         report = (yield).get_result()
         if report.when == "setup":
+
             # when function pytest_setup is called,
-            # test item session is started in RP
+            # test item session will be started in RP
             PyTestService.start_pytest_item(item)
 
         if report.when == "call":
@@ -41,65 +43,74 @@ def pytest_runtest_makereport(self, item, call):
                 # the TestItem will reported as SKIPPED
                 item_result = "SKIPPED"
             PyTestService.finish_pytest_item(item_result)
+            self.called = None
 
 
 def pytest_sessionstart(session):
     config = session.config
     if config.option.rp_launch:
         # get config parameters if rp_launch option is set
-        rp_uuid = config.getini('rp_uuid')
-        rp_project = config.getini('rp_project')
-        rp_endpoint = config.getini('rp_endpoint')
+        rp_uuid = config.getini("rp_uuid")
+        rp_project = config.getini("rp_project")
+        rp_endpoint = config.getini("rp_endpoint")
+        rp_launch_tags = config.getini("rp_launch_tags")
         # initialize PyTest
         PyTestService.init_service(
                 project=rp_project,
                 endpoint=rp_endpoint,
                 uuid=rp_uuid)
-        launch_name = config.getoption('rp_launch')
+        launch_name = config.getoption("rp_launch")
 
-        PyTestService.start_launch(launch_name)
+        PyTestService.start_launch(launch_name, tags=rp_launch_tags)
 
 
 def pytest_sessionfinish(session):
     config = session.config
     if config.option.rp_launch:
-        PyTestService.finish_launch("RP_Launch")
+        # FixMe: currently method of RP api takes the string parameter
+        # so it is hardcoded
+        PyTestService.finish_launch(status="RP_Launch")
 
 
 def pytest_configure(config):
-    rp_launch = config.getoption('rp_launch')
+    rp_launch = config.getoption("rp_launch")
     if rp_launch:
         # set Pytest_Reporter and configure it
         config._reporter = RP_Report_Listener()
 
-        if hasattr(config, '_reporter'):
+        if hasattr(config, "_reporter"):
             config.pluginmanager.register(config._reporter)
 
 
 def pytest_unconfigure(config):
-    if hasattr(config, '_reporter'):
+    if hasattr(config, "_reporter"):
         reporter = config._reporter
         del config._reporter
         config.pluginmanager.unregister(reporter)
-        logging.debug('ReportPortal is unconfigured')
+        logging.debug("RP is unconfigured")
 
 
 def pytest_addoption(parser):
     group = parser.getgroup("reporting")
     group.addoption(
-        '--rp_launch',
-        action='store',
-        dest=None,
-        help="The Launch name for ReportPortal, i.e PyTest")
+        "--rp-launch",
+        action="store",
+        dest="rp_launch",
+        help="The Launch name for RP")
 
     parser.addini(
-        'rp_uuid',
+        "rp_uuid",
         help="Uid of RP user")
 
     parser.addini(
-        'rp_endpoint',
-        help="Report portal server")
+        "rp_endpoint",
+        help="RP server")
 
     parser.addini(
         "rp_project",
-        help='Report Portal Project')
+        help="RP Project")
+
+    parser.addini(
+        "rp_launch_tags",
+        type="args",
+        help="Tags for of RP Launch, i.e Perfomance Regression")

pytest_reportportal/service.py

@@ -1,5 +1,6 @@
 import logging
 from time import time
+from six import with_metaclass
 from reportportal_client import (
     ReportPortalService, FinishExecutionRQ, StartLaunchRQ, StartTestItemRQ,
     FinishTestItemRQ, SaveLogRQ)
@@ -9,127 +10,141 @@ def timestamp():
     return str(int(time() * 1000))
 
 
-class PyTestService(object):
+class Singleton(type):
+    _instances = {}
 
-    RP = None
-    TEST_ITEM_STACK = []
-    launch_id = None
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(
+                *args, **kwargs)
+        return cls._instances[cls]
 
-    @staticmethod
-    def init_service(endpoint, project, uuid):
 
-        if PyTestService.RP is None:
+class PyTestServiceClass(with_metaclass(Singleton, object)):
+
+    def __init__(self):
+
+        self.RP = None
+        self.TEST_ITEM_STACK = []
+        self.launch_id = None
+        self.loglevel_map = {
+            0: "TRACE",
+            10: "DEBUG",
+            20: "INFO",
+            30: "WARNING",
+            40: "ERROR"}
+
+    def init_service(self, endpoint, project, uuid):
+
+        if self.RP is None:
             logging.debug(
                 msg="ReportPortal - Init service: "
                     "endpoint={0}, project={1}, uuid={2}".
                     format(endpoint, project, uuid))
-            PyTestService.RP = ReportPortalService(
+            self.RP = ReportPortalService(
                 endpoint=endpoint,
                 project=project,
                 token=uuid)
         else:
-            raise Exception("PyTest is initialized")
-        return PyTestService.RP
+            logging.debug("The pytest is already initialized")
+        return self.RP
 
-    @staticmethod
-    def start_launch(launch_name=None, mode=None):
-        # TODO: Tags, values could be moved to separated module like helper
-        tags = ["PyTest"]
+    def start_launch(
+            self, launch_name=None, mode=None, tags=None, launch=None):
+        # In next versions launch object(suite, testcase)
+        # could be set as parameter
         sl_pt = StartLaunchRQ(
             name=launch_name,
             start_time=timestamp(),
-            description="PyTest_Launch",
+            description='Pytest Launch',
             mode=mode,
             tags=tags)
         logging.debug(msg="ReportPortal - Start launch: "
                           "request_body={0}".format(sl_pt.data))
-        req_data = PyTestService.RP.start_launch(sl_pt)
+        req_data = self.RP.start_launch(sl_pt)
         logging.debug(msg="ReportPortal - Launch started: "
                           "response_body={0}".format(req_data.raw))
-        PyTestService.launch_id = req_data.id
+        self.launch_id = req_data.id
 
-        PyTestService.TEST_ITEM_STACK.append((None, "SUITE"))
+        self.TEST_ITEM_STACK.append((None, "SUITE"))
         logging.debug(
             msg="ReportPortal - Stack: {0}".
-                format(PyTestService.TEST_ITEM_STACK))
+                format(self.TEST_ITEM_STACK))
 
-    @staticmethod
-    def start_pytest_item(test_item=None):
+    def start_pytest_item(self, test_item=None):
         try:
             # for common items
             item_description = test_item.function.__doc__
         except AttributeError:
             # doctest  has no `function` attribute
             item_description = test_item.reportinfo()[2]
-
         start_rq = StartTestItemRQ(
             name=test_item.name,
             description=item_description,
             tags=['PyTest Item Tag'],
             start_time=timestamp(),
-            launch_id=PyTestService.launch_id,
+            launch_id=self.launch_id,
             type="TEST")
 
-        parent_item_id = PyTestService._get_top_id_from_stack()
+        parent_item_id = self._get_top_id_from_stack()
 
         logging.debug(
             msg="ReportPortal - Start TestItem: "
             "request_body={0}, parent_item={1}".format(
                 start_rq.data, parent_item_id))
 
-        req_data = PyTestService.RP.start_test_item(
+        req_data = self.RP.start_test_item(
             parent_item_id=parent_item_id, start_test_item_rq=start_rq)
 
-        PyTestService.TEST_ITEM_STACK.append((req_data.id, "TEST"))
+        self.TEST_ITEM_STACK.append((req_data.id, "TEST"))
         logging.debug(
             msg="ReportPortal - Stack: {0}".
-                format(PyTestService.TEST_ITEM_STACK))
+                format(self.TEST_ITEM_STACK))
 
-    @staticmethod
-    def finish_pytest_item(status, issue=None):
+    def finish_pytest_item(self, status, issue=None):
         fta_rq = FinishTestItemRQ(end_time=timestamp(),
                                   status=status,
                                   issue=issue)
 
-        test_item_id = PyTestService._get_top_id_from_stack()
+        test_item_id = self._get_top_id_from_stack()
         logging.debug(
             msg="ReportPortal - Finish TetsItem:"
                 " request_body={0}, test_id={1}".
                 format(fta_rq.data, test_item_id))
-        PyTestService.RP.finish_test_item(
+        self.RP.finish_test_item(
             item_id=test_item_id,
             finish_test_item_rq=fta_rq)
-        PyTestService.TEST_ITEM_STACK.pop()
+        self.TEST_ITEM_STACK.pop()
         logging.debug(
             msg="ReportPortal - Stack: {0}".
-                format(PyTestService.TEST_ITEM_STACK))
+                format(self.TEST_ITEM_STACK))
 
-    @staticmethod
-    def finish_launch(status):
+    def finish_launch(self, launch=None, status="rp_launch"):
+        # TO finish launch session str parameter is needed
         fl_rq = FinishExecutionRQ(
             end_time=timestamp(),
             status=status)
-        launch_id = PyTestService.launch_id
+        launch_id = self.launch_id
         logging.debug(msg="ReportPortal - Finish launch: "
                           "request_body={0}, launch_id={1}".format(fl_rq.data,
                                                                    launch_id))
-        PyTestService.RP.finish_launch(launch_id, fl_rq)
-        PyTestService.TEST_ITEM_STACK.pop()
+        self.RP.finish_launch(launch_id, fl_rq)
+        self.TEST_ITEM_STACK.pop()
         logging.debug(
             msg="ReportPortal - Stack: {0}".
-                format(PyTestService.TEST_ITEM_STACK))
+                format(self.TEST_ITEM_STACK))
 
-    @staticmethod
-    def _get_top_id_from_stack():
+    def _get_top_id_from_stack(self):
         try:
-            return PyTestService.TEST_ITEM_STACK[-1][0]
+            return self.TEST_ITEM_STACK[-1][0]
         except IndexError:
             return None
 
-    @staticmethod
-    def post_log(message, log_level="DEBUG"):
-        sl_rq = SaveLogRQ(item_id=PyTestService._get_top_id_from_stack(),
+    def post_log(self, message, log_level="INFO"):
+        sl_rq = SaveLogRQ(item_id=self._get_top_id_from_stack(),
                           time=timestamp(), message=message,
-                          level=log_level)
-        PyTestService.RP.log(sl_rq)
+                          level=self.loglevel_map[log_level])
+        self.RP.log(sl_rq)
+
 
+PyTestService = PyTestServiceClass()