KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use create_external_boto_client method in localstack
Best Python code snippet using localstack_python
test_lambda_whitebox.py
Source:test_lambda_whitebox.py 
...33)34class TestLambdaFallbackUrl(unittest.TestCase):35 @staticmethod36 def _run_forward_to_fallback_url(url, fallback=True, lambda_name=None, num_requests=3):37 lambda_client = aws_stack.create_external_boto_client("lambda")38 if fallback:39 config.LAMBDA_FALLBACK_URL = url40 else:41 config.LAMBDA_FORWARD_URL = url42 try:43 result = []44 for i in range(num_requests):45 lambda_name = lambda_name or "non-existing-lambda-%s" % i46 ctx = {"env": "test"}47 tmp = lambda_client.invoke(48 FunctionName=lambda_name,49 Payload=b'{"foo":"bar"}',50 InvocationType="RequestResponse",51 ClientContext=to_str(base64.b64encode(to_bytes(json.dumps(ctx)))),52 )53 result.append(tmp)54 return result55 finally:56 if fallback:57 config.LAMBDA_FALLBACK_URL = ""58 else:59 config.LAMBDA_FORWARD_URL = ""60 def test_forward_to_fallback_url_dynamodb(self):61 db_table = "lambda-records"62 ddb_client = aws_stack.create_external_boto_client("dynamodb")63 def num_items():64 return len((run_safe(ddb_client.scan, TableName=db_table) or {"Items": []})["Items"])65 items_before = num_items()66 self._run_forward_to_fallback_url("dynamodb://%s" % db_table)67 items_after = num_items()68 self.assertEqual(items_before + 3, items_after)69 def test_forward_to_fallback_url_http(self):70 class MyUpdateListener(ProxyListener):71 def forward_request(self, method, path, data, headers):72 records.append({"data": data, "headers": headers, "method": method, "path": path})73 return lambda_result74 lambda_result = {"result": "test123"}75 local_port = get_free_tcp_port()76 proxy = start_proxy(local_port, backend_url=None, update_listener=MyUpdateListener())77 local_url = "%s://localhost:%s" % (get_service_protocol(), local_port)78 # test 1: forward to LAMBDA_FALLBACK_URL79 records = []80 self._run_forward_to_fallback_url(local_url)81 items_after = len(records)82 for record in records:83 self.assertIn("non-existing-lambda", record["headers"]["lambda-function-name"])84 self.assertEqual(3, items_after)85 # create test Lambda86 lambda_name = "test-%s" % short_uid()87 testutil.create_lambda_function(88 handler_file=TEST_LAMBDA_PYTHON,89 func_name=lambda_name,90 libs=TEST_LAMBDA_LIBS,91 )92 # test 2: forward to LAMBDA_FORWARD_URL93 records = []94 inv_results = self._run_forward_to_fallback_url(95 local_url, lambda_name=lambda_name, fallback=False96 )97 items_after = len(records)98 for record in records:99 headers = record["headers"]100 self.assertIn("/lambda/", headers["Authorization"])101 self.assertEqual("POST", record["method"])102 self.assertIn("/functions/%s/invocations" % lambda_name, record["path"])103 self.assertTrue(headers.get("X-Amz-Client-Context"))104 self.assertEqual("RequestResponse", headers.get("X-Amz-Invocation-Type"))105 self.assertEqual({"foo": "bar"}, json.loads(to_str(record["data"])))106 self.assertEqual(3, items_after)107 # assert result payload matches108 response_payload = inv_results[0]["Payload"].read()109 self.assertEqual(lambda_result, json.loads(response_payload))110 # clean up / shutdown111 lambda_client = aws_stack.create_external_boto_client("lambda")112 lambda_client.delete_function(FunctionName=lambda_name)113 proxy.stop()114 def test_adding_fallback_function_name_in_headers(self):115 lambda_client = aws_stack.create_external_boto_client("lambda")116 ddb_client = aws_stack.create_external_boto_client("dynamodb")117 db_table = "lambda-records"118 config.LAMBDA_FALLBACK_URL = "dynamodb://%s" % db_table119 lambda_client.invoke(120 FunctionName="non-existing-lambda",121 Payload=b"{}",122 InvocationType="RequestResponse",123 )124 result = run_safe(ddb_client.scan, TableName=db_table)125 self.assertEqual("non-existing-lambda", result["Items"][0]["function_name"]["S"])126class TestDockerExecutors(unittest.TestCase):127 @classmethod128 def setUpClass(cls):129 cls.lambda_client = aws_stack.create_external_boto_client("lambda")130 cls.s3_client = aws_stack.create_external_boto_client("s3")131 @pytest.mark.skipif(not use_docker(), reason="Only applicable with docker executor")132 def test_additional_docker_flags(self):133 flags_before = config.LAMBDA_DOCKER_FLAGS134 env_value = short_uid()135 config.LAMBDA_DOCKER_FLAGS = f"-e Hello={env_value}"136 function_name = "flags-{}".format(short_uid())137 try:138 testutil.create_lambda_function(139 handler_file=TEST_LAMBDA_ENV,140 libs=TEST_LAMBDA_LIBS,141 func_name=function_name,142 )143 lambda_client = aws_stack.create_external_boto_client("lambda")144 result = lambda_client.invoke(FunctionName=function_name, Payload="{}")145 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])146 result_data = result["Payload"].read()147 result_data = json.loads(to_str(result_data))148 self.assertEqual({"Hello": env_value}, result_data)149 finally:150 config.LAMBDA_DOCKER_FLAGS = flags_before151 # clean up152 lambda_client.delete_function(FunctionName=function_name)153 def test_code_updated_on_redeployment(self):154 lambda_api.LAMBDA_EXECUTOR.cleanup()155 func_name = "test_code_updated_on_redeployment"156 # deploy function for the first time157 testutil.create_lambda_function(158 func_name=func_name,159 handler_file=TEST_LAMBDA_ENV,160 libs=TEST_LAMBDA_LIBS,161 envvars={"Hello": "World"},162 )163 # test first invocation164 result = self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")165 payload = json.loads(to_str(result["Payload"].read()))166 assert payload["Hello"] == "World"167 # replacement code168 updated_handler = "handler = lambda event, context: {'Hello': 'Elon Musk'}"169 updated_handler = testutil.create_lambda_archive(170 updated_handler, libs=TEST_LAMBDA_LIBS, get_content=True171 )172 self.lambda_client.update_function_code(FunctionName=func_name, ZipFile=updated_handler)173 # second invocation should exec updated lambda code174 result = self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")175 payload = json.loads(to_str(result["Payload"].read()))176 assert payload["Hello"] == "Elon Musk"177 @pytest.mark.skipif(178 condition=not isinstance(179 lambda_api.LAMBDA_EXECUTOR, lambda_executors.LambdaExecutorReuseContainers180 ),181 reason="Test only applicable if docker-reuse executor is selected",182 )183 def test_prime_and_destroy_containers(self):184 executor = lambda_api.LAMBDA_EXECUTOR185 func_name = "test_prime_and_destroy_containers"186 func_arn = lambda_api.func_arn(func_name)187 # make sure existing containers are gone188 executor.cleanup()189 self.assertEqual(0, len(executor.get_all_container_names()))190 # deploy and invoke lambda without Docker191 testutil.create_lambda_function(192 func_name=func_name,193 handler_file=TEST_LAMBDA_ENV,194 libs=TEST_LAMBDA_LIBS,195 envvars={"Hello": "World"},196 )197 self.assertEqual(0, len(executor.get_all_container_names()))198 self.assertDictEqual({}, executor.function_invoke_times)199 # invoke a few times.200 durations = []201 num_iterations = 3202 for i in range(0, num_iterations + 1):203 prev_invoke_time = None204 if i > 0:205 prev_invoke_time = executor.function_invoke_times[func_arn]206 start_time = time.time()207 self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")208 duration = time.time() - start_time209 self.assertEqual(1, len(executor.get_all_container_names()))210 # ensure the last invoke time is being updated properly.211 if i > 0:212 self.assertGreater(executor.function_invoke_times[func_arn], prev_invoke_time)213 else:214 self.assertGreater(executor.function_invoke_times[func_arn], 0)215 durations.append(duration)216 # the first call would have created the container. subsequent calls would reuse and be faster.217 for i in range(1, num_iterations + 1):218 self.assertLess(durations[i], durations[0])219 status = executor.get_docker_container_status(func_arn)220 self.assertEqual(1, status)221 container_network = executor.get_docker_container_network(func_arn)222 self.assertEqual("bridge", container_network)223 executor.cleanup()224 status = executor.get_docker_container_status(func_arn)225 self.assertEqual(0, status)226 self.assertEqual(0, len(executor.get_all_container_names()))227 # clean up228 testutil.delete_lambda_function(func_name)229 @pytest.mark.skipif(230 condition=not isinstance(231 lambda_api.LAMBDA_EXECUTOR, lambda_executors.LambdaExecutorReuseContainers232 ),233 reason="Test only applicable if docker-reuse executor is selected",234 )235 def test_destroy_idle_containers(self):236 executor = lambda_api.LAMBDA_EXECUTOR237 func_name = "test_destroy_idle_containers"238 func_arn = lambda_api.func_arn(func_name)239 # make sure existing containers are gone240 executor.destroy_existing_docker_containers()241 self.assertEqual(0, len(executor.get_all_container_names()))242 # deploy and invoke lambda without Docker243 testutil.create_lambda_function(244 func_name=func_name,245 handler_file=TEST_LAMBDA_ENV,246 libs=TEST_LAMBDA_LIBS,247 envvars={"Hello": "World"},248 )249 self.assertEqual(0, len(executor.get_all_container_names()))250 self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")251 self.assertEqual(1, len(executor.get_all_container_names()))252 # try to destroy idle containers.253 executor.idle_container_destroyer()254 self.assertEqual(1, len(executor.get_all_container_names()))255 # simulate an idle container256 executor.function_invoke_times[func_arn] = (257 int(time.time() * 1000) - lambda_executors.MAX_CONTAINER_IDLE_TIME_MS258 )259 executor.idle_container_destroyer()260 def assert_container_destroyed():261 self.assertEqual(0, len(executor.get_all_container_names()))262 retry(assert_container_destroyed, retries=3)263 # clean up264 testutil.delete_lambda_function(func_name)265class TestLocalExecutors(unittest.TestCase):266 def test_python3_runtime_multiple_create_with_conflicting_module(self):267 lambda_client = aws_stack.create_external_boto_client("lambda")268 original_do_use_docker = lambda_api.DO_USE_DOCKER269 try:270 # always use the local runner271 lambda_api.DO_USE_DOCKER = False272 python3_with_settings1 = load_file(TEST_LAMBDA_PYTHON3_MULTIPLE_CREATE1, mode="rb")273 python3_with_settings2 = load_file(TEST_LAMBDA_PYTHON3_MULTIPLE_CREATE2, mode="rb")274 lambda_name1 = "test1-%s" % short_uid()275 testutil.create_lambda_function(276 func_name=lambda_name1,277 zip_file=python3_with_settings1,278 runtime=LAMBDA_RUNTIME_PYTHON36,279 handler="handler1.handler",280 )281 lambda_name2 = "test2-%s" % short_uid()...
test_events.py
Source:test_events.py
...8 sqs_client=None,9 log_group_name=None,10 logs_client=None,11):12 events_client = events_client or aws_stack.create_external_boto_client("events")13 kwargs = {"EventBusName": bus_name} if bus_name else {}14 if target_ids:15 target_ids = target_ids if isinstance(target_ids, list) else [target_ids]16 events_client.remove_targets(Rule=rule_name, Ids=target_ids, Force=True, **kwargs)17 if rule_name:18 events_client.delete_rule(Name=rule_name, Force=True, **kwargs)19 if bus_name:20 events_client.delete_event_bus(Name=bus_name)21 if queue_url:22 sqs_client = sqs_client or aws_stack.create_external_boto_client("sqs")23 sqs_client.delete_queue(QueueUrl=queue_url)24 if log_group_name:25 logs_client = logs_client or aws_stack.create_external_boto_client("logs")26 log_streams = logs_client.describe_log_streams(logGroupName=log_group_name)27 for log_stream in log_streams["logStreams"]:28 logs_client.delete_log_stream(29 logGroupName=log_group_name, logStreamName=log_stream["logStreamName"]30 )...
test_org.py
Source:test_org.py
...8 sqs_client=None,9 log_group_name=None,10 logs_client=None,11):12 events_client = events_client or aws_stack.create_external_boto_client("events")13 kwargs = {"EventBusName": bus_name} if bus_name else {}14 if target_ids:15 target_ids = target_ids if isinstance(target_ids, list) else [target_ids]16 events_client.remove_targets(Rule=rule_name, Ids=target_ids, Force=True, **kwargs)17 if rule_name:18 events_client.delete_rule(Name=rule_name, Force=True, **kwargs)19 if bus_name:20 events_client.delete_event_bus(Name=bus_name)21 if queue_url:22 sqs_client = sqs_client or aws_stack.create_external_boto_client("sqs")23 sqs_client.delete_queue(QueueUrl=queue_url)24 if log_group_name:25 logs_client = logs_client or aws_stack.create_external_boto_client("logs")26 log_streams = logs_client.describe_log_streams(logGroupName=log_group_name)27 for log_stream in log_streams["logStreams"]:28 logs_client.delete_log_stream(29 logGroupName=log_group_name, logStreamName=log_stream["logStreamName"]30 )...
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?
