All methods link to APIs
def send_api_request(method,
url,
status_code=None,
nb_retry=5,
wait_time=10,
timeout=60,
**kwargs)
Constructs and sends a Request object.
Arguments:
method
str - Method for the new Request object: 'GET', 'OPTIONS', 'HEAD', 'POST', 'PUT', 'PATCH', or 'DELETE'.url
str - URL for the new Request object.status_code
List[int] - List to get the expected status code.nb_retry
int - Integer to set the number of possible retries.wait_time
int, optional - Integer to set the waiting time between each try.params
dict, optional - Dictionary, list of tuples or bytes to send in the query string for the Request.data
dict, optional - Dictionary, list of tuples, bytes, or file-like object to send in the body of the Request.json
optional - A JSON serializable Python object to send in the body of the Request.headers
dict, optional - Dictionary of HTTP Headers to send with the Request.cookies
optional - Dict or CookieJar object to send with the Request.files
dict, optional - Dictionary of 'name': file-like-objects (or {'name': file-tuple'}) for multipart encoding upload. 'file-tuple' can be a 2-tuple ('filename', fileobj'), 3-tuple ('filename', fileobj', 'content_type')' or a 4-tuple ('filename', fileobj', 'content_type', custom_headers'), where 'content-type' is a string defining the content type of the given file and 'custom_headers' a dict-like object containing additional headers to add for the file.auth
optional - Auth tuple to enable Basic/Digest/Custom HTTP Auth.timeout
float or tuple, optional - How many seconds to wait for the server to send data before giving up, as a float, or a (connect timeout, read timeout) tuple.allow_redirects
bool, optional - Boolean. Enable/disable GET/OPTIONS/POST/PUT/PATCH/DELETE/HEAD redirection. Defaults to True.proxies
optional - Dictionary mapping protocol to the URL of the proxy.verify
optional - Either a boolean, in which case it controls whether we verify the server's TLS certificate, or a string, in which case it must be a path to a CA bundle to use. Defaults to True.stream
optional - If False, the response content will be immediately downloaded.cert
optional - If String, path to SSL client cert file (.pem). If Tuple, ('cert', 'key') pair.
Returns:
Response
- Response object.
All methods link to the local environment
def compare_values(symbol, arg1, arg2) -> bool
Compare two values for equality. Return a Syntax Error if the symbol is not found.
Arguments:
symbol
str - Symbol to use for comparison ('==', '<', '<=', '>=', '>').arg1
Any - First value to compare.arg2
Any - Second value to compare.
Returns:
bool
- True if the comparison is successful, False otherwise.
Raises:
SyntaxError
- If the symbol variable in the argument is not found.
def find_resource(name: str, filepath: str = 'resources/') -> Path
Help to find a resource in a project.
Arguments:
name
str - Name of the resource.filepath
str, optional - Default file path of the resource.
Returns:
Path
- The path of the resource.
Raises:
FileNotFoundError
- If the resource is not found.
def import_recurcively_modules(path_file: str) -> None
Import all modules in a package.
Arguments:
path_file
str - Path of the file/module.
Returns:
None
def get_timezone_paris(timezone: str = 'Europe/Paris') -> datetime
Get the current datetime in the Paris timezone.
Arguments:
timezone
str, optional - Timezone to use.
Returns:
datetime
- Current datetime in the specified timezone.
def response_message(message: str,
response: requests.Response,
password: bool = False,
truncate_content: int = 0,
truncate_body: int = 0) -> str
Show the response with all information to have a better debug.
Arguments:
message
str - A message to explain the request or the error.response
requests.Response - The request response.password
bool, optional - If the header contains secure data, it will not be shown. Defaults to False.truncate_content
int, optional - Truncate content to the defined size, -1 to deactivate, 0 to show all content. Defaults to 0.truncate_body
int, optional - Truncate body to the defined size, -1 to deactivate, 0 to deactivate. Defaults to 0.
Returns:
str
- The formatted response message.
from quality_toolkit.helpers.local_functions import response_message
...
assert response.status_code == 204, response_message('Failed to get information', response, truncate_content=20, is_header_secure=False)
...
Result :
Assertion Failed: Failed to get information from SAP,
- code: 200
- content: {"d":{"results":[{"_
- method: GET
- url: http://localhost?test=1
- header: the header contain a secure data, it can't to be showed
- body: No body
All methods link to the ui
def install_selenium_webdriver(remote_url='http://127.0.0.1:4444/wd/hub',
browser='chrome',
headless=True)
Method to install the Selenium WebDriver.
Arguments:
remote_url
str, optional - Either a string representing the URL of the remote server or a custom RemoteConnection object. Defaults to 'http://127.0.0.1:4444/wd/hub'.browser
str, optional - Define which type of browser to instantiate. Possible values are 'chrome', 'firefox', or 'edge'. Opera is not supported by Selenium: robotframework/SeleniumLibrary#1782. Defaults to 'chrome'.headless
bool, optional - Set the headless option, which is only available for Chrome and Firefox. Defaults to True.
def install_sync_playwright(browser='chromium', **kwargs)
Method to install the Sync Playwright browser.
Arguments:
browser
str, optional - The browser name, possible values are 'chrome', 'firefox', or 'webkit'. Defaults to 'chrome'.executable_path
Union[pathlib.Path, str, None], optional - Path to a browser executable to run instead of the bundled one, ifexecutable_path
is a relative path, then it is resolved relative to the current working directory. Note that Playwright only works with the bundled Chromium, Firefox, or WebKit, use at your own risk. Defaults to None.channel
Union[str, None], optional - Browser distribution channel, supported values are 'chrome', 'chrome-beta', 'chrome-dev', 'chrome-canary', 'msedge', 'msedge-beta', 'msedge-dev', 'msedge-canary'. Read more about using Google Chrome and Microsoft Edge. Defaults to None.args
Union[List[str], None], optional - Additional arguments to pass to the browser instance, the list of Chromium flags can be found here. Defaults to None.ignore_default_args
Union[List[str], bool, None], optional - If True, Playwright does not pass its own configuration args and only uses the ones fromargs
. If a list is given, it filters out the given default arguments. Dangerous option; use with care. Defaults to False.handle_sigint
Union[bool, None], optional - Close the browser process on Ctrl-C. Defaults to True.handle_sigterm
Union[bool, None], optional - Close the browser process on SIGTERM. Defaults to True.handle_sighup
Union[bool, None], optional - Close the browser process on SIGHUP. Defaults to True.timeout
Union[float, None], optional - Maximum time in milliseconds to wait for the browser instance to start. Defaults to 30000 (30 seconds). Pass 0 to disable timeout.env
Union[Dict[str, Union[bool, float, str]], None], optional - Specify environment variables that will be visible to the browser. Defaults toprocess.env
.headless
Union[bool, None], optional - Whether to run the browser in headless mode. Defaults to True unless thedevtools
option is True.devtools
Union[bool, None], optional - Chromium-only Whether to auto-open a Developer Tools panel for each tab. If True, theheadless
option will be set to False. proxy (Union[{'server'
- str, 'bypass': Union[str, None], 'username': Union[str, None], 'password': Union[str, None]}, None ], optional): Network proxy settings.downloads_path
Union[pathlib.Path, str, None], optional - If specified, accepted downloads are downloaded into this directory. Otherwise, a temporary directory is created and is deleted when the browser is closed. In either case, the downloads are deleted when the browser context they were created in is closed.slow_mo
Union[float, None], optional - Slows down Playwright operations by the specified amount of milliseconds. Useful for debugging. Defaults to None.traces_dir
Union[pathlib.Path, str, None], optional - If specified, traces are saved into this directory.chromium_sandbox
Union[bool, None], optional - Enable Chromium sandboxing. Defaults to None.firefox_user_prefs
Union[Dict[str, Union[bool, float, str]], None], optional - Firefox user preferences. Learn more about the Firefox user preferences atabout:config
.
async def install_async_playwright_browser(browser: str = 'chrome', **kwargs)
Method to install the Async Playwright browser.
Arguments:
browser
str, optional - The browser name, possible values are 'chrome', 'firefox', or 'webkit'. Defaults to 'chrome'.executable_path
Union[pathlib.Path, str, None], optional - Path to a browser executable to run instead of the bundled one, ifexecutable_path
is a relative path, then it is resolved relative to the current working directory. Note that Playwright only works with the bundled Chromium, Firefox, or WebKit, use at your own risk. Defaults to None.channel
Union[str, None], optional - Browser distribution channel, supported values are 'chrome', 'chrome-beta', 'chrome-dev', 'chrome-canary', 'msedge', 'msedge-beta', 'msedge-dev', 'msedge-canary'. Read more about using Google Chrome and Microsoft Edge. Defaults to None.args
Union[List[str], None], optional - Additional arguments to pass to the browser instance, the list of Chromium flags can be found here. Defaults to None.ignore_default_args
Union[List[str], bool, None], optional - If True, Playwright does not pass its own configuration args and only uses the ones fromargs
. If a list is given, it filters out the given default arguments. Dangerous option; use with care. Defaults to False.handle_sigint
Union[bool, None], optional - Close the browser process on Ctrl-C. Defaults to True.handle_sigterm
Union[bool, None], optional - Close the browser process on SIGTERM. Defaults to True.handle_sighup
Union[bool, None], optional - Close the browser process on SIGHUP. Defaults to True.timeout
Union[float, None], optional - Maximum time in milliseconds to wait for the browser instance to start. Defaults to 30000 (30 seconds). Pass 0 to disable timeout.env
Union[Dict[str, Union[bool, float, str]], None], optional - Specify environment variables that will be visible to the browser. Defaults toprocess.env
.headless
Union[bool, None], optional - Whether to run the browser in headless mode. Defaults to True unless thedevtools
option is True.devtools
Union[bool, None], optional - Chromium-only Whether to auto-open a Developer Tools panel for each tab. If True, theheadless
option will be set to False. proxy (Union[{'server'
- str, 'bypass': Union[str, None], 'username': Union[str, None], 'password': Union[str, None]}, None ], optional): Network proxy settings.downloads_path
Union[pathlib.Path, str, None], optional - If specified, accepted downloads are downloaded into this directory. Otherwise, a temporary directory is created and is deleted when the browser is closed. In either case, the downloads are deleted when the browser context they were created in is closed.slow_mo
Union[float, None], optional - Slows down Playwright operations by the specified amount of milliseconds. Useful for debugging. Defaults to None.traces_dir
Union[pathlib.Path, str, None], optional - If specified, traces are saved into this directory.chromium_sandbox
Union[bool, None], optional - Enable Chromium sandboxing. Defaults to None.firefox_user_prefs
Union[Dict[str, Union[bool, float, str]], None], optional - Firefox user preferences. Learn more about the Firefox user preferences atabout:config
.
Generic BaseSql Extension
class BaseSql(ABC)
Base class for SQL operations.
def fetch_all(query, params=None, nb_retry=5, wait_time=5)
Get a list of dictionaries.
Arguments:
query
str - The SQL query to execute.params
tuple, optional - The parameters to substitute in the query. Defaults to None.nb_retry
int, optional - The number of retries in case of query failure. Defaults to 5.wait_time
int, optional - The waiting time between retries in seconds. Defaults to 5.
Returns:
list
- A list of dictionaries representing the query result.
def fetch_one(query, params=None, nb_retry=5, wait_time=5)
Get only one dictionary.
Arguments:
query
str - The SQL query to execute.params
tuple, optional - The parameters to substitute in the query. Defaults to None.nb_retry
int, optional - The number of retries in case of query failure. Defaults to 5.wait_time
int, optional - The waiting time between retries in seconds. Defaults to 5.
Returns:
dict
- A dictionary representing the first row of the query result.
def close()
Close the database connection and cursor.
Service MSSQL
class ConnectionMssql(BaseSql)
A class representing a connection to a Microsoft SQL Server.
Arguments:
db_config
dict - A dictionary containing the configuration details for the database connection.
Attributes:
connection
- A pytds.Connection object representing the connection to the database.cursor
- A pytds.Cursor object representing the cursor used to execute SQL queries.
Methods:
execute_script: Executes a SQL script from a file.
Arguments:
script_name
str - The name of the script.script_path
str, optional - The path where the script is located. Defaults to 'resources/scripts/'.params
Union[dict, None], optional - The parameters to be substituted into the script. Defaults to None.
Returns:
-
int
- A return value indicating the status of the execution. (0 indicates success)execute_ps: Executes a stored procedure.
Arguments:
ps_name
str - The name of the stored procedure.params
Union[list, None], optional - The parameters to be passed to the stored procedure. Defaults to None.
Returns:
-
any
- The result of the execution, typically the first row returned by the stored procedure.execute_query: Executes a SQL query.
Arguments:
query
str - The SQL query to be executed.params
Union[tuple, None], optional - The parameters to be passed to the query. Defaults to None.
Returns:
int
- A return value indicating the status of the execution. (0 indicates success)
Notes:
- The
db_config
dictionary should contain the necessary connection details, such as host, port, database name, username, and password. - The
as_dict
andautocommit
parameters are set to True by default. - The
execute_script
,execute_ps
, andexecute_query
methods execute the provided SQL script, stored procedure, and SQL query, respectively. - The
execute_script
method can optionally accept parameters to be substituted in the script using string formatting. - The
execute_ps
andexecute_query
methods can optionally accept parameters to be passed to the stored procedure or query, respectively.
Usage Example: db_config = {
-
"host"
- "localhost", -
"port"
- 1433, -
"database"
- "mydatabase", -
"user"
- "myusername", -
"password"
- "mypassword" }connection = ConnectionMssql(db_config) connection.execute_script("myscript.sql") result = connection.execute_ps("mystoredprocedure", [param1, param2]) connection.execute_query("SELECT * FROM mytable")
def __init__(db_config)
Initializes a new instance of the ConnectionMssql class.
Arguments:
db_config
dict - A dictionary containing the configuration details for the database connection.
def execute_script(script_name, script_path='resources/scripts/', params=None)
Executes a SQL script from a file.
Arguments:
script_name
str - The name of the script.script_path
str, optional - The path where the script is located. Defaults to 'resources/scripts/'.params
Union[dict, None], optional - The parameters to be substituted into the script. Defaults to None.
Returns:
int
- A return value indicating the status of the execution. (0 indicates success)
def execute_ps(ps_name, params=None)
Executes a stored procedure.
Arguments:
ps_name
str - The name of the stored procedure.params
Union[list, None], optional - The parameters to be passed to the stored procedure. Defaults to None.
Returns:
any
- The result of the execution, typically the first row returned by the stored procedure.
def execute_query(query, params=None)
Executes a SQL query.
Arguments:
query
str - The SQL query to be executed.params
Union[tuple, None], optional - The parameters to be passed to the query. Defaults to None.
Returns:
int
- A return value indicating the status of the execution. (0 indicates success)
Service PSQL
class ConnectionPsql(BaseSql)
ConnectionPsql class for establishing a connection and executing queries on a PostgreSQL database.
Arguments:
db_config
dict - A dictionary containing the database configuration parameters.
Methods:
init: Initializes the ConnectionPsql object.
execute_query: Executes a query on the PostgreSQL database.
def __init__(db_config)
Initializes the ConnectionPsql object.
Arguments:
db_config
dict - A dictionary containing the database configuration parameters.
Returns:
None
def execute_query(query, params=None)
Executes a query on the PostgreSQL database.
Arguments:
query
str - The SQL query to be executed.params
tuple - Optional parameters to be used in the query.
Returns:
None
class Sftp()
Sftp class for performing SFTP operations.
Arguments:
host
str - The hostname or IP address of the remote server.username
str - The username to authenticate with on the remote server.password
str - The password to authenticate with on the remote server.
Methods:
init: Initializes the Sftp object.
upload_file: Uploads a local file to the specified remote path on the server.
list_files: Lists files in the specified remote path on the server.
close: Closes the SFTP connection.
def __init__(host, username, password)
Initializes the Sftp object.
Arguments:
host
str - The hostname or IP address of the remote server.username
str - The username to authenticate with on the remote server.password
str - The password to authenticate with on the remote server.
def upload_file(local_file, remote_path)
Uploads a local file to the specified remote path on the server.
Arguments:
local_file
str - The path to the local file to be uploaded.remote_path
str - The path on the server where the file should be uploaded.
def list_files(remote_path, file)
Lists files in the specified remote path on the server.
Arguments:
remote_path
str - The path on the server to list files from.file
str - The name of the file to filter the list by.
Returns:
list_files
list - A list of file names in the remote path that match the provided file name.
def close()
Closes the SFTP connection.
Service SSO
class Sso()
Sso class for handling authentication and authorization using Single Sign-On (SSO).
Arguments:
-
server_url
str - The URL of the SSO server. -
realm_name
str - The name of the realm. -
client_id
str - The client ID for authentication. -
client_secret
str - The client secret for authentication. -
**kwargs
- Additional optional arguments.Optional Args:
- username (str): The username for password grant.
- password (str): The password for password grant.
- audience (str): The audience for token exchange.
Methods:
init: Initializes the Sso object.
__initialize_access_token: Initializes the access token.
__refresh_access_token: Refreshes the JWT token if necessary and returns it.
jwt_token: Refreshes the JWT token if necessary and returns it.
logout: Invalidates the JWT token and logs out the user.
def __init__(server_url, realm_name, client_id, client_secret, **kwargs)
Initializes the Sso object.
Arguments:
server_url
str - The URL of the SSO server.realm_name
str - The name of the realm.client_id
str - The client ID for authentication.client_secret
str - The client secret for authentication.**kwargs
- Additional optional arguments.
Returns:
None
Example:
from quality_toolkit.services.sso import Sso
# Create your own resources for your application
from resources.environment.sso import sso_env
...
context.sso1 = Sso(
sso_env.SERVER_URL,
sso_env.REALM_NAME,
sso_env.CLIENT_ID,
sso_env.CLIENT_SECRET,
audience = sso_env.AUDIENCE)
context.sso2 = Sso(
sso_env.SERVER_URL,
sso_env.REALM_NAME,
sso_env.CLIENT_ID,
sso_env.CLIENT_SECRET,
username = sso_env.USERNAME,
password = sso_env.USERNAME)
...
token1 = context.sso1.jwt_token()
token2 = context.sso2.jwt_token()
...
def jwt_token()
Refreshes the JWT token if necessary and returns it.
Returns:
str
- The JWT token.
def logout()
Invalidates the JWT token and logs out the user.
Returns:
None