How to use _prepare_invocation_for_java_lambda method in localstack

Best Python code snippet using localstack_python

thundra.py

Source:thundra.py

...101    return runtime == LAMBDA_RUNTIME_JAVA8 or runtime == LAMBDA_RUNTIME_JAVA8_AL2102def _is_java_lambda_with_support_version(lambda_details):103    runtime = getattr(lambda_details, "runtime", lambda_details)104    return runtime in [LAMBDA_RUNTIME_JAVA8, LAMBDA_RUNTIME_JAVA8_AL2, LAMBDA_RUNTIME_JAVA11]105def _prepare_invocation_for_java_lambda(context: InvocationContext) -> AdditionalInvocationOptions:106    # Download and initialize Java agent107    _ensure_java_agent_initialized()108    # If agent could not be initialized, skip here109    if not THUNDRA_JAVA_AGENT_INITIALIZED:110        return None111    result = AdditionalInvocationOptions()112    environment = context.environment113    agent_flag = "-javaagent:{agent_path}"114    # Inject Thundra agent path into "JAVA_TOOL_OPTIONS" env var,115    # so it will be automatically loaded on JVM startup116    java_tool_opts = environment.get("JAVA_TOOL_OPTIONS", "")117    if agent_flag not in java_tool_opts:118        java_tool_opts += f" {agent_flag}"119    result.env_updates["JAVA_TOOL_OPTIONS"] = java_tool_opts.strip()120    # Disable CDS (Class Data Sharing),121    # because "-javaagent" cannot be enabled when CDS is enabled on JDK 8.122    # CDS can only be disabled by "_JAVA_OPTIONS" env var,123    # because by default it is enabled ("-Xshare:on")124    # on Lambci by command line parameters and125    # "_JAVA_OPTIONS" has precedence over command line parameters126    # but "JAVA_TOOL_OPTIONS" is not.127    if _is_java8_lambda(context.lambda_function):128        java_opts = environment.get("_JAVA_OPTIONS", "")129        java_opts += " -Xshare:off"130        result.env_updates["_JAVA_OPTIONS"] = java_opts.strip()131    # If log disable is not configured explicitly, set it to false to enable log capturing by default132    log_disabled = environment.get(THUNDRA_AGENT_LOG_DISABLE_ENV_VAR_NAME)133    if not log_disabled:134        result.env_updates[THUNDRA_AGENT_LOG_DISABLE_ENV_VAR_NAME] = "false"135    # Make sure API key is contained in environment136    result.env_updates[THUNDRA_APIKEY_ENV_VAR_NAME] = _get_apikey(environment)137    # Note: The code below doesn't seem to be required, as LAMBDA_EXECUTOR=local also picks up $JAVA_TOOL_OPTIONS138    # if context.lambda_command:139    #     result.updated_command = context.lambda_command.replace(140    #         "java ", f"java {agent_flag} ", 1141    #     )142    # construct agent file path mapping143    result.files_to_add["agent_path"] = THUNDRA_JAVA_AGENT_LOCAL_PATH144    return result145################146# NODE AGENT147################148def _ensure_node_agent_initialized():149    global THUNDRA_NODE_AGENT_INITIALIZED150    if not THUNDRA_NODE_AGENT_INITIALIZED:151        if _init_node_agent_configs() and _install_node_agent():152            THUNDRA_NODE_AGENT_INITIALIZED = True153def _get_latest_node_agent_version():154    try:155        return common.run("npm view @thundra/core version".split())156    except Exception as e:157        print("Unable to get latest version of Thundra Node agent: %s" % e)158        return None159def _init_node_agent_configs() -> bool:160    global THUNDRA_NODE_AGENT_VERSION161    global THUNDRA_NODE_AGENT_LOCAL_PATH162    global THUNDRA_NODE_AGENT_LOCAL_PATH_ON_HOST163    latest_version = _get_latest_node_agent_version()164    version = os.getenv("THUNDRA_AGENT_NODE_VERSION", latest_version)165    if not version:166        return False167    THUNDRA_NODE_AGENT_VERSION = version.strip()168    THUNDRA_NODE_AGENT_LOCAL_PATH = "%s/thundra/node/%s/" % (169        config.TMP_FOLDER,170        THUNDRA_NODE_AGENT_VERSION,171    )172    THUNDRA_NODE_AGENT_LOCAL_PATH_ON_HOST = "%s/thundra/node/%s/" % (173        config.HOST_TMP_FOLDER,174        THUNDRA_NODE_AGENT_VERSION,175    )176    return True177def _install_node_agent() -> bool:178    # Install Thundra Node agent NPM package179    if not os.path.exists(THUNDRA_NODE_AGENT_LOCAL_PATH):180        install.log_install_msg("Thundra Node agent", verbatim=True)181        try:182            install_thundra_cmd = "npm install --prefix %s @thundra/core@%s --no-save" % (183                THUNDRA_NODE_AGENT_LOCAL_PATH,184                THUNDRA_NODE_AGENT_VERSION,185            )186            common.run(install_thundra_cmd.split())187        except Exception as e:188            print("Unable to install Thundra Node agent: %s" % e)189            return False190    return True191def _is_node_lambda_with_support_version(func_details):192    runtime = getattr(func_details, "runtime", func_details)193    return runtime in [194        LAMBDA_RUNTIME_NODEJS10X,195        LAMBDA_RUNTIME_NODEJS12X,196        LAMBDA_RUNTIME_NODEJS14X,197    ]198def _prepare_invocation_for_node_lambda(context: InvocationContext) -> AdditionalInvocationOptions:199    # Download and initialize Node agent200    _ensure_node_agent_initialized()201    # If agent could not be initialized, skip here202    if not THUNDRA_NODE_AGENT_INITIALIZED:203        return None204    result = AdditionalInvocationOptions()205    # Make sure API key is contained in environment206    result.env_updates[THUNDRA_APIKEY_ENV_VAR_NAME] = _get_apikey(context.environment)207    # Switch handler to Thundra and pass original handler to Thundra through environment variable208    result.env_updates[THUNDRA_AGENT_LAMBDA_HANDLER_ENV_VAR_NAME] = context.handler209    result.updated_handler = "/opt/nodejs/node_modules/@thundra/core/dist/handler.wrapper"210    # If log disable is not configured explicitly, set it to false to enable log capturing by default211    log_disabled = context.environment.get(THUNDRA_AGENT_LOG_DISABLE_ENV_VAR_NAME)212    if not log_disabled:213        result.env_updates[THUNDRA_AGENT_LOG_DISABLE_ENV_VAR_NAME] = "false"214    # Map Thundra agent path into container so it will be accessible by Lambda function Node environment215    agent_path_mapping = (216        "-v %s/node_modules/:/opt/nodejs/node_modules/" % THUNDRA_NODE_AGENT_LOCAL_PATH_ON_HOST217    )218    if context.docker_flags:219        context.docker_flags = f"{context.docker_flags} {agent_path_mapping}"220    else:221        context.docker_flags = agent_path_mapping222    return result223################224# PYTHON AGENT225################226def _ensure_python_agent_initialized():227    global THUNDRA_PYTHON_AGENT_INITIALIZED228    if not THUNDRA_PYTHON_AGENT_INITIALIZED:229        if _init_python_agent_configs() and _install_python_agent():230            THUNDRA_PYTHON_AGENT_INITIALIZED = True231def _get_latest_python_agent_version():232    try:233        from distutils.version import StrictVersion234        import requests235        response = requests.get("https://pypi.org/pypi/thundra/json")236        data = json.loads(response.content.decode())237        versions = sorted(list(data["releases"].keys()), key=StrictVersion, reverse=True)238        return versions[0]239    except Exception as e:240        print("Unable to get latest version of Thundra Python agent: %s" % e)241        return None242def _init_python_agent_configs() -> bool:243    global THUNDRA_PYTHON_AGENT_VERSION244    global THUNDRA_PYTHON_AGENT_LOCAL_PATH245    global THUNDRA_PYTHON_AGENT_LOCAL_PATH_ON_HOST246    latest_version = _get_latest_python_agent_version()247    version = os.getenv("THUNDRA_AGENT_PYTHON_VERSION", latest_version)248    if not version:249        return False250    THUNDRA_PYTHON_AGENT_VERSION = version.strip()251    THUNDRA_PYTHON_AGENT_LOCAL_PATH = "%s/thundra/python/%s/" % (252        config.TMP_FOLDER,253        THUNDRA_PYTHON_AGENT_VERSION,254    )255    THUNDRA_PYTHON_AGENT_LOCAL_PATH_ON_HOST = "%s/thundra/python/%s/" % (256        config.HOST_TMP_FOLDER,257        THUNDRA_PYTHON_AGENT_VERSION,258    )259    return True260def _install_python_agent() -> bool:261    # Install Thundra Python agent PIP package262    if not os.path.exists(THUNDRA_PYTHON_AGENT_LOCAL_PATH):263        install.log_install_msg("Thundra Python agent", verbatim=True)264        try:265            install_thundra_cmd = "pip install --target=%s thundra==%s --no-warn-conflicts" % (266                THUNDRA_PYTHON_AGENT_LOCAL_PATH,267                THUNDRA_PYTHON_AGENT_VERSION,268            )269            common.run(install_thundra_cmd.split())270        except Exception as e:271            print("Unable to install Thundra Python agent: %s" % e)272            return False273    return True274def _is_python_lambda_with_support_version(func_details):275    runtime = getattr(func_details, "runtime", func_details)276    return runtime in [277        LAMBDA_RUNTIME_PYTHON36,278        LAMBDA_RUNTIME_PYTHON37,279        LAMBDA_RUNTIME_PYTHON38,280    ]281def _prepare_invocation_for_python_lambda(282    context: InvocationContext,283) -> AdditionalInvocationOptions:284    # Download and initialize Python agent285    _ensure_python_agent_initialized()286    # If agent could not be initialized, skip here287    if not THUNDRA_PYTHON_AGENT_INITIALIZED:288        return None289    result = AdditionalInvocationOptions()290    # Make sure API key is contained in environment291    result.env_updates[THUNDRA_APIKEY_ENV_VAR_NAME] = _get_apikey(context.environment)292    # Switch handler to Thundra and pass original handler to Thundra through environment variable293    result.env_updates[THUNDRA_AGENT_LAMBDA_HANDLER_ENV_VAR_NAME] = context.handler294    result.updated_handler = "thundra.handler.wrapper"295    # If log disable is not configured explicitly, set it to false to enable log capturing by default296    log_disabled = context.environment.get(THUNDRA_AGENT_LOG_DISABLE_ENV_VAR_NAME)297    if not log_disabled:298        result.env_updates[THUNDRA_AGENT_LOG_DISABLE_ENV_VAR_NAME] = "false"299    # Map Thundra agent path into container so it will be accessible by Lambda function Python environment300    agent_path_mapping = "-v %s/:/opt/python/" % THUNDRA_PYTHON_AGENT_LOCAL_PATH_ON_HOST301    if context.docker_flags:302        context.docker_flags = f"{context.docker_flags} {agent_path_mapping}"303    else:304        context.docker_flags = agent_path_mapping305    return result306################307# THUNDRA PLUGIN308################309class LambdaExecutorPluginThundra(LambdaExecutorPlugin):310    def should_apply(self, context: InvocationContext) -> bool:311        # Local executor is not supported yet312        if "local" in config.LAMBDA_EXECUTOR:313            return False314        # Plugin can only applied if LAMBDA_REMOTE_DOCKER=0315        if "docker" in config.LAMBDA_EXECUTOR and config.LAMBDA_REMOTE_DOCKER:316            return False317        # Plugin can only applied if API key is configured318        thundra_apikey = _get_apikey(context.environment)319        if not thundra_apikey:320            return False321        # Plugin can be applied for Java Lambdas with supported versions322        if _is_java_lambda_with_support_version(context.lambda_function.runtime):323            return True324        # Plugin can be applied for Node Lambdas with supported versions325        if _is_node_lambda_with_support_version(context.lambda_function.runtime):326            return True327        # Plugin can be applied for Python Lambdas with supported versions328        if _is_python_lambda_with_support_version(context.lambda_function.runtime):329            return True330        # Not applicable for Thundra plugin331        return False332    def prepare_invocation(333        self, context: InvocationContext334    ) -> Optional[Union[AdditionalInvocationOptions, InvocationResult]]:335        if is_java_lambda(context.lambda_function):336            return _prepare_invocation_for_java_lambda(context)337        elif is_nodejs_runtime(context.lambda_function):338            return _prepare_invocation_for_node_lambda(context)339        elif is_python_runtime(context.lambda_function):340            return _prepare_invocation_for_python_lambda(context)...

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.