def METHOD_NAME():
    flow, output_name = generate_data_ingestion_flow_from_s3_input(
        "test-s3",
        s3_uri="test_uri",
        s3_content_type="csv",
        s3_has_header=True,
        operator_version="0.2",
        schema={"k1": "string", "k2": "long"},
    )
    assert flow["metadata"] == {"version": 1, "disable_limits": False}
    assert len(flow["nodes"]) == 2
    assert output_name.endswith(flow["nodes"][1]["outputs"][0]["name"])
    # reset node id
    flow["nodes"][0]["node_id"] = "123456"
    flow["nodes"][1]["node_id"] = "7891011"
    flow["nodes"][1]["inputs"][0]["node_id"] = "123456"
    assert flow["nodes"] == [
        {
            "node_id": "123456",
            "type": "SOURCE",
            "operator": "sagemaker.s3_source_0.2",
            "parameters": {
                "dataset_definition": {
                    "datasetSourceType": "S3",
                    "name": "test-s3",
                    "s3ExecutionContext": {
                        "s3Uri": "test_uri",
                        "s3ContentType": "csv",
                        "s3HasHeader": True,
                    },
                }
            },
            "inputs": [],
            "outputs": [{"name": "default"}],
        },
        {
            "node_id": "7891011",
            "type": "TRANSFORM",
            "operator": "sagemaker.spark.infer_and_cast_type_0.2",
            "trained_parameters": {"k1": "string", "k2": "long"},
            "parameters": {},
            "inputs": [{"name": "default", "node_id": "123456", "output_name": "default"}],
            "outputs": [{"name": "default"}],
        },
    ]