Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use localstack method in localstack

Best Python code snippet using localstack_python

__init__.py

Source:__init__.py

1"""2Wait for Localstack to be ready.3Localstack publishes which services it has been configured to execute and4their status (e.g. "running").  This module provides a class that will5check that the status of all services are "running".6Notes7-----8The service health checks for Localstack are described in detail in the9Localstack documentation see10https://github.com/localstack/localstack#service-health-checks11Examples12--------13To run with defaults simply have:14>>> from wait4localstack import Wait4Localstack15>>> wait4localstack = Wait4Localstack()16>>> wait4localstack.wait_for_all_services()17"""18import argparse19import logging20import sys21import time22import wait4localstack.localstack23class Wait4Localstack:24    """A class for waiting for Localstack to be ready."""25    def __init__(self, localstack_endpoint='http://localhost:4566/health',26                 maximum_retries=0, test_interval=2, exponential_backoff=False, log_level='WARN'):27        """28        Create a Wait4Localstack object.29        Parameters30        ----------31        localstack_endpoint : str,optional32            The url of the localstack endpoint to be checked (e.g. http://localhost:4566/health).  Default is33            http://localhost:4566/health34        maximum_retries : int,optional35            The max number of retries for attempting to check the endpoint.  If set to zero, will try for infinity.36            Default is zero.37        test_interval : int,optional38            The time in seconds between attempts.  Default is two.39        exponential_backoff : bool,optional40            If the number of seconds between attempts is to be doubled.  Default is False.41        log_level : str,optional42            What the log level should be for the utility.  Can be DEBUG, INFO or WARN.  Default is WARN.43        """44        self._exponential_backoff = None45        self._localstack_endpoint = None46        self._logger = None47        self._maximum_retries = None48        self._test_interval = None49        logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')50        logger = logging.getLogger('wait4localstack')51        logger.setLevel(log_level)52        self.logger(logger)53        self.exponential_backoff(exponential_backoff)54        self.localstack_endpoint(localstack_endpoint)55        self.maximum_retries(maximum_retries)56        self.test_interval(test_interval)57    def exponential_backoff(self, exponential_backoff=None):58        """59        Get or set exponential backoff within the class.60        Parameters61        ----------62        exponential_backoff : bool,optional63            If provided, set if exponential backoff is True or False.64        Returns65        -------66        bool67            If exponential backoff is true or false.68        """69        logger = self.logger()70        if exponential_backoff is not None:71            logger.debug(f'Setting exponential backoff to {exponential_backoff}')72            self._exponential_backoff = exponential_backoff73        return self._exponential_backoff74    def localstack_endpoint(self, localstack_endpoint=None):75        """76        Get or set the localstack endpoint.77        Parameters78        ----------79        localstack_endpoint : str,optional80            The URL of the localstack endpoint (e.g. http://localstack:4566/health).81        Returns82        -------83        str84            The URL of the localstack endpoint.85        """86        logger = self.logger()87        if localstack_endpoint is not None:88            logger.debug(f'Setting localstack endpoint to {localstack_endpoint}')89            self._localstack_endpoint = localstack_endpoint90        return self._localstack_endpoint91    def logger(self, logger=None):92        """93        Get or set the logger.94        Parameters95        ----------96        logger : logging.Logger97            The logger to use for logging.98        Returns99        -------100        logging.Logger101            The logger to use for logging.102        """103        if logger is not None:104            self._logger = logger105        return self._logger106    def maximum_retries(self, maximum_retries=None):107        """108        Get or set the maximum number of retries.109        Parameters110        ----------111        maximum_retries : int,optional112            The maximum number of retries.  If set to zero, then will try for infinity.113        Returns114        -------115        int116            The maximum number of retries.117        """118        logger = self.logger()119        if maximum_retries is not None:120            logger.debug(f'Setting maximum number of retries to {maximum_retries}')121            self._maximum_retries = maximum_retries122        return self._maximum_retries123    def test_interval(self, test_interval=None):124        """125        Get or set the interval between tests.126        If exponential backoff is enabled then this number will be doubled each time127        it is called.128        Parameters129        ----------130        test_interval : int,optional131            Set the interval between tests or if exponential backoff is enabled, set the initial wait period.132        Returns133        -------134        int135            The interval to the next test.136        """137        logger = self.logger()138        if test_interval is not None:139            logger.debug(f'Setting the test interval to {test_interval}')140            self._test_interval = test_interval141        response = self._test_interval142        if self.exponential_backoff():143            new_time_interval = self._test_interval * 2144            logger.debug(f'Setting new test level of {new_time_interval}')145            self._test_interval = new_time_interval146        else:147            logger.debug(f'Keeping test interval at {self._test_interval}')148        return response149    def wait_for_all_services(self):150        """Check the health endpoint until it is successful or max attempts has been reached."""151        logger = self.logger()152        connected = False153        attempts = 0154        max_retries = self.maximum_retries()155        while not connected:156            localstack = wait4localstack.localstack.LocalStack(self.localstack_endpoint(), logger)157            connected = localstack.is_live()158            if max_retries and attempts >= max_retries:159                logger.error(f'Localstack is not ready after {max_retries} attempts.')160                sys.exit(1)161            sleep_time = self.test_interval()162            if not connected:163                logger.debug(f'Will retry in {sleep_time} seconds.')164                attempts += 1165                time.sleep(sleep_time)166def command_line_interface(args):167    """168    Process arguments provided by the command line.169    Parameters170    ----------171    args : list of str172        The arguments to be processed.173    Returns174    -------175    args176        The command line arguments provided.177    """178    parser = argparse.ArgumentParser(args, description='Wait for Localstack to be ready.')179    parser.prog = 'wait4localstack'180    group = parser.add_mutually_exclusive_group()181    group.add_argument('-d', '--debug', help='Should log level be DEBUG.', action='store_true')182    parser.add_argument('-e', '--endpoint',183                        help='The endpoint for the Localstack healthcheck (default http://localhost:4566/health).',184                        dest='localstack_endpoint', default='http://localhost:4566/health')185    help_str = 'The maximum retries.  If zero, try infinitely (default is zero).'186    parser.add_argument('-m', '--maximum-retries', help=help_str, type=int)187    help_str = 'The time in seconds to wait between retries (default is 2).'188    parser.add_argument('-t', '--test-interval', help=help_str, default=2, type=int)189    group.add_argument('-v', '--verbose', help='Should log level be INFO.', action='store_true')190    parser.add_argument('-x', '--exponential_backoff',191                        help='Should the time between retries by doubled at each attempt.',192                        dest='exponential_backoff', action='store_true')193    return parser.parse_args()194def main():195    """196    Provide an entry point for Wait4Localstack.197    This is the entrypoint for the executable script that then creates and198    consumes a Wait4Localstack object.199    """200    args = command_line_interface(sys.argv)201    if args.debug:202        log_level = 'DEBUG'203    elif args.verbose:204        log_level = 'INFO'205    else:206        log_level = 'WARN'207    widget = Wait4Localstack(args.localstack_endpoint,208                             args.maximum_retries,209                             args.test_interval,210                             args.exponential_backoff,211                             log_level)...

e2e_test.py

Source:e2e_test.py

1import os2import unittest3from runner import Runner4MODULE_NAME = "e2e"5RESOURCE_TYPE = "aws_vpc.this"6class TestE2E(unittest.TestCase):7    @classmethod8    def setUpClass(self):9        self.snippet = """10        provider "aws" {11              access_key                  = "test"12              region                      = "us-east-1"13              s3_force_path_style         = true14              secret_key                  = "test"15              skip_credentials_validation = true16              skip_metadata_api_check     = true17              skip_requesting_account_id  = true18              endpoints {19                apigateway     = "http://localstack:4567"20                cloudformation = "http://localstack:4581"21                cloudwatch     = "http://localstack:4582"22                dynamodb       = "http://localstack:4569"23                ec2            = "http://localstack:4597"24                es             = "http://localstack:4578"25                firehose       = "http://localstack:4573"26                iam            = "http://localstack:4593"27                kinesis        = "http://localstack:4568"28                lambda         = "http://localstack:4574"29                route53        = "http://localstack:4580"30                redshift       = "http://localstack:4577"31                s3             = "http://localstack:4572"32                secretsmanager = "http://localstack:4584"33                ses            = "http://localstack:4579"34                sns            = "http://localstack:4575"35                sqs            = "http://localstack:4576"36                ssm            = "http://localstack:4583"37                stepfunctions  = "http://localstack:4585"38                sts            = "http://localstack:4592"39              }40            }41            module "e2e" {42              source = "./mymodule"43              providers = {44                aws = aws45              }46            }47        """48        self.runner = Runner(self.snippet)49        self.result = self.runner.result50    def test_vpc_cidr(self):51        self.assertEqual(self.runner.get_value(f"module.{MODULE_NAME}.{RESOURCE_TYPE}", "cidr_block"), "10.0.0.0/16")52if __name__ == '__main__':...

setup.py

Source:setup.py

...11  "port": IMPOSTER_PORT,12  "protocol": "http",13  "recordRequests": True14}15def wait_for_localstack():16	logging.info("Waiting for localstack")17	healthcheck_url = "".join([LOCALSTACK_URL, "/health"])18	counter = 019	while counter < 42:20		try:21			response = requests.get(healthcheck_url)22			s3_status = response.json().get("services", {}).get("s3")23			if s3_status == "running":24				return25		except requests.exceptions.ConnectionError:26			pass27		counter += 128		sleep(5)29	raise Exception("Localstack not available")30def create_bucket(bucket_name:str) -> None:31	logging.info(f"Creating S3 bucket {bucket_name}")32	s3_client = boto3.client("s3", endpoint_url=LOCALSTACK_URL)33	s3_client.create_bucket(Bucket=bucket_name)34def create_imposter() -> None:35	logging.info(f"Creating imposter at port {IMPOSTER_PORT}")36	_ = requests.post(f"{MOUNTEBANK_URL}/imposters", json=IMPOSTER_REQUEST)37if __name__ == "__main__":38	logging.info("Starting local/testing setup script")39	wait_for_localstack()40	create_bucket(S3_BUCKET)41	create_imposter()...

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.