Skip to content

Commit

Permalink
Pydocs and type hinting fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
@HardNorth
HardNorth committed Sep 25, 2023
1 parent ffafa27 commit 97afbc9
Showing 1 changed file with 66 additions and 23 deletions.
89 changes: 66 additions & 23 deletions reportportal_client/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
import warnings
from abc import abstractmethod
from os import getenv
from typing import Union, Tuple, List, Dict, Any, Optional, TextIO
from typing import Union, Tuple, Any, Optional, TextIO, List, Dict

import requests
from requests.adapters import HTTPAdapter, Retry, DEFAULT_RETRIES
Expand Down Expand Up @@ -49,6 +49,7 @@ class RP(metaclass=AbstractBaseClass):
This abstract class serves as common interface for different Report Portal clients. It's implemented to
ease migration from version to version and to ensure that each particular client has the same methods.
"""

__metaclass__ = AbstractBaseClass

@property
Expand Down Expand Up @@ -77,11 +78,19 @@ def launch_id(self) -> Optional[Union[str, Task[str]]]:
@property
@abstractmethod
def endpoint(self) -> str:
"""Return current base URL.
:return: base URL string
"""
raise NotImplementedError('"endpoint" property is not implemented!')

@property
@abstractmethod
def project(self) -> str:
"""Return current Project name.
:return: Project name string
"""
raise NotImplementedError('"project" property is not implemented!')

@property
Expand All @@ -98,7 +107,7 @@ def start_launch(self,
name: str,
start_time: str,
description: Optional[str] = None,
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
rerun: bool = False,
rerun_of: Optional[str] = None,
**kwargs) -> Union[Optional[str], Task[str]]:
Expand All @@ -122,8 +131,8 @@ def start_test_item(self,
item_type: str,
*,
description: Optional[str] = None,
attributes: Optional[List[Dict]] = None,
parameters: Optional[Dict] = None,
attributes: Optional[List[dict]] = None,
parameters: Optional[dict] = None,
parent_item_id: Union[Optional[str], Task[str]] = None,
has_stats: bool = True,
code_ref: Optional[str] = None,
Expand Down Expand Up @@ -160,7 +169,7 @@ def finish_test_item(self,
*,
status: str = None,
issue: Optional[Issue] = None,
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
description: str = None,
retry: bool = False,
**kwargs: Any) -> Union[Optional[str], Task[str]]:
Expand All @@ -185,7 +194,7 @@ def finish_test_item(self,
def finish_launch(self,
end_time: str,
status: str = None,
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
**kwargs: Any) -> Union[Optional[str], Task[str]]:
"""Finish current launch.
Expand All @@ -200,7 +209,7 @@ def finish_launch(self,
@abstractmethod
def update_test_item(self,
item_uuid: Union[Optional[str], Task[str]],
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
description: Optional[str] = None) -> Union[Optional[str], Task[str]]:
"""Update existing test item at the Report Portal.
Expand All @@ -213,27 +222,56 @@ def update_test_item(self,

@abstractmethod
def get_launch_info(self) -> Union[Optional[dict], Task[str]]:
"""Get the current launch information.
:returns Launch information in dictionary
"""
raise NotImplementedError('"get_launch_info" method is not implemented!')

@abstractmethod
def get_item_id_by_uuid(self, item_uuid: Union[str, Task[str]]) -> Optional[str]:
"""Get test item ID by the given UUID.
:param item_uuid: UUID returned on the item start
:returns Test item ID
"""
raise NotImplementedError('"get_item_id_by_uuid" method is not implemented!')

@abstractmethod
def get_launch_ui_id(self) -> Optional[int]:
"""Get UI ID of the current launch.
:returns UI ID of the given launch. None if UI ID has not been found.
"""
raise NotImplementedError('"get_launch_ui_id" method is not implemented!')

@abstractmethod
def get_launch_ui_url(self) -> Optional[str]:
"""Get full quality URL of the current launch.
:returns launch URL string
"""
raise NotImplementedError('"get_launch_ui_id" method is not implemented!')

@abstractmethod
def get_project_settings(self) -> Optional[Dict]:
def get_project_settings(self) -> Union[Optional[dict], Task[dict]]:
"""Get project settings.
:returns HTTP response in dictionary
"""
raise NotImplementedError('"get_project_settings" method is not implemented!')

@abstractmethod
def log(self, datetime: str, message: str, level: Optional[Union[int, str]] = None,
attachment: Optional[Dict] = None, item_id: Union[Optional[str], Task[str]] = None) -> None:
attachment: Optional[dict] = None, item_id: Union[Optional[str], Task[str]] = None) -> None:
"""Send log message to the Report Portal.
:param datetime: Time in UTC
:param message: Log message text
:param level: Message's log level
:param attachment: Message's attachments
:param item_id: ID of the RP item the message belongs to
"""
raise NotImplementedError('"log" method is not implemented!')

def start(self) -> None:
Expand All @@ -260,7 +298,11 @@ def current_item(self) -> Optional[Union[str, Task[str]]]:
raise NotImplementedError('"current_item" method is not implemented!')

@abstractmethod
def clone(self) -> 'RPClient':
def clone(self) -> 'RP':
"""Clone the client object, set current Item ID as cloned root Item ID.
:returns: Cloned client object
"""
raise NotImplementedError('"clone" method is not implemented!')


Expand All @@ -274,6 +316,7 @@ class RPClient(RP):
NOTICE: the class is not thread-safe, use new class instance for every new
thread to avoid request/response messing and other issues.
"""

api_v1: str
api_v2: str
base_url_v1: str
Expand Down Expand Up @@ -434,7 +477,7 @@ def start_launch(self,
name: str,
start_time: str,
description: Optional[str] = None,
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
rerun: bool = False,
rerun_of: Optional[str] = None,
**kwargs) -> Optional[str]:
Expand Down Expand Up @@ -482,7 +525,7 @@ def start_test_item(self,
item_type: str,
description: Optional[str] = None,
attributes: Optional[List[Dict]] = None,
parameters: Optional[Dict] = None,
parameters: Optional[dict] = None,
parent_item_id: Optional[str] = None,
has_stats: bool = True,
code_ref: Optional[str] = None,
Expand Down Expand Up @@ -550,7 +593,7 @@ def finish_test_item(self,
end_time: str,
status: str = None,
issue: Optional[Issue] = None,
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
description: str = None,
retry: bool = False,
**kwargs: Any) -> Optional[str]:
Expand Down Expand Up @@ -595,7 +638,7 @@ def finish_test_item(self,
def finish_launch(self,
end_time: str,
status: str = None,
attributes: Optional[Union[List, Dict]] = None,
attributes: Optional[Union[list, dict]] = None,
**kwargs: Any) -> Optional[str]:
"""Finish launch.
Expand Down Expand Up @@ -627,7 +670,7 @@ def finish_launch(self,
logger.debug('response message: %s', response.message)
return response.message

def update_test_item(self, item_uuid: str, attributes: Optional[Union[List, Dict]] = None,
def update_test_item(self, item_uuid: str, attributes: Optional[Union[list, dict]] = None,
description: Optional[str] = None) -> Optional[str]:
"""Update existing test item at the Report Portal.
Expand Down Expand Up @@ -657,7 +700,7 @@ def _log(self, batch: Optional[List[RPRequestLog]]) -> Optional[Tuple[str, ...]]
return response.messages

def log(self, time: str, message: str, level: Optional[Union[int, str]] = None,
attachment: Optional[Dict] = None, item_id: Optional[str] = None) -> Optional[Tuple[str, ...]]:
attachment: Optional[dict] = None, item_id: Optional[str] = None) -> Optional[Tuple[str, ...]]:
"""Send log message to the Report Portal.
:param time: Time in UTC
Expand All @@ -673,21 +716,21 @@ def log(self, time: str, message: str, level: Optional[Union[int, str]] = None,
rp_log = RPRequestLog(self.launch_uuid, time, rp_file, item_id, level, message)
return self._log(self._log_batcher.append(rp_log))

def get_item_id_by_uuid(self, uuid: str) -> Optional[str]:
def get_item_id_by_uuid(self, item_uuid: str) -> Optional[str]:
"""Get test item ID by the given UUID.
:param uuid: UUID returned on the item start
:return: Test item ID
:param item_uuid: UUID returned on the item start
:returns Test item ID
"""
url = uri_join(self.base_url_v1, 'item', 'uuid', uuid)
url = uri_join(self.base_url_v1, 'item', 'uuid', item_uuid)
response = HttpRequest(self.session.get, url=url,
verify_ssl=self.verify_ssl).make()
return response.id if response else None

def get_launch_info(self) -> Optional[Dict]:
def get_launch_info(self) -> Optional[dict]:
"""Get the current launch information.
:return dict: Launch information in dictionary
:returns Launch information in dictionary
"""
if self.launch_uuid is None:
return {}
Expand Down Expand Up @@ -737,7 +780,7 @@ def get_launch_ui_url(self) -> Optional[str]:
logger.debug('get_launch_ui_url - UUID: %s', self.launch_uuid)
return url

def get_project_settings(self) -> Optional[Dict]:
def get_project_settings(self) -> Optional[dict]:
"""Get project settings.
:return: HTTP response in dictionary
Expand Down

0 comments on commit 97afbc9

Please sign in to comment.