tests.system.amazon.aws.utils

Package Contents

Classes

Variable

Stores metadata about a variable to be fetched for AWS System Tests.

SystemTestContextBuilder

This builder class ultimately constructs a TaskFlow task which is run at

Functions

fetch_variable(key[, default_value, test_name, optional])

Given a Parameter name: first check for an existing Environment Variable,

set_env_id()

Retrieves or generates an Environment ID, validate that it is suitable,

all_tasks_passed(ti)

prune_logs(logs[, force_delete, retry, retry_times, ti])

If all tasks in this dagrun have succeeded, then delete the associated logs.

split_string(string)

Attributes

ENV_ID_ENVIRON_KEY

ENV_ID_KEY

DEFAULT_ENV_ID_PREFIX

DEFAULT_ENV_ID_LEN

DEFAULT_ENV_ID

PURGE_LOGS_INTERVAL_PERIOD

TEST_FILE_IDENTIFIERS

INVALID_ENV_ID_MSG

LOWERCASE_ENV_ID_MSG

NO_VALUE_MSG

log

tests.system.amazon.aws.utils.ENV_ID_ENVIRON_KEY: str = 'SYSTEM_TESTS_ENV_ID'[source]
tests.system.amazon.aws.utils.ENV_ID_KEY: str = 'ENV_ID'[source]
tests.system.amazon.aws.utils.DEFAULT_ENV_ID_PREFIX: str = 'env'[source]
tests.system.amazon.aws.utils.DEFAULT_ENV_ID_LEN: int = 8[source]
tests.system.amazon.aws.utils.DEFAULT_ENV_ID: str[source]
tests.system.amazon.aws.utils.PURGE_LOGS_INTERVAL_PERIOD = 5[source]
tests.system.amazon.aws.utils.TEST_FILE_IDENTIFIERS: list[str] = ['example_', 'test_'][source]
tests.system.amazon.aws.utils.INVALID_ENV_ID_MSG: str = 'To maximize compatibility, the SYSTEM_TESTS_ENV_ID must be an alphanumeric string which starts...'[source]
tests.system.amazon.aws.utils.LOWERCASE_ENV_ID_MSG: str = 'The provided Environment ID contains uppercase letters and will be converted to lowercase for...'[source]
tests.system.amazon.aws.utils.NO_VALUE_MSG: str = 'No Value Found: Variable {key} could not be found, and no default value was provided.'[source]
tests.system.amazon.aws.utils.log[source]
class tests.system.amazon.aws.utils.Variable(name, to_split=False, delimiter=None, test_name=None, optional=False)[source]

Stores metadata about a variable to be fetched for AWS System Tests.

Parameters

  • name (str) – The name of the variable to be fetched.

  • to_split (bool) – If True, the input is a string-formatted List and needs to be split. Defaults to False.

  • delimiter (str | None) – If to_split is true, this will be used to split the string. Defaults to ‘,’.

  • test_name (str | None) – The name of the system test that the variable is associated with.

get_value()[source]
set_default(default)[source]
class tests.system.amazon.aws.utils.SystemTestContextBuilder[source]

This builder class ultimately constructs a TaskFlow task which is run at runtime (task execution time). This task generates and stores the test ENV_ID as well as any external resources requested (e.g.g IAM Roles, VPC, etc)

add_variable(variable_name, split_string=False, delimiter=None, optional=False, **kwargs)[source]

Register a variable to fetch from environment or cloud parameter store

build()[source]

Build and return a TaskFlow task which will create an env_id and fetch requested variables. Storing everything in xcom for downstream tasks to use.

tests.system.amazon.aws.utils.fetch_variable(key, default_value=None, test_name=None, optional=False)[source]

Given a Parameter name: first check for an existing Environment Variable, then check SSM for a value. If neither are available, fall back on the optional default value.

Parameters

  • key (str) – The name of the Parameter to fetch a value for.

  • default_value (str | None) – The default value to use if no value can be found.

  • test_name (str | None) – The system test name.

  • optional (bool) – Whether the variable is optional. If True, does not raise ValueError if variable does not exist

Returns

The value of the parameter.

Return type

str | None

tests.system.amazon.aws.utils.set_env_id()[source]

Retrieves or generates an Environment ID, validate that it is suitable, export it as an Environment Variable, and return it.

If an Environment ID has already been generated, use that. Otherwise, try to fetch it and export it as an Environment Variable. If there is not one available to fetch then generate one and export it as an Environment Variable.

Returns

A valid System Test Environment ID.

Return type

str

tests.system.amazon.aws.utils.all_tasks_passed(ti)[source]
tests.system.amazon.aws.utils.prune_logs(logs, force_delete=False, retry=False, retry_times=3, ti=None)[source]

If all tasks in this dagrun have succeeded, then delete the associated logs. Otherwise, append the logs with a retention policy. This allows the logs to be used for troubleshooting but assures they won’t build up indefinitely.

Parameters

  • logs (list[tuple[str, str | None]]) – A list of log_group/stream_prefix tuples to delete.

  • force_delete (bool) – Whether to check log streams within the log group before removal. If True, removes the log group and all its log streams inside it.

  • retry (bool) – Whether to retry if the log group/stream was not found. In some cases, the log group/stream is created seconds after the main resource has been created. By default, it retries for 3 times with a 5s waiting period.

  • retry_times (int) – Number of retries.

  • ti – Used to check the status of the tasks. This gets pulled from the DAG’s context and does not need to be passed manually.

tests.system.amazon.aws.utils.split_string(string)[source]

Was this entry helpful?