Skip to content
This repository was archived by the owner on Feb 27, 2025. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions xoa_driver/internals/hli_v1/testers/_base_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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,
)
"""
Expand Down
13 changes: 9 additions & 4 deletions xoa_driver/internals/hli_v1/testers/l23_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
13 changes: 9 additions & 4 deletions xoa_driver/internals/hli_v1/testers/l23ve_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
13 changes: 9 additions & 4 deletions xoa_driver/internals/hli_v1/testers/l47_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
13 changes: 9 additions & 4 deletions xoa_driver/internals/hli_v1/testers/l47ve_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
"""
Expand Down
22 changes: 21 additions & 1 deletion xoa_driver/internals/hli_v2/testers/_base_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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,
)
"""
Expand Down
24 changes: 21 additions & 3 deletions xoa_driver/internals/hli_v2/testers/l23_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
24 changes: 21 additions & 3 deletions xoa_driver/internals/hli_v2/testers/l23ve_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
24 changes: 21 additions & 3 deletions xoa_driver/internals/hli_v2/testers/l47_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
24 changes: 21 additions & 3 deletions xoa_driver/internals/hli_v2/testers/l47ve_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
"""
Expand Down
11 changes: 10 additions & 1 deletion xoa_driver/internals/utils/session.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import asyncio
from contextlib import suppress
from typing import (
TYPE_CHECKING,
Tuple
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down