--flatten-keywords feature initial implementation

Author: HardNorth

robotframework_reportportal/listener.py

@@ -67,6 +67,7 @@ class __DummyContext:
 DEFAULT_BINARY_FILE_TYPE = "application/octet-stream"
 TRUNCATION_SIGN = "...'"
 REMOVED_KEYWORD_CONTENT_LOG = "Content removed using the --remove-keywords option."
+FLATTENED_KEYWORD_CONTENT_LOG = "Content flattened."
 REMOVED_WUKS_KEYWORD_LOG = "{number} failing items removed using the --remove-keywords option."
 REMOVED_FOR_WHILE_KEYWORD_LOG = "{number} passing items removed using the --remove-keywords option."
 WUKS_KEYWORD_NAME = "BuiltIn.Wait Until Keyword Succeeds"
@@ -441,6 +442,13 @@ def start_test(self, name: str, attributes: Dict, ts: Optional[Any] = None) -> N
         test.rp_item_id = self.service.start_test(test=test, ts=ts)
         self._add_current_item(test)
 
+    def _log_data_removed(self, item_id: str, timestamp: str, message: str) -> None:
+        msg = LogMessage(message)
+        msg.level = "DEBUG"
+        msg.item_id = item_id
+        msg.timestamp = timestamp
+        self.__post_log_message(msg)
+
     def _log_keyword_content_removed(self, item_id: str, timestamp: str) -> None:
         self._log_data_removed(item_id, timestamp, REMOVED_KEYWORD_CONTENT_LOG)
 
@@ -464,13 +472,6 @@ def end_test(self, _: Optional[str], attributes: Dict, ts: Optional[Any] = None)
         self._remove_current_item()
         self.service.finish_test(test=test, ts=ts)
 
-    def _log_data_removed(self, item_id: str, timestamp: str, message: str) -> None:
-        msg = LogMessage(message)
-        msg.level = "DEBUG"
-        msg.item_id = item_id
-        msg.timestamp = timestamp
-        self.__post_log_message(msg)
-
     def _do_start_keyword(self, keyword: Keyword, ts: Optional[str] = None) -> None:
         logger.debug(f"ReportPortal - Start Keyword: {keyword.robot_attributes}")
         keyword.rp_item_id = self.service.start_keyword(keyword=keyword, ts=ts)
@@ -483,8 +484,13 @@ def _should_remove(self, keyword: Keyword) -> Optional[KeywordMatch]:
         return None
 
     def _should_flatten(self, keyword: Keyword) -> bool:
+        if not isinstance(keyword, Keyword):
+            return False
         return any(matcher.match(keyword) for matcher in self._flatten_keyword_filters)
 
+    def _log_keyword_content_flattened(self, item_id: str, timestamp: str) -> None:
+        self._log_data_removed(item_id, timestamp, FLATTENED_KEYWORD_CONTENT_LOG)
+
     @check_rp_enabled
     def start_keyword(self, name: str, attributes: Dict, ts: Optional[Any] = None) -> None:
         """Start a new keyword(test step) at the ReportPortal.
@@ -514,7 +520,13 @@ def start_keyword(self, name: str, attributes: Dict, ts: Optional[Any] = None) -
             parent.skipped_keywords.append(kwd)
             kwd.posted = False
         else:
-            self._do_start_keyword(kwd, ts)
+            if parent.flattened or self._should_flatten(parent):
+                kwd.rp_item_id = parent.rp_item_id
+                kwd.flattened = True
+            else:
+                self._do_start_keyword(kwd, ts)
+                if not kwd.flattened and self._should_flatten(kwd):
+                    self._log_keyword_content_flattened(kwd.rp_item_id, kwd.start_time)
             if skip_data:
                 kwd.remove_origin = kwd
             if self._remove_data_passed_tests:
@@ -575,7 +587,7 @@ def end_keyword(self, _: Optional[str], attributes: Dict, ts: Optional[Any] = No
                 self._log_keyword_content_removed(kwd.rp_item_id, kwd.start_time)
 
         self._remove_current_item()
-        if not kwd.posted:
+        if not kwd.posted or kwd.flattened:
             return
         self._do_end_keyword(kwd, ts)
 

tests/integration/test_flatten_keywords.py

@@ -0,0 +1,116 @@
+#  Copyright 2025 EPAM Systems
+#
+#  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 pytest
+from unittest import mock
+from tests import REPORT_PORTAL_SERVICE
+from tests.helpers import utils
+from tests.helpers.utils import DEFAULT_VARIABLES
+
+
+@pytest.mark.parametrize(
+    "file, keyword_to_flatten, exit_code, expected_statuses, log_number, skip_idx, skip_message",
+    [
+        (
+            "examples/rkie_keyword_error.robot",
+            "name: BuiltIn.Run Keyword And Ignore Error",
+            0,
+            ["PASSED"] * 3 + ["SKIPPED"] * 3 + ["PASSED"] * 3,
+            7,
+            0,
+            "Content flattened.",
+        ),
+        # (
+        #     "examples/for_keyword.robot",
+        #     "FOR",
+        #     0,
+        #     ["PASSED"] * 5,
+        #     2,
+        #     0,
+        #     "2 passing items flattened using the --flatten-keywords option.",
+        # ),
+        # (
+        #     "examples/while_keyword.robot",
+        #     "WHILE",
+        #     0,
+        #     ["PASSED"] * 7,
+        #     5,
+        #     2,
+        #     "2 passing items flattened using the --flatten-keywords option.",
+        # ),
+        # (
+        #     "examples/for_keyword_failed.robot",
+        #     "FOR",
+        #     1,
+        #     ["PASSED"] + ["FAILED"] * 6,
+        #     3,
+        #     0,
+        #     "1 passing items flattened using the --flatten-keywords option.",
+        # ),
+        # (
+        #     "examples/while_keyword_failed.robot",
+        #     "WHILE",
+        #     1,
+        #     ["PASSED"] * 3 + ["FAILED"] * 6,
+        #     6,
+        #     2,
+        #     "1 passing items flattened using the --flatten-keywords option.",
+        # ),
+        # (
+        #     "examples/wuks_keyword.robot",
+        #     "WUKS",
+        #     0,
+        #     ["PASSED"] * 2 + ["FAILED"] * 3 + ["PASSED"] * 2 + ["SKIPPED"] * 2 + ["PASSED"] * 4,
+        #     6,
+        #     0,
+        #     "1 failing items flattened using the --flatten-keywords option.",
+        # ),
+        # (
+        #     "examples/wuks_keyword_failed.robot",
+        #     "WUKS",
+        #     1,
+        #     ["PASSED"] * 2 + ["FAILED"] * 6,
+        #     4,
+        #     0,
+        #     "2 failing items flattened using the --flatten-keywords option.",
+        # ),
+    ],
+)
+@mock.patch(REPORT_PORTAL_SERVICE)
+def test_keyword_flatten(
+    mock_client_init, file, keyword_to_flatten, exit_code, expected_statuses, log_number, skip_idx, skip_message
+):
+    mock_client = mock_client_init.return_value
+    mock_client.start_test_item.side_effect = utils.item_id_gen
+
+    variables = DEFAULT_VARIABLES.copy()
+    variables["RP_FLATTEN_KEYWORDS"] = True
+    result = utils.run_robot_tests([file], variables=variables, arguments={"--flatten-keywords": keyword_to_flatten})
+    assert result == exit_code
+
+    launch_start = mock_client.start_launch.call_args_list
+    launch_finish = mock_client.finish_launch.call_args_list
+    assert len(launch_start) == len(launch_finish) == 1
+
+    item_start_calls = mock_client.start_test_item.call_args_list
+    item_finish_calls = mock_client.finish_test_item.call_args_list
+    assert len(item_start_calls) == len(item_finish_calls)
+    assert len(item_finish_calls) == len(expected_statuses)
+
+    statuses = [finish[1]["status"] for finish in item_finish_calls]
+    assert statuses == expected_statuses
+
+    log_calls = utils.get_log_calls(mock_client)
+    assert len(log_calls) == log_number
+    assert sorted(log_calls, key=lambda x: x[1]["time"])[skip_idx][1]["message"] == skip_message