Merge pull request #202 from reportportal/EPMRPP-95939_seamless_log_s…

…creenshots

[EPMRPP-95939] Seamless screenshot logs

examples/library/Log.py

@@ -14,35 +14,19 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-import os
-
 from robotframework_reportportal import logger
 
 
-def screenshot_log(level, message, screenshot_file):
-    """
-    Attach a screenshot file into a log entry on ReportPortal.
-
-    :param level: log entry level
-    :param message: screenshot description
-    :param screenshot_file: path to image file
-    """
-    with open(screenshot_file, "rb") as image_file:
-        file_data = image_file.read()
-    item_log(level, message, {"name": screenshot_file.split(os.path.sep)[-1],
-                              "data": file_data,
-                              "mime": "image/png"})
-
-
-def item_log(level, message, attachment=None):
+def item_log(level, message, attachment=None, html=False):
     """
     Post a log entry on which will be attached to the current processing item.
 
     :param level: log entry level
     :param message: message to post
     :param attachment: path to attachment file
+    :param html: format or not format the message as html
     """
-    logger.write(message, level, attachment=attachment)
+    logger.write(message, level, attachment=attachment, html=html)
 
 
 def launch_log(level, message, attachment=None):

examples/screenshot.robot

@@ -0,0 +1,8 @@
+*** Settings ***
+Library       library/Log.py
+
+*** Test Cases ***
+Selenium Screenshot test
+    Item Log  INFO  </td></tr><tr><td colspan="3"><a href="examples/res/selenium-screenshot-1.png"><img src="examples/res/selenium-screenshot-1.png" width="800px"></a>  None  True
+Playwright Screenshot test
+    Item Log  INFO  </td></tr><tr><td colspan="3"><a href="examples/res/Screenshot_test_FAILURE_SCREENSHOT_1.png" target="_blank"><img src="examples/res/Screenshot_test_FAILURE_SCREENSHOT_1.png" width="800px"/></a>  None  True

robotframework_reportportal/listener.py

@@ -31,7 +31,12 @@
 from robotframework_reportportal.variables import Variables
 
 logger = logging.getLogger(__name__)
-VARIABLE_PATTERN = r'^\s*\${[^}]*}\s*=\s*'
+VARIABLE_PATTERN = re.compile(r'^\s*\${[^}]*}\s*=\s*')
+IMAGE_PATTERN = re.compile(
+    r'</td></tr><tr><td colspan="\d+"><a href="[^"]+"(?: target="_blank")?>'
+    r'<img src="([^"]+)" width="\d+(?:px|pt)?"/?></a>')
+
+DEFAULT_BINARY_FILE_TYPE = 'application/octet-stream'
 TRUNCATION_SIGN = "...'"
 
 
@@ -98,7 +103,7 @@ def __init__(self) -> None:
         self._service = None
         self._variables = None
 
-    def _build_msg_struct(self, message: Dict) -> LogMessage:
+    def _build_msg_struct(self, message: Dict[str, Any]) -> LogMessage:
         """Check if the given message comes from our custom logger or not.
 
         :param message: Message passed by the Robot Framework
@@ -110,6 +115,44 @@ def _build_msg_struct(self, message: Dict) -> LogMessage:
             msg.level = message['level']
         if not msg.launch_log:
             msg.item_id = getattr(self.current_item, 'rp_item_id', None)
+
+        message_str = msg.message
+        if is_binary(message_str):
+            variable_match = VARIABLE_PATTERN.search(message_str)
+            if variable_match:
+                # Treat as partial binary data
+                msg_content = message_str[variable_match.end():]
+                # remove trailing `'"...`, add `...'`
+                msg.message = (message_str[variable_match.start():variable_match.end()]
+                               + str(msg_content.encode('utf-8'))[:-5] + TRUNCATION_SIGN)
+            else:
+                # Do not log full binary data, since it's usually corrupted
+                content_type = guess_content_type_from_bytes(_unescape(message_str, 128))
+                msg.message = (f'Binary data of type "{content_type}" logging skipped, as it was processed as text and'
+                               ' hence corrupted.')
+                msg.level = 'WARN'
+        elif message.get('html', 'no') == 'yes':
+            image_match = IMAGE_PATTERN.match(message_str)
+            if image_match:
+                image_path = image_match.group(1)
+                msg.message = f'Image attached: {image_path}'
+                if os.path.exists(image_path):
+                    image_type_by_name = guess_type(image_path)[0]
+                    with open(image_path, 'rb') as fh:
+                        image_data = fh.read()
+                        image_type_by_data = guess_content_type_from_bytes(image_data)
+                        if image_type_by_name and image_type_by_data and image_type_by_name != image_type_by_data:
+                            logger.warning(
+                                f'Image type mismatch: type by file name "{image_type_by_name}" '
+                                f'!= type by file content "{image_type_by_data}"')
+                            mime_type = DEFAULT_BINARY_FILE_TYPE
+                        else:
+                            mime_type = image_type_by_name or image_type_by_data or DEFAULT_BINARY_FILE_TYPE
+                        msg.attachment = {
+                            'name': os.path.basename(image_path),
+                            'data': image_data,
+                            'mime': mime_type
+                        }
         return msg
 
     def _add_current_item(self, item: Union[Keyword, Launch, Suite, Test]) -> None:
@@ -132,20 +175,6 @@ def log_message(self, message: Dict) -> None:
         :param message: Message passed by the Robot Framework
         """
         msg = self._build_msg_struct(message)
-        if is_binary(msg.message):
-            variable_match = re.search(VARIABLE_PATTERN, msg.message)
-            if variable_match:
-                # Treat as partial binary data
-                msg_content = msg.message[variable_match.end():]
-                # remove trailing `'"...`, add `...'`
-                msg.message = (msg.message[variable_match.start():variable_match.end()]
-                               + str(msg_content.encode('utf-8'))[:-5] + TRUNCATION_SIGN)
-            else:
-                # Do not log full binary data, since it's usually corrupted
-                content_type = guess_content_type_from_bytes(_unescape(msg.message, 128))
-                msg.message = (f'Binary data of type "{content_type}" logging skipped, as it was processed as text and'
-                               ' hence corrupted.')
-                msg.level = 'WARN'
         logger.debug(f'ReportPortal - Log Message: {message}')
         self.service.log(message=msg)
 
@@ -161,7 +190,7 @@ def log_message_with_image(self, msg: Dict, image: str):
             mes.attachment = {
                 'name': os.path.basename(image),
                 'data': fh.read(),
-                'mime': guess_type(image)[0] or 'application/octet-stream'
+                'mime': guess_type(image)[0] or DEFAULT_BINARY_FILE_TYPE
             }
         logger.debug(f'ReportPortal - Log Message with Image: {mes} {image}')
         self.service.log(message=mes)

tests/integration/test_screenshot.py

@@ -0,0 +1,44 @@
+#  Copyright 2023 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.
+
+from tests.helpers import utils
+from unittest import mock
+
+from tests import REPORT_PORTAL_SERVICE
+
+EXAMPLE_TEST = 'examples/screenshot.robot'
+SELENIUM_SCREENSHOT = 'examples/res/selenium-screenshot-1.png'
+PLAYWRIGHT_SCREENSHOT = 'examples/res/Screenshot_test_FAILURE_SCREENSHOT_1.png'
+SCREENSHOTS = [SELENIUM_SCREENSHOT, PLAYWRIGHT_SCREENSHOT]
+
+
+@mock.patch(REPORT_PORTAL_SERVICE)
+def test_screenshot_log(mock_client_init):
+    result = utils.run_robot_tests([EXAMPLE_TEST])
+    assert result == 0  # the test successfully passed
+
+    mock_client = mock_client_init.return_value
+    calls = utils.get_log_calls(mock_client)
+    assert len(calls) == 2
+
+    for i, call in enumerate(calls):
+        message = call[1]['message']
+        assert message == f'Image attached: {SCREENSHOTS[i]}'
+
+        attachment = call[1]['attachment']
+
+        assert attachment['name'] == SCREENSHOTS[i].split('/')[-1]
+        assert attachment['mime'] == 'image/png'
+        with open(SCREENSHOTS[i], 'rb') as file:
+            assert attachment['data'] == file.read()