test(extras): add an end-to-end test for the microvm operation

packages/aws-durable-execution-sdk-python-extras/e2e/.gitignore

@@ -0,0 +1 @@
+build/

packages/aws-durable-execution-sdk-python-extras/e2e/README.md

@@ -0,0 +1,68 @@
+# End-to-end test for the microvm operation
+
+`run_e2e.py` tests the `microvm` operation and the MicroVM worker against a real AWS account. It builds both from this repository, so a run tests the branch you are on.
+
+The scenarios and the durable function's events are the same as those of the JavaScript SDK's `extras/e2e`. So the two SDKs are checked against one list of cases.
+
+## What it does
+
+1. Builds the core SDK, this package, and the MicroVM worker as wheels.
+2. Creates or reuses an S3 bucket and three IAM roles: one for the image build, one for the MicroVM, and one for the function. Every resource is named with the `dex-microvm-e2e-py` prefix.
+3. Builds a MicroVM image from `microvm/` with the worker, unless one with the same sources exists.
+4. Deploys `function/index.py` as a durable function on Python 3.13, arm64, and publishes a version.
+5. Starts one durable execution per scenario, and checks its outcome, its operation history, and that every MicroVM it launched was terminated.
+
+## Running it
+
+You need AWS credentials for the target account, boto3 1.43.35 or later, and `uv` or `pip`. From the repository root:
+
+```
+python packages/aws-durable-execution-sdk-python-extras/e2e/run_e2e.py
+```
+
+| Variable | Effect |
+|---|---|
+| `AWS_REGION` | The Region. Defaults to `us-east-1`. |
+| `E2E_SCENARIOS` | A comma-separated list of scenario IDs. It also selects optional and pending scenarios. |
+| `E2E_MAX_MICROVMS` | How many MicroVMs run at once. Defaults to 4: a new account may allocate 8 GB of MicroVM memory, and each MicroVM has 2 GB. A scenario starts only when its MicroVMs fit. |
+| `E2E_IMAGE_ARN` | Use this MicroVM image instead of building one. For example, an image with the JavaScript worker checks the operation against both workers. |
+| `E2E_FUNCTION_ARN` | Use this published function version instead of deploying one. |
+| `E2E_SUB_TYPES` | `custom` expects the JavaScript operations' own subtypes, such as `MicrovmLaunch`. By default the runner expects the core subtypes that the Python operation records today. |
+
+The runner prints one JSON report per scenario and exits with 1 when any scenario fails.
+
+## Scenarios
+
+| ID | Checks |
+|---|---|
+| `succeed` | A job in the `run` hook returns its result. |
+| `fail` | A failing job raises `MicrovmJobFailedError` with the job's message. |
+| `crash` | A MicroVM that exits mid-job fails on the 15-second heartbeat timeout, not the 900-second timeout. |
+| `too-large-for-the-run-hook` | A job past the `run` hook limit fails with `MicrovmDeliveryError` before the callback exists. |
+| `run-hook-cjk`, `run-hook-emoji` | 3,400 code points of multi-byte text fit the `run` hook, which counts code points, and arrive unchanged. |
+| `parallel` | Two `microvm` calls in a `map` launch two MicroVMs. |
+| `long-job` | Optional. A 10-minute job keeps its MicroVM running. |
+
+The checks that apply to every scenario: the history has the `.callback`, `.launch`, and `.terminate` operations with their subtypes, and every launched MicroVM ends `TERMINATED`.
+
+The JavaScript operations record their own subtypes, such as `Microvm` and `MicrovmLaunch`. The Python core SDK accepts only its own subtypes until issue #749 lands. So the Python operation records `RunInChildContext`, `Callback`, and `Step`, and the runner expects those unless `E2E_SUB_TYPES=custom`.
+
+### Pending scenarios
+
+These need changes that have not landed yet. The runner skips them unless `E2E_SCENARIOS` names them.
+
+| ID | Needs |
+|---|---|
+| `terminated` | The worker's terminate action, which fails the callback when the MicroVM is terminated mid-job. |
+| `request-succeed`, `request-fail`, `auto-http`, `auto-http-cjk` | Delivery over HTTP. When it lands, remove `too-large-for-the-run-hook`. |
+| `session`, `suspend-session`, `idle-probe` | `microvm_session`, and the worker's suspend and resume. |
+
+When a change lands, remove `pending` from its scenarios in `run_e2e.py`, and add the scenario's branch to `function/index.py`.
+
+## Error types
+
+The Python SDK reports the fully-qualified class name as the execution's error type, such as `aws_durable_execution_sdk_python_extras.microvm.errors.MicrovmJobFailedError`. The JavaScript SDK reports `MicrovmJobFailedError`. The runner compares the class name alone, so a scenario passes with either SDK's function.
+
+## Cleaning up
+
+The runner keeps its resources for the next run. A MicroVM image costs nothing while no MicroVM runs. To remove everything, delete the function `dex-microvm-e2e-py-fn`, the images named `dex-microvm-e2e-py-*`, the three roles named `dex-microvm-e2e-py-*`, and the bucket `dex-microvm-e2e-py-<account>-<region>`.

packages/aws-durable-execution-sdk-python-extras/e2e/function/index.py

@@ -0,0 +1,82 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+"""The durable function that the end-to-end test deploys.
+
+Each event selects one scenario, so one deployment covers every case. The
+events are the same as those of the JavaScript SDK's ``extras/e2e``. So the two
+runners share their scenarios.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from aws_durable_execution_sdk_python.concurrency.models import BatchResult
+from aws_durable_execution_sdk_python.config import Duration
+from aws_durable_execution_sdk_python.context import DurableContext
+from aws_durable_execution_sdk_python.execution import durable_execution
+
+from aws_durable_execution_sdk_python_extras.microvm import MicrovmConfig, microvm
+
+
+# The scenarios that need an operation this package does not have yet. The
+# runner skips them unless E2E_SCENARIOS selects them.
+NOT_AVAILABLE = {
+    "session": "microvm_session",
+    "suspend-session": "microvm_session",
+    "idle-session": "microvm_session",
+}
+
+
+@durable_execution
+def handler(event: dict[str, Any], context: DurableContext) -> Any:
+    scenario = event["scenario"]
+    if scenario in NOT_AVAILABLE:
+        msg = f"the {scenario} scenario needs {NOT_AVAILABLE[scenario]}"
+        raise NotImplementedError(msg)
+    if event.get("requestPath") is not None:
+        msg = "delivery over HTTP is not available yet"
+        raise NotImplementedError(msg)
+
+    heartbeat_timeout = event.get("heartbeatTimeoutSeconds")
+    config = MicrovmConfig(
+        image_identifier=_require_env("MICROVM_IMAGE_ARN"),
+        execution_role_arn=_require_env("MICROVM_ROLE_ARN"),
+        timeout=Duration.from_seconds(event["timeoutSeconds"]),
+        heartbeat_timeout=(
+            None
+            if heartbeat_timeout is None
+            else Duration.from_seconds(heartbeat_timeout)
+        ),
+    )
+    job = dict(event["job"])
+    if event.get("paddingLength"):
+        job["padding"] = (event.get("paddingChar") or "x") * event["paddingLength"]
+
+    if scenario == "parallel":
+        # Each map item is a child context. The operation takes the item's
+        # context, so each job runs inside its item.
+        batch: BatchResult[Any] = context.map(
+            ["a", "b"],
+            lambda item_context, label, _index, _items: microvm(
+                item_context,
+                f"job-{label}",
+                job_input={**job, "label": label},
+                config=config,
+            ),
+            name="jobs",
+        )
+        batch.throw_if_error()
+        return batch.get_results()
+
+    return microvm(context, "job", job_input=job, config=config)
+
+
+def _require_env(name: str) -> str:
+    value = os.environ.get(name)
+    if not value:
+        msg = f"{name} is not set"
+        raise RuntimeError(msg)
+    return value

packages/aws-durable-execution-sdk-python-extras/e2e/microvm/Dockerfile

@@ -0,0 +1,8 @@
+FROM public.ecr.aws/docker/library/python:3.13-slim
+WORKDIR /app
+# The runner builds the worker from this repository and adds its wheel.
+COPY *.whl app.py ./
+RUN pip install --no-cache-dir ./*.whl
+EXPOSE 8080
+# Lambda snapshots the running process after it starts.
+CMD ["python", "app.py"]

packages/aws-durable-execution-sdk-python-extras/e2e/microvm/app.py

@@ -0,0 +1,137 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+"""The application inside the MicroVM image that the end-to-end test builds.
+
+Lambda starts it at image build and snapshots it. The run hook then delivers
+each job. The routes and the results are the same as those of the JavaScript
+SDK's ``extras/e2e/microvm-app.ts``.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import platform
+import threading
+import time
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from aws_durable_execution_sdk_python_microvm_worker import (
+    MicrovmJobContext,
+    start_microvm_worker,
+)
+
+
+SESSION_DIR = Path("/tmp/session")  # noqa: S108
+STARTED = time.monotonic()
+
+# Only the process memory holds this value. A job that reads it back after a
+# suspend shows that the resume restored the memory.
+remembered: dict[str, str] = {}
+
+# Counts the resume lifecycle hooks that the worker logged. A job that reports
+# a count above zero shows that the service called the hook after the worker
+# suspended its own MicroVM.
+resume_hooks = 0
+
+
+class CountingLogger:
+    """The worker's logger, which also counts the resume hooks."""
+
+    def __init__(self, logger: logging.Logger) -> None:
+        self._logger = logger
+
+    def info(
+        self, msg: object, *args: object, extra: Mapping[str, object] | None = None
+    ) -> None:
+        global resume_hooks  # noqa: PLW0603
+        if msg == "lifecycle hook" and (extra or {}).get("hook") == "resume":
+            resume_hooks += 1
+        self._logger.info(msg, *args, extra=extra)
+
+    def warning(
+        self, msg: object, *args: object, extra: Mapping[str, object] | None = None
+    ) -> None:
+        self._logger.warning(msg, *args, extra=extra)
+
+    def error(
+        self, msg: object, *args: object, extra: Mapping[str, object] | None = None
+    ) -> None:
+        self._logger.error(msg, *args, extra=extra)
+
+
+def run_job(job: dict[str, Any], context: MicrovmJobContext) -> dict[str, Any]:
+    context.logger.info("job received")
+    if job.get("mode") == "crash":
+        # Exits after the first heartbeat. The durable function must then fail
+        # on the heartbeat timeout, well before the callback timeout.
+        threading.Timer(2.0, lambda: os._exit(1)).start()
+        threading.Event().wait()
+    context.cancelled.wait(job.get("sleepSeconds", 0))
+    if job.get("mode") == "fail":
+        msg = f"job {job.get('label')} failed on purpose"
+        raise RuntimeError(msg)
+    return {
+        "label": job.get("label"),
+        "microvmId": context.microvm_id,
+        "python": platform.python_version(),
+        # The length of the job as JSON, in characters, as the JavaScript app
+        # measures it.
+        "inputLength": len(json.dumps(job, separators=(",", ":"), ensure_ascii=False)),
+        # Echoed so that a test can compare what the MicroVM received.
+        "padding": job.get("padding"),
+        "finishedAt": datetime.now(UTC).isoformat(),
+    }
+
+
+# A session writes a file with one job and reads it with the next.
+def write(job: dict[str, Any], context: MicrovmJobContext) -> dict[str, Any]:
+    SESSION_DIR.mkdir(parents=True, exist_ok=True)
+    (SESSION_DIR / job["key"]).write_text(job["value"])
+    return {"microvmId": context.microvm_id}
+
+
+def read(job: dict[str, Any], context: MicrovmJobContext) -> dict[str, Any]:
+    return {
+        "microvmId": context.microvm_id,
+        "value": (SESSION_DIR / job["key"]).read_text(),
+    }
+
+
+def remember(job: dict[str, Any], context: MicrovmJobContext) -> dict[str, Any]:
+    remembered.update(value=job["value"], storedAt=datetime.now(UTC).isoformat())
+    return {"microvmId": context.microvm_id, "pid": os.getpid()}
+
+
+def recall(_job: dict[str, Any], context: MicrovmJobContext) -> dict[str, Any]:
+    return {
+        "microvmId": context.microvm_id,
+        "pid": os.getpid(),
+        "uptimeSeconds": round(time.monotonic() - STARTED),
+        "resumeHooks": resume_hooks,
+        **remembered,
+    }
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    # The worker runs `handler` for a job that names no route: a small job
+    # arrives in the run hook, and a large one over HTTP on the worker's
+    # default job path. The /job route receives jobs that name it.
+    worker = start_microvm_worker(
+        handler=run_job,
+        routes={
+            "/job": run_job,
+            "/write": write,
+            "/read": read,
+            "/remember": remember,
+            "/recall": recall,
+        },
+        logger=CountingLogger(logging.getLogger("microvm-worker")),
+    )
+    worker.wait()