Add custom log level handling

Author: HardNorth

examples/test_rp_custom_logging.py

@@ -0,0 +1,25 @@
+#  Copyright (c) 2022 https://reportportal.io .
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License
+
+import logging
+
+logging.basicConfig(level=logging.INFO)
+
+logger = logging.getLogger(__name__)
+
+LOG_LEVEL: int = 35
+LOG_MESSAGE: str = "Assertion error"
+
+
+def test_report_portal_logging():
+    logger.log(LOG_LEVEL, LOG_MESSAGE)

pytest_reportportal/config.py

@@ -81,6 +81,9 @@ class AgentConfig:
     rp_http_timeout: Optional[Union[tuple[float, float], float]]
     rp_report_fixtures: bool
 
+    # Custom log levels and overrides
+    rp_log_custom_levels: Optional[dict[int, str]]
+
     def __init__(self, pytest_config: Config) -> None:
         """Initialize required attributes."""
         self.rp_enabled = to_bool(getattr(pytest_config.option, "rp_enabled", True))
@@ -177,6 +180,16 @@ def __init__(self, pytest_config: Config) -> None:
             self.rp_http_timeout = connect_timeout or read_timeout
         self.rp_report_fixtures = to_bool(self.find_option(pytest_config, "rp_report_fixtures", False))
 
+        # Custom log levels and overrides
+        log_custom_levels = self.find_option(pytest_config, "rp_log_custom_levels")
+        self.rp_log_custom_levels = None
+        if log_custom_levels:
+            levels = {}
+            for custom_level in log_custom_levels:
+                level, level_name = str(custom_level).split(":", maxsplit=1)
+                levels[int(level)] = level_name
+            self.rp_log_custom_levels = levels
+
     # noinspection PyMethodMayBeStatic
     def find_option(self, pytest_config: Config, option_name: str, default: Any = None) -> Any:
         """

pytest_reportportal/plugin.py

@@ -254,6 +254,7 @@ def pytest_runtest_protocol(item: Item) -> Generator[None, Any, None]:
         filter_client_logs=True,
         endpoint=agent_config.rp_endpoint,
         ignored_record_names=("reportportal_client", "pytest_reportportal"),
+        custom_levels=agent_config.rp_log_custom_levels,
     )
     log_format = agent_config.rp_log_format
     if log_format:
@@ -600,6 +601,12 @@ def add_shared_option(name, help_str, default=None, action="store"):
         "rp_log_batch_payload_size",
         help="DEPRECATED: Maximum payload size in bytes of async batch log requests",
     )
+    parser.addini(
+        "rp_log_custom_levels",
+        type="args",
+        help="Custom log levels specified as 'int level:string'. E.G.: '35:ASSERTION'. Overrides existing level if int"
+        " level matches.",
+    )
     parser.addini("rp_ignore_attributes", type="args", help="Ignore specified pytest markers, i.e parametrize")
     parser.addini(
         "rp_is_skipped_an_issue", default=True, type="bool", help="Treat skipped tests as required investigation"

pytest_reportportal/service.py

@@ -631,7 +631,7 @@ def _get_test_case_id(
 
     def _get_issue_ids(self, mark):
         issue_ids = mark.kwargs.get("issue_id", [])
-        if not isinstance(issue_ids, List):
+        if not isinstance(issue_ids, list):
             issue_ids = [issue_ids]
         return issue_ids
 

tests/integration/__init__.py

@@ -12,6 +12,11 @@
 #  limitations under the License
 
 """This package contains integration tests for the project."""
+from collections import defaultdict
+from typing import Optional
+
+from reportportal_client import set_current
+from reportportal_client.steps import StepReporter
 
 from tests.helpers import utils
 
@@ -272,3 +277,48 @@
     (test, HIERARCHY_TEST_VARIABLES[idx], HIERARCHY_TEST_EXPECTED_ITEMS[idx])
     for idx, test in enumerate(HIERARCHY_TESTS)
 ]
+
+ITEM_ID_DICT = defaultdict(lambda: 0)
+ITEM_ID_LIST = []
+
+
+def generate_item_id(*args, **kwargs) -> str:
+    global ITEM_ID_DICT
+    global ITEM_ID_LIST
+    if args:
+        name = args[0]
+    else:
+        name = kwargs["name"]
+    count = ITEM_ID_DICT[name]
+    count += 1
+    ITEM_ID_DICT[name] = count
+    item_id = f"{name}_{count}"
+    ITEM_ID_LIST.append(item_id)
+    return item_id
+
+
+def get_last_item_id() -> Optional[str]:
+    global ITEM_ID_LIST
+    if len(ITEM_ID_LIST) > 0:
+        return ITEM_ID_LIST[-1]
+
+
+def remove_last_item_id(*_, **__) -> Optional[str]:
+    global ITEM_ID_LIST
+    if len(ITEM_ID_LIST) > 0:
+        return ITEM_ID_LIST.pop()
+
+
+def setup_mock(mock_client_init):
+    mock_client = mock_client_init.return_value
+    mock_client.step_reporter = StepReporter(mock_client)
+    set_current(mock_client)
+    return mock_client
+
+
+def setup_mock_for_logging(mock_client_init):
+    mock_client = setup_mock(mock_client_init)
+    mock_client.start_test_item.side_effect = generate_item_id
+    mock_client.finish_test_item.side_effect = remove_last_item_id
+    mock_client.current_item.side_effect = get_last_item_id
+    return mock_client

tests/integration/test_bdd.py

@@ -18,59 +18,13 @@
 from unittest import mock
 
 import pytest
-from reportportal_client import set_current
-from reportportal_client.steps import StepReporter
+from integration import setup_mock, setup_mock_for_logging
 
 from tests import REPORT_PORTAL_SERVICE
 from tests.helpers import utils
 
 pytest_bdd_version = [int(p) for p in importlib.metadata.version("pytest-bdd").split(".")]
 
-ITEM_ID_DICT = defaultdict(lambda: 0)
-ITEM_ID_LIST = []
-
-
-def generate_item_id(*args, **kwargs) -> str:
-    global ITEM_ID_DICT
-    global ITEM_ID_LIST
-    if args:
-        name = args[0]
-    else:
-        name = kwargs["name"]
-    count = ITEM_ID_DICT[name]
-    count += 1
-    ITEM_ID_DICT[name] = count
-    item_id = f"{name}_{count}"
-    ITEM_ID_LIST.append(item_id)
-    return item_id
-
-
-def get_last_item_id() -> Optional[str]:
-    global ITEM_ID_LIST
-    if len(ITEM_ID_LIST) > 0:
-        return ITEM_ID_LIST[-1]
-
-
-def remove_last_item_id(*_, **__) -> Optional[str]:
-    global ITEM_ID_LIST
-    if len(ITEM_ID_LIST) > 0:
-        return ITEM_ID_LIST.pop()
-
-
-def setup_mock(mock_client_init):
-    mock_client = mock_client_init.return_value
-    mock_client.step_reporter = StepReporter(mock_client)
-    set_current(mock_client)
-    return mock_client
-
-
-def setup_mock_for_logging(mock_client_init):
-    mock_client = setup_mock(mock_client_init)
-    mock_client.start_test_item.side_effect = generate_item_id
-    mock_client.finish_test_item.side_effect = remove_last_item_id
-    mock_client.current_item.side_effect = get_last_item_id
-    return mock_client
-
 
 STEP_NAMES = [
     "Given there are 5 cucumbers",

tests/integration/test_config_handling.py

@@ -17,7 +17,9 @@
 from unittest import mock
 
 import pytest
+import test_rp_custom_logging
 from delayed_assert import assert_expectations, expect
+from integration import setup_mock_for_logging
 from reportportal_client import OutputType
 
 from examples.test_rp_logging import LOG_MESSAGE
@@ -268,3 +270,19 @@ def test_client_timeouts(mock_client_init, connect_value, read_value, expected_r
     assert int(result) == 0, "Exit code should be 0 (no errors)"
     assert mock_client_init.call_count == 1
     assert mock_client_init.call_args_list[0][1]["http_timeout"] == expected_result
+
+
+@mock.patch(REPORT_PORTAL_SERVICE)
+def test_rp_log_custom_levels(mock_client_init):
+    setup_mock_for_logging(mock_client_init)
+    custom_log_level = test_rp_custom_logging.LOG_LEVEL
+    custom_log_name = "ASSERTION"
+    variables = dict(utils.DEFAULT_VARIABLES)
+    variables.update({"rp_log_custom_levels": str(custom_log_level) + ":" + custom_log_name})
+
+    result = utils.run_pytest_tests(["examples/test_rp_custom_logging.py"], variables=variables)
+    assert int(result) == 0, "Exit code should be 0 (no errors)"
+
+    mock_client = mock_client_init.return_value
+    assert mock_client.log.call_count == 1
+    assert mock_client.log.call_args_list[0][1]["level"] == custom_log_name

tests/unit/test_plugin.py

@@ -259,6 +259,7 @@ def test_pytest_addoption_adds_correct_ini_file_arguments():
         "rp_log_batch_size",
         "rp_log_batch_payload_limit",
         "rp_log_batch_payload_size",
+        "rp_log_custom_levels",
         "rp_ignore_attributes",
         "rp_is_skipped_an_issue",
         "rp_hierarchy_code",