diff --git a/xoa_driver/internals/hli_v1/testers/_base_tester.py b/xoa_driver/internals/hli_v1/testers/_base_tester.py index 02a99d12..23f99ffa 100644 --- a/xoa_driver/internals/hli_v1/testers/_base_tester.py +++ b/xoa_driver/internals/hli_v1/testers/_base_tester.py @@ -52,11 +52,15 @@ class BaseTester(ABC, Generic[TesterStateStorage]): :type password: str, optional :param port: the port number for connection establishment, default to 22606 :type port: int, optional - :param debug: `True` if debug log output from the tester is needed, and `False` otherwise - :type debug: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: self.__host = host self.__port = port self._conn = TransportationHandler( @@ -67,6 +71,7 @@ def __init__(self, host: str, username: str, password: str = "xena", port: int = self._conn, username, password=password, + timeout_seconds=session_timeout, keepalive=True, ) """ diff --git a/xoa_driver/internals/hli_v1/testers/l23_tester.py b/xoa_driver/internals/hli_v1/testers/l23_tester.py index 991610b7..16c32202 100644 --- a/xoa_driver/internals/hli_v1/testers/l23_tester.py +++ b/xoa_driver/internals/hli_v1/testers/l23_tester.py @@ -67,18 +67,23 @@ class L23Tester(BaseTester["testers_state.GenuineTesterLocalState"]): :type password: str, optional :param port: the port number for connection establishment, default to 22606 :type port: int, optional - :param debug: `True` if debug log output from the tester is needed, and `False` otherwise - :type debug: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self._local_states = testers_state.GenuineTesterLocalState(host, port) diff --git a/xoa_driver/internals/hli_v1/testers/l23ve_tester.py b/xoa_driver/internals/hli_v1/testers/l23ve_tester.py index 35480f1d..f30eca46 100644 --- a/xoa_driver/internals/hli_v1/testers/l23ve_tester.py +++ b/xoa_driver/internals/hli_v1/testers/l23ve_tester.py @@ -37,18 +37,23 @@ class L23VeTester(BaseTester["testers_state.TesterLocalState"]): :type password: str, optional :param port: the port number for connection establishment, default to 22606 :type port: int, optional - :param debug: `True` if debug log output from the tester is needed, and `False` otherwise - :type debug: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self._local_states = testers_state.TesterLocalState(host, port) diff --git a/xoa_driver/internals/hli_v1/testers/l47_tester.py b/xoa_driver/internals/hli_v1/testers/l47_tester.py index f75cad18..b879f386 100644 --- a/xoa_driver/internals/hli_v1/testers/l47_tester.py +++ b/xoa_driver/internals/hli_v1/testers/l47_tester.py @@ -44,18 +44,23 @@ class L47Tester(BaseTester["testers_state.GenuineTesterLocalState"]): :type password: str, optional :param port: the port number for connection establishment, default to 22606 :type port: int, optional - :param debug: `True` if debug log output from the tester is needed, and `False` otherwise - :type debug: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self._local_states = testers_state.GenuineTesterLocalState(host, port) diff --git a/xoa_driver/internals/hli_v1/testers/l47ve_tester.py b/xoa_driver/internals/hli_v1/testers/l47ve_tester.py index a86b632b..a077eef0 100644 --- a/xoa_driver/internals/hli_v1/testers/l47ve_tester.py +++ b/xoa_driver/internals/hli_v1/testers/l47ve_tester.py @@ -21,18 +21,23 @@ class L47VeTester(L47Tester): :type password: str, optional :param port: the port number for connection establishment, default to 22606 :type port: int, optional - :param debug: `True` if debug log output from the tester is needed, and `False` otherwise - :type debug: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self.version_no_minor = C_VERSIONNO_MINOR(self._conn) """ diff --git a/xoa_driver/internals/hli_v2/testers/_base_tester.py b/xoa_driver/internals/hli_v2/testers/_base_tester.py index f7d36570..05ef2711 100644 --- a/xoa_driver/internals/hli_v2/testers/_base_tester.py +++ b/xoa_driver/internals/hli_v2/testers/_base_tester.py @@ -41,7 +41,26 @@ # min version = rel v83.2 eq fw v446.5 class BaseTester(ABC, Generic[TesterStateStorage]): - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + """ + Basic Tester class of APIv2. + + :param host: tester's address/hostname + :type host: str + :param username: username of the user + :type username: str + :param password: login password of the tester, defaults to "xena" + :type password: str, optional + :param port: the port number for connection establishment, default to 22606 + :type port: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int + """ + + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: self.__host = host self.__port = port self._conn = TransportationHandler( @@ -52,6 +71,7 @@ def __init__(self, host: str, username: str, password: str = "xena", port: int = self._conn, username, password=password, + timeout_seconds=session_timeout, keepalive=True, ) """ diff --git a/xoa_driver/internals/hli_v2/testers/l23_tester.py b/xoa_driver/internals/hli_v2/testers/l23_tester.py index 082b81cb..552d83e1 100644 --- a/xoa_driver/internals/hli_v2/testers/l23_tester.py +++ b/xoa_driver/internals/hli_v2/testers/l23_tester.py @@ -55,17 +55,35 @@ def get_module_type(revision: str) -> Type: class L23Tester(BaseTester["testers_state.GenuineTesterLocalState"]): """ - Representation of a physical Xena Valkyrie Tester. + This is a conceptual class of Xena Valkyrie Tester. + It is essentially an extended :class:`BaseTester`. + + + :param host: tester's address/hostname + :type host: str + :param username: username of the user + :type username: str + :param password: login password of the tester, defaults to "xena" + :type password: str, optional + :param port: the port number for connection establishment, default to 22606 + :type port: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self._local_states = testers_state.GenuineTesterLocalState(host, port) diff --git a/xoa_driver/internals/hli_v2/testers/l23ve_tester.py b/xoa_driver/internals/hli_v2/testers/l23ve_tester.py index 3e7af826..01afcb3f 100644 --- a/xoa_driver/internals/hli_v2/testers/l23ve_tester.py +++ b/xoa_driver/internals/hli_v2/testers/l23ve_tester.py @@ -25,17 +25,35 @@ def get_module_type(revision: str) -> Type: class L23VeTester(BaseTester["testers_state.TesterLocalState"]): """ - Representation of a virtual Xena Valkyrie Tester. + This is a conceptual class of Xena ValkyrieVE Tester. + It is essentially an extended :class:`BaseTester`. + + + :param host: tester's address/hostname + :type host: str + :param username: username of the user + :type username: str + :param password: login password of the tester, defaults to "xena" + :type password: str, optional + :param port: the port number for connection establishment, default to 22606 + :type port: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self._local_states = testers_state.TesterLocalState(host, port) diff --git a/xoa_driver/internals/hli_v2/testers/l47_tester.py b/xoa_driver/internals/hli_v2/testers/l47_tester.py index 63868131..ce3a204c 100644 --- a/xoa_driver/internals/hli_v2/testers/l47_tester.py +++ b/xoa_driver/internals/hli_v2/testers/l47_tester.py @@ -32,17 +32,35 @@ def get_module_type(revision: str) -> "Type": class L47Tester(BaseTester["testers_state.GenuineTesterLocalState"]): """ - Representation of a physical Xena Vulcan Tester. + This is a conceptual class of Xena Vulcan Tester. + It is essentially an extended :class:`BaseTester`. + + + :param host: tester's address/hostname + :type host: str + :param username: username of the user + :type username: str + :param password: login password of the tester, defaults to "xena" + :type password: str, optional + :param port: the port number for connection establishment, default to 22606 + :type port: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self._local_states = testers_state.GenuineTesterLocalState(host, port) diff --git a/xoa_driver/internals/hli_v2/testers/l47ve_tester.py b/xoa_driver/internals/hli_v2/testers/l47ve_tester.py index ef8f1872..a2325be1 100644 --- a/xoa_driver/internals/hli_v2/testers/l47ve_tester.py +++ b/xoa_driver/internals/hli_v2/testers/l47ve_tester.py @@ -9,17 +9,35 @@ class L47VeTester(L47Tester): """ - Representation of a virtual Xena Vulcan Tester. + This is a conceptual class of Xena VulcanVE Tester. + It is essentially an extended :class:`BaseTester`. + + + :param host: tester's address/hostname + :type host: str + :param username: username of the user + :type username: str + :param password: login password of the tester, defaults to "xena" + :type password: str, optional + :param port: the port number for connection establishment, default to 22606 + :type port: int, optional + :param enable_logging: `True` if debug log output from the tester is needed, and `False` otherwise + :type enable_logging: bool + :param custom_logger: `None` allow to specify logger with custom configurations + :type enable_logging: CustomLogger | None + :param session_timeout: `130` Session timeout in seconds + :type session_timeout: int """ - def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None) -> None: + def __init__(self, host: str, username: str, password: str = "xena", port: int = 22606, *, enable_logging: bool = False, custom_logger: CustomLogger | None = None, session_timeout: int = 130) -> None: super().__init__( host=host, username=username, password=password, port=port, enable_logging=enable_logging, - custom_logger=custom_logger + custom_logger=custom_logger, + session_timeout=session_timeout ) self.version_no_minor = C_VERSIONNO_MINOR(self._conn) """ diff --git a/xoa_driver/internals/utils/session.py b/xoa_driver/internals/utils/session.py index 044587da..2e4ebc3f 100644 --- a/xoa_driver/internals/utils/session.py +++ b/xoa_driver/internals/utils/session.py @@ -1,4 +1,5 @@ import asyncio +from contextlib import suppress from typing import ( TYPE_CHECKING, Tuple @@ -61,6 +62,13 @@ async def logon(self) -> "TesterSession": ) return self + async def chang_timeout(self, seconds: int = 130) -> None: + """Modify session timeout, """ + if not self.is_online: + raise RuntimeError("Timeout can be changet only after connection is established.") + self.timeout = seconds + await C_TIMEOUT(self._conn).set(self.timeout) + def __handle_exceptions(self, fut: asyncio.Future) -> None: if fut.cancelled(): return None @@ -69,7 +77,8 @@ def __handle_exceptions(self, fut: asyncio.Future) -> None: async def __do_keepalive(self) -> None: while self._conn.is_connected: - await C_KEEPALIVE(self._conn).get() + with suppress(Exception): + await C_KEEPALIVE(self._conn).get() await asyncio.sleep(max(0, self.timeout - 5)) @property