22
22
import time
23
23
import uuid
24
24
from platform import machine , processor , system
25
- from typing import Optional , Any , List , Dict , Callable , Tuple , Union , TypeVar , Generic
25
+ from types import MappingProxyType
26
+ from typing import Optional , Any , List , Dict , Callable , Tuple , Union , TypeVar , Generic , Iterable
26
27
27
28
from reportportal_client .core .rp_file import RPFile
28
29
29
30
logger : logging .Logger = logging .getLogger (__name__ )
30
31
_T = TypeVar ('_T' )
31
32
ATTRIBUTE_LENGTH_LIMIT : int = 128
32
33
TRUNCATE_REPLACEMENT : str = '...'
34
+ BYTES_TO_READ_FOR_DETECTION = 128
35
+
36
+ CONTENT_TYPE_TO_EXTENSIONS = MappingProxyType ({
37
+ 'application/pdf' : 'pdf' ,
38
+ 'application/zip' : 'zip' ,
39
+ 'application/java-archive' : 'jar' ,
40
+ 'image/jpeg' : 'jpg' ,
41
+ 'image/png' : 'png' ,
42
+ 'image/gif' : 'gif' ,
43
+ 'image/bmp' : 'bmp' ,
44
+ 'image/vnd.microsoft.icon' : 'ico' ,
45
+ 'image/webp' : 'webp' ,
46
+ 'audio/mpeg' : 'mp3' ,
47
+ 'audio/wav' : 'wav' ,
48
+ 'video/mpeg' : 'mpeg' ,
49
+ 'video/avi' : 'avi' ,
50
+ 'video/webm' : 'webm' ,
51
+ 'text/plain' : 'txt' ,
52
+ 'application/octet-stream' : 'bin'
53
+ })
33
54
34
55
35
56
class LifoQueue (Generic [_T ]):
@@ -123,15 +144,15 @@ def dict_to_payload(dictionary: Optional[dict]) -> Optional[List[dict]]:
123
144
return result
124
145
125
146
126
- def gen_attributes (rp_attributes : List [str ]) -> List [Dict [str , str ]]:
147
+ def gen_attributes (rp_attributes : Iterable [str ]) -> List [Dict [str , str ]]:
127
148
"""Generate list of attributes for the API request.
128
149
129
150
Example of input list:
130
151
['tag_name:tag_value1', 'tag_value2']
131
152
Output of the function for the given input list:
132
153
[{'key': 'tag_name', 'value': 'tag_value1'}, {'value': 'tag_value2'}]
133
154
134
- :param rp_attributes: List of attributes(tags)
155
+ :param rp_attributes: Iterable of attributes(tags)
135
156
:return: Correctly created list of dictionaries
136
157
to be passed to RP
137
158
"""
@@ -147,8 +168,7 @@ def gen_attributes(rp_attributes: List[str]) -> List[Dict[str, str]]:
147
168
if all (attr_dict .values ()):
148
169
attrs .append (attr_dict )
149
170
continue
150
- logger .debug ('Failed to process "{0}" attribute, attribute value'
151
- ' should not be empty.' .format (rp_attr ))
171
+ logger .debug (f'Failed to process "{ rp_attr } " attribute, attribute value should not be empty.' )
152
172
return attrs
153
173
154
174
@@ -309,31 +329,22 @@ def get_function_params(func: Callable, args: tuple, kwargs: Dict[str, Any]) ->
309
329
310
330
311
331
TYPICAL_MULTIPART_BOUNDARY : str = '--972dbca3abacfd01fb4aea0571532b52'
312
-
313
332
TYPICAL_JSON_PART_HEADER : str = TYPICAL_MULTIPART_BOUNDARY + '''\r
314
333
Content-Disposition: form-data; name="json_request_part"\r
315
334
Content-Type: application/json\r
316
335
\r
317
336
'''
318
-
319
337
TYPICAL_FILE_PART_HEADER : str = TYPICAL_MULTIPART_BOUNDARY + '''\r
320
338
Content-Disposition: form-data; name="file"; filename="{0}"\r
321
339
Content-Type: {1}\r
322
340
\r
323
341
'''
324
-
325
342
TYPICAL_JSON_PART_HEADER_LENGTH : int = len (TYPICAL_JSON_PART_HEADER )
326
-
327
343
TYPICAL_MULTIPART_FOOTER : str = '\r \n ' + TYPICAL_MULTIPART_BOUNDARY + '--'
328
-
329
344
TYPICAL_MULTIPART_FOOTER_LENGTH : int = len (TYPICAL_MULTIPART_FOOTER )
330
-
331
345
TYPICAL_JSON_ARRAY : str = '[]'
332
-
333
346
TYPICAL_JSON_ARRAY_LENGTH : int = len (TYPICAL_JSON_ARRAY )
334
-
335
347
TYPICAL_JSON_ARRAY_ELEMENT : str = ','
336
-
337
348
TYPICAL_JSON_ARRAY_ELEMENT_LENGTH : int = len (TYPICAL_JSON_ARRAY_ELEMENT )
338
349
339
350
@@ -419,8 +430,8 @@ def guess_content_type_from_bytes(data: Union[bytes, bytearray, List[int]]) -> s
419
430
if isinstance (data , list ):
420
431
my_data = bytes (my_data )
421
432
422
- if len (my_data ) >= 128 :
423
- my_data = my_data [:128 ]
433
+ if len (my_data ) >= BYTES_TO_READ_FOR_DETECTION :
434
+ my_data = my_data [:BYTES_TO_READ_FOR_DETECTION ]
424
435
425
436
if not is_binary (my_data ):
426
437
return 'text/plain'
0 commit comments