Maintenance: Refactor E2E mechanism for extensibility, ease test writing, and unblock integ tests

[heitorlessa]

Summary

We defined the E2E mechanism on #1226 and recently implemented a POC. This issue tracks the extensibility work required to allow E2E to own infrastructure. This includes making it easier for tests to be able to control the payload each E2E test uses to invoke their respective Lambda function.

Tasks

• Refactor E2E: Encapsulate CDK Assets logic #1436
• Refactor E2E: Allow tests to own Infrastructure and invoke payload #1437
• Refactor E2E: Safely parallelize tests defined within feature group #1438
• Refactor E2E: Refactor Metrics test and increase coverage #1439
• Refactor E2E: Refactor Tracer test and increase coverage #1440
• Refactor E2E: Refactor Logger test and increase coverage #1441
• Refactor E2E: Build Lambda Layer once and expose to children stacks #1464

Why is this needed?

When trying to increase E2E coverage, we identified a gap in the mechanism as it doesn't allow each feature group (e.g., Metrics) to customize their own infrastructure.

This made it difficult to add E2E tests for Idempotency where we wanted to (1) create a DynamoDB table and reference it in Lambda functions env var, and (2) send different payloads to test idempotency results.

Once this is complete, we can resume increasing coverage for other features, including defining test strategies (when to use which) and resume integration tests.

Which area does this relate to?

Tests

Solution

This is how it looks like on my fork using the new refactoring (one or more PRs depending on final size).

Metrics infrastructure

tests/e2e/metrics/infrastructure.py fully controls what infrastructure Metrics should have. It keeps the original mechanism of creating Lambda functions within the handlers directory but now more explicit. It also allows to override any CDK Lambda function prop.

from pathlib import Path

from tests.e2e.utils.infrastructure import BaseInfrastructureV2


class MetricsStack(BaseInfrastructureV2):
    def __init__(self, handlers_dir: Path, feature_name: str = "metrics") -> None:
        super().__init__(feature_name, handlers_dir)

    def create_resources(self):
        self.create_lambda_functions()

Metrics infrastructure parallelization

tests/e2e/metrics/conftest.py handles infrastructure deployment so tests can run in parallel after deployment. It now handles stack deletion in case of failures too.

import json
from pathlib import Path
from typing import Dict

import pytest
from _pytest import fixtures
from filelock import FileLock

from tests.e2e.metrics.infrastructure import MetricsStack


@pytest.fixture(autouse=True, scope="module")
def infrastructure(request: fixtures.SubRequest, tmp_path_factory: pytest.TempPathFactory, worker_id) -> MetricsStack:
    """Setup and teardown logic for E2E test infrastructure

    Parameters
    ----------
    request : fixtures.SubRequest
        test fixture containing metadata about test execution

    Returns
    -------
    MetricsStack
        Metrics Stack to deploy infrastructure

    Yields
    ------
    Iterator[MetricsStack]
        Deployed Infrastructure
    """
    stack = MetricsStack(handlers_dir=Path(f"{request.fspath.dirname}/handlers"))

    # NOTE: This will be encapsulated as it's reusable
    try:
        if worker_id == "master":
            # no parallelization, deploy stack and let fixture be cached
            yield stack.deploy()
        else:
            # tmp dir shared by all workers
            root_tmp_dir = tmp_path_factory.getbasetemp().parent

            cache = root_tmp_dir / "cache.json"
            with FileLock(f"{cache}.lock"):
                # If cache exists, return stack outputs back
                # otherwise it's the first run by the main worker
                # deploy and return stack outputs so subsequent workers can reuse
                if cache.is_file():
                    stack_outputs = json.loads(cache.read_text())
                else:
                    stack_outputs: Dict = stack.deploy()
                    cache.write_text(json.dumps(stack_outputs))
            yield stack_outputs
    finally:
        stack.delete()

Metrics test including cold start

tests/e2e/metrics/test_metrics.py now uses CDK Stack outputs as separate fixtures for each function ARN. It creates a standard for outputs so any function will become two outputs in PascalCase: Name and NameArn. Additional helpers were added to make explicit tests easier to create. The main focus is ease of writing tests and allowing any new or existing contributor to understand what's going on.

Functions are fully isolated which allows us to safely parallelize tests - Before ~175s and now it is ~94s even with an additional Lambda function (ColdStart).

import json

import pytest

from tests.e2e.utils import helpers


@pytest.fixture
def basic_handler_fn(infrastructure: dict) -> str:
    return infrastructure.get("BasicHandler", "")


@pytest.fixture
def basic_handler_fn_arn(infrastructure: dict) -> str:
    return infrastructure.get("BasicHandlerArn", "")


@pytest.fixture
def cold_start_fn(infrastructure: dict) -> str:
    return infrastructure.get("ColdStart", "")


@pytest.fixture
def cold_start_fn_arn(infrastructure: dict) -> str:
    return infrastructure.get("ColdStartArn", "")


METRIC_NAMESPACE = "powertools-e2e-metric"


def test_basic_lambda_metric_is_visible(basic_handler_fn: str, basic_handler_fn_arn: str):
    # GIVEN
    metric_name = helpers.build_metric_name()
    service = helpers.build_service_name()
    dimensions = helpers.build_add_dimensions_input(service=service)
    metrics = helpers.build_multiple_add_metric_input(metric_name=metric_name, value=1, quantity=3)

    # WHEN
    event = json.dumps({"metrics": metrics, "service": service, "namespace": METRIC_NAMESPACE})
    _, execution_time = helpers.trigger_lambda(lambda_arn=basic_handler_fn_arn, payload=event)

    metrics = helpers.get_metrics(
        namespace=METRIC_NAMESPACE, start_date=execution_time, metric_name=metric_name, dimensions=dimensions
    )

    # THEN
    metric_data = metrics.get("Values", [])
    assert metric_data and metric_data[0] == 3.0


def test_cold_start_metric(cold_start_fn_arn: str, cold_start_fn: str):
    # GIVEN
    metric_name = "ColdStart"
    service = helpers.build_service_name()
    dimensions = helpers.build_add_dimensions_input(function_name=cold_start_fn, service=service)

    # WHEN
    event = json.dumps({"service": service, "namespace": METRIC_NAMESPACE})
    _, execution_time = helpers.trigger_lambda(lambda_arn=cold_start_fn_arn, payload=event)

    metrics = helpers.get_metrics(
        namespace=METRIC_NAMESPACE, start_date=execution_time, metric_name=metric_name, dimensions=dimensions
    )

    # THEN
    metric_data = metrics.get("Values", [])
    assert metric_data and metric_data[0] == 1.0

Acknowledgment

• This request meets Lambda Powertools Tenets
• Should this be considered in other Lambda Powertools languages? i.e. Java, TypeScript

[heitorlessa]

@mploski turns out the PR it's much smaller than I thought - ~800 LOC addition, ~300 deletion. Given that you're mostly familiar with it, I'm inclined to make a single PR instead of 3 smaller ones (more coordination).

I'll create a proper description with the changes and send a PR tomorrow for review before you go on holidays

[heitorlessa]

Final result with the new parallelization and Lambda Layer being built once. Last step is to enable at CI level and measure whether retry/jitter numbers are sufficient for the random hardware we might get.

graph TD
    A[make e2e test] -->Spawn{"Split and group tests <br>by feature and CPU"}

    Spawn -->|Worker0| Worker0_Start["Load tests"]
    Spawn -->|Worker1| Worker1_Start["Load tests"]
    Spawn -->|WorkerN| WorkerN_Start["Load tests"]

    Worker0_Start -->|Wait| LambdaLayerStack["Lambda Layer Stack Deployment"]
    Worker1_Start -->|Wait| LambdaLayerStack["Lambda Layer Stack Deployment"]
    WorkerN_Start -->|Wait| LambdaLayerStack["Lambda Layer Stack Deployment"]

    LambdaLayerStack -->|Worker0| Worker0_Deploy["Launch feature stack"]
    LambdaLayerStack -->|Worker1| Worker1_Deploy["Launch feature stack"]
    LambdaLayerStack -->|WorkerN| WorkerN_Deploy["Launch feature stack"]

    Worker0_Deploy -->|Worker0| Worker0_Tests["Run tests"]
    Worker1_Deploy -->|Worker1| Worker1_Tests["Run tests"]
    WorkerN_Deploy -->|WorkerN| WorkerN_Tests["Run tests"]

    Worker0_Tests --> ResultCollection
    Worker1_Tests --> ResultCollection
    WorkerN_Tests --> ResultCollection

    ResultCollection{"Wait for workers<br/>Collect test results"}
    ResultCollection --> TestEnd["Report results"]
    ResultCollection --> DeployEnd["Delete Stacks"]

Loading