Best Python code snippet using localstack_python
version_manager.py
Source:version_manager.py
...263 invocation=invocation,264 )265 self.queued_invocations.put(invocation_storage)266 return invocation_storage.result_future267 def set_environment_ready(self, executor_id: str) -> None:268 environment = self.all_environments.get(executor_id)269 if not environment:270 raise Exception(271 "Inconsistent state detected: Non existing environment '%s' reported error.",272 executor_id,273 )274 environment.set_ready()275 self.available_environments.put(environment)276 def set_environment_failed(self, executor_id: str) -> None:277 environment = self.all_environments.get(executor_id)278 if not environment:279 raise Exception(280 "Inconsistent state detected: Non existing environment '%s' reported error.",281 executor_id,282 )283 environment.errored()284 def store_logs(self, invocation_result: InvocationResult, executor: RuntimeEnvironment) -> None:285 if invocation_result.logs:286 log_item = LogItem(287 executor.get_log_group_name(),288 executor.get_log_stream_name(),289 invocation_result.logs,290 )291 self.log_handler.add_logs(log_item)292 else:293 LOG.warning(294 "Received no logs from invocation with id %s for lambda %s",295 invocation_result.invocation_id,296 self.function_arn,297 )298 def invocation_response(299 self, invoke_id: str, invocation_result: Union[InvocationResult, InvocationError]300 ) -> None:301 running_invocation = self.running_invocations.pop(invoke_id, None)302 if running_invocation is None:303 raise Exception(f"Cannot map invocation result {invoke_id} to invocation")304 if not invocation_result.logs:305 invocation_result.logs = running_invocation.logs306 executor = running_invocation.executor307 running_invocation.invocation.result_future.set_result(invocation_result)308 # mark executor available again309 executor.invocation_done()310 self.available_environments.put(executor)311 self.store_logs(invocation_result=invocation_result, executor=executor)312 # Service Endpoint implementation313 def invocation_result(self, invoke_id: str, invocation_result: InvocationResult) -> None:314 LOG.debug("Got invocation result for invocation '%s'", invoke_id)315 self.invocation_response(invoke_id=invoke_id, invocation_result=invocation_result)316 def invocation_error(self, invoke_id: str, invocation_error: InvocationError) -> None:317 LOG.debug("Got invocation error for invocation '%s'", invoke_id)318 self.invocation_response(invoke_id=invoke_id, invocation_result=invocation_error)319 def invocation_logs(self, invoke_id: str, invocation_logs: InvocationLogs) -> None:320 LOG.debug("Got logs for invocation '%s'", invoke_id)321 for log_line in invocation_logs.logs.splitlines():322 LOG.debug("> %s", log_line)323 running_invocation = self.running_invocations.get(invoke_id, None)324 if running_invocation is None:325 raise Exception(f"Cannot map invocation result {invoke_id} to invocation")326 running_invocation.logs = invocation_logs.logs327 def status_ready(self, executor_id: str) -> None:328 self.set_environment_ready(executor_id=executor_id)329 def status_error(self, executor_id: str) -> None:...
TestEnvironment.py
Source:TestEnvironment.py
1from Environment import Environment2env = Environment()3env.add_connections({0:[1,2,3], 1:[4,5,6,7,8]})4env.set_environment_ready()5a = env.reset()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!