# Copyright (c) 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

"""This module contains asynchronous implementations of ReportPortal Client."""

# mypy: ignore-errors

import asyncio

import logging

import ssl

import threading

import time

import warnings

from datetime import datetime

from os import getenv

from typing import Any, Coroutine, Optional, TypeVar, Union, cast

import aiohttp

import certifi

# noinspection PyProtectedMember

from reportportal_client._internal.aio.http import ClientSession, RetryingClientSession

# noinspection PyProtectedMember

from reportportal_client._internal.aio.tasks import (

DEFAULT_TASK_TRIGGER_INTERVAL,

DEFAULT_TASK_TRIGGER_NUM,

BackgroundTaskList,

BatchedTaskFactory,

ThreadedTaskFactory,

TriggerTaskBatcher,

)

# noinspection PyProtectedMember

from reportportal_client._internal.local import set_current

# noinspection PyProtectedMember

from reportportal_client._internal.logs.batcher import LogBatcher

# noinspection PyProtectedMember

from reportportal_client._internal.services.auth import ApiKeyAuthAsync, AuthAsync, OAuthPasswordGrantAsync

# noinspection PyProtectedMember

from reportportal_client._internal.services.statistics import async_send_event

# noinspection PyProtectedMember

from reportportal_client._internal.static.abstract import AbstractBaseClass, abstractmethod

# noinspection PyProtectedMember

from reportportal_client._internal.static.defines import DEFAULT_LOG_LEVEL, MICROSECONDS_MIN_VERSION

# noinspection PyProtectedMember

from reportportal_client.aio.tasks import EmptyTask, Task

from reportportal_client.aio.util import await_if_necessary

from reportportal_client.client import RP, OutputType

from reportportal_client.core.rp_issues import Issue

from reportportal_client.core.rp_requests import (

AsyncHttpRequest,

AsyncItemFinishRequest,

AsyncItemStartRequest,

AsyncRPLogBatch,

AsyncRPRequestLog,

ErrorPrintingAsyncHttpRequest,

ItemUpdateRequest,

LaunchFinishRequest,

LaunchStartRequest,

RPFile,

)

from reportportal_client.helpers import (

ITEM_DESCRIPTION_LENGTH_LIMIT,

ITEM_NAME_LENGTH_LIMIT,

LAUNCH_DESCRIPTION_LENGTH_LIMIT,

LAUNCH_NAME_LENGTH_LIMIT,

LifoQueue,

agent_name_version,

compare_semantic_versions,

extract_server_version,

root_uri_join,

uri_join,

)

from reportportal_client.logs import MAX_LOG_BATCH_PAYLOAD_SIZE

from reportportal_client.steps import StepReporter

logger = logging.getLogger(__name__)

logger.addHandler(logging.NullHandler())

_T = TypeVar("_T")

DEFAULT_TASK_TIMEOUT: float = 60.0

DEFAULT_SHUTDOWN_TIMEOUT: float = 120.0

class Client:

"""Stateless asynchronous ReportPortal Client.

This class intentionally made to not store any data or context from ReportPortal. It provides basic

reporting and data read functions in asynchronous manner. Use it whenever you want to handle item IDs, log

batches, future tasks on your own.

"""

api_v1: str

api_v2: str

base_url_v1: str

base_url_v2: str

endpoint: str

is_skipped_an_issue: bool

project: str

api_key: Optional[str]

oauth_uri: Optional[str]

oauth_username: Optional[str]

oauth_password: Optional[str]

oauth_client_id: Optional[str]

oauth_client_secret: Optional[str]

oauth_scope: Optional[str]

auth: AuthAsync

verify_ssl: Union[bool, str]

retries: Optional[int]

max_pool_size: int

http_timeout: Optional[Union[float, tuple[float, float]]]

keepalive_timeout: Optional[float]

mode: str

launch_uuid_print: bool

print_output: OutputType

truncate_attributes: bool

truncate_fields: bool

replace_binary_chars: bool

launch_name_length_limit: int

item_name_length_limit: int

launch_description_length_limit: int

item_description_length_limit: int

_skip_analytics: Optional[str]

_session: Optional[ClientSession]

__stat_task: Optional[asyncio.Task]

def __init__(

self,

endpoint: str,

project: str,

*,

api_key: Optional[str] = None,

is_skipped_an_issue: bool = True,

verify_ssl: Union[bool, str] = True,

retries: Optional[int] = -1,

max_pool_size: int = 50,

http_timeout: Optional[Union[float, tuple[float, float]]] = (10, 10),

keepalive_timeout: Optional[float] = None,

mode: str = "DEFAULT",

launch_uuid_print: bool = False,

print_output: OutputType = OutputType.STDOUT,

truncate_attributes: bool = True,

truncate_fields: bool = True,

replace_binary_chars: bool = True,

launch_name_length_limit: int = LAUNCH_NAME_LENGTH_LIMIT,

item_name_length_limit: int = ITEM_NAME_LENGTH_LIMIT,

launch_description_length_limit: int = LAUNCH_DESCRIPTION_LENGTH_LIMIT,

item_description_length_limit: int = ITEM_DESCRIPTION_LENGTH_LIMIT,

# OAuth 2.0 Password Grant parameters

oauth_uri: Optional[str] = None,

oauth_username: Optional[str] = None,

oauth_password: Optional[str] = None,

oauth_client_id: Optional[str] = None,

oauth_client_secret: Optional[str] = None,

oauth_scope: Optional[str] = None,

**kwargs: Any,

) -> None:

"""Initialize the class instance with arguments.

:param endpoint: Endpoint of the ReportPortal service.

:param project: Project name to report to.

:param api_key: Authorization API key.

:param oauth_uri: OAuth 2.0 token endpoint URI (for OAuth authentication).

:param oauth_username: Username for OAuth 2.0 authentication.

:param oauth_password: Password for OAuth 2.0 authentication.

:param oauth_client_id: OAuth 2.0 client ID.

:param oauth_client_secret: OAuth 2.0 client secret (optional).

:param oauth_scope: OAuth 2.0 scope (optional).

:param is_skipped_an_issue: Option to mark skipped tests as not 'To Investigate' items on the

server side.

:param verify_ssl: Option to skip ssl verification.

:param retries: Number of retry attempts to make in case of connection / server

errors.

:param max_pool_size: Option to set the maximum number of connections to save the pool.

:param http_timeout: A float in seconds for connect and read timeout. Use a Tuple to

specific connect and read separately.

:param keepalive_timeout: Maximum amount of idle time in seconds before force connection

closing.

:param mode: Launch mode, all Launches started by the client will be in that mode.

:param launch_uuid_print: Print Launch UUID into passed TextIO or by default to stdout.

:param print_output: Set output stream for Launch UUID printing.

:param truncate_attributes: Truncate test item attributes to default maximum length.

:param truncate_fields: Truncate request fields to configured limits.

:param replace_binary_chars: Toggle replacement of basic binary characters with \ufffd char.

:param launch_name_length_limit: Maximum allowed launch name length.

:param item_name_length_limit: Maximum allowed test item name length.

:param launch_description_length_limit: Maximum allowed launch description length.

:param item_description_length_limit: Maximum allowed test item description length.

"""

self.api_v1, self.api_v2 = "v1", "v2"

self.endpoint = endpoint

self.project = project

self.base_url_v1 = root_uri_join(f"api/{self.api_v1}", self.project)

self.base_url_v2 = root_uri_join(f"api/{self.api_v2}", self.project)

self.is_skipped_an_issue = is_skipped_an_issue

self.verify_ssl = verify_ssl

self.retries = retries

self.max_pool_size = max_pool_size

self.http_timeout = http_timeout

self.keepalive_timeout = keepalive_timeout

self.mode = mode

self._skip_analytics = getenv("AGENT_NO_ANALYTICS")

self.launch_uuid_print = launch_uuid_print

self.print_output = print_output

self._session = None

self.__stat_task = None

self.truncate_attributes = truncate_attributes

self.truncate_fields = truncate_fields

self.replace_binary_chars = replace_binary_chars

self.launch_name_length_limit = launch_name_length_limit

self.item_name_length_limit = item_name_length_limit

self.launch_description_length_limit = launch_description_length_limit

self.item_description_length_limit = item_description_length_limit

self.api_key = api_key

# Handle deprecated token argument

if not self.api_key and "token" in kwargs:

warnings.warn(

message="Argument `token` is deprecated since 5.3.5 and will be subject for removing in "

"the next major version. Use `api_key` argument instead.",

category=DeprecationWarning,

stacklevel=2,

)

self.api_key = kwargs["token"]

self.oauth_uri = oauth_uri

self.oauth_username = oauth_username

self.oauth_password = oauth_password

self.oauth_client_id = oauth_client_id

self.oauth_client_secret = oauth_client_secret

self.oauth_scope = oauth_scope

# Initialize authentication

oauth_params = [oauth_uri, oauth_username, oauth_password, oauth_client_id]

oauth_provided = all(oauth_params)

if oauth_provided:

# Use OAuth 2.0 Password Grant authentication

# These params will be defined, since we checked them with "all" keyword above

# These 'or ""' just to mute dump type checkers

self.auth = OAuthPasswordGrantAsync(

oauth_uri=oauth_uri or "",

username=oauth_username or "",

password=oauth_password or "",

client_id=oauth_client_id or "",

client_secret=oauth_client_secret,

scope=oauth_scope,

)

elif self.api_key:

# Use API key authentication

self.auth = ApiKeyAuthAsync(self.api_key)

else:

# Neither OAuth nor API key provided

raise ValueError(

"Authentication credentials are required. Please provide either:\n"

"1. OAuth 2.0 parameters: oauth_uri, username, password, and client_id\n"

" (with optional client_secret and scope), or\n"

"2. api_key parameter for API key authentication.\n"

"\n"

"Example for OAuth:\n"

" Client(endpoint='...', project='...', oauth_oauth_uri='https://example.com/oauth/token',\n"

" oauth_username='user', oauth_password='pass', oauth_client_id='client_id')\n"

"\n"

"Example for API key:\n"

" Client(endpoint='...', project='...', api_key='your_api_key')"

)

async def session(self) -> ClientSession:

"""Return ClientSession class instance, initialize it if necessary.

:return: ClientSession instance.

"""

if self._session:

return self._session

if self.verify_ssl is None or (type(self.verify_ssl) is bool and not self.verify_ssl):

ssl_config = False

else:

if type(self.verify_ssl) is str:

ssl_config = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=self.verify_ssl)

else:

ssl_config = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=certifi.where())

connection_params: dict[str, Any] = {"ssl": ssl_config, "limit": self.max_pool_size}

if self.keepalive_timeout:

connection_params["keepalive_timeout"] = self.keepalive_timeout

connector = aiohttp.TCPConnector(**connection_params)

session_params: dict[str, Any] = {"connector": connector}

if self.http_timeout:

if type(self.http_timeout) is tuple:

connect_timeout, read_timeout = self.http_timeout

else:

connect_timeout, read_timeout = self.http_timeout, self.http_timeout

session_params["timeout"] = aiohttp.ClientTimeout(connect=connect_timeout, sock_read=read_timeout)

retries_set = self.retries is not None and self.retries > 0

# Use retries with default parameters if not set, but don't use retries if it's `None`

use_retries = self.retries == -1 or retries_set

if retries_set:

session_params["max_retry_number"] = self.retries

wrapped_session: Union[aiohttp.ClientSession, RetryingClientSession]

if use_retries:

wrapped_session = RetryingClientSession(self.endpoint, **session_params)

else:

# noinspection PyTypeChecker

wrapped_session = aiohttp.ClientSession(self.endpoint, **session_params)

my_session = ClientSession(wrapped=wrapped_session, auth=self.auth)

self._session = my_session

return my_session

async def close(self) -> None:

"""Gracefully close internal aiohttp.ClientSession class instance and reset it."""

if self._session:

await self._session.close()

self._session = None

async def __get_item_url(self, item_id_future: Union[Optional[str], Task[Optional[str]]]) -> Optional[str]:

item_id = await await_if_necessary(item_id_future)

if not item_id:

logger.warning("Attempt to make request for non-existent id.")

return None

return root_uri_join(self.base_url_v2, "item", item_id)

async def __get_launch_url(self, launch_uuid_future: Union[Optional[str], Task[Optional[str]]]) -> Optional[str]:

launch_uuid = await await_if_necessary(launch_uuid_future)

if not launch_uuid:

logger.warning("Attempt to make request for non-existent launch.")

return None

return root_uri_join(self.base_url_v2, "launch", launch_uuid, "finish")

async def start_launch(

self,

name: str,

start_time: str,

*,

description: Optional[str] = None,

attributes: Optional[Union[list, dict]] = None,

rerun: bool = False,

rerun_of: Optional[str] = None,

**_,

) -> Optional[str]:

"""Start a new Launch with the given arguments.

:param name: Launch name.

:param start_time: Launch start time.

:param description: Launch description.

:param attributes: Launch attributes.

:param rerun: Start launch in rerun mode.

:param rerun_of: For rerun mode specifies which launch will be re-run. Should be used with the

'rerun' option.

:return: Launch UUID if successfully started or None.

"""

url = root_uri_join(self.base_url_v2, "launch")

request_payload = LaunchStartRequest(

name=name,

start_time=start_time,

attributes=attributes,

truncate_attributes_enabled=self.truncate_attributes,

truncate_fields_enabled=self.truncate_fields,

replace_binary_characters=self.replace_binary_chars,

description=description,

mode=self.mode,

rerun=rerun,

rerun_of=rerun_of,

).payload

response = await AsyncHttpRequest(

(await self.session()).post, url=url, json=request_payload, name="start_launch"

).make()

if not response:

return None

if not self._skip_analytics:

stat_coro = async_send_event("start_launch", *agent_name_version(attributes))

self.__stat_task = asyncio.create_task(stat_coro)

launch_uuid = await response.id

logger.debug(f"start_launch - ID: {launch_uuid}")

if self.launch_uuid_print and self.print_output:

print(f"ReportPortal Launch UUID: {launch_uuid}", file=self.print_output.get_output())

return launch_uuid

async def start_test_item(

self,

launch_uuid: Union[str, Task[str]],

name: str,

start_time: str,

item_type: str,

*,

parent_item_id: Optional[Union[str, Task[str]]] = None,

description: Optional[str] = None,

attributes: Optional[Union[list[dict], dict]] = None,

parameters: Optional[dict] = None,

code_ref: Optional[str] = None,

test_case_id: Optional[str] = None,

has_stats: Optional[bool] = True,

retry: Optional[bool] = False,

retry_of: Optional[str] = None,

uuid: Optional[str] = None,

**_: Any,

) -> Optional[str]:

"""Start Test Case/Suite/Step/Nested Step Item.

:param launch_uuid: A launch UUID where to start the Test Item.

:param name: Name of the Test Item.

:param start_time: The Item start time.

:param item_type: Type of the Test Item. Allowed values:

"suite", "story", "test", "scenario", "step", "before_class", "before_groups",

"before_method", "before_suite", "before_test", "after_class", "after_groups",

"after_method", "after_suite", "after_test".

:param parent_item_id: A UUID of a parent SUITE / STEP.

:param description: The Item description.

:param attributes: Test Item attributes.

:param parameters: Set of parameters (for parametrized Test Items).

:param code_ref: Physical location of the Test Item.

:param test_case_id: A unique ID of the current Step.

:param has_stats: Set to False if test item is a Nested Step.

:param retry: Used to report retry of the test. Allowed values: "True" or "False".

:param retry_of: For retry mode specifies which test item will be marked as retried. Should be used

with the 'retry' parameter.

:param uuid: Test Item UUID to use on start (overrides server one, should be globally unique).

:return: Test Item UUID if successfully started or None.

"""

if parent_item_id:

url = self.__get_item_url(parent_item_id)

else:

url = root_uri_join(self.base_url_v2, "item")

request_payload = AsyncItemStartRequest(

name=name,

start_time=start_time,

type_=item_type,

launch_uuid=launch_uuid,

attributes=attributes,

truncate_attributes_enabled=self.truncate_attributes,

truncate_fields_enabled=self.truncate_fields,

replace_binary_characters=self.replace_binary_chars,

code_ref=code_ref,

description=description,

has_stats=has_stats,

parameters=parameters,

retry=retry,

test_case_id=test_case_id,

retry_of=retry_of,

uuid=uuid,

).payload

response = await AsyncHttpRequest(

(await self.session()).post, url=url, json=request_payload, name="start_test_item"

).make()

if not response:

return None

item_id = await response.id

if not item_id:

logger.warning("start_test_item - invalid response: %s", str(await response.json))

else:

logger.debug("start_test_item - ID: %s", item_id)

return item_id

async def finish_test_item(

self,

launch_uuid: Union[str, Task[str]],

item_id: Union[str, Task[str]],

end_time: str,

*,

status: Optional[str] = None,

description: Optional[str] = None,

attributes: Optional[Union[list, dict]] = None,

test_case_id: Optional[str] = None,

issue: Optional[Issue] = None,

retry: Optional[bool] = False,

retry_of: Optional[str] = None,

**_: Any,

) -> Optional[str]:

"""Finish Test Suite/Case/Step/Nested Step Item.

:param launch_uuid: A launch UUID where to finish the Test Item.

:param item_id: ID of the Test Item.

:param end_time: The Item end time.

:param status: Test status. Allowed values:

PASSED, FAILED, STOPPED, SKIPPED, INTERRUPTED, CANCELLED, INFO, WARN or None.

:param description: Test Item description. Overrides description from start request.

:param attributes: Test Item attributes(tags). Pairs of key and value. These attributes override

attributes on start Test Item call.

:param test_case_id: A unique ID of the current Step. Overrides passed on start request.

:param issue: Issue which will be attached to the current Item.

:param retry: Used to report retry of the test. Allowed values: "True" or "False".

:param retry_of: For retry mode specifies which test item will be marked as retried. Should be used

with the 'retry' parameter.

:return: Response message.

"""

url = self.__get_item_url(item_id)

request_payload = AsyncItemFinishRequest(

end_time=end_time,

launch_uuid=launch_uuid,

status=status,

attributes=attributes,

truncate_attributes_enabled=self.truncate_attributes,

truncate_fields_enabled=self.truncate_fields,

replace_binary_characters=self.replace_binary_chars,

description=description,

test_case_id=test_case_id,

is_skipped_an_issue=self.is_skipped_an_issue,

issue=issue,

retry=retry,

retry_of=retry_of,

).payload

response = await AsyncHttpRequest(

(await self.session()).put, url=url, json=request_payload, name="finish_test_item"

).make()

if not response:

return None

message = await response.message

logger.debug("finish_test_item - ID: %s", await await_if_necessary(item_id))

logger.debug("response message: %s", message)

return message

async def finish_launch(

self,

launch_uuid: Union[str, Task[str]],

end_time: str,

*,

status: Optional[str] = None,

attributes: Optional[Union[list, dict]] = None,

**kwargs: Any,

) -> Optional[str]:

"""Finish a Launch.

:param launch_uuid: A Launch UUID to finish.

:param end_time: Launch end time.

:param status: Launch status. Can be one of the followings:

PASSED, FAILED, STOPPED, SKIPPED, INTERRUPTED, CANCELLED or None.

:param attributes: Launch attributes. These attributes override attributes on Start Launch call.

:return: Response message or None.

"""

url = self.__get_launch_url(launch_uuid)

request_payload = LaunchFinishRequest(

end_time=end_time,

status=status,

attributes=attributes,

truncate_attributes_enabled=self.truncate_attributes,

truncate_fields_enabled=self.truncate_fields,

replace_binary_characters=self.replace_binary_chars,

description=kwargs.get("description"),

).payload

response = await AsyncHttpRequest(

(await self.session()).put, url=url, json=request_payload, name="finish_launch"

).make()

if not response:

return None

logger.debug("finish_launch - ID: %s", await await_if_necessary(launch_uuid))

return None

async def update_test_item(

self,

item_uuid: Union[str, Task[str]],

*,

attributes: Optional[Union[list, dict]] = None,

description: Optional[str] = None,

) -> Optional[str]:

"""Update existing Test Item at the ReportPortal.

:param item_uuid: Test Item UUID returned on the item start.

:param attributes: Test Item attributes: [{'key': 'k_name', 'value': 'k_value'}, ...].

:param description: Test Item description.

:return: Response message or None.

"""

data = ItemUpdateRequest(

description=description,

attributes=attributes,

truncate_attributes_enabled=self.truncate_attributes,

truncate_fields_enabled=self.truncate_fields,

replace_binary_characters=self.replace_binary_chars,

).payload

item_id = await self.get_item_id_by_uuid(item_uuid)

url = root_uri_join(self.base_url_v1, "item", item_id, "update")

response = await AsyncHttpRequest(

(await self.session()).put, url=url, json=data, name="update_test_item"

).make()

if not response:

return None

logger.debug("update_test_item - Item: %s", item_id)

return await response.message

async def __get_launch_uuid_url(self, launch_uuid_future: Union[str, Task[str]]) -> Optional[str]:

launch_uuid = await await_if_necessary(launch_uuid_future)

if not launch_uuid:

logger.warning("Attempt to make request for non-existent Launch UUID.")

return None

logger.debug("get_launch_info - ID: %s", launch_uuid)

return root_uri_join(self.base_url_v1, "launch", "uuid", launch_uuid)

async def get_launch_info(self, launch_uuid_future: Union[str, Task[str]]) -> Optional[dict]:

"""Get Launch information by Launch UUID.

:param launch_uuid_future: Str or Task UUID returned on the Launch start.

:return: Launch information in dictionary.

"""

url = self.__get_launch_uuid_url(launch_uuid_future)

response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_launch_info").make()

if not response:

return None

launch_info = None

if response.is_success:

launch_info = await response.json

logger.debug("get_launch_info - Launch info: %s", launch_info)

else:

logger.warning("get_launch_info - Launch info: Failed to fetch launch ID from the API.")

return launch_info

async def __get_item_uuid_url(self, item_uuid_future: Union[Optional[str], Task[Optional[str]]]) -> Optional[str]:

item_uuid = await await_if_necessary(item_uuid_future)

if not item_uuid:

logger.warning("Attempt to make request for non-existent UUID.")

return None

return root_uri_join(self.base_url_v1, "item", "uuid", item_uuid)

async def get_item_id_by_uuid(self, item_uuid_future: Union[str, Task[str]]) -> Optional[str]:

"""Get Test Item ID by the given Item UUID.

:param item_uuid_future: Str or Task UUID returned on the Item start.

:return: Test Item ID.

"""

url = self.__get_item_uuid_url(item_uuid_future)

response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_item_id").make()

return await response.id if response else None

async def get_launch_ui_id(self, launch_uuid_future: Union[str, Task[str]]) -> Optional[int]:

"""Get Launch ID of the given Launch.

:param launch_uuid_future: Str or Task UUID returned on the Launch start.

:return: Launch ID of the Launch. None if not found.

"""

launch_info = await self.get_launch_info(launch_uuid_future)

return launch_info.get("id") if launch_info else None

async def get_launch_ui_url(self, launch_uuid_future: Union[str, Task[str]]) -> Optional[str]:

"""Get full quality URL of the given Launch.

:param launch_uuid_future: Str or Task UUID returned on the Launch start.

:return: Launch URL string.

"""

launch_uuid = await await_if_necessary(launch_uuid_future)

launch_info = await self.get_launch_info(launch_uuid)

launch_id = launch_info.get("id") if launch_info else None

if not launch_id:

return None

mode = launch_info.get("mode") if launch_info else None

if not mode:

mode = self.mode

launch_type = "launches" if mode.upper() == "DEFAULT" else "userdebug"

path = f"ui/#{self.project.lower()}/{launch_type}/all/{launch_id}"

url = uri_join(self.endpoint, path)

logger.debug("get_launch_ui_url - ID: %s", launch_uuid)

return url

async def get_project_settings(self) -> Optional[dict]:

"""Get settings of the current Project.

:return: Settings response in Dictionary.

"""

url = root_uri_join(self.base_url_v1, "settings")

response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_project_settings").make()

return await response.json if response else None

async def get_api_info(self) -> Optional[dict]:

"""Get server information, like version.

:return: server information.

"""

url = root_uri_join("api/info")

response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_api_info").make()

return await response.json if response else None

async def log_batch(self, log_batch: Optional[list[AsyncRPRequestLog]]) -> Optional[tuple[str, ...]]:

"""Send batch logging message to the ReportPortal.

:param log_batch: A list of log message objects.

:return: Completion message tuple of variable size (depending on request size).

"""

if not log_batch:

return None

url = root_uri_join(self.base_url_v2, "log")

response = await ErrorPrintingAsyncHttpRequest(

(await self.session()).post,

url=url,

data=AsyncRPLogBatch(

truncate_attributes_enabled=None,

truncate_fields_enabled=None,

replace_binary_characters=None,

log_reqs=log_batch,

).payload,

name="log",

).make()

return await response.messages if response else None

def clone(self) -> "Client":

"""Clone the client object, set current Item ID as cloned item ID.

:return: Cloned client object

:rtype: Client

"""

cloned = Client(

endpoint=self.endpoint,

project=self.project,

api_key=self.api_key,

is_skipped_an_issue=self.is_skipped_an_issue,

verify_ssl=self.verify_ssl,

retries=self.retries,

max_pool_size=self.max_pool_size,

http_timeout=self.http_timeout,

keepalive_timeout=self.keepalive_timeout,

mode=self.mode,

launch_uuid_print=self.launch_uuid_print,

print_output=self.print_output,

truncate_fields=self.truncate_fields,

truncate_attributes=self.truncate_attributes,

replace_binary_chars=self.replace_binary_chars,

launch_name_length_limit=self.launch_name_length_limit,

item_name_length_limit=self.item_name_length_limit,

launch_description_length_limit=self.launch_description_length_limit,

item_description_length_limit=self.item_description_length_limit,

oauth_uri=self.oauth_uri,

oauth_username=self.oauth_username,

oauth_password=self.oauth_password,

oauth_client_id=self.oauth_client_id,

oauth_client_secret=self.oauth_client_secret,

oauth_scope=self.oauth_scope,

)

return cloned

def __getstate__(self) -> dict[str, Any]:

"""Control object pickling and return object fields as Dictionary.

:return: object state dictionary

:rtype: dict

"""

state = self.__dict__.copy()

# Don't pickle 'session' field, since it contains unpickling 'socket'

del state["_session"]

return state

def __setstate__(self, state: dict[str, Any]) -> None:

"""Control object pickling, receives object state as Dictionary.

:param dict state: object state dictionary

"""

self.__dict__.update(state)

self._session = None

class AsyncRPClient(RP):

"""Asynchronous ReportPortal Client.

This class implements common RP client interface but all its methods are async, so it capable to use in

asynchronous ReportPortal agents. It handles HTTP request and response bodies generation and

serialization, connection retries and log batching.

"""

log_batch_size: int

log_batch_payload_limit: int

_item_stack: LifoQueue

_log_batcher: LogBatcher

__client: Client

__launch_uuid: Optional[str]

__step_reporter: StepReporter

use_own_launch: bool

_api_info_task: Optional[asyncio.Task[Optional[dict]]]

_api_info_cache: Optional[dict]

_use_microseconds: Optional[bool]

@property

def client(self) -> Client:

"""Return current Client instance.

:return: Client instance.

"""

return self.__client

@property

def launch_uuid(self) -> Optional[str]:

"""Return current Launch UUID.

:return: UUID string.

"""

return self.__launch_uuid

@property

def endpoint(self) -> str:

"""Return current base URL.

:return: base URL string.

"""

return self.__endpoint

@property

def project(self) -> str:

"""Return current Project name.

:return: Project name string.

"""

return self.__project

@property

def step_reporter(self) -> StepReporter:

"""Return StepReporter object for the current launch.

:return: StepReporter to report steps.

"""

return self.__step_reporter

def __init__(

self,

endpoint: str,

project: str,

*,

client: Optional[Client] = None,

launch_uuid: Optional[str] = None,

log_batch_size: int = 20,

log_batch_payload_limit: int = MAX_LOG_BATCH_PAYLOAD_SIZE,

log_batcher: Optional[LogBatcher] = None,

**kwargs: Any,

) -> None:

"""Initialize the class instance with arguments.

:param endpoint: Endpoint of the ReportPortal service.

:param project: Project name to report to.

:param api_key: Authorization API key.

:param oauth_uri: OAuth 2.0 token endpoint URI (for OAuth authentication).

:param oauth_username: Username for OAuth 2.0 authentication.

:param oauth_password: Password for OAuth 2.0 authentication.

:param oauth_client_id: OAuth 2.0 client ID.

:param oauth_client_secret: OAuth 2.0 client secret (optional).

:param oauth_scope: OAuth 2.0 scope (optional).

:param is_skipped_an_issue: Option to mark skipped tests as not 'To Investigate' items on the

server side.

:param verify_ssl: Option to skip ssl verification.

:param retries: Number of retry attempts to make in case of connection / server

errors.

:param max_pool_size: Option to set the maximum number of connections to save the pool.

:param http_timeout: A float in seconds for connect and read timeout. Use a Tuple to

specific connect and read separately.

:param keepalive_timeout: Maximum amount of idle time in seconds before force connection

closing.

:param mode: Launch mode, all Launches started by the client will be in that mode.

:param launch_uuid_print: Print Launch UUID into passed TextIO or by default to stdout.

:param print_output: Set output stream for Launch UUID printing.

:param truncate_attributes: Truncate test item attributes to default maximum length.

:param client: ReportPortal async Client instance to use. If set, all above arguments

will be ignored.

:param launch_uuid: A launch UUID to use instead of starting own one.

:param log_batch_size: Option to set the maximum number of logs that can be processed in one

batch.

:param log_batch_payload_limit: maximum size in bytes of logs that can be processed in one batch

:param log_batcher: ReportPortal log batcher instance to use. If set, 'log_batch'

arguments above will be ignored.

"""

self.__endpoint = endpoint

self.__project = project

self.__step_reporter = StepReporter(self)

self._item_stack = LifoQueue()

self.log_batch_size = log_batch_size

self.log_batch_payload_limit = log_batch_payload_limit

self._log_batcher = log_batcher or LogBatcher(log_batch_size, log_batch_payload_limit)

if client:

self.__client = client

else:

self.__client = Client(endpoint, project, **kwargs)

self.__launch_uuid = launch_uuid

if launch_uuid:

self.use_own_launch = False

else:

self.use_own_launch = True

self._api_info_task = None

self._api_info_cache = None

self._use_microseconds = None

set_current(self)

def __cache_api_info(self, api_info: Optional[dict]) -> Optional[dict]:

if not isinstance(api_info, dict):

return None

self._api_info_cache = api_info

version = extract_server_version(api_info)

self._use_microseconds = bool(version and compare_semantic_versions(version, MICROSECONDS_MIN_VERSION) >= 0)

return api_info

async def __prefetch_api_info(self) -> Optional[dict]:

try:

api_info = await self.__client.get_api_info()

return self.__cache_api_info(api_info)

except Exception as exc:

logger.warning("Unable to prefetch API info in background: %s", exc)

return None

def __init_api_info_prefetch(self) -> None:

try:

loop = asyncio.get_running_loop()

self._api_info_task = loop.create_task(self.__prefetch_api_info())

except RuntimeError:

# Construction may happen without an active loop.

self._api_info_task = None

async def start_launch(

self,

name: str,

start_time: Union[str, datetime],

description: Optional[str] = None,

attributes: Optional[Union[list, dict]] = None,

rerun: bool = False,

rerun_of: Optional[str] = None,

**kwargs,

) -> Optional[str]:

"""Start a new Launch with the given arguments.

:param name: Launch name.

:param start_time: Launch start time.

:param description: Launch description.

:param attributes: Launch attributes.

:param rerun: Start launch in rerun mode.

:param rerun_of: For rerun mode specifies which launch will be re-run. Should be used with the

'rerun' option.

:return: Launch UUID if successfully started or None.

"""

if not self.use_own_launch:

return self.launch_uuid

launch_uuid = await self.__client.start_launch(

name,

await self._convert_time(start_time),

description=description,

attributes=attributes,

rerun=rerun,

rerun_of=rerun_of,

**kwargs,

)

self.__launch_uuid = launch_uuid

return self.launch_uuid

async def start_test_item(

self,

name: str,

start_time: Union[str, datetime],

item_type: str,

description: Optional[str] = None,

attributes: Optional[list[dict]] = None,

parameters: Optional[dict] = None,

parent_item_id: Optional[str] = None,

has_stats: bool = True,

code_ref: Optional[str] = None,

retry: bool = False,

test_case_id: Optional[str] = None,

retry_of: Optional[str] = None,

uuid: Optional[str] = None,

**kwargs: Any,

) -> Optional[str]:

"""Start Test Case/Suite/Step/Nested Step Item.

:param name: Name of the Test Item.

:param start_time: The Item start time.

:param item_type: Type of the Test Item. Allowed values:

"suite", "story", "test", "scenario", "step", "before_class", "before_groups",

"before_method", "before_suite", "before_test", "after_class", "after_groups",

"after_method", "after_suite", "after_test".

:param description: The Item description.

:param attributes: Test Item attributes.

:param parameters: Set of parameters (for parametrized Test Items).

:param parent_item_id: A UUID of a parent SUITE / STEP.

:param has_stats: Set to False if test item is a Nested Step.

:param code_ref: Physical location of the Test Item.

:param retry: Used to report retry of the test. Allowed values: "True" or "False".

:param test_case_id: A unique ID of the current Step.

:param retry_of: For retry mode specifies which test item will be marked as retried. Should be used

with the 'retry' parameter.

:param uuid: Test Item UUID to use on start (overrides server one, should be globally unique).

:return: Test Item UUID if successfully started or None.

"""