StepFunctions: tasks with apigateway:invoke fail without issuing an HTTP request #10163

States.Runtime | TypeError(FailureEvent.__init__() missing 1 required positional argument: 'env')

import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import boto3
from moto import mock_aws

RECEIVED = []


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def do_POST(self):
        length = int(self.headers.get("Content-Length") or 0)
        body = self.rfile.read(length) if length else b""
        RECEIVED.append({"path": self.path, "body": body.decode()})
        payload = b'{"ok": true}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
PORT = server.server_address[1]
threading.Thread(target=server.serve_forever, daemon=True).start()

DEFN = {
    "StartAt": "Call",
    "States": {
        "Call": {
            "Type": "Task",
            "Resource": "arn:aws:states:::apigateway:invoke",
            "Parameters": {
                "ApiEndpoint": f"http://127.0.0.1:{PORT}",
                "Method": "POST",
                "Stage": "prod",
                "Path": "/things",
                "QueryParameters": {"mode": ["fast"]},
                "RequestBody": {"hello": "world"},
                "Headers": {"Content-Type": ["application/json"]},
            },
            "End": True,
        }
    },
}


@mock_aws(config={"stepfunctions": {"execute_state_machine": True}})
def run():
    sfn = boto3.client("stepfunctions", region_name="us-east-1")
    arn = sfn.create_state_machine(
        name="apigw-repro",
        definition=json.dumps(DEFN),
        roleArn="arn:aws:iam::123456789012:role/sf",
    )["stateMachineArn"]

    ex = sfn.start_execution(stateMachineArn=arn, input="{}")["executionArn"]
    for _ in range(50):
        d = sfn.describe_execution(executionArn=ex)
        if d["status"] != "RUNNING":
            break
        time.sleep(0.2)

    print("status :", d["status"])
    print("error  :", d.get("error"))
    print("cause  :", d.get("cause"))
    print("requests received by the API:", RECEIVED)

    assert d["status"] == "SUCCEEDED", d.get("cause")
    assert RECEIVED, "the task never issued an HTTP request"
    assert RECEIVED[0]["path"] == "/prod/_user_request_/things/?mode=fast", RECEIVED[0]["path"]
    assert json.loads(RECEIVED[0]["body"]) == {"hello": "world"}, RECEIVED[0]["body"]
    print("\nOK")


run()