Release

.github/workflows/release.yml

@@ -36,10 +36,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Set up Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: '3.8'
 
@@ -76,7 +76,7 @@ jobs:
           git push --tags
 
       - name: Checkout develop branch
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           ref: 'develop'
           fetch-depth: 0

.github/workflows/tests.yml

@@ -37,10 +37,10 @@ jobs:
         python-version: [ '3.7', '3.8', '3.9', '3.10', '3.11' ]
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python-version }}
 
@@ -55,8 +55,9 @@ jobs:
 
       - name: Upload coverage to Codecov
         if: matrix.python-version == 3.8 && success()
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         with:
+          token: ${{ secrets.CODECOV_TOKEN }}
           files: coverage.xml
           flags: unittests
           name: codecov-client-reportportal

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 ### Added
+- `is_binary` method in `helpers` module, by @HardNorth
+- `guess_content_type_from_bytes` method in `helpers` module, by @HardNorth
+
+## [5.5.4]
+### Added
 - Issue [#225](https://github.com/reportportal/client-Python/issues/225): JSON decoding error logging, by @HardNorth
 ### Fixed
 - Issue [#226](https://github.com/reportportal/client-Python/issues/226): Logging batch flush on client close, by @HardNorth

README.md

@@ -4,7 +4,7 @@
 [![Python versions](https://img.shields.io/pypi/pyversions/reportportal-client.svg)](https://pypi.org/project/reportportal-client)
 [![Build Status](https://github.com/reportportal/client-Python/actions/workflows/tests.yml/badge.svg)](https://github.com/reportportal/client-Python/actions/workflows/tests.yml)
 [![codecov.io](https://codecov.io/gh/reportportal/client-Python/branch/develop/graph/badge.svg)](https://codecov.io/gh/reportportal/client-Python)
-[![Join Slack chat!](https://slack.epmrpp.reportportal.io/badge.svg)](https://slack.epmrpp.reportportal.io/)
+[![Join Slack chat!](https://img.shields.io/badge/slack-join-brightgreen.svg)](https://slack.epmrpp.reportportal.io/)
 [![stackoverflow](https://img.shields.io/badge/reportportal-stackoverflow-orange.svg?style=flat)](http://stackoverflow.com/questions/tagged/reportportal)
 [![Build with Love](https://img.shields.io/badge/build%20with-❤%EF%B8%8F%E2%80%8D-lightgrey.svg)](http://reportportal.io?style=flat)
 

reportportal_client/_internal/static/abstract.py

@@ -40,16 +40,14 @@ class Implementation(Interface):
     def __call__(cls, *args, **kwargs):
         """Disable instantiation for the interface classes."""
         if cls.__name__ in AbstractBaseClass._abc_registry:
-            raise TypeError("No instantiation allowed for Interface-Class"
-                            " '{}'. Please inherit.".format(cls.__name__))
+            raise TypeError("No instantiation allowed for Interface-Class '{}'. Please inherit.".format(cls.__name__))
 
         result = super(AbstractBaseClass, cls).__call__(*args, **kwargs)
         return result
 
     def __new__(mcs, name, bases, namespace):
         """Register instance of the implementation class."""
-        class_ = super(AbstractBaseClass, mcs).__new__(mcs, name,
-                                                       bases, namespace)
+        class_ = super(AbstractBaseClass, mcs).__new__(mcs, name, bases, namespace)
         if namespace.get("__metaclass__") is AbstractBaseClass:
             mcs._abc_registry.append(name)
         return class_

reportportal_client/helpers.py

@@ -391,3 +391,78 @@ async def await_if_necessary(obj: Optional[Any]) -> Optional[Any]:
         elif asyncio.iscoroutinefunction(obj):
             return await obj()
     return obj
+
+
+def is_binary(iterable: Union[bytes, bytearray, str]) -> bool:
+    """Check if given iterable is binary.
+
+    :param iterable: iterable to check
+    :return: True if iterable contains binary bytes, False otherwise
+    """
+    if isinstance(iterable, str):
+        byte_iterable = iterable.encode('utf-8')
+    else:
+        byte_iterable = iterable
+
+    if 0x00 in byte_iterable:
+        return True
+    return False
+
+
+def guess_content_type_from_bytes(data: Union[bytes, bytearray, List[int]]) -> str:
+    """Guess content type from bytes.
+
+    :param data: bytes or bytearray
+    :return: content type
+    """
+    my_data = data
+    if isinstance(data, list):
+        my_data = bytes(my_data)
+
+    if len(my_data) >= 128:
+        my_data = my_data[:128]
+
+    if not is_binary(my_data):
+        return 'text/plain'
+
+    # images
+    if my_data.startswith(b'\xff\xd8\xff'):
+        return 'image/jpeg'
+    if my_data.startswith(b'\x89PNG\r\n\x1a\n'):
+        return 'image/png'
+    if my_data.startswith(b'GIF8'):
+        return 'image/gif'
+    if my_data.startswith(b'BM'):
+        return 'image/bmp'
+    if my_data.startswith(b'\x00\x00\x01\x00'):
+        return 'image/vnd.microsoft.icon'
+    if my_data.startswith(b'RIFF') and b'WEBP' in my_data:
+        return 'image/webp'
+
+    # audio
+    if my_data.startswith(b'ID3'):
+        return 'audio/mpeg'
+    if my_data.startswith(b'RIFF') and b'WAVE' in my_data:
+        return 'audio/wav'
+
+    # video
+    if my_data.startswith(b'\x00\x00\x01\xba'):
+        return 'video/mpeg'
+    if my_data.startswith(b'RIFF') and b'AVI LIST' in my_data:
+        return 'video/avi'
+    if my_data.startswith(b'\x1aE\xdf\xa3'):
+        return 'video/webm'
+
+    # archives
+    if my_data.startswith(b'PK\x03\x04'):
+        if my_data.startswith(b'PK\x03\x04\x14\x00\x08'):
+            return 'application/java-archive'
+        return 'application/zip'
+    if my_data.startswith(b'PK\x05\x06'):
+        return 'application/zip'
+
+    # office
+    if my_data.startswith(b'%PDF'):
+        return 'application/pdf'
+
+    return 'application/octet-stream'

setup.py

@@ -4,7 +4,7 @@
 
 from setuptools import setup, find_packages
 
-__version__ = '5.5.4'
+__version__ = '5.5.5'
 
 TYPE_STUBS = ['*.pyi']
 

test_res/files/simple.txt

@@ -0,0 +1,8 @@
+HTTP/1.1 407 Proxy Authentication Required
+Cache-Control: no-cache, no-store, max-age=0, must-revalidate
+Expires: 0
+Pragma: no-cache
+X-Content-Type-Options: nosniff
+X-Frame-Options: DENY
+X-Xss-Protection: 1; mode=block
+Content-Length: 0

tests/test_helpers.py

@@ -21,7 +21,7 @@
 from reportportal_client.helpers import (
     gen_attributes,
     get_launch_sys_attrs,
-    verify_value_length, ATTRIBUTE_LENGTH_LIMIT, TRUNCATE_REPLACEMENT
+    verify_value_length, ATTRIBUTE_LENGTH_LIMIT, TRUNCATE_REPLACEMENT, guess_content_type_from_bytes, is_binary
 )
 
 
@@ -94,3 +94,43 @@ def test_verify_value_length(attributes, expected_attributes):
         assert element.get('key') == expected.get('key')
         assert element.get('value') == expected.get('value')
         assert element.get('system') == expected.get('system')
+
+
+@pytest.mark.parametrize(
+    'file, expected_is_binary',
+    [
+        ('test_res/pug/lucky.jpg', True),
+        ('test_res/pug/unlucky.jpg', True),
+        ('test_res/files/image.png', True),
+        ('test_res/files/demo.zip', True),
+        ('test_res/files/test.jar', True),
+        ('test_res/files/test.pdf', True),
+        ('test_res/files/test.bin', True),
+        ('test_res/files/simple.txt', False),
+    ]
+)
+def test_binary_content_detection(file, expected_is_binary):
+    """Test for validate binary content detection."""
+    with open(file, 'rb') as f:
+        content = f.read()
+    assert is_binary(content) == expected_is_binary
+
+
+@pytest.mark.parametrize(
+    'file, expected_type',
+    [
+        ('test_res/pug/lucky.jpg', 'image/jpeg'),
+        ('test_res/pug/unlucky.jpg', 'image/jpeg'),
+        ('test_res/files/image.png', 'image/png'),
+        ('test_res/files/demo.zip', 'application/zip'),
+        ('test_res/files/test.jar', 'application/java-archive'),
+        ('test_res/files/test.pdf', 'application/pdf'),
+        ('test_res/files/test.bin', 'application/octet-stream'),
+        ('test_res/files/simple.txt', 'text/plain'),
+    ]
+)
+def test_binary_content_type_detection(file, expected_type):
+    """Test for validate binary content type detection."""
+    with open(file, 'rb') as f:
+        content = f.read()
+    assert guess_content_type_from_bytes(content) == expected_type