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

How to use _execute_in_custom_runtime method in localstack

Best Python code snippet using localstack_python

lambda_executors.py

Source:lambda_executors.py

...949                mount_volumes=mount_volumes,950                stdin=stdin,951            )952class LambdaExecutorLocal(LambdaExecutor):953    def _execute_in_custom_runtime(954        self, cmd: Union[str, List[str]], lambda_function: LambdaFunction = None955    ) -> InvocationResult:956        """957        Generic run function for executing lambdas in custom runtimes.958        :param cmd: the command to execute959        :param lambda_function: function details960        :return: the InvocationResult961        """962        env_vars = lambda_function and lambda_function.envvars963        kwargs = {"stdin": True, "inherit_env": True, "asynchronous": True, "env_vars": env_vars}964        process = run(cmd, stderr=subprocess.PIPE, outfile=subprocess.PIPE, **kwargs)965        result, log_output = process.communicate()966        try:967            result = to_str(result).strip()968        except Exception:969            pass970        log_output = to_str(log_output).strip()971        return_code = process.returncode972        # Note: The user's code may have been logging to stderr, in which case the logs973        # will be part of the "result" variable here. Hence, make sure that we extract974        # only the *last* line of "result" and consider anything above that as log output.975        # TODO: not sure if this code is needed/used976        if isinstance(result, str) and "\n" in result:977            lines = result.split("\n")978            idx = last_index_of(979                lines, lambda line: line and not line.startswith(INTERNAL_LOG_PREFIX)980            )981            if idx >= 0:982                result = lines[idx]983                additional_logs = "\n".join(lines[:idx] + lines[idx + 1 :])984                log_output += "\n%s" % additional_logs985        log_formatted = log_output.strip().replace("\n", "\n> ")986        func_arn = lambda_function and lambda_function.arn()987        LOG.debug(988            "Lambda %s result / log output:\n%s\n> %s" % (func_arn, result.strip(), log_formatted)989        )990        # store log output - TODO get live logs from `process` above?991        # store_lambda_logs(lambda_function, log_output)992        if return_code != 0:993            raise InvocationException(994                "Lambda process returned error status code: %s. Result: %s. Output:\n%s"995                % (return_code, result, log_output),996                log_output,997                result,998            )999        invocation_result = InvocationResult(result, log_output=log_output)1000        return invocation_result1001    def _execute(1002        self, lambda_function: LambdaFunction, inv_context: InvocationContext1003    ) -> InvocationResult:1004        # apply plugin patches to prepare invocation context1005        result = self.apply_plugin_patches(inv_context)1006        if isinstance(result, InvocationResult):1007            return result1008        lambda_cwd = lambda_function.cwd1009        environment = self._prepare_environment(lambda_function)1010        if lambda_function.timeout:1011            environment["AWS_LAMBDA_FUNCTION_TIMEOUT"] = str(lambda_function.timeout)1012        context = inv_context.context1013        if context:1014            environment["AWS_LAMBDA_FUNCTION_NAME"] = context.function_name1015            environment["AWS_LAMBDA_FUNCTION_VERSION"] = context.function_version1016            environment["AWS_LAMBDA_FUNCTION_INVOKED_ARN"] = context.invoked_function_arn1017            environment["AWS_LAMBDA_FUNCTION_MEMORY_SIZE"] = str(context.memory_limit_in_mb)1018        # execute the Lambda function in a forked sub-process, sync result via queue1019        queue = Queue()1020        lambda_function_callable = lambda_function.function(inv_context.function_version)1021        def do_execute():1022            # now we're executing in the child process, safe to change CWD and ENV1023            result = None1024            try:1025                if lambda_cwd:1026                    os.chdir(lambda_cwd)1027                    sys.path.insert(0, "")1028                if environment:1029                    os.environ.update(environment)1030                # set default env variables required for most Lambda handlers1031                self.set_default_env_variables()1032                # run the actual handler function1033                result = lambda_function_callable(inv_context.event, context)1034            except Exception as e:1035                result = str(e)1036                sys.stderr.write("%s %s" % (e, traceback.format_exc()))1037                raise1038            finally:1039                queue.put(result)1040        process = Process(target=do_execute)1041        start_time = now(millis=True)1042        error = None1043        with CaptureOutput() as c:1044            try:1045                process.run()1046            except Exception as e:1047                error = e1048        result = queue.get()1049        end_time = now(millis=True)1050        # Make sure to keep the log line below, to ensure the log stream gets created1051        request_id = long_uid()1052        log_output = 'START %s: Lambda %s started via "local" executor ...' % (1053            request_id,1054            lambda_function.arn(),1055        )1056        # TODO: Interweaving stdout/stderr currently not supported1057        for stream in (c.stdout(), c.stderr()):1058            if stream:1059                log_output += ("\n" if log_output else "") + stream1060        if isinstance(result, InvocationResult) and result.log_output:1061            log_output += "\n" + result.log_output1062        log_output += "\nEND RequestId: %s" % request_id1063        log_output += "\nREPORT RequestId: %s Duration: %s ms" % (1064            request_id,1065            int((end_time - start_time) * 1000),1066        )1067        # store logs to CloudWatch1068        store_lambda_logs(lambda_function, log_output)1069        result = result.result if isinstance(result, InvocationResult) else result1070        if error:1071            LOG.info(1072                'Error executing Lambda "%s": %s %s',1073                lambda_function.arn(),1074                error,1075                "".join(traceback.format_tb(error.__traceback__)),1076            )1077            raise InvocationException(result, log_output)1078        # construct final invocation result1079        invocation_result = InvocationResult(result, log_output=log_output)1080        # run plugins post-processing logic1081        invocation_result = self.process_result_via_plugins(inv_context, invocation_result)1082        return invocation_result1083    def provide_file_to_lambda(self, local_file: str, inv_context: InvocationContext) -> str:1084        # This is a no-op for local executors - simply return the given local file path1085        return local_file1086    def execute_java_lambda(1087        self, event, context, main_file, lambda_function: LambdaFunction = None1088    ) -> InvocationResult:1089        lambda_function.envvars = lambda_function.envvars or {}1090        java_opts = config.LAMBDA_JAVA_OPTS or ""1091        handler = lambda_function.handler1092        lambda_function.envvars[LAMBDA_HANDLER_ENV_VAR_NAME] = handler1093        event_file = EVENT_FILE_PATTERN.replace("*", short_uid())1094        save_file(event_file, json.dumps(json_safe(event)))1095        TMP_FILES.append(event_file)1096        classpath = "%s:%s:%s" % (1097            main_file,1098            Util.get_java_classpath(main_file),1099            LAMBDA_EXECUTOR_JAR,1100        )1101        cmd = "java %s -cp %s %s %s" % (1102            java_opts,1103            classpath,1104            LAMBDA_EXECUTOR_CLASS,1105            event_file,1106        )1107        # apply plugin patches1108        inv_context = InvocationContext(1109            lambda_function, event, environment=lambda_function.envvars, lambda_command=cmd1110        )1111        result = self.apply_plugin_patches(inv_context)1112        if isinstance(result, InvocationResult):1113            return result1114        cmd = inv_context.lambda_command1115        LOG.info(cmd)1116        # execute Lambda and get invocation result1117        invocation_result = self._execute_in_custom_runtime(cmd, lambda_function=lambda_function)1118        return invocation_result1119    def execute_javascript_lambda(1120        self, event, context, main_file, lambda_function: LambdaFunction = None1121    ):1122        handler = lambda_function.handler1123        function = handler.split(".")[-1]1124        event_json_string = "%s" % (json.dumps(json_safe(event)) if event else "{}")1125        context_json_string = "%s" % (json.dumps(context.__dict__) if context else "{}")1126        cmd = [1127            "node",1128            "-e",1129            'require("%s").%s(%s,%s).then(r => process.stdout.write(JSON.stringify(r)))'1130            % (1131                main_file,1132                function,1133                event_json_string,1134                context_json_string,1135            ),1136        ]1137        LOG.info(cmd)1138        result = self._execute_in_custom_runtime(cmd, lambda_function=lambda_function)1139        return result1140    @staticmethod1141    def set_default_env_variables():1142        # set default env variables required for most Lambda handlers1143        default_env_vars = {"AWS_DEFAULT_REGION": aws_stack.get_region()}1144        env_vars_before = {var: os.environ.get(var) for var in default_env_vars}1145        os.environ.update({k: v for k, v in default_env_vars.items() if not env_vars_before.get(k)})1146        return env_vars_before1147    @staticmethod1148    def reset_default_env_variables(env_vars_before):1149        for env_name, env_value in env_vars_before.items():1150            env_value_before = env_vars_before.get(env_name)1151            os.environ[env_name] = env_value_before or ""1152            if env_value_before is None:1153                os.environ.pop(env_name, None)1154    def execute_go_lambda(self, event, context, main_file, lambda_function: LambdaFunction = None):1155        if lambda_function:1156            lambda_function.envvars["AWS_LAMBDA_FUNCTION_HANDLER"] = main_file1157            lambda_function.envvars["AWS_LAMBDA_EVENT_BODY"] = json.dumps(json_safe(event))1158        else:1159            LOG.warning("Unable to get function details for local execution of Golang Lambda")1160        cmd = GO_LAMBDA_RUNTIME1161        LOG.info(cmd)1162        result = self._execute_in_custom_runtime(cmd, lambda_function=lambda_function)1163        return result1164class Util:1165    debug_java_port = False1166    @classmethod1167    def get_java_opts(cls):1168        opts = config.LAMBDA_JAVA_OPTS or ""1169        # Replace _debug_port_ with a random free port1170        if "_debug_port_" in opts:1171            if not cls.debug_java_port:1172                cls.debug_java_port = get_free_tcp_port()1173            opts = opts.replace("_debug_port_", ("%s" % cls.debug_java_port))1174        else:1175            # Parse the debug port from opts1176            m = re.match(".*address=(.+:)?(\\d+).*", opts)...

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.