Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use set_environment_failed method in localstack

Best Python code snippet using localstack_python

version_manager.py

Source:version_manager.py

...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:...

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.