KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
