KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use is_env_true method in localstack

Best Python code snippet using localstack_python
config.py
Source:config.py
...207def eval_log_type(env_var_name):208    """get the log type from environment variable"""209    ls_log = os.environ.get(env_var_name, "").lower().strip()210    return ls_log if ls_log in LOG_LEVELS else False211def is_env_true(env_var_name):212    """Whether the given environment variable has a truthy value."""213    return os.environ.get(env_var_name, "").lower().strip() in TRUE_STRINGS214def is_env_not_false(env_var_name):215    """Whether the given environment variable is empty or has a truthy value."""216    return os.environ.get(env_var_name, "").lower().strip() not in FALSE_STRINGS217def load_environment(profile: str = None):218    """Loads the environment variables from ~/.localstack/{profile}.env219    :param profile: the profile to load (defaults to "default")220    """221    if not profile:222        profile = "default"223    path = os.path.join(CONFIG_DIR, f"{profile}.env")224    if not os.path.exists(path):225        return226    import dotenv227    dotenv.load_dotenv(path, override=False)228def is_linux():229    return platform.system() == "Linux"230def ping(host):231    """Returns True if host responds to a ping request"""232    is_windows = platform.system().lower() == "windows"233    ping_opts = "-n 1" if is_windows else "-c 1"234    args = "ping %s %s" % (ping_opts, host)235    return (236        subprocess.call(args, shell=not is_windows, stdout=subprocess.PIPE, stderr=subprocess.PIPE)237        == 0238    )239def in_docker():240    """241    Returns True if running in a docker container, else False242    Ref. https://docs.docker.com/config/containers/runmetrics/#control-groups243    """244    if OVERRIDE_IN_DOCKER:245        return True246    # details: https://github.com/localstack/localstack/pull/4352247    if os.path.exists("/.dockerenv"):248        return True249    if os.path.exists("/run/.containerenv"):250        return True251    if not os.path.exists("/proc/1/cgroup"):252        return False253    try:254        if any(255            [256                os.path.exists("/sys/fs/cgroup/memory/docker/"),257                any(258                    "docker-" in file_names259                    for file_names in os.listdir("/sys/fs/cgroup/memory/system.slice")260                ),261                os.path.exists("/sys/fs/cgroup/docker/"),262                any(263                    "docker-" in file_names264                    for file_names in os.listdir("/sys/fs/cgroup/system.slice/")265                ),266            ]267        ):268            return False269    except Exception:270        pass271    with open("/proc/1/cgroup", "rt") as ifh:272        content = ifh.read()273        if "docker" in content:274            return True275        os_hostname = socket.gethostname()276        if os_hostname and os_hostname in content:277            return True278    return False279# whether the in_docker check should always return true280OVERRIDE_IN_DOCKER = is_env_true("OVERRIDE_IN_DOCKER")281is_in_docker = in_docker()282is_in_linux = is_linux()283# CLI specific: the configuration profile to load284CONFIG_PROFILE = os.environ.get("CONFIG_PROFILE", "").strip()285# CLI specific: host configuration directory286CONFIG_DIR = os.environ.get("CONFIG_DIR", os.path.expanduser("~/.localstack"))287# keep this on top to populate environment288try:289    load_environment(CONFIG_PROFILE)290except ImportError:291    # dotenv may not be available in lambdas or other environments where config is loaded292    pass293# default AWS region294if "DEFAULT_REGION" not in os.environ:295    os.environ["DEFAULT_REGION"] = os.environ.get("AWS_DEFAULT_REGION") or AWS_REGION_US_EAST_1296DEFAULT_REGION = os.environ["DEFAULT_REGION"]297# expose services on a specific host externally298HOSTNAME_EXTERNAL = os.environ.get("HOSTNAME_EXTERNAL", "").strip() or LOCALHOST299# name of the host under which the LocalStack services are available300LOCALSTACK_HOSTNAME = os.environ.get("LOCALSTACK_HOSTNAME", "").strip() or LOCALHOST301# directory for persisting data (TODO: deprecated, simply use PERSISTENCE=1)302DATA_DIR = os.environ.get("DATA_DIR", "").strip()303# whether localstack should persist service state across localstack runs304PERSISTENCE = is_env_true("PERSISTENCE")305# whether to clear config.dirs.tmp on startup and shutdown306CLEAR_TMP_FOLDER = is_env_not_false("CLEAR_TMP_FOLDER")307# folder for temporary files and data308TMP_FOLDER = os.path.join(tempfile.gettempdir(), "localstack")309# this is exclusively for the CLI to configure the container mount into /var/lib/localstack310VOLUME_DIR = os.environ.get("LOCALSTACK_VOLUME_DIR", "").strip() or TMP_FOLDER311# fix for Mac OS, to be able to mount /var/folders in Docker312if TMP_FOLDER.startswith("/var/folders/") and os.path.exists("/private%s" % TMP_FOLDER):313    TMP_FOLDER = "/private%s" % TMP_FOLDER314# temporary folder of the host (required when running in Docker). Fall back to local tmp folder if not set315HOST_TMP_FOLDER = os.environ.get("HOST_TMP_FOLDER", TMP_FOLDER)316# whether to use the old directory structure and mounting config317LEGACY_DIRECTORIES = is_env_true("LEGACY_DIRECTORIES")318# whether to enable verbose debug logging319LS_LOG = eval_log_type("LS_LOG")320DEBUG = is_env_true("DEBUG") or LS_LOG in TRACE_LOG_LEVELS321# whether to enable debugpy322DEVELOP = is_env_true("DEVELOP")323# PORT FOR DEBUGGER324DEVELOP_PORT = int(os.environ.get("DEVELOP_PORT", "").strip() or DEFAULT_DEVELOP_PORT)325# whether to make debugpy wait for a debbuger client326WAIT_FOR_DEBUGGER = is_env_true("WAIT_FOR_DEBUGGER")327# whether to use SSL encryption for the services328# TODO: this is deprecated and should be removed (edge port supports HTTP/HTTPS multiplexing)329USE_SSL = is_env_true("USE_SSL")330# whether to use the legacy edge proxy or the newer Gateway/HandlerChain framework331LEGACY_EDGE_PROXY = is_env_true("LEGACY_EDGE_PROXY")332# Whether to report internal failures as 500 or 501 errors.333FAIL_FAST = is_env_true("FAIL_FAST")334# whether to use the legacy single-region mode, defined via DEFAULT_REGION335USE_SINGLE_REGION = is_env_true("USE_SINGLE_REGION")336# whether to run in TF compatibility mode for TF integration tests337# (e.g., returning verbatim ports for ELB resources, rather than edge port 4566, etc.)338TF_COMPAT_MODE = is_env_true("TF_COMPAT_MODE")339# default encoding used to convert strings to byte arrays (mainly for Python 3 compatibility)340DEFAULT_ENCODING = "utf-8"341# path to local Docker UNIX domain socket342DOCKER_SOCK = os.environ.get("DOCKER_SOCK", "").strip() or "/var/run/docker.sock"343# additional flags to pass to "docker run" when starting the stack in Docker344DOCKER_FLAGS = os.environ.get("DOCKER_FLAGS", "").strip()345# command used to run Docker containers (e.g., set to "sudo docker" to run as sudo)346DOCKER_CMD = os.environ.get("DOCKER_CMD", "").strip() or "docker"347# use the command line docker client instead of the new sdk version, might get removed in the future348LEGACY_DOCKER_CLIENT = is_env_true("LEGACY_DOCKER_CLIENT")349# whether to forward edge requests in-memory (instead of via proxy servers listening on backend ports)350# TODO: this will likely become the default and may get removed in the future351FORWARD_EDGE_INMEM = True352# Default bind address for the edge service353EDGE_BIND_HOST = os.environ.get("EDGE_BIND_HOST", "").strip() or "127.0.0.1"354# port number for the edge service, the main entry point for all API invocations355EDGE_PORT = int(os.environ.get("EDGE_PORT") or 0) or DEFAULT_PORT_EDGE356# fallback port for non-SSL HTTP edge service (in case HTTPS edge service cannot be used)357EDGE_PORT_HTTP = int(os.environ.get("EDGE_PORT_HTTP") or 0)358# optional target URL to forward all edge requests to359EDGE_FORWARD_URL = os.environ.get("EDGE_FORWARD_URL", "").strip()360# IP of the docker bridge used to enable access between containers361DOCKER_BRIDGE_IP = os.environ.get("DOCKER_BRIDGE_IP", "").strip()362# whether to enable API-based updates of configuration variables at runtime363ENABLE_CONFIG_UPDATES = is_env_true("ENABLE_CONFIG_UPDATES")364# CORS settings365DISABLE_CORS_HEADERS = is_env_true("DISABLE_CORS_HEADERS")366DISABLE_CORS_CHECKS = is_env_true("DISABLE_CORS_CHECKS")367DISABLE_CUSTOM_CORS_S3 = is_env_true("DISABLE_CUSTOM_CORS_S3")368DISABLE_CUSTOM_CORS_APIGATEWAY = is_env_true("DISABLE_CUSTOM_CORS_APIGATEWAY")369EXTRA_CORS_ALLOWED_HEADERS = os.environ.get("EXTRA_CORS_ALLOWED_HEADERS", "").strip()370EXTRA_CORS_EXPOSE_HEADERS = os.environ.get("EXTRA_CORS_EXPOSE_HEADERS", "").strip()371EXTRA_CORS_ALLOWED_ORIGINS = os.environ.get("EXTRA_CORS_ALLOWED_ORIGINS", "").strip()372# whether to disable publishing events to the API373DISABLE_EVENTS = is_env_true("DISABLE_EVENTS")374DEBUG_ANALYTICS = is_env_true("DEBUG_ANALYTICS")375# whether to eagerly start services376EAGER_SERVICE_LOADING = is_env_true("EAGER_SERVICE_LOADING")377# whether to enable multi-accounts378MULTI_ACCOUNTS = is_env_true("MULTI_ACCOUNTS")379# Whether to skip downloading additional infrastructure components (e.g., custom Elasticsearch versions)380SKIP_INFRA_DOWNLOADS = os.environ.get("SKIP_INFRA_DOWNLOADS", "").strip()381# Whether to skip downloading our signed SSL cert.382SKIP_SSL_CERT_DOWNLOAD = is_env_true("SKIP_SSL_CERT_DOWNLOAD")383# name of the main Docker container384MAIN_CONTAINER_NAME = os.environ.get("MAIN_CONTAINER_NAME", "").strip() or "localstack_main"385# the latest commit id of the repository when the docker image was created386LOCALSTACK_BUILD_GIT_HASH = os.environ.get("LOCALSTACK_BUILD_GIT_HASH", "").strip() or None387# the date on which the docker image was created388LOCALSTACK_BUILD_DATE = os.environ.get("LOCALSTACK_BUILD_DATE", "").strip() or None389# Equivalent to HTTP_PROXY, but only applicable for external connections390OUTBOUND_HTTP_PROXY = os.environ.get("OUTBOUND_HTTP_PROXY", "")391# Equivalent to HTTPS_PROXY, but only applicable for external connections392OUTBOUND_HTTPS_PROXY = os.environ.get("OUTBOUND_HTTPS_PROXY", "")393# Whether to enable the partition adjustment listener (in order to support other partitions that the default)394ARN_PARTITION_REWRITING = is_env_true("ARN_PARTITION_REWRITING")395# whether to skip waiting for the infrastructure to shut down, or exit immediately396FORCE_SHUTDOWN = is_env_not_false("FORCE_SHUTDOWN")397# whether to return mocked success responses for still unimplemented API methods398MOCK_UNIMPLEMENTED = is_env_true("MOCK_UNIMPLEMENTED")399# set variables no_proxy, i.e., run internal service calls directly400no_proxy = ",".join([LOCALSTACK_HOSTNAME, LOCALHOST, LOCALHOST_IP, "[::1]"])401if os.environ.get("no_proxy"):402    os.environ["no_proxy"] += "," + no_proxy403elif os.environ.get("NO_PROXY"):404    os.environ["NO_PROXY"] += "," + no_proxy405else:406    os.environ["no_proxy"] = no_proxy407# additional CLI commands, can be set by plugins408CLI_COMMANDS = {}409# determine IP of Docker bridge410if not DOCKER_BRIDGE_IP:411    DOCKER_BRIDGE_IP = "172.17.0.1"412    if is_in_docker:413        candidates = (DOCKER_BRIDGE_IP, "172.18.0.1")414        for ip in candidates:415            if ping(ip):416                DOCKER_BRIDGE_IP = ip417                break418# determine route to Docker host from container419try:420    DOCKER_HOST_FROM_CONTAINER = DOCKER_BRIDGE_IP421    if not is_in_docker and not is_in_linux:422        # If we're running outside docker, and would like the Lambda containers to be able423        # to access services running on the local machine, set DOCKER_HOST_FROM_CONTAINER accordingly424        if LOCALSTACK_HOSTNAME == LOCALHOST:425            DOCKER_HOST_FROM_CONTAINER = "host.docker.internal"426    # update LOCALSTACK_HOSTNAME if host.docker.internal is available427    if is_in_docker:428        DOCKER_HOST_FROM_CONTAINER = socket.gethostbyname("host.docker.internal")429        if LOCALSTACK_HOSTNAME == DOCKER_BRIDGE_IP:430            LOCALSTACK_HOSTNAME = DOCKER_HOST_FROM_CONTAINER431except socket.error:432    pass433# -----434# SERVICE-SPECIFIC CONFIGS BELOW435# -----436# port ranges for external service instances (f.e. elasticsearch clusters, opensearch clusters,...)437EXTERNAL_SERVICE_PORTS_START = int(438    os.environ.get("EXTERNAL_SERVICE_PORTS_START")439    or os.environ.get("SERVICE_INSTANCES_PORTS_START")440    or 4510441)442EXTERNAL_SERVICE_PORTS_END = int(443    os.environ.get("EXTERNAL_SERVICE_PORTS_END")444    or os.environ.get("SERVICE_INSTANCES_PORTS_END")445    or (EXTERNAL_SERVICE_PORTS_START + 50)446)447# java options to Lambda448LAMBDA_JAVA_OPTS = os.environ.get("LAMBDA_JAVA_OPTS", "").strip()449# limit in which to kinesalite will start throwing exceptions450KINESIS_SHARD_LIMIT = os.environ.get("KINESIS_SHARD_LIMIT", "").strip() or "100"451# delay in kinesalite response when making changes to streams452KINESIS_LATENCY = os.environ.get("KINESIS_LATENCY", "").strip() or "500"453# Delay between data persistence (in seconds)454KINESIS_MOCK_PERSIST_INTERVAL = os.environ.get("KINESIS_MOCK_PERSIST_INTERVAL", "").strip() or "5s"455# Kinesis provider - either "kinesis-mock" or "kinesalite"456KINESIS_PROVIDER = os.environ.get("KINESIS_PROVIDER") or "kinesis-mock"457# Whether or not to handle lambda event sources as synchronous invocations458SYNCHRONOUS_SNS_EVENTS = is_env_true("SYNCHRONOUS_SNS_EVENTS")459SYNCHRONOUS_SQS_EVENTS = is_env_true("SYNCHRONOUS_SQS_EVENTS")460SYNCHRONOUS_API_GATEWAY_EVENTS = is_env_not_false("SYNCHRONOUS_API_GATEWAY_EVENTS")461SYNCHRONOUS_KINESIS_EVENTS = is_env_not_false("SYNCHRONOUS_KINESIS_EVENTS")462SYNCHRONOUS_DYNAMODB_EVENTS = is_env_not_false("SYNCHRONOUS_DYNAMODB_EVENTS")463# randomly inject faults to Kinesis464KINESIS_ERROR_PROBABILITY = float(os.environ.get("KINESIS_ERROR_PROBABILITY", "").strip() or 0.0)465# randomly inject faults to DynamoDB466DYNAMODB_ERROR_PROBABILITY = float(os.environ.get("DYNAMODB_ERROR_PROBABILITY", "").strip() or 0.0)467DYNAMODB_READ_ERROR_PROBABILITY = float(468    os.environ.get("DYNAMODB_READ_ERROR_PROBABILITY", "").strip() or 0.0469)470DYNAMODB_WRITE_ERROR_PROBABILITY = float(471    os.environ.get("DYNAMODB_WRITE_ERROR_PROBABILITY", "").strip() or 0.0472)473# JAVA EE heap size for dynamodb474DYNAMODB_HEAP_SIZE = os.environ.get("DYNAMODB_HEAP_SIZE", "").strip() or "256m"475# single DB instance across multiple credentials are regions476DYNAMODB_SHARE_DB = int(os.environ.get("DYNAMODB_SHARE_DB") or 0)477# Used to toggle QueueDeletedRecently errors when re-creating a queue within 60 seconds of deleting it478SQS_DELAY_RECENTLY_DELETED = is_env_true("SQS_DELAY_RECENTLY_DELETED")479# expose SQS on a specific port externally480SQS_PORT_EXTERNAL = int(os.environ.get("SQS_PORT_EXTERNAL") or 0)481# Strategy used when creating SQS queue urls. can be "off", "domain", or "path"482SQS_ENDPOINT_STRATEGY = os.environ.get("SQS_ENDPOINT_STRATEGY", "") or "off"483# host under which the LocalStack services are available from Lambda Docker containers484HOSTNAME_FROM_LAMBDA = os.environ.get("HOSTNAME_FROM_LAMBDA", "").strip()485# whether to remotely copy the Lambda code or locally mount a volume486LAMBDA_REMOTE_DOCKER = is_env_true("LAMBDA_REMOTE_DOCKER")487# make sure we default to LAMBDA_REMOTE_DOCKER=true if running in Docker488if is_in_docker and not os.environ.get("LAMBDA_REMOTE_DOCKER", "").strip():489    LAMBDA_REMOTE_DOCKER = True490# Marker name to indicate that a bucket represents the local file system. This is used for testing491# Serverless applications where we mount the Lambda code directly into the container from the host OS.492BUCKET_MARKER_LOCAL = (493    os.environ.get("BUCKET_MARKER_LOCAL", "").strip() or DEFAULT_BUCKET_MARKER_LOCAL494)495# network that the docker lambda container will be joining496LAMBDA_DOCKER_NETWORK = os.environ.get("LAMBDA_DOCKER_NETWORK", "").strip()497# custom DNS server that the docker lambda container will use498LAMBDA_DOCKER_DNS = os.environ.get("LAMBDA_DOCKER_DNS", "").strip()499# additional flags passed to Lambda Docker run/create commands500LAMBDA_DOCKER_FLAGS = os.environ.get("LAMBDA_DOCKER_FLAGS", "").strip()501# prebuild images before execution? Increased cold start time on the tradeoff of increased time until lambda is ACTIVE502LAMBDA_PREBUILD_IMAGES = is_env_true("LAMBDA_PREBUILD_IMAGES")503# default container registry for lambda execution images504LAMBDA_CONTAINER_REGISTRY = (505    os.environ.get("LAMBDA_CONTAINER_REGISTRY", "").strip() or DEFAULT_LAMBDA_CONTAINER_REGISTRY506)507# whether to remove containers after Lambdas finished executing508LAMBDA_REMOVE_CONTAINERS = (509    os.environ.get("LAMBDA_REMOVE_CONTAINERS", "").lower().strip() not in FALSE_STRINGS510)511# Adding Stepfunctions default port512LOCAL_PORT_STEPFUNCTIONS = int(os.environ.get("LOCAL_PORT_STEPFUNCTIONS") or 8083)513# Stepfunctions lambda endpoint override514STEPFUNCTIONS_LAMBDA_ENDPOINT = os.environ.get("STEPFUNCTIONS_LAMBDA_ENDPOINT", "").strip()515# path prefix for windows volume mounting516WINDOWS_DOCKER_MOUNT_PREFIX = os.environ.get("WINDOWS_DOCKER_MOUNT_PREFIX", "/host_mnt")517# whether to skip S3 presign URL signature validation (TODO: currently enabled, until all issues are resolved)518S3_SKIP_SIGNATURE_VALIDATION = is_env_not_false("S3_SKIP_SIGNATURE_VALIDATION")519# user-defined lambda executor mode520LAMBDA_EXECUTOR = os.environ.get("LAMBDA_EXECUTOR", "").strip()521# Fallback URL to use when a non-existing Lambda is invoked. If this matches522# `dynamodb://<table_name>`, then the invocation is recorded in the corresponding523# DynamoDB table. If this matches `http(s)://...`, then the Lambda invocation is524# forwarded as a POST request to that URL.525LAMBDA_FALLBACK_URL = os.environ.get("LAMBDA_FALLBACK_URL", "").strip()526# Forward URL used to forward any Lambda invocations to an external527# endpoint (can use useful for advanced test setups)528LAMBDA_FORWARD_URL = os.environ.get("LAMBDA_FORWARD_URL", "").strip()529# Time in seconds to wait at max while extracting Lambda code.530# By default, it is 25 seconds for limiting the execution time531# to avoid client/network timeout issues532LAMBDA_CODE_EXTRACT_TIME = int(os.environ.get("LAMBDA_CODE_EXTRACT_TIME") or 25)533# whether lambdas should use stay open mode if executed in "docker-reuse" executor534LAMBDA_STAY_OPEN_MODE = is_in_docker and is_env_not_false("LAMBDA_STAY_OPEN_MODE")535# truncate output string slices value536LAMBDA_TRUNCATE_STDOUT = int(os.getenv("LAMBDA_TRUNCATE_STDOUT") or 2000)537# A comma-delimited string of stream names and its corresponding shard count to538# initialize during startup.539# For example: "my-first-stream:1,my-other-stream:2,my-last-stream:1"540KINESIS_INITIALIZE_STREAMS = os.environ.get("KINESIS_INITIALIZE_STREAMS", "").strip()541# KMS provider - can be either "local-kms" or "moto"542KMS_PROVIDER = (os.environ.get("KMS_PROVIDER") or "").strip() or "moto"543# URL to a custom OpenSearch/Elasticsearch backend cluster. If this is set to a valid URL, then localstack will not544# create OpenSearch/Elasticsearch cluster instances, but instead forward all domains to the given backend.545OPENSEARCH_CUSTOM_BACKEND = (546    os.environ.get("OPENSEARCH_CUSTOM_BACKEND", "").strip()547    or os.environ.get("ES_CUSTOM_BACKEND", "").strip()548)549# Strategy used when creating OpenSearch/Elasticsearch domain endpoints routed through the edge proxy550# valid values: domain | path | port (off)551OPENSEARCH_ENDPOINT_STRATEGY = (552    os.environ.get("OPENSEARCH_ENDPOINT_STRATEGY", "").strip()553    or os.environ.get("ES_ENDPOINT_STRATEGY", "").strip()554    or "domain"555)556if OPENSEARCH_ENDPOINT_STRATEGY == "off":557    OPENSEARCH_ENDPOINT_STRATEGY = "port"558# Whether to start one cluster per domain (default), or multiplex opensearch domains to a single clusters559OPENSEARCH_MULTI_CLUSTER = is_env_not_false("OPENSEARCH_MULTI_CLUSTER") or is_env_true(560    "ES_MULTI_CLUSTER"561)562# list of environment variable names used for configuration.563# Make sure to keep this in sync with the above!564# Note: do *not* include DATA_DIR in this list, as it is treated separately565CONFIG_ENV_VARS = [566    "BUCKET_MARKER_LOCAL",567    "DEBUG",568    "DEFAULT_REGION",569    "DEVELOP",570    "DEVELOP_PORT",571    "DISABLE_CORS_CHECKS",572    "DISABLE_CORS_HEADERS",573    "DISABLE_CUSTOM_CORS_APIGATEWAY",574    "DISABLE_CUSTOM_CORS_S3",575    "DISABLE_EVENTS",576    "DOCKER_BRIDGE_IP",577    "DYNAMODB_ERROR_PROBABILITY",578    "DYNAMODB_HEAP_SIZE",579    "DYNAMODB_SHARE_DB",580    "DYNAMODB_READ_ERROR_PROBABILITY",581    "DYNAMODB_WRITE_ERROR_PROBABILITY",582    "EAGER_SERVICE_LOADING",583    "EDGE_BIND_HOST",584    "EDGE_FORWARD_URL",585    "EDGE_PORT",586    "EDGE_PORT_HTTP",587    "ENABLE_CONFIG_UPDATES",588    "ES_CUSTOM_BACKEND",589    "ES_ENDPOINT_STRATEGY",590    "ES_MULTI_CLUSTER",591    "EXTRA_CORS_ALLOWED_HEADERS",592    "EXTRA_CORS_ALLOWED_ORIGINS",593    "EXTRA_CORS_EXPOSE_HEADERS",594    "HOSTNAME",595    "HOSTNAME_EXTERNAL",596    "HOSTNAME_FROM_LAMBDA",597    "KINESIS_ERROR_PROBABILITY",598    "KINESIS_INITIALIZE_STREAMS",599    "KINESIS_MOCK_PERSIST_INTERVAL",600    "LAMBDA_CODE_EXTRACT_TIME",601    "LAMBDA_CONTAINER_REGISTRY",602    "LAMBDA_DOCKER_DNS",603    "LAMBDA_DOCKER_FLAGS",604    "LAMBDA_DOCKER_NETWORK",605    "LAMBDA_EXECUTOR",606    "LAMBDA_FALLBACK_URL",607    "LAMBDA_FORWARD_URL",608    "LAMBDA_JAVA_OPTS",609    "LAMBDA_REMOTE_DOCKER",610    "LAMBDA_REMOVE_CONTAINERS",611    "LAMBDA_STAY_OPEN_MODE",612    "LAMBDA_TRUNCATE_STDOUT",613    "LEGACY_DIRECTORIES",614    "LEGACY_DOCKER_CLIENT",615    "LEGACY_EDGE_PROXY",616    "LOCALSTACK_API_KEY",617    "LOCALSTACK_HOSTNAME",618    "LOG_LICENSE_ISSUES",619    "LS_LOG",620    "MAIN_CONTAINER_NAME",621    "MULTI_ACCOUNTS",622    "OPENSEARCH_ENDPOINT_STRATEGY",623    "OUTBOUND_HTTP_PROXY",624    "OUTBOUND_HTTPS_PROXY",625    "PERSISTENCE",626    "REQUESTS_CA_BUNDLE",627    "S3_SKIP_SIGNATURE_VALIDATION",628    "SERVICES",629    "SKIP_INFRA_DOWNLOADS",630    "SKIP_SSL_CERT_DOWNLOAD",631    "SQS_DELAY_RECENTLY_DELETED",632    "SQS_ENDPOINT_STRATEGY",633    "SQS_PORT_EXTERNAL",634    "STEPFUNCTIONS_LAMBDA_ENDPOINT",635    "SYNCHRONOUS_API_GATEWAY_EVENTS",636    "SYNCHRONOUS_DYNAMODB_EVENTS",637    "SYNCHRONOUS_KINESIS_EVENTS",638    "SYNCHRONOUS_SNS_EVENTS",639    "SYNCHRONOUS_SQS_EVENTS",640    "TEST_AWS_ACCOUNT_ID",641    "TEST_IAM_USER_ID",642    "TEST_IAM_USER_NAME",643    "TF_COMPAT_MODE",644    "THUNDRA_APIKEY",645    "THUNDRA_AGENT_JAVA_VERSION",646    "THUNDRA_AGENT_NODE_VERSION",647    "THUNDRA_AGENT_PYTHON_VERSION",648    "USE_SINGLE_REGION",649    "USE_SSL",650    "WAIT_FOR_DEBUGGER",651    "WINDOWS_DOCKER_MOUNT_PREFIX",652]653def is_local_test_mode() -> bool:654    """Returns True if we are running in the context of our local integration tests."""655    return is_env_true(ENV_INTERNAL_TEST_RUN)656def is_collect_metrics_mode() -> bool:657    """Returns True if metric collection is enabled."""658    return is_env_true(ENV_INTERNAL_TEST_COLLECT_METRIC)659def collect_config_items() -> List[Tuple[str, Any]]:660    """Returns a list of key-value tuples of LocalStack configuration values."""661    none = object()  # sentinel object662    # collect which keys to print663    keys = []664    keys.extend(CONFIG_ENV_VARS)665    keys.append("DATA_DIR")666    keys.sort()667    values = globals()668    result = []669    for k in keys:670        v = values.get(k, none)671        if v is none:672            continue673        result.append((k, v))674    result.sort()675    return result676def is_trace_logging_enabled():677    if LS_LOG:678        log_level = str(LS_LOG).upper()679        return log_level.lower() in TRACE_LOG_LEVELS680    return False681# set log levels immediately, but will be overwritten later by setup_logging682if DEBUG:683    logging.getLogger("").setLevel(logging.DEBUG)684    logging.getLogger("localstack").setLevel(logging.DEBUG)685LOG = logging.getLogger(__name__)686if is_trace_logging_enabled():687    load_end_time = time.time()688    LOG.debug(689        "Initializing the configuration took %s ms", int((load_end_time - load_start_time) * 1000)690    )691def parse_service_ports() -> Dict[str, int]:692    """Parses the environment variable $SERVICES with a comma-separated list of services693    and (optional) ports they should run on: 'service1:port1,service2,service3:port3'"""694    service_ports = os.environ.get("SERVICES", "").strip()695    if service_ports and not is_env_true("EAGER_SERVICE_LOADING"):696        LOG.warning("SERVICES variable is ignored if EAGER_SERVICE_LOADING=0.")697        service_ports = None  # TODO remove logic once we clear up the service ports stuff698    if not service_ports:699        return DEFAULT_SERVICE_PORTS700    result = {}701    for service_port in re.split(r"\s*,\s*", service_ports):702        parts = re.split(r"[:=]", service_port)703        service = parts[0]704        key_upper = service.upper().replace("-", "_")705        port_env_name = "%s_PORT" % key_upper706        # (1) set default port number707        port_number = DEFAULT_SERVICE_PORTS.get(service)708        # (2) set port number from <SERVICE>_PORT environment, if present709        if os.environ.get(port_env_name):...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.