Add plug points for code/config initializations in LambdaExecutorPlugin
diff --git a/localstack/services/awslambda/lambda_api.py b/localstack/services/awslambda/lambda_api.py
index 2fb94d851fd28..a56fc1f24d078 100644
--- a/localstack/services/awslambda/lambda_api.py
+++ b/localstack/services/awslambda/lambda_api.py
@@ -224,14 +224,6 @@ def check_batch_size_range(source_arn, batch_size=None):
     return batch_size
 
 
-def add_function_mapping(lambda_name, lambda_handler, lambda_cwd=None):
-    region = LambdaRegion.get()
-    arn = func_arn(lambda_name)
-    lambda_details = region.lambdas[arn]
-    lambda_details.versions.get(VERSION_LATEST)["Function"] = lambda_handler
-    lambda_details.cwd = lambda_cwd or lambda_details.cwd
-
-
 def build_mapping_obj(data) -> Dict:
     mapping = {}
     function_name = data["FunctionName"]
@@ -963,18 +955,21 @@ def set_archive_code(code, lambda_name, zip_file_content=None):
     return tmp_dir
 
 
-def set_function_code(code, lambda_name, lambda_cwd=None):
+def set_function_code(lambda_function: LambdaFunction):
     def _set_and_configure():
-        lambda_handler = do_set_function_code(code, lambda_name, lambda_cwd=lambda_cwd)
-        add_function_mapping(lambda_name, lambda_handler, lambda_cwd)
+        lambda_handler = do_set_function_code(lambda_function)
+        lambda_function.versions.get(VERSION_LATEST)["Function"] = lambda_handler
+        # initialize function code via plugins
+        for plugin in lambda_executors.LambdaExecutorPlugin.get_plugins():
+            plugin.init_function_code(lambda_function)
 
     # unzipping can take some time - limit the execution time to avoid client/network timeout issues
     run_for_max_seconds(25, _set_and_configure)
-    return {"FunctionName": lambda_name}
+    return {"FunctionName": lambda_function.name()}
 
 
-def do_set_function_code(code, lambda_name, lambda_cwd=None):
-    def generic_handler(event, context):
+def do_set_function_code(lambda_function: LambdaFunction):
+    def generic_handler(*_):
         raise ClientError(
             (
                 'Unable to find executor for Lambda function "%s". Note that '
@@ -984,15 +979,16 @@ def generic_handler(event, context):
         )
 
     region = LambdaRegion.get()
-    arn = func_arn(lambda_name)
+    lambda_name = lambda_function.name()
+    arn = lambda_function.arn()
     lambda_details = region.lambdas[arn]
     runtime = get_lambda_runtime(lambda_details)
     lambda_environment = lambda_details.envvars
     handler_name = lambda_details.handler = lambda_details.handler or LAMBDA_DEFAULT_HANDLER
-    code_passed = code
-    code = code or lambda_details.code
-    is_local_mount = code.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
+    code_passed = lambda_function.code
+    is_local_mount = code_passed.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
     zip_file_content = None
+    lambda_cwd = lambda_function.cwd
 
     LAMBDA_EXECUTOR.cleanup(arn)
 
@@ -1361,7 +1357,7 @@ def create_function():
         func_details.image_config = data.get("ImageConfig", {})
         func_details.tracing_config = data.get("TracingConfig", {})
         func_details.set_dead_letter_config(data)
-        result = set_function_code(func_details.code, lambda_name)
+        result = set_function_code(func_details)
         if isinstance(result, Response):
             del region.lambdas[arn]
             return result
@@ -1460,13 +1456,14 @@ def update_function_code(function):
     """
     region = LambdaRegion.get()
     arn = func_arn(function)
-    if arn not in region.lambdas:
+    func_details = region.lambdas.get(arn)
+    if not func_details:
         return not_found_error("Function not found: %s" % arn)
     data = json.loads(to_str(request.data))
-    result = set_function_code(data, function)
+    func_details.code = data
+    result = set_function_code(func_details)
     if isinstance(result, Response):
         return result
-    func_details = region.lambdas.get(arn)
     func_details.last_modified = datetime.utcnow()
     result.update(format_func_details(func_details))
     if data.get("Publish"):
@@ -1558,6 +1555,10 @@ def update_function_configuration(function):
     func_details = region.lambdas.get(arn)
     result.update(format_func_details(func_details))
 
+    # initialize plugins
+    for plugin in lambda_executors.LambdaExecutorPlugin.get_plugins():
+        plugin.init_function_configuration(func_details)
+
     return jsonify(result)
 
 
diff --git a/localstack/services/awslambda/lambda_executors.py b/localstack/services/awslambda/lambda_executors.py
index 80e1c95d02981..da2ab05310cc9 100644
--- a/localstack/services/awslambda/lambda_executors.py
+++ b/localstack/services/awslambda/lambda_executors.py
@@ -11,7 +11,7 @@
 import time
 import traceback
 from multiprocessing import Process, Queue
-from typing import Any, Dict, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from localstack import config
 from localstack.services.awslambda.lambda_utils import (
@@ -145,7 +145,7 @@ def __init__(
 class LambdaExecutorPlugin:
     """Plugin abstraction that allows to hook in additional functionality into the Lambda executors."""
 
-    INSTANCES = []
+    INSTANCES: List["LambdaExecutorPlugin"] = []
 
     def initialize(self):
         """Called once, for any active plugin to run initialization logic (e.g., downloading dependencies).
@@ -168,8 +168,16 @@ def process_result(
         """Optionally modify the result returned from the given Lambda invocation."""
         return result
 
+    def init_function_configuration(self, lambda_function: LambdaFunction):
+        """Initialize the configuration of the given function upon creation or function update."""
+        pass
+
+    def init_function_code(self, lambda_function: LambdaFunction):
+        """Initialize the code of the given function upon creation or function update."""
+        pass
+
     @classmethod
-    def get_plugins(cls):
+    def get_plugins(cls) -> List["LambdaExecutorPlugin"]:
         if not cls.INSTANCES:
             classes = get_all_subclasses(LambdaExecutorPlugin)
             cls.INSTANCES = [clazz() for clazz in classes]
diff --git a/localstack/utils/testutil.py b/localstack/utils/testutil.py
index e737c792e0be0..a96dcab3b4e4f 100644
--- a/localstack/utils/testutil.py
+++ b/localstack/utils/testutil.py
@@ -1,5 +1,6 @@
 import glob
 import importlib
+import io
 import json
 import os
 import shutil
@@ -7,7 +8,7 @@
 import time
 import zipfile
 from contextlib import contextmanager
-from typing import Dict
+from typing import Callable, Dict, List
 
 import requests
 from six import iteritems
@@ -35,14 +36,18 @@
     load_file,
     mkdir,
     poll_condition,
+    rm_rf,
     run,
     save_file,
+    short_uid,
     to_str,
 )
 
 ARCHIVE_DIR_PREFIX = "lambda.archive."
 DEFAULT_GET_LOG_EVENTS_DELAY = 3
 LAMBDA_TIMEOUT_SEC = 8
+LAMBDA_ASSETS_BUCKET_NAME = "ls-test-lambda-assets-bucket"
+MAX_LAMBDA_ARCHIVE_UPLOAD_SIZE = 50_000_000
 
 
 def is_local_test_mode():
@@ -64,48 +69,63 @@ def rm_dir(dir):
     shutil.rmtree(dir)
 
 
-def create_lambda_archive(script, get_content=False, libs=[], runtime=None, file_name=None):
+def create_lambda_archive(
+    script: str,
+    get_content: bool = False,
+    libs: List[str] = [],
+    runtime: str = None,
+    file_name: str = None,
+    exclude_func: Callable[[str], bool] = None,
+):
     """Utility method to create a Lambda function archive"""
     runtime = runtime or LAMBDA_DEFAULT_RUNTIME
-    tmp_dir = tempfile.mkdtemp(prefix=ARCHIVE_DIR_PREFIX)
-    TMP_FILES.append(tmp_dir)
-    file_name = file_name or get_handler_file_from_name(LAMBDA_DEFAULT_HANDLER, runtime=runtime)
-    script_file = os.path.join(tmp_dir, file_name)
-    if os.path.sep in script_file:
-        mkdir(os.path.dirname(script_file))
-        # create __init__.py files along the path to allow Python imports
-        path = file_name.split(os.path.sep)
-        for i in range(1, len(path)):
-            save_file(os.path.join(tmp_dir, *(path[:i] + ["__init__.py"])), "")
-    save_file(script_file, script)
-    chmod_r(script_file, 0o777)
-    # copy libs
-    for lib in libs:
-        paths = [lib, "%s.py" % lib]
-        try:
-            module = importlib.import_module(lib)
-            paths.append(module.__file__)
-        except Exception:
-            pass
-        target_dir = tmp_dir
-        root_folder = os.path.join(LOCALSTACK_VENV_FOLDER, "lib/python*/site-packages")
-        if lib == "localstack":
-            paths = ["localstack/*.py", "localstack/utils"]
-            root_folder = LOCALSTACK_ROOT_FOLDER
-            target_dir = os.path.join(tmp_dir, lib)
-            mkdir(target_dir)
-        for path in paths:
-            file_to_copy = path if path.startswith("/") else os.path.join(root_folder, path)
-            for file_path in glob.glob(file_to_copy):
-                name = os.path.join(target_dir, file_path.split(os.path.sep)[-1])
-                if os.path.isdir(file_path):
-                    copy_dir(file_path, name)
-                else:
-                    shutil.copyfile(file_path, name)
 
-    # create zip file
-    result = create_zip_file(tmp_dir, get_content=get_content)
-    return result
+    with tempfile.TemporaryDirectory(prefix=ARCHIVE_DIR_PREFIX) as tmp_dir:
+        file_name = file_name or get_handler_file_from_name(LAMBDA_DEFAULT_HANDLER, runtime=runtime)
+        script_file = os.path.join(tmp_dir, file_name)
+        if os.path.sep in script_file:
+            mkdir(os.path.dirname(script_file))
+            # create __init__.py files along the path to allow Python imports
+            path = file_name.split(os.path.sep)
+            for i in range(1, len(path)):
+                save_file(os.path.join(tmp_dir, *(path[:i] + ["__init__.py"])), "")
+        save_file(script_file, script)
+        chmod_r(script_file, 0o777)
+        # copy libs
+        for lib in libs:
+            paths = [lib, "%s.py" % lib]
+            try:
+                module = importlib.import_module(lib)
+                paths.append(module.__file__)
+            except Exception:
+                pass
+            target_dir = tmp_dir
+            root_folder = os.path.join(LOCALSTACK_VENV_FOLDER, "lib/python*/site-packages")
+            if lib == "localstack":
+                paths = ["localstack/*.py", "localstack/utils"]
+                root_folder = LOCALSTACK_ROOT_FOLDER
+                target_dir = os.path.join(tmp_dir, lib)
+                mkdir(target_dir)
+            for path in paths:
+                file_to_copy = path if path.startswith("/") else os.path.join(root_folder, path)
+                for file_path in glob.glob(file_to_copy):
+                    name = os.path.join(target_dir, file_path.split(os.path.sep)[-1])
+                    if os.path.isdir(file_path):
+                        copy_dir(file_path, name)
+                    else:
+                        shutil.copyfile(file_path, name)
+
+        if exclude_func:
+            for dirpath, folders, files in os.walk(tmp_dir):
+                for name in list(folders) + list(files):
+                    full_name = os.path.join(dirpath, name)
+                    relative = os.path.relpath(full_name, start=tmp_dir)
+                    if exclude_func(relative):
+                        rm_rf(full_name)
+
+        # create zip file
+        result = create_zip_file(tmp_dir, get_content=get_content)
+        return result
 
 
 def delete_lambda_function(name):
@@ -125,7 +145,7 @@ def create_zip_file_python(source_path, base_dir, zip_file):
         for root, dirs, files in os.walk(base_dir):
             for name in files:
                 full_name = os.path.join(root, name)
-                relative = root[len(base_dir) :].lstrip(os.path.sep)
+                relative = os.path.relpath(root, start=base_dir)
                 dest = os.path.join(relative, name)
                 zip_file.write(full_name, dest)
 
@@ -198,6 +218,16 @@ def create_lambda_function(
         except Exception:
             pass
 
+    lambda_code = {"ZipFile": zip_file}
+    if len(zip_file) > MAX_LAMBDA_ARCHIVE_UPLOAD_SIZE:
+        s3 = aws_stack.connect_to_service("s3")
+        aws_stack.get_or_create_bucket(LAMBDA_ASSETS_BUCKET_NAME)
+        asset_key = f"{short_uid()}.zip"
+        s3.upload_fileobj(
+            Fileobj=io.BytesIO(zip_file), Bucket=LAMBDA_ASSETS_BUCKET_NAME, Key=asset_key
+        )
+        lambda_code = {"S3Bucket": LAMBDA_ASSETS_BUCKET_NAME, "S3Key": asset_key}
+
     # create function
     additional_kwargs = kwargs
     kwargs = {
@@ -205,7 +235,7 @@ def create_lambda_function(
         "Runtime": runtime,
         "Handler": handler,
         "Role": LAMBDA_TEST_ROLE,
-        "Code": {"ZipFile": zip_file},
+        "Code": lambda_code,
         "Timeout": LAMBDA_TIMEOUT_SEC,
         "Environment": dict(Variables=envvars),
         "Tags": tags,

* Add plug points for code/config initializations in LambdaExecutorPlugin
* Streamline parameters for `set_function_code(..)` functions - replacing individual args with single `LambdaFunction` arg
* Refactor `testutil.py` to deploy large Lambda zip files automatically via S3 (avoiding `RequestEntityTooLargeException`)
