build_integ_base.py

import os
import posixpath
import sys
import uuid
import shutil
import tempfile
import time
import logging
import json
from datetime import datetime, timezone
from typing import Optional
from unittest import TestCase

import jmespath
from pathlib import Path

import pytest
from parameterized import parameterized_class

from samcli.lib.utils import osutils
from samcli.lib.utils.architecture import ARM64, X86_64
from samcli.local.docker.lambda_build_container import LambdaBuildContainer
from samcli.local.docker.utils import get_validated_container_client
from samcli.yamlhelper import yaml_parse
from tests.testing_utils import (
    IS_WINDOWS,
    CommandResult,
    run_command,
    SKIP_DOCKER_TESTS,
    SKIP_DOCKER_MESSAGE,
    SKIP_DOCKER_BUILD,
    get_sam_command,
    run_command_with_input,
)
from samcli.commands.build.utils import MountMode

LOG = logging.getLogger(__name__)


def show_container_in_test_name(testcase_func, param_num, param):
    """
    Generates a custom name for parameterized test cases.
    Adds '_in_container' suffix when any parameter contains 'container' in its string representation.
    """
    # Get the base test name
    base_name = f"{testcase_func.__name__}_{param_num}"

    # Check if any parameter contains "container" in its string representation
    for arg in param.args:
        if isinstance(arg, str) and "container" in arg.lower():
            base_name += "_in_container"
            break
        elif arg is True:  # Also check for boolean True which might indicate use_container
            # Check if this might be a use_container parameter by position
            # This is a fallback for cases where True is used instead of "use_container"
            continue

    return base_name


class BuildIntegBase(TestCase):
    template: Optional[str] = "template.yaml"

    @classmethod
    def setUpClass(cls):
        cls.cmd = get_sam_command()
        integration_dir = Path(__file__).resolve().parents[1]
        cls.test_data_path = str(Path(integration_dir, "testdata", "buildcmd"))
        cls.template_path = str(Path(cls.test_data_path, cls.template)) if cls.template else None

    def setUp(self):
        # To invoke a function created by the build command, we need the built artifacts to be in a
        # location that is shared in Docker. Most temp directories are not shared. Therefore we are
        # using a scratch space within the test folder that is .gitignored. Contents of this folder
        # is also deleted after every test run
        self.scratch_dir = str(Path(__file__).resolve().parent.joinpath("tmp", str(uuid.uuid4()).replace("-", "")[:10]))
        shutil.rmtree(self.scratch_dir, ignore_errors=True)
        os.makedirs(self.scratch_dir)

        self.working_dir = tempfile.mkdtemp(dir=self.scratch_dir)
        self.custom_build_dir = tempfile.mkdtemp(dir=self.scratch_dir)

        self.default_build_dir = Path(self.working_dir, ".aws-sam", "build")
        self.built_template = self.default_build_dir.joinpath("template.yaml")

    def tearDown(self):
        self.custom_build_dir and shutil.rmtree(self.custom_build_dir, ignore_errors=True)
        self.working_dir and shutil.rmtree(self.working_dir, ignore_errors=True)
        self.scratch_dir and shutil.rmtree(self.scratch_dir, ignore_errors=True)

    def get_command_list(
        self,
        build_dir=None,
        base_dir=None,
        manifest_path=None,
        use_container=None,
        parameter_overrides=None,
        mode=None,
        function_identifier=None,
        debug=False,
        cached=False,
        cache_dir=None,
        parallel=False,
        container_env_var=None,
        container_env_var_file=None,
        build_image=None,
        exclude=None,
        region=None,
        hook_name=None,
        beta_features=None,
        build_in_source=None,
        mount_with=None,
        config_file=None,
        save_params=False,
        project_root_dir=None,
        use_buildkit=False,
    ):
        command_list = [self.cmd, "build"]

        if function_identifier:
            command_list += [function_identifier]

        command_list += ["-t", self.template_path] if self.template_path else []

        if parameter_overrides:
            command_list += ["--parameter-overrides", self._make_parameter_override_arg(parameter_overrides)]

        if build_dir:
            command_list += ["-b", build_dir]

        if base_dir:
            command_list += ["-s", base_dir]

        if manifest_path:
            command_list += ["-m", manifest_path]

        if use_container:
            command_list += ["--use-container"]

        if debug:
            command_list += ["--debug"]

        if use_buildkit:
            command_list += ["--use-buildkit"]

        if cached:
            command_list += ["--cached"]

        if cache_dir:
            command_list += ["-cd", cache_dir]

        if parallel:
            command_list += ["--parallel"]

        if container_env_var:
            command_list += ["--container-env-var", container_env_var]

        if container_env_var_file:
            command_list += ["--container-env-var-file", container_env_var_file]

        if build_image:
            command_list += ["--build-image", build_image]

        if mount_with:
            command_list += ["--mount-with", mount_with.value]

        if exclude:
            for f in exclude:
                command_list += ["--exclude", f]

        if region:
            command_list += ["--region", region]

        if beta_features is not None:
            command_list += ["--beta-features"] if beta_features else ["--no-beta-features"]

        if hook_name:
            command_list += ["--hook-name", hook_name]

        if build_in_source is not None:
            command_list += ["--build-in-source"] if build_in_source else ["--no-build-in-source"]

        if config_file is not None:
            command_list += ["--config-file", config_file]

        if save_params:
            command_list += ["--save-params"]

        if project_root_dir is not None:
            command_list += ["--terraform-project-root-path", project_root_dir]

        return command_list

    def verify_docker_container_cleanedup(self, runtime):
        if IS_WINDOWS:
            time.sleep(1)

        docker_client = get_validated_container_client()

        # Use strategy pattern method for unified container listing across runtimes
        samcli_containers = docker_client.list_containers_by_image(
            f"{LambdaBuildContainer._IMAGE_URI_PREFIX}-{runtime}", all_containers=True
        )

        self.assertFalse(bool(samcli_containers), "Build containers have not been removed")

    def get_number_of_created_containers(self):
        if IS_WINDOWS:
            time.sleep(1)

        docker_client = get_validated_container_client()
        containers = docker_client.containers.list(all=True, filters={"status": "exited"})
        return len(containers)

    def verify_pulled_image(self, runtime, architecture=X86_64):
        docker_client = get_validated_container_client()
        image_name = f"{LambdaBuildContainer._IMAGE_URI_PREFIX}-{runtime}"
        architecture = architecture if architecture and "provided" not in runtime else X86_64
        tag_name = LambdaBuildContainer.get_image_tag(architecture)

        # Use strategy pattern method for unified image validation across runtimes
        is_valid = docker_client.validate_image_count(image_name, expected_count_range=(1, 2))
        self.assertTrue(is_valid, f"Image validation failed for {image_name}")

        # Verify the specific tag exists by getting the images and checking tags
        images = docker_client.images.list(name=image_name)
        image_tag = f"{image_name}:{tag_name}"
        for t in [tag for image in images for tag in image.tags]:
            if t == image_tag:
                return  # Found, pass
        self.fail(f"{image_tag} was not pulled")

    def _make_parameter_override_arg(self, overrides):
        return " ".join(["ParameterKey={},ParameterValue={}".format(key, value) for key, value in overrides.items()])

    def _verify_image_build_artifact(self, template_path, image_function_logical_id, property, image_uri):
        self.assertTrue(template_path.exists(), "Build directory should be created")

        build_dir = template_path.parent
        build_dir_files = os.listdir(str(build_dir))
        self.assertNotIn(image_function_logical_id, build_dir_files)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), image_function_logical_id, property, image_uri)

    def _verify_resource_property(self, template_path, logical_id, property, expected_value):
        with open(template_path, "r") as fp:
            template_dict = yaml_parse(fp.read())

            self.assertEqual(
                expected_value, jmespath.search(f"Resources.{logical_id}.Properties.{property}", template_dict)
            )

    def _verify_invoke_built_function(self, template_path, function_logical_id, overrides, expected_result):
        LOG.info("Invoking built function '{}'".format(function_logical_id))

        cmdlist = [
            self.cmd,
            "local",
            "invoke",
            function_logical_id,
            "-t",
            str(template_path),
            "--no-event",
        ]

        if overrides:
            cmdlist += [
                "--parameter-overrides",
                overrides,
            ]

        for i in range(5):
            process_execute = run_command(cmdlist)
            process_stdout = process_execute.stdout.decode("utf-8")
            process_stderr = process_execute.stderr.decode("utf-8")
            LOG.info("Local invoke stdout: %s", process_stdout)
            LOG.info("Local invoke stderr: %s", process_stderr)

            if "timed out after" in process_stderr:
                LOG.info("Function timed out, retrying")
                continue

            if json.loads(process_stdout) == expected_result:
                LOG.info("Expected result found, succeeded!")
                # success
                return

        self.fail("Failed to invoke function & get expected result")

    def get_override(self, runtime, code_uri, architecture, handler):
        overrides = {"Runtime": runtime, "CodeUri": code_uri, "Handler": handler}
        if architecture:
            overrides["Architectures"] = architecture
        return overrides


@pytest.mark.ruby
class BuildIntegRubyBase(BuildIntegBase):
    EXPECTED_FILES_PROJECT_MANIFEST = {"app.rb"}
    EXPECTED_RUBY_GEM = "aws-eventstream"

    FUNCTION_LOGICAL_ID = "Function"

    def _test_with_default_gemfile(self, runtime, use_container, code_uri, relative_path, architecture=None):
        overrides = self.get_override(runtime, code_uri, architecture, "ignored")
        cmdlist = self.get_command_list(use_container=use_container, parameter_overrides=overrides)

        run_command(cmdlist, cwd=self.working_dir)

        self._verify_built_artifact(
            self.default_build_dir,
            self.FUNCTION_LOGICAL_ID,
            self.EXPECTED_FILES_PROJECT_MANIFEST,
            self.EXPECTED_RUBY_GEM,
        )

        self._verify_resource_property(
            str(self.built_template),
            "OtherRelativePathResource",
            "BodyS3Location",
            os.path.relpath(
                os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        self._verify_resource_property(
            str(self.built_template),
            "GlueResource",
            "Command.ScriptLocation",
            os.path.relpath(
                os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        if use_container:
            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files, expected_modules):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)

        ruby_version = None
        ruby_bundled_path = None

        # Walk through ruby version to get to the gem path
        for dirpath, dirname, _ in os.walk(str(resource_artifact_dir.joinpath("vendor", "bundle", "ruby"))):
            ruby_version = dirname
            ruby_bundled_path = Path(dirpath)
            break

        # If Gemfile is in root folder, vendor folder will also be created there
        if ruby_version is None:
            for dirpath, dirname, _ in os.walk(str(Path(self.working_dir).joinpath("vendor", "bundle", "ruby"))):
                ruby_version = dirname
                ruby_bundled_path = Path(dirpath)
                break

        gem_path = ruby_bundled_path.joinpath(ruby_version[0], "gems")

        self.assertTrue(any([True if self.EXPECTED_RUBY_GEM in gem else False for gem in os.listdir(str(gem_path))]))


@pytest.mark.nodejs
class BuildIntegEsbuildBase(BuildIntegBase):
    FUNCTION_LOGICAL_ID = "Function"
    # Everything should be minifed to one line and a second line for the sourcemap mapping
    MAX_MINIFIED_LINE_COUNT = 2
    MANIFEST_PATH: Optional[str] = None

    def _test_with_default_package_json(
        self, runtime, use_container, code_uri, expected_files, handler, architecture=None, build_in_source=None
    ):
        if use_container and (SKIP_DOCKER_TESTS or SKIP_DOCKER_BUILD):
            self.skipTest(SKIP_DOCKER_MESSAGE)
        overrides = self.get_override(runtime, code_uri, architecture, handler)
        manifest_path = str(Path(self.test_data_path, self.MANIFEST_PATH)) if self.MANIFEST_PATH else None
        cmdlist = self.get_command_list(
            use_container=use_container,
            parameter_overrides=overrides,
            build_in_source=build_in_source,
            manifest_path=manifest_path,
        )
        run_command(cmdlist, cwd=self.working_dir)

        self._verify_built_artifact(
            self.default_build_dir,
            self.FUNCTION_LOGICAL_ID,
            expected_files,
        )

        expected = {"body": '{"message":"hello world!"}', "statusCode": 200}
        if not SKIP_DOCKER_TESTS and architecture == X86_64:
            # ARM64 is not supported yet for invoking
            self._verify_invoke_built_function(
                self.built_template, self.FUNCTION_LOGICAL_ID, self._make_parameter_override_arg(overrides), expected
            )

        self._verify_esbuild_properties(self.default_build_dir, self.FUNCTION_LOGICAL_ID, handler)

        if use_container:
            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _test_with_various_properties(self, overrides, runtime):
        overrides = self.get_override(**overrides)
        cmdlist = self.get_command_list(parameter_overrides=overrides)

        run_command(cmdlist, cwd=self.working_dir)

        expected = {"body": '{"message":"hello world!"}', "statusCode": 200}
        if not SKIP_DOCKER_TESTS and overrides["Architectures"] == X86_64:
            # ARM64 is not supported yet for invoking
            self._verify_invoke_built_function(
                self.built_template, self.FUNCTION_LOGICAL_ID, self._make_parameter_override_arg(overrides), expected
            )

        self._verify_esbuild_properties(self.default_build_dir, self.FUNCTION_LOGICAL_ID, overrides["Handler"])

    def _verify_esbuild_properties(self, build_dir, function_logical_id, handler):
        filename = self._extract_filename_from_handler(handler)
        resource_artifact_dir = build_dir.joinpath(function_logical_id)
        self._verify_sourcemap_created(filename, resource_artifact_dir)
        self._verify_function_minified(filename, resource_artifact_dir)

    @staticmethod
    def _extract_filename_from_handler(handler):
        # Takes a handler in the form /a/b/c/file.function and returns file
        return str(Path(handler).stem)

    def _verify_function_minified(self, filename, resource_artifact_dir):
        with open(Path(resource_artifact_dir, f"{filename}.js"), "r") as handler_file:
            x = len(handler_file.readlines())
        self.assertLessEqual(x, self.MAX_MINIFIED_LINE_COUNT)

    def _verify_sourcemap_created(self, filename, resource_artifact_dir):
        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        self.assertIn(f"{filename}.js.map", all_artifacts)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)


@pytest.mark.nodejs
class BuildIntegNodeBase(BuildIntegBase):
    EXPECTED_FILES_PROJECT_MANIFEST = {"node_modules", "main.js"}
    EXPECTED_NODE_MODULES = {"minimal-request-promise"}
    CODE_URI = "Node"
    FUNCTION_LOGICAL_ID = "Function"
    TEST_INVOKE = False
    MANIFEST_PATH: Optional[str] = None

    def _test_with_default_package_json(self, runtime, use_container, relative_path, architecture=None):
        if use_container and (SKIP_DOCKER_TESTS or SKIP_DOCKER_BUILD):
            self.skipTest(SKIP_DOCKER_MESSAGE)
        overrides = self.get_override(runtime, self.CODE_URI, architecture, "main.lambdaHandler")
        manifest_path = str(Path(self.test_data_path, self.MANIFEST_PATH)) if self.MANIFEST_PATH else None
        cmdlist = self.get_command_list(
            use_container=use_container, parameter_overrides=overrides, manifest_path=manifest_path
        )

        run_command(cmdlist, cwd=self.working_dir)

        self._verify_built_artifact(
            self.default_build_dir,
            self.FUNCTION_LOGICAL_ID,
            self.EXPECTED_FILES_PROJECT_MANIFEST,
            self.EXPECTED_NODE_MODULES,
        )

        expected = {"body": '{"message":"hello world!"}', "statusCode": 200}
        if not SKIP_DOCKER_TESTS and self.TEST_INVOKE:
            self._verify_invoke_built_function(
                self.built_template, self.FUNCTION_LOGICAL_ID, self._make_parameter_override_arg(overrides), expected
            )

        self._verify_resource_property(
            str(self.built_template),
            "OtherRelativePathResource",
            "BodyS3Location",
            os.path.relpath(
                os.path.normpath(os.path.join(str(str(relative_path)), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        self._verify_resource_property(
            str(self.built_template),
            "GlueResource",
            "Command.ScriptLocation",
            os.path.relpath(
                os.path.normpath(os.path.join(str(str(relative_path)), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        if use_container:
            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files, expected_modules):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)

        all_modules = set(os.listdir(str(resource_artifact_dir.joinpath("node_modules"))))
        actual_files = all_modules.intersection(expected_modules)
        self.assertEqual(actual_files, expected_modules)


@pytest.mark.golang
class BuildIntegGoBase(BuildIntegBase):
    FUNCTION_LOGICAL_ID = "Function"
    EXPECTED_FILES_PROJECT_MANIFEST = {"hello-world"}

    def _test_with_go(self, runtime, code_uri, mode, relative_path, architecture=None, use_container=False):
        overrides = self.get_override(runtime, code_uri, architecture, "hello-world")
        cmdlist = self.get_command_list(use_container=use_container, parameter_overrides=overrides)

        # Need to pass GOPATH ENV variable to match the test directory when running build
        LOG.info("Running with SAM_BUILD_MODE={}".format(mode))

        newenv = os.environ.copy()
        if mode:
            newenv["SAM_BUILD_MODE"] = mode

        newenv["GOPROXY"] = "direct"
        newenv["GOPATH"] = str(self.working_dir)

        # Build with musl target to avoid glibc compatibility issues
        # This ensures the binary works in the Lambda execution environment
        newenv["GOOS"] = "linux"
        newenv["GOARCH"] = "arm64" if architecture == ARM64 else "amd64"
        newenv["CGO_ENABLED"] = "0"

        run_command(cmdlist, cwd=self.working_dir, env=newenv)

        self._verify_built_artifact(
            self.default_build_dir, self.FUNCTION_LOGICAL_ID, self.EXPECTED_FILES_PROJECT_MANIFEST
        )

        self._verify_resource_property(
            str(self.built_template),
            "OtherRelativePathResource",
            "BodyS3Location",
            os.path.relpath(
                os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        expected = "{'message': 'Hello World'}"
        if not SKIP_DOCKER_TESTS and architecture == X86_64:
            # ARM64 is not supported yet for invoking
            self._verify_invoke_built_function(
                self.built_template, self.FUNCTION_LOGICAL_ID, self._make_parameter_override_arg(overrides), expected
            )

        if use_container:
            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)


@pytest.mark.java
class BuildIntegJavaBase(BuildIntegBase):
    # Run gradlew & gradle-kotlin containerized tests only with latest java version
    LATEST_JAVA_VERSION = "java21"
    SKIP_ARM64_EARLIER_JAVA_TESTS = "Skipping gradlew & gradle-kotlin test for this Java version"

    FUNCTION_LOGICAL_ID = "Function"
    EXPECTED_FILES_PROJECT_MANIFEST_GRADLE = {"aws", "lib", "META-INF"}
    EXPECTED_FILES_PROJECT_MANIFEST_MAVEN = {"aws", "lib"}
    EXPECTED_GRADLE_DEPENDENCIES = {"annotations-2.1.0.jar", "aws-lambda-java-core-1.1.0.jar"}
    EXPECTED_MAVEN_DEPENDENCIES = {
        "software.amazon.awssdk.annotations-2.1.0.jar",
        "com.amazonaws.aws-lambda-java-core-1.1.0.jar",
    }

    FUNCTION_LOGICAL_ID = "Function"
    USING_GRADLE_PATH = os.path.join("Java", "gradle")
    USING_GRADLEW_PATH = os.path.join("Java", "gradlew")
    USING_GRADLEW_IN_CONTAINER_PATH = os.path.join("Java", "gradlew-in-container")
    USING_GRADLE_KOTLIN_PATH = os.path.join("Java", "gradle-kotlin")
    USING_MAVEN_PATH = os.path.join("Java", "maven")

    def _test_with_building_java(
        self,
        runtime,
        code_path,
        expected_files,
        expected_dependencies,
        use_container,
        relative_path,
        architecture=None,
    ):
        if use_container and (SKIP_DOCKER_TESTS or SKIP_DOCKER_BUILD):
            self.skipTest(SKIP_DOCKER_MESSAGE)

        # skip arm64 use container tests for gradlew and gradle-kotlin
        if (
            use_container
            and runtime != self.LATEST_JAVA_VERSION
            and architecture == ARM64
            and any(path in code_path for path in [self.USING_GRADLEW_PATH, self.USING_GRADLE_KOTLIN_PATH])
        ):
            self.skipTest(self.SKIP_ARM64_EARLIER_JAVA_TESTS)

        overrides = self.get_override(runtime, code_path, architecture, "aws.example.Hello::myHandler")
        if use_container and str(runtime).lower() in ["java21", "java25"] and self.USING_MAVEN_PATH not in code_path:
            container_env = "GRADLE_USER_HOME=/tmp/.gradle"
            mount_with = MountMode.WRITE
        else:
            container_env = None
            mount_with = None
        cmdlist = self.get_command_list(
            use_container=use_container,
            parameter_overrides=overrides,
            mount_with=mount_with,
            container_env_var=container_env,
        )
        cmdlist += ["--skip-pull-image"]
        if code_path == self.USING_GRADLEW_PATH and use_container and IS_WINDOWS:
            osutils.convert_to_unix_line_ending(os.path.join(self.test_data_path, self.USING_GRADLEW_PATH, "gradlew"))
        # Use shorter timeout in GitHub Actions to fail faster;
        # Putting 1800 because Windows Canary Instances takes longer
        timeout = 1800 if os.environ.get("BY_CANARY") else 180
        run_command(cmdlist, cwd=self.working_dir, timeout=timeout)

        self._verify_built_artifact(
            self.default_build_dir, self.FUNCTION_LOGICAL_ID, expected_files, expected_dependencies
        )

        self._verify_resource_property(
            str(self.built_template),
            "OtherRelativePathResource",
            "BodyS3Location",
            os.path.relpath(
                os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        self._verify_resource_property(
            str(self.built_template),
            "GlueResource",
            "Command.ScriptLocation",
            os.path.relpath(
                os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        # If we are testing in the container, invoke the function as well. Otherwise we cannot guarantee docker is on appveyor
        if use_container:
            expected = "Hello World"
            if not SKIP_DOCKER_TESTS:
                self._verify_invoke_built_function(
                    self.built_template,
                    self.FUNCTION_LOGICAL_ID,
                    self._make_parameter_override_arg(overrides),
                    expected,
                )

            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files, expected_modules):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)

        lib_dir_contents = set(os.listdir(str(resource_artifact_dir.joinpath("lib"))))
        self.assertEqual(lib_dir_contents, expected_modules)


@pytest.mark.python
class BuildIntegPythonBase(BuildIntegBase):
    EXPECTED_FILES = {
        "__init__.py",
        "main.py",
        "numpy",
        # 'cryptography',
    }

    FUNCTION_LOGICAL_ID = "Function"
    prop = "CodeUri"

    def _test_with_default_requirements(
        self,
        runtime,
        codeuri,
        use_container,
        relative_path,
        manifest="requirements.txt",
        do_override=True,
        check_function_only=False,
        architecture=None,
        beta_features=False,
    ):
        if use_container and (SKIP_DOCKER_TESTS or SKIP_DOCKER_BUILD):
            self.skipTest(SKIP_DOCKER_MESSAGE)
        overrides = self.get_override(runtime, codeuri, architecture, "main.handler") if do_override else None
        cmdlist = self.get_command_list(
            use_container=use_container, parameter_overrides=overrides, beta_features=beta_features
        )

        run_command(cmdlist, cwd=self.working_dir)

        self._verify_built_artifact(
            self.default_build_dir, self.FUNCTION_LOGICAL_ID, (self.EXPECTED_FILES | {manifest})
        )

        if not check_function_only:
            self._verify_resource_property(
                str(self.built_template),
                "OtherRelativePathResource",
                "BodyS3Location",
                os.path.relpath(
                    os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                    str(self.default_build_dir),
                ),
            )

            self._verify_resource_property(
                str(self.built_template),
                "GlueResource",
                "Command.ScriptLocation",
                os.path.relpath(
                    os.path.normpath(os.path.join(str(relative_path), "SomeRelativePath")),
                    str(self.default_build_dir),
                ),
            )

            self._verify_resource_property(
                str(self.built_template),
                "ExampleNestedStack",
                "TemplateURL",
                "https://s3.amazonaws.com/examplebucket/exampletemplate.yml",
            )

        expected = {"pi": "3.14"}
        if not SKIP_DOCKER_TESTS:
            self._verify_invoke_built_function(
                self.built_template,
                self.FUNCTION_LOGICAL_ID,
                self._make_parameter_override_arg(overrides) if do_override else None,
                expected,
            )
            if use_container:
                self.verify_docker_container_cleanedup(runtime)
                self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, self.prop, function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)

    def _get_python_version(self):
        """
        This method is used to override python version of some of the tests which is run with Makefile
        """
        return "python3.10"


@pytest.mark.provided
class BuildIntegProvidedBase(BuildIntegBase):
    EXPECTED_FILES_PROJECT_MANIFEST = {"__init__.py", "main.py", "requests", "requirements.txt"}

    FUNCTION_LOGICAL_ID = "Function"

    def _test_with_Makefile(
        self, runtime, use_container, manifest, architecture=None, code_uri="Provided", build_in_source=None
    ):
        if use_container and (SKIP_DOCKER_TESTS or SKIP_DOCKER_BUILD):
            self.skipTest(SKIP_DOCKER_MESSAGE)

        overrides = self.get_override(runtime, code_uri, architecture, "main.handler")
        manifest_path = None
        if manifest:
            manifest_path = os.path.join(self.test_data_path, "Provided", manifest)

        cmdlist = self.get_command_list(
            use_container=use_container,
            parameter_overrides=overrides,
            manifest_path=manifest_path,
            build_in_source=build_in_source,
        )

        # Built using Makefile for a python project.
        run_command(cmdlist, cwd=self.working_dir)

        if self.is_nested_parent:
            self._verify_built_artifact_in_subapp(
                self.default_build_dir, "SubApp", self.FUNCTION_LOGICAL_ID, self.EXPECTED_FILES_PROJECT_MANIFEST
            )
        else:
            self._verify_built_artifact(
                self.default_build_dir, self.FUNCTION_LOGICAL_ID, self.EXPECTED_FILES_PROJECT_MANIFEST
            )

        expected = "2.23.0"
        # Building was done with a makefile, but invoke should be checked with corresponding python image.
        overrides["Runtime"] = self._get_python_version()
        if not SKIP_DOCKER_TESTS:
            self._verify_invoke_built_function(
                self.built_template, self.FUNCTION_LOGICAL_ID, self._make_parameter_override_arg(overrides), expected
            )
        if use_container:
            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)

    def _verify_built_artifact_in_subapp(self, build_dir, subapp_path, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")
        subapp_build_dir = Path(build_dir, subapp_path)
        self.assertTrue(subapp_build_dir.exists(), f"Build directory for sub app {subapp_path} should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)

        subapp_build_dir_files = os.listdir(str(subapp_build_dir))
        self.assertIn("template.yaml", subapp_build_dir_files)
        self.assertIn(function_logical_id, subapp_build_dir_files)

        template_path = subapp_build_dir.joinpath("template.yaml")
        resource_artifact_dir = subapp_build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)

    def _get_python_version(self):
        return "python{}.{}".format(sys.version_info.major, sys.version_info.minor)


class DedupBuildIntegBase(BuildIntegBase):
    def _verify_build_and_invoke_functions(self, expected_messages, command_result, overrides):
        self._verify_process_code_and_output(command_result)
        for expected_message in expected_messages:
            expected = f"Hello {expected_message}"
            function_id = f"Hello{expected_message}Function"
            self._verify_build_artifact(self.default_build_dir, function_id)
            if not SKIP_DOCKER_TESTS:
                self._verify_invoke_built_function(self.built_template, function_id, overrides, expected)

    def _verify_build_artifact(self, build_dir, function_logical_id):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)

        # confirm function logical id is in the built template
        template_dict = {}
        with open(Path(build_dir).joinpath("template.yaml"), "r") as template_file:
            template_dict = yaml_parse(template_file.read())
        self.assertIn(function_logical_id, template_dict.get("Resources", {}).keys())

        # confirm build folder for the function exist in the build directory
        built_folder = (
            template_dict.get("Resources", {}).get(function_logical_id, {}).get("Properties", {}).get("CodeUri")
        )
        if not built_folder:
            built_folder = (
                template_dict.get("Resources", {}).get(function_logical_id, {}).get("Properties", {}).get("ContentUri")
            )
        self.assertIn(built_folder, build_dir_files)

    def _verify_process_code_and_output(self, command_result):
        self.assertEqual(command_result.process.returncode, 0)
        # check HelloWorld and HelloMars functions are built in the same build
        self.assertRegex(
            command_result.stderr.decode("utf-8"),
            "Building codeuri: .* runtime: .* functions: " "HelloWorldFunction, HelloMarsFunction",
        )


class CachedBuildIntegBase(DedupBuildIntegBase):
    def _verify_cached_artifact(self, cache_dir):
        self.assertTrue(cache_dir.exists(), "Cache directory should be created")


class NestedBuildIntegBase(BuildIntegBase):
    def _verify_build(self, function_full_paths, stack_paths, command_result):
        self._verify_process_code_and_output(command_result, function_full_paths)
        for function_full_path in function_full_paths:
            self._verify_build_artifact(self.default_build_dir, function_full_path)
        for stack_path in stack_paths:
            self._verify_move_template(self.default_build_dir, stack_path)

    def _verify_build_artifact(self, build_dir, function_full_path):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        # full_path is always posix path
        path_components = posixpath.split(function_full_path)
        artifact_path = Path(build_dir, *path_components)
        self.assertTrue(artifact_path.exists())

    def _verify_move_template(self, build_dir, stack_path):
        path_components = posixpath.split(stack_path)
        stack_build_dir_path = Path(build_dir, Path(*path_components), "template.yaml")
        self.assertTrue(stack_build_dir_path.exists())

    def _verify_process_code_and_output(self, command_result, function_full_paths):
        self.assertEqual(command_result.process.returncode, 0)
        # check HelloWorld and HelloMars functions are built in the same build
        for function_full_path in function_full_paths:
            self.assertRegex(
                command_result.stderr.decode("utf-8"),
                f"Building codeuri: .* runtime: .* functions: .*{function_full_path}.*",
            )

    def _verify_invoke_built_functions(self, template_path, overrides, function_and_expected):
        for function_identifier, expected in function_and_expected:
            self._verify_invoke_built_function(template_path, function_identifier, overrides, expected)


class IntrinsicIntegBase(BuildIntegBase):
    """
    Currently sam-cli does not have great support for intrinsic functions,
    in this kind of integ tests, there are functions that are buildable but not invocable.
    """

    def _verify_build(self, function_full_paths, layer_full_path, stack_paths, command_result):
        """
        Verify resources have their build artifact folders, stack has their own template.yaml, and command succeeds.
        """
        self._verify_process_code_and_output(command_result, function_full_paths, layer_full_path)
        for function_full_path in function_full_paths:
            self._verify_build_artifact(self.default_build_dir, function_full_path)
        for stack_path in stack_paths:
            self._verify_move_template(self.default_build_dir, stack_path)

    def _verify_build_artifact(self, build_dir, function_full_path):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        # full_path is always posix path
        path_components = posixpath.split(function_full_path)
        artifact_path = Path(build_dir, *path_components)
        self.assertTrue(artifact_path.exists())

    def _verify_move_template(self, build_dir, stack_path):
        path_components = posixpath.split(stack_path)
        stack_build_dir_path = Path(build_dir, Path(*path_components), "template.yaml")
        self.assertTrue(stack_build_dir_path.exists())

    def _verify_process_code_and_output(self, command_result, function_full_paths, layer_full_path):
        self.assertEqual(command_result.process.returncode, 0)
        # check HelloWorld and HelloMars functions are built in the same build
        for function_full_path in function_full_paths:
            self.assertRegex(
                command_result.stderr.decode("utf-8"),
                f"Building codeuri: .* runtime: .* functions:.*{function_full_path}.*",
            )
        self.assertIn(
            f"Building layer '{layer_full_path}'",
            command_result.stderr.decode("utf-8"),
        )

    def _verify_invoke_built_functions(self, template_path, functions, error_message):
        """
        Invoke the function, if error_message is not None, the invoke should fail.
        """
        for function_logical_id in functions:
            LOG.info("Invoking built function '{}'".format(function_logical_id))

            cmdlist = [
                self.cmd,
                "local",
                "invoke",
                function_logical_id,
                "-t",
                str(template_path),
                "--no-event",
            ]

            process_execute = run_command(cmdlist)
            process_execute.process.wait()

            process_stderr = process_execute.stderr.decode("utf-8")
            if error_message:
                self.assertNotEqual(0, process_execute.process.returncode)
                self.assertIn(error_message, process_stderr)
            else:
                self.assertEqual(0, process_execute.process.returncode)


def rust_parameterized_class(cls):
    """
    common class parameterize for rust integration tests
    """
    cls = parameterized_class(
        ("template", "code_uri", "binary", "expected_invoke_result"),
        [
            (
                "template_build_method_rust_single_function.yaml",
                "Rust/single-function",
                None,
                {"req_id": "34", "msg": "Hello World"},
            ),
            (
                "template_build_method_rust_binary.yaml",
                "Rust/multi-binaries",
                "function_a",
                {"req_id": "63", "msg": "Hello FunctionA"},
            ),
            (
                "template_build_method_rust_binary.yaml",
                "Rust/multi-binaries",
                "function_b",
                {"req_id": "99", "msg": "Hello FunctionB"},
            ),
            (
                "template_build_method_rust_cargo_workspace.yaml",
                "Rust/cargo_workspace/function_b",
                None,
                {"req_id": "281", "msg": "Hello World B"},
            ),
        ],
    )(cls)
    return cls


@pytest.mark.provided
class BuildIntegRustBase(BuildIntegBase):
    FUNCTION_LOGICAL_ID = "Function"
    EXPECTED_FILES_PROJECT_MANIFEST = {"bootstrap"}

    def setUp(self):
        super().setUp()
        # Copy source code to working_dir to allow tests run in parallel, as Cargo Lambda generates artifacts in source code dir
        osutils.copytree(
            Path(self.template_path).parent.joinpath(self.code_uri),
            Path(self.working_dir).joinpath(self.code_uri),
        )
        # copy template path
        tmp_template_path = Path(self.working_dir).joinpath(self.template)
        shutil.copyfile(Path(self.template_path), tmp_template_path)
        self.template_path = str(tmp_template_path)

    def _test_with_rust_cargo_lambda(
        self,
        runtime,
        code_uri,
        handler="bootstap",
        binary=None,
        build_mode=None,
        architecture=None,
        use_container=False,
        expected_invoke_result=None,
    ):
        if use_container and (SKIP_DOCKER_TESTS or SKIP_DOCKER_BUILD):
            self.skipTest(SKIP_DOCKER_MESSAGE)
        overrides = self.get_override(runtime, code_uri, architecture, handler)
        if binary:
            overrides["Binary"] = binary
        cmdlist = self.get_command_list(use_container=use_container, parameter_overrides=overrides, beta_features=True)

        newenv = os.environ.copy()
        if build_mode:
            newenv["SAM_BUILD_MODE"] = build_mode

        run_command(cmdlist, cwd=self.working_dir, env=newenv)

        self._verify_built_artifact(
            self.default_build_dir, self.FUNCTION_LOGICAL_ID, self.EXPECTED_FILES_PROJECT_MANIFEST
        )

        if expected_invoke_result and not SKIP_DOCKER_TESTS and architecture == X86_64:
            # ARM64 is not supported yet for local invoke
            self._verify_invoke_built_function(
                self.built_template,
                self.FUNCTION_LOGICAL_ID,
                self._make_parameter_override_arg(overrides),
                expected_invoke_result,
            )

        if use_container:
            self.verify_docker_container_cleanedup(runtime)
            self.verify_pulled_image(runtime, architecture)

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)


@pytest.mark.dotnet
class BuildIntegDotnetBase(BuildIntegBase):
    FUNCTION_LOGICAL_ID = "Function"
    EXPECTED_FILES_PROJECT_MANIFEST = {
        "Amazon.Lambda.APIGatewayEvents.dll",
        "Amazon.Lambda.Core.dll",
        "HelloWorld.runtimeconfig.json",
        "Amazon.Lambda.Serialization.Json.dll",
        "Newtonsoft.Json.dll",
        "HelloWorld.deps.json",
        "HelloWorld.dll",
    }

    EXPECTED_FILES_PROJECT_MANIFEST_PROVIDED = {
        "bootstrap",
    }

    def validate_build_command(self, overrides, mode, mount_mode=None, use_container=False, input=None) -> None:
        do_use_container = mount_mode != None or use_container

        cmdlist = self.get_command_list(
            use_container=do_use_container, parameter_overrides=overrides, mount_with=mount_mode
        )

        if do_use_container:
            cmdlist += ["--container-env-var", "DOTNET_CLI_HOME=/tmp/dotnet"]
            cmdlist += ["--container-env-var", "XDG_DATA_HOME=/tmp/xdg"]

        LOG.info("Running with SAM_BUILD_MODE={}".format(mode))

        newenv = os.environ.copy()
        if mode:
            newenv["SAM_BUILD_MODE"] = mode

        if input:
            command_result = run_command_with_input(cmdlist, input.encode(), cwd=self.working_dir, env=newenv)
        else:
            command_result = run_command(cmdlist, cwd=self.working_dir, env=newenv)

        self.assertEqual(command_result.process.returncode, 0)

    def validate_invoke_command(self, overrides, runtime):
        expected = "{'message': 'Hello World'}"

        self._verify_invoke_built_function(
            self.built_template, self.FUNCTION_LOGICAL_ID, self._make_parameter_override_arg(overrides), expected
        )
        self.verify_docker_container_cleanedup(runtime)

    def validate_build_artifacts(self, expected_manifest):
        self._verify_built_artifact(
            self.default_build_dir,
            self.FUNCTION_LOGICAL_ID,
            expected_manifest,
        )

        self._verify_built_properties()

    def _verify_built_properties(self):
        self._verify_resource_property(
            str(self.built_template),
            "OtherRelativePathResource",
            "BodyS3Location",
            os.path.relpath(
                os.path.normpath(os.path.join(str(self.test_data_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

        self._verify_resource_property(
            str(self.built_template),
            "GlueResource",
            "Command.ScriptLocation",
            os.path.relpath(
                os.path.normpath(os.path.join(str(self.test_data_path), "SomeRelativePath")),
                str(self.default_build_dir),
            ),
        )

    def _verify_built_artifact(self, build_dir, function_logical_id, expected_files):
        self.assertTrue(build_dir.exists(), "Build directory should be created")

        build_dir_files = os.listdir(str(build_dir))
        self.assertIn("template.yaml", build_dir_files)
        self.assertIn(function_logical_id, build_dir_files)

        template_path = build_dir.joinpath("template.yaml")
        resource_artifact_dir = build_dir.joinpath(function_logical_id)

        # Make sure the template has correct CodeUri for resource
        self._verify_resource_property(str(template_path), function_logical_id, "CodeUri", function_logical_id)

        all_artifacts = set(os.listdir(str(resource_artifact_dir)))
        actual_files = all_artifacts.intersection(expected_files)
        self.assertEqual(actual_files, expected_files)