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

Copied to Clipboard!

How to use localstack_config method in localstack

Best Python code snippet using localstack_python

localstack_dynamodb_s3_sqs_sns.py

Source:localstack_dynamodb_s3_sqs_sns.py

...19    if 'LOCALSTACK_HOST' in os.environ:20        return "http://" + str(os.environ.get('LOCALSTACK_HOST')) + ":4566"21    else:22        return "http://localhost:4566"23def localstack_config():24    return LocalstackConfig(endpoint_url=localstack_url(),25                            use_ssl=False,26                            aws_access_key_id='test',27                            aws_secret_access_key='test',28                            region_name='us-east-1')29def localstack_config_for_service(service, config):30    return boto3.client(service,31                        endpoint_url=config.endpoint_url,32                        use_ssl=config.use_ssl,33                        aws_access_key_id=config.aws_access_key_id,34                        aws_secret_access_key=config.aws_secret_access_key,35                        region_name=config.region_name)36# Examples of usage each below service is based on:37# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/index.html38def s3_list_buckets(s3):39    return s3.list_buckets()40def s3_create_bucket(s3, bucket):41    s3.create_bucket(Bucket=bucket)42def s3_upload_file(s3, bucket):43    binary_data = b'Binary data stored in S3'44    s3.put_object(Body=binary_data, Bucket=bucket, Key='simple_file_with_binary_data.txt')45def s3_list_object_in_bucket(s3, bucket):46    return s3.list_objects(Bucket=bucket)['Contents']47def sns_list_topics(sns):48    return sns.list_topics()49def sns_create_topic(sns, topic):50    sns.create_topic(Name=topic)51def sqs_list_queues(sqs):52    return sqs.list_queues()53def sqs_create_queue(sqs, queue):54    sqs.create_queue(QueueName=queue)55def dynamodb_delete_table(dynamodb, table_name):56    dynamodb.delete_table(57        TableName=table_name58    )59def dynamodb_create_table(dynamodb, table_name):60    dynamodb.create_table(61        AttributeDefinitions=[62            {63                'AttributeName': 'FirstName',64                'AttributeType': 'S'65            },66            {67                'AttributeName': 'LastName',68                'AttributeType': 'S'69            },70        ],71        TableName=table_name,72        KeySchema=[73            {74                'AttributeName': 'FirstName',75                'KeyType': 'HASH'76            },77            {78                'AttributeName': 'LastName',79                'KeyType': 'RANGE'80            },81        ],82        ProvisionedThroughput={83            'ReadCapacityUnits': 123,84            'WriteCapacityUnits': 12385        },86        Tags=[87            {88                'Key': 'Owner',89                'Value': 'Sebastian'90            },91        ]92    )93def dynamodb_list_tables(dynamodb):94    return dynamodb.list_tables()95if __name__ == "__main__":96    # Print host, config and time97    print("HOST: " + info_hostname())98    print("LOCALSTACK: " + localstack_url())99    print("CONFIG: " + info_config())100    print("TIME: ", info_time())101    # Create dynamodb table102    dynamodb_create_table(localstack_config_for_service('dynamodb', localstack_config()), 'localstack_dynamodb_python')103    # Print out dynamodb tables names104    print("\nDynamoDB tables:")105    for table in dynamodb_list_tables(localstack_config_for_service('dynamodb', localstack_config()))['TableNames']:106        print("- " + table)107    # Delete dynamodb table108    dynamodb_delete_table(localstack_config_for_service('dynamodb', localstack_config()), 'localstack_dynamodb_python')109    # Create bucket if not exists110    s3_create_bucket(localstack_config_for_service('s3', localstack_config()), 'localstack-s3-python')111    # Upload binary data112    s3_upload_file(localstack_config_for_service('s3', localstack_config()), 'localstack-s3-python')113    # Print out bucket names114    print("\nS3 buckets:")115    for bucket in s3_list_buckets(localstack_config_for_service('s3', localstack_config()))['Buckets']:116        print("- " + bucket['Name'])117    # Print out bucket files118    print("\nFiles in localstack-s3-python:")119    for obj in s3_list_object_in_bucket(localstack_config_for_service('s3', localstack_config()), 'localstack-s3-python'):120        print("- " + obj['Key'])121    # Create SNS topic122    sns_create_topic(localstack_config_for_service('sns', localstack_config()), 'localstack_sns_python')123    # Print out topics names124    print("\nSNS topics:")125    for topic in sns_list_topics(localstack_config_for_service('sns', localstack_config()))['Topics']:126        print("- " + topic['TopicArn'])127    # Create SQS queue128    sqs_create_queue(localstack_config_for_service('sqs', localstack_config()), 'localstack_sqs_python')129    # Print out queues names130    print("\nSQS queues:")131    for queue in sqs_list_queues(localstack_config_for_service('sqs', localstack_config()))['QueueUrls']:...

main_boto3.py

Source:main_boto3.py

...19    if 'LOCALSTACK_HOST' in os.environ:20        return "http://" + str(os.environ.get('LOCALSTACK_HOST')) + ":4566"21    else:22        return "http://localhost:4566"23def localstack_config():24    return LocalstackConfig(endpoint_url=localstack_url(),25                            use_ssl=False,26                            aws_access_key_id='test',27                            aws_secret_access_key='test',28                            region_name='us-east-1')29def localstack_config_for_service(service, config):30    return boto3.client(service,31                        endpoint_url=config.endpoint_url,32                        use_ssl=config.use_ssl,33                        aws_access_key_id=config.aws_access_key_id,34                        aws_secret_access_key=config.aws_secret_access_key,35                        region_name=config.region_name)36# Examples of usage each below service is based on:37# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/index.html38def s3_list_buckets(s3):39    return s3.list_buckets()40def s3_create_bucket(s3, bucket):41    s3.create_bucket(Bucket=bucket)42def s3_upload_file(s3, bucket):43    binary_data = b'Binary data stored in S3'44    s3.put_object(Body=binary_data, Bucket=bucket, Key='simple_file_with_binary_data.txt')45def s3_list_object_in_bucket(s3, bucket):46    return s3.list_objects(Bucket=bucket)['Contents']47def sns_list_topics(sns):48    return sns.list_topics()49def sns_create_topic(sns, topic):50    sns.create_topic(Name=topic)51def sqs_list_queues(sqs):52    return sqs.list_queues()53def sqs_create_queue(sqs, queue):54    sqs.create_queue(QueueName=queue)55def dynamodb_delete_table(dynamodb, table_name):56    dynamodb.delete_table(57        TableName=table_name58    )59def dynamodb_create_table(dynamodb, table_name):60    dynamodb.create_table(61        AttributeDefinitions=[62            {63                'AttributeName': 'FirstName',64                'AttributeType': 'S'65            },66            {67                'AttributeName': 'LastName',68                'AttributeType': 'S'69            },70        ],71        TableName=table_name,72        KeySchema=[73            {74                'AttributeName': 'FirstName',75                'KeyType': 'HASH'76            },77            {78                'AttributeName': 'LastName',79                'KeyType': 'RANGE'80            },81        ],82        ProvisionedThroughput={83            'ReadCapacityUnits': 123,84            'WriteCapacityUnits': 12385        },86        Tags=[87            {88                'Key': 'Owner',89                'Value': 'Sebastian'90            },91        ]92    )93def dynamodb_list_tables(dynamodb):94    return dynamodb.list_tables()95if __name__ == "__main__":96    # Print host, config and time97    print("HOST: " + info_hostname())98    print("LOCALSTACK: " + localstack_url())99    print("CONFIG: " + info_config())100    print("TIME: ", info_time())101    # Create dynamodb table102    dynamodb_create_table(localstack_config_for_service('dynamodb', localstack_config()), 'demo-dynamodb-py')103    # Print out dynamodb tables names104    print("\nDynamoDB tables:")105    for table in dynamodb_list_tables(localstack_config_for_service('dynamodb', localstack_config()))['TableNames']:106        print("- " + table)107    # Delete dynamodb table108    dynamodb_delete_table(localstack_config_for_service('dynamodb', localstack_config()), 'demo-dynamodb-py')109    # Create bucket if not exists110    s3_create_bucket(localstack_config_for_service('s3', localstack_config()), 'demo-bucket-py')111    # Upload binary data112    s3_upload_file(localstack_config_for_service('s3', localstack_config()), 'demo-bucket-py')113    # Print out bucket names114    print("\nS3 buckets:")115    for bucket in s3_list_buckets(localstack_config_for_service('s3', localstack_config()))['Buckets']:116        print("- " + bucket['Name'])117    # Print out bucket files118    print("\nFiles in demo-bucket-py:")119    for obj in s3_list_object_in_bucket(localstack_config_for_service('s3', localstack_config()), 'demo-bucket-py'):120        print("- " + obj['Key'])121    # Create SNS topic122    sns_create_topic(localstack_config_for_service('sns', localstack_config()), 'demo-sns-py')123    # Print out topics names124    print("\nSNS topics:")125    for topic in sns_list_topics(localstack_config_for_service('sns', localstack_config()))['Topics']:126        print("- " + topic['TopicArn'])127    # Create SQS queue128    sqs_create_queue(localstack_config_for_service('sqs', localstack_config()), 'demo-sqs-py')129    # Print out queues names130    print("\nSQS queues:")131    for queue in sqs_list_queues(localstack_config_for_service('sqs', localstack_config()))['QueueUrls']:...

multiserver.py

Source:multiserver.py

1import sys2import json3import logging4import subprocess5import requests6from flask_cors import CORS7from moto import server as moto_server8from requests.models import Response9from localstack import constants, config as localstack_config10from localstack.utils.common import (11    FuncThread, ShellCommandThread, TMP_THREADS, to_str, json_safe,12    wait_for_port_open, is_port_open, get_free_tcp_port)13from localstack.utils.bootstrap import setup_logging14from localstack.services.generic_proxy import ProxyListener, start_proxy_server15LOG = logging.getLogger('localstack.multiserver')16# maps API names to server details17API_SERVERS = {}18# network port for multiserver instance (lazily initialized on startup)19MULTI_SERVER_PORT = None20# network port for moto server instance (lazily initialized on startup)21MOTO_SERVER_PORT = None22# API paths23API_PATH_SERVERS = '/servers'24# whether to start the multiserver in a separate process25RUN_SERVER_IN_PROCESS = False26def patch_moto_server():27    def create_backend_app(service):28        if not service:29            LOG.warning('Unable to create moto backend app for empty service: "%s"' % service)30            return None31        backend_app = create_backend_app_orig(service)32        CORS(backend_app)33        return backend_app34    create_backend_app_orig = moto_server.create_backend_app35    moto_server.create_backend_app = create_backend_app36def start_api_server_locally(request):37    if localstack_config.FORWARD_EDGE_INMEM:38        if '__started__' in API_SERVERS:39            return40        API_SERVERS['__started__'] = True41    api = request.get('api')42    port = request.get('port')43    if api in API_SERVERS:44        return API_SERVERS[api]45    result = API_SERVERS[api] = {}46    def thread_func(params):47        if localstack_config.FORWARD_EDGE_INMEM:48            return moto_server.main(['-p', str(port), '-H', constants.BIND_HOST])49        return moto_server.main([api, '-p', str(port), '-H', constants.BIND_HOST])50    thread = FuncThread(thread_func)51    thread.start()52    TMP_THREADS.append(thread)53    result['port'] = port54    result['thread'] = thread55    return result56def start_server(port, asynchronous=False):57    if is_port_open(port):58        LOG.debug('API Multiserver appears to be already running.')59        return60    class ConfigListener(ProxyListener):61        def forward_request(self, method, path, data, **kwargs):62            response = Response()63            response.status_code = 20064            response._content = '{}'65            try:66                if path == API_PATH_SERVERS:67                    if method == 'POST':68                        start_api_server_locally(json.loads(to_str(data)))69                    elif method == 'GET':70                        response._content = json.dumps(json_safe(API_SERVERS))71            except Exception as e:72                LOG.error('Unable to process request: %s' % e)73                response.status_code = 50074                response._content = str(e)75            return response76    proxy = start_proxy_server(port, update_listener=ConfigListener())77    if asynchronous:78        return proxy79    proxy.join()80def start_api_server(api, port, server_port=None):81    server_port = server_port or get_multi_server_port()82    thread = start_server_process(server_port)83    url = 'http://localhost:%s%s' % (server_port, API_PATH_SERVERS)84    if localstack_config.FORWARD_EDGE_INMEM:85        port = get_moto_server_port()86    payload = {87        'api': api,88        'port': port89    }90    result = requests.post(url, json=payload)91    if result.status_code >= 400:92        raise Exception('Unable to start API in multi server (%s): %s' %93                        (result.status_code, result.content))94    return thread95def start_server_process(port):96    if '__server__' in API_SERVERS:97        return API_SERVERS['__server__']['thread']98    port = port or get_multi_server_port()99    API_SERVERS['__server__'] = config = {'port': port}100    LOG.info('Starting multi API server process on port %s' % port)101    if RUN_SERVER_IN_PROCESS:102        cmd = '"%s" "%s" %s' % (sys.executable, __file__, port)103        env_vars = {104            'PYTHONPATH': '.:%s' % constants.LOCALSTACK_ROOT_FOLDER105        }106        thread = ShellCommandThread(cmd, outfile=subprocess.PIPE, env_vars=env_vars, inherit_cwd=True)107        thread.start()108    else:109        thread = start_server(port, asynchronous=True)110    TMP_THREADS.append(thread)111    config['thread'] = thread112    wait_for_port_open(port, retries=20, sleep_time=1)113    return thread114def get_multi_server_port():115    global MULTI_SERVER_PORT116    MULTI_SERVER_PORT = MULTI_SERVER_PORT or get_free_tcp_port()117    return MULTI_SERVER_PORT118def get_moto_server_port():119    global MOTO_SERVER_PORT120    MOTO_SERVER_PORT = MOTO_SERVER_PORT or get_free_tcp_port()121    return MOTO_SERVER_PORT122def main():123    setup_logging()124    port = int(sys.argv[1]) if len(sys.argv) > 1 else get_multi_server_port()125    start_server(port)126patch_moto_server()127if __name__ == '__main__':...

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.