/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nifi.processors.aws.wag;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processors.aws.AbstractAWSProcessor;
import org.apache.nifi.processors.aws.credentials.provider.service.AWSCredentialsProviderControllerService;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.web.util.TestServer;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.handler.AbstractHandler;
import org.junit.Assert;
import org.junit.Test;

public abstract class TestInvokeAWSGatewayApiCommon {

    public static TestServer server;
    public static String url;
    public TestRunner runner;

    public void setupControllerService() throws InitializationException {
        final AWSCredentialsProviderControllerService serviceImpl = new AWSCredentialsProviderControllerService();
        runner.addControllerService("awsCredentialsProvider", serviceImpl);
        runner.setProperty(serviceImpl, InvokeAWSGatewayApi.ACCESS_KEY, "awsAccessKey");
        runner.setProperty(serviceImpl, InvokeAWSGatewayApi.SECRET_KEY, "awsSecretKey");
        runner.enableControllerService(serviceImpl);
        runner.setProperty(InvokeAWSGatewayApi.AWS_CREDENTIALS_PROVIDER_SERVICE,
                           "awsCredentialsProvider");

    }

    public void setupAuth() {
        runner.setProperty(InvokeAWSGatewayApi.ACCESS_KEY, "testAccessKey");
        runner.setProperty(InvokeAWSGatewayApi.SECRET_KEY, "testSecretKey");
    }

    public void setupCredFile() {
        runner.setProperty(AbstractAWSProcessor.CREDENTIALS_FILE,
                           "src/test/resources/mock-aws-credentials.properties");
    }

    public void setupEndpointAndRegion() {
        runner.setProperty(InvokeAWSGatewayApi.PROP_AWS_GATEWAY_API_REGION, "us-east-1");
        runner.setProperty(InvokeAWSGatewayApi.PROP_AWS_API_KEY, "abcd");
        runner.setProperty(InvokeAWSGatewayApi.PROP_AWS_GATEWAY_API_ENDPOINT, url);
    }

    public void addHandler(Handler handler) {
        server.addHandler(handler);
    }

    @Test
    public void test200() throws Exception {
        addHandler(new GetOrHeadHandler());
        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        createFlowFiles(runner);

        // verify that call works with or without flowfile being sent for GET
        // there should only be 1 REQ, but 2 RESPONSE
        runner.run();
        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 2);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");

        // expected in response of each message
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("{\"status\":\"200\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "application/json");

        final MockFlowFile bundle2 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(1);
        bundle1.assertContentEquals("{\"status\":\"200\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "application/json");

        final List<ProvenanceEventRecord> provEvents = runner.getProvenanceEvents();
        assertEquals(3, provEvents.size());
        boolean forkEvent = false;
        boolean fetchEvent = false;
        boolean recieveEvent = false;
        for (final ProvenanceEventRecord event : provEvents) {
            if (event.getEventType() == ProvenanceEventType.FORK) {
                forkEvent = true;
            } else if (event.getEventType() == ProvenanceEventType.FETCH) {
                fetchEvent = true;
            } else if (event.getEventType() == ProvenanceEventType.RECEIVE) {
                recieveEvent = true;
            }
        }

        assertTrue(forkEvent);
        assertTrue(fetchEvent);
        assertTrue(recieveEvent);
    }

    @Test
    public void testOutputResponseRegardless() throws Exception {
        addHandler(new GetOrHeadHandler(true));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_OUTPUT_RESPONSE_REGARDLESS, "true");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "404");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Found");
        bundle.assertAttributeEquals("Foo", "Bar");

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("{ \"error\": \"oops\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "404");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Found");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "application/json");
    }

    @Test
    public void testOutputResponseRegardlessWithOutputInAttribute() throws Exception {
        addHandler(new GetOrHeadHandler(true));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_OUTPUT_RESPONSE_REGARDLESS, "true");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PUT_OUTPUT_IN_ATTRIBUTE, "outputBody");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "404");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Found");
        bundle.assertAttributeEquals("outputBody", "{ \"error\": \"oops\"}");
        bundle.assertAttributeEquals("Foo", "Bar");

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("{ \"error\": \"oops\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "404");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Found");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "application/json");
    }

    @Test
    public void testOutputResponseSetMimeTypeToResponseContentType() throws Exception {
        addHandler(new GetOrHeadHandler());

        String statusUrl = "/status/200";

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        createFlowFiles(runner);
        runner.setProperty(InvokeAWSGatewayApi.PROP_OUTPUT_RESPONSE_REGARDLESS, "true");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PUT_OUTPUT_IN_ATTRIBUTE, "outputBody");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("outputBody", "{\"status\":\"200\"}");
        bundle.assertAttributeEquals("Foo", "Bar");

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("{\"status\":\"200\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "application/json");
        bundle1.assertAttributeEquals("mime.type", "application/json");
    }

    @Test
    public void testOutputResponseRegardlessWithOutputInAttributeLarge() throws Exception {
        addHandler(new GetLargeHandler(true));
        String statusUrl = "/status/200";

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        createFlowFiles(runner);
        runner.setProperty(InvokeAWSGatewayApi.PROP_OUTPUT_RESPONSE_REGARDLESS, "true");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PUT_OUTPUT_IN_ATTRIBUTE, "outputBody");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PUT_ATTRIBUTE_MAX_LENGTH, "11");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "404");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Found");
        bundle.assertAttributeEquals("outputBody", "{\"name\":\"Lo");
        bundle.assertAttributeEquals("Foo", "Bar");

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals(
            "{\"name\":\"Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. "
                + "Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor "
                + "in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, "
                + "sunt in culpa qui officia deserunt mollit anim id est laborum.\"}");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "404");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Found");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "application/json");
    }


    @Test
    // NOTE : Amazon does not support multiple headers with the same name!!!
    public void testMultipleSameHeaders() throws Exception {
        addHandler(new GetMultipleHeaderHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("/status/200".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        // amazon does not support headers with the same name, we'll only get 2 here
        // this is in the amazon layer, we only get Map<String,String> for headers so the 1 has been stripped
        // already
        bundle1.assertAttributeEquals("double", "2");
        bundle1.assertAttributeEquals("Content-Type", "text/plain;charset=iso-8859-1");
    }

    @Test
    public void testPutResponseHeadersInRequest() throws Exception {
        addHandler(new GetMultipleHeaderHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_ADD_HEADERS_TO_REQUEST, "true");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code and status.message
        // original flow file (+all attributes from response)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");
        bundle.assertAttributeEquals("double", "2");
        bundle.assertAttributeEquals("Content-Type", "text/plain;charset=iso-8859-1");

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("/status/200".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("double", "2");
        bundle1.assertAttributeEquals("Content-Type", "text/plain;charset=iso-8859-1");
    }

    @Test
    public void testToRequestAttribute() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PUT_OUTPUT_IN_ATTRIBUTE, "outputBody");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in request status.code, status.message and body of response in attribute
        // original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals("outputBody", "{\"status\":\"200\"}");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");
    }

    @Test
    public void testNoInput() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setIncomingConnection(false);
        runner.setNonLoopConnection(false);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("{\"status\":\"200\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Content-Type", "application/json");
    }

    @Test
    public void testNoInputWithAttributes() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_ATTRIBUTES_TO_SEND, "myAttribute");
        runner.setIncomingConnection(false);
        runner.setNonLoopConnection(false);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in response
        // status code, status message, all headers from server response --> ff attributes
        // server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("{\"status\":\"200\"}".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Content-Type", "application/json");
    }

    @Test
    public void testNoInputSendToAttribute() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PUT_OUTPUT_IN_ATTRIBUTE, "outputBody");
        runner.setIncomingConnection(false);
        runner.setNonLoopConnection(false);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE, 0);
        runner.assertPenalizeCount(0);

        // expected in request
        // status code, status message, no ff content
        // server response message body into attribute of ff
        final MockFlowFile bundle1 = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ).get(0);
        bundle1.assertContentEquals("".getBytes("UTF-8"));
        bundle1.assertAttributeEquals("outputBody", "{\"status\":\"200\"}");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
    }

    @Test
    public void test500() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/500");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(1);

        // expected in response
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RETRY_NAME).get(0);
        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "500");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Internal Server Error");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.RESPONSE_BODY, "{\"status\":\"500\"}");

        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");

    }

    @Test
    public void test300() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/302");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in response
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);

        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "302");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Moved Temporarily");
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");

    }

    @Test
    public void test304() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/304");

        createFlowFiles(runner);

        //assertTrue(server.jetty.isRunning());
        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in response
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);

        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "304");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Not Modified");
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");

    }

    @Test
    public void test400() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/400");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);
        // expected in response
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);

        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "400");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Bad Request");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.RESPONSE_BODY, "{\"status\":\"400\"}");
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");

    }

    @Test
    public void test400WithPenalizeNoRetry() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/400");
        runner.setProperty(InvokeAWSGatewayApi.PROP_PENALIZE_NO_RETRY, "true");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(1);
        // expected in response
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);

        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "400");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Bad Request");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.RESPONSE_BODY, "{\"status\":\"400\"}");
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");
    }

    @Test
    public void test412() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/412");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        // expected in response
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_NO_RETRY_NAME).get(0);
        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);

        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "412");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "Precondition Failed");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.RESPONSE_BODY, "{\"status\":\"412\"}");
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");

    }

    @Test
    public void testHead() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "HEAD");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");

        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        final String actual1 = new String(bundle1.toByteArray(), StandardCharsets.UTF_8);
        final String expected1 = "";
        Assert.assertEquals(expected1, actual1);
    }

    @Test
    public void testPost() throws Exception {
        addHandler(new MutativeMethodHandler(MutativeMethod.POST));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "POST");

        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");

        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeNotExists("Content-Type");

        final String actual1 = new String(bundle1.toByteArray(), StandardCharsets.UTF_8);
        final String expected1 = "";
        Assert.assertEquals(expected1, actual1);
    }

    @Test
    public void testPostWithMimeType() throws Exception {
        final String suppliedMimeType = "text/plain";
        addHandler(new MutativeMethodHandler(MutativeMethod.POST, suppliedMimeType));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "POST");

        final Map<String, String> attrs = new HashMap<>();

        attrs.put(CoreAttributes.MIME_TYPE.key(), suppliedMimeType);
        runner.enqueue("Hello".getBytes(), attrs);

        runner.run(1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
    }

    @Test
    public void testPostWithEmptyELExpression() throws Exception {
        addHandler(new MutativeMethodHandler(MutativeMethod.POST,
                                             InvokeAWSGatewayApi.DEFAULT_CONTENT_TYPE));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "POST");

        final Map<String, String> attrs = new HashMap<>();
        attrs.put(CoreAttributes.MIME_TYPE.key(), "");
        runner.enqueue("Hello".getBytes(), attrs);

        runner.run(1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
    }

    @Test
    public void testPostWithContentTypeProperty() throws Exception {
        final String suppliedMimeType = "text/plain";
        addHandler(new MutativeMethodHandler(MutativeMethod.POST, suppliedMimeType));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "POST");
        runner.setProperty(InvokeAWSGatewayApi.PROP_CONTENT_TYPE, suppliedMimeType);

        final Map<String, String> attrs = new HashMap<>();
        attrs.put(CoreAttributes.MIME_TYPE.key(), "text/csv");
        runner.enqueue("Hello".getBytes(), attrs);

        runner.run(1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
    }

    @Test
    public void testPostWithEmptyBodySet() throws Exception {
        final String suppliedMimeType = "";
        addHandler(new MutativeMethodHandler(MutativeMethod.POST, suppliedMimeType));

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "POST");
        runner.setProperty(InvokeAWSGatewayApi.PROP_CONTENT_TYPE, suppliedMimeType);
        runner.setProperty(InvokeAWSGatewayApi.PROP_SEND_BODY, "false");

        final Map<String, String> attrs = new HashMap<>();
        attrs.put(CoreAttributes.MIME_TYPE.key(), suppliedMimeType);
        runner.enqueue("Hello".getBytes(), attrs);

        runner.run(1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
    }


        @Test
        public void testPutWithMimeType() throws Exception {
            final String suppliedMimeType = "text/plain";
            addHandler(new MutativeMethodHandler(MutativeMethod.PUT, suppliedMimeType));

            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PUT");

            final Map<String, String> attrs = new HashMap<>();

            attrs.put(CoreAttributes.MIME_TYPE.key(), suppliedMimeType);
            runner.enqueue("Hello".getBytes(), attrs);

            runner.run(1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        }

        @Test
        public void testPutWithEmptyELExpression() throws Exception {
            addHandler(new MutativeMethodHandler(MutativeMethod.PUT, InvokeAWSGatewayApi.DEFAULT_CONTENT_TYPE));
            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PUT");

            final Map<String, String> attrs = new HashMap<>();
            attrs.put(CoreAttributes.MIME_TYPE.key(), "");
            runner.enqueue("Hello".getBytes(), attrs);

            runner.run(1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        }

        @Test
        public void testPutWithContentTypeProperty() throws Exception {
            final String suppliedMimeType = "text/plain";
            addHandler(new MutativeMethodHandler(MutativeMethod.PUT, suppliedMimeType));
            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PUT");
            runner.setProperty(InvokeAWSGatewayApi.PROP_CONTENT_TYPE, suppliedMimeType);

            final Map<String, String> attrs = new HashMap<>();
            attrs.put(CoreAttributes.MIME_TYPE.key(), "text/csv");
            runner.enqueue("Hello".getBytes(), attrs);

            runner.run(1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        }

        @Test
        public void testPut() throws Exception {
            addHandler(new MutativeMethodHandler(MutativeMethod.PUT));

            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/post");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PUT");

            createFlowFiles(runner);

            runner.run();
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE, 0);
            runner.assertPenalizeCount(0);

            final MockFlowFile bundle = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ).get(0);
            bundle.assertContentEquals("Hello".getBytes("UTF-8"));
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle.assertAttributeEquals("Foo", "Bar");

            final MockFlowFile bundle1 = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE).get(0);
            bundle1.assertContentEquals("".getBytes("UTF-8"));
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle1.assertAttributeEquals("Foo", "Bar");
            bundle1.assertAttributeNotExists("Content-Type");

            final String actual1 = new String(bundle1.toByteArray(), StandardCharsets.UTF_8);
            final String expected1 = "";
            Assert.assertEquals(expected1, actual1);
        }

        @Test
        public void testPatch() throws Exception {
            addHandler(new MutativeMethodHandler(MutativeMethod.PATCH));
            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/patch");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PATCH");


            createFlowFiles(runner);

            runner.run();
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE, 0);
            runner.assertPenalizeCount(0);

            final MockFlowFile bundle = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ).get(0);
            bundle.assertContentEquals("Hello".getBytes("UTF-8"));
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle.assertAttributeEquals("Foo", "Bar");

            final MockFlowFile bundle1 = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE).get(0);
            bundle1.assertContentEquals("".getBytes("UTF-8"));
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle1.assertAttributeEquals("Foo", "Bar");
            bundle1.assertAttributeNotExists("Content-Type");

            final String actual1 = new String(bundle1.toByteArray(), StandardCharsets.UTF_8);
            final String expected1 = "";
            Assert.assertEquals(expected1, actual1);
        }

        @Test
        public void testPatchWithMimeType() throws Exception {
            final String suppliedMimeType = "text/plain";
            addHandler(new MutativeMethodHandler(MutativeMethod.PATCH, suppliedMimeType));

            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/patch");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PATCH");

            final Map<String, String> attrs = new HashMap<>();

            attrs.put(CoreAttributes.MIME_TYPE.key(), suppliedMimeType);
            runner.enqueue("Hello".getBytes(), attrs);

            runner.run(1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        }

        @Test
        public void testPatchWithEmptyELExpression() throws Exception {
            addHandler(new MutativeMethodHandler(MutativeMethod.PATCH, InvokeAWSGatewayApi.DEFAULT_CONTENT_TYPE));

            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/patch");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PATCH");

            final Map<String, String> attrs = new HashMap<>();
            attrs.put(CoreAttributes.MIME_TYPE.key(), "");
            runner.enqueue("Hello".getBytes(), attrs);

            runner.run(1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        }

        @Test
        public void testPatchWithContentTypeProperty() throws Exception {
            final String suppliedMimeType = "text/plain";
            addHandler(new MutativeMethodHandler(MutativeMethod.PATCH, suppliedMimeType));

            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/patch");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "PATCH");
            runner.setProperty(InvokeAWSGatewayApi.PROP_CONTENT_TYPE, suppliedMimeType);

            final Map<String, String> attrs = new HashMap<>();
            attrs.put(CoreAttributes.MIME_TYPE.key(), "text/csv");
            runner.enqueue("Hello".getBytes(), attrs);

            runner.run(1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        }

        @Test
        public void testDelete() throws Exception {
            addHandler(new DeleteHandler());
            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "DELETE");

            createFlowFiles(runner);

            runner.run();
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE, 0);
            runner.assertPenalizeCount(0);

            final MockFlowFile bundle = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ).get(0);
            bundle.assertContentEquals("Hello".getBytes("UTF-8"));
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle.assertAttributeEquals("Foo", "Bar");

            final MockFlowFile bundle1 = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE).get(0);
            bundle1.assertContentEquals("".getBytes("UTF-8"));
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle1.assertAttributeEquals("Foo", "Bar");
            final String actual1 = new String(bundle1.toByteArray(), StandardCharsets.UTF_8);
            final String expected1 = "";
            Assert.assertEquals(expected1, actual1);
        }

        @Test
        public void testOptions() throws Exception {
            addHandler(new OptionsHandler());
            setupEndpointAndRegion();

            runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
            runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "OPTIONS");

            createFlowFiles(runner);

            runner.run();
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY, 0);
            runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE, 0);
            runner.assertPenalizeCount(0);

            final MockFlowFile bundle = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ).get(0);
            bundle.assertContentEquals("Hello".getBytes("UTF-8"));
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle.assertAttributeEquals("Foo", "Bar");

            final MockFlowFile bundle1 = runner.getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE).get(0);
            bundle1.assertContentEquals("{\"status\":\"200\"}".getBytes("UTF-8"));
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
            bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
            bundle1.assertAttributeEquals("Foo", "Bar");
        }

    @Test
    public void testSendAttributes() throws Exception {
        addHandler(new AttributesSentHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_ATTRIBUTES_TO_SEND, "F.*");
        runner.setProperty("dynamicHeader", "yes!");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 0);
        runner.assertPenalizeCount(0);

        //expected in request status.code and status.message
        //original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");

        //expected in response
        //status code, status message, all headers from server response --> ff attributes
        //server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE_NAME).get(0);
        bundle1.assertContentEquals("Bar".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("dynamicHeader", "yes!");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "text/plain;charset=iso-8859-1");
    }

    @Test
    public void testReadTimeout() throws Exception {
        addHandler(new ReadTimeoutHandler());

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.TIMEOUT, "5 secs");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 1);
        runner.assertPenalizeCount(1);

        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_FAILURE_NAME).get(0);

        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");
    }

    @Test
    public void testConnectFailBadPort() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();
        // this is the bad urls
        final String badurlport = "http://localhost:" + 445;

        runner.setProperty(InvokeAWSGatewayApi.PROP_AWS_GATEWAY_API_ENDPOINT, badurlport);
        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/doesnotExist");
        runner.setProperty(InvokeAWSGatewayApi.TIMEOUT, "1 sec");
        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 1);
        runner.assertPenalizeCount(1);

        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_FAILURE_NAME).get(0);

        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");
    }

    @Test
    public void testConnectFailBadHost() throws Exception {
        addHandler(new GetOrHeadHandler());

        setupEndpointAndRegion();
        final String badurlhost = "http://localhOOst:" + 445;

        runner.setProperty(InvokeAWSGatewayApi.PROP_AWS_GATEWAY_API_ENDPOINT, badurlhost);
        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/doesnotExist");
        runner.setProperty(InvokeAWSGatewayApi.TIMEOUT, "1 sec");
        createFlowFiles(runner);

        runner.run();
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY_NAME, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE_NAME, 1);
        runner.assertPenalizeCount(1);

        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_FAILURE_NAME).get(0);

        final String actual = new String(bundle.toByteArray(), StandardCharsets.UTF_8);
        final String expected = "Hello";
        Assert.assertEquals(expected, actual);
        bundle.assertAttributeEquals("Foo", "Bar");
    }

    @Test(expected = java.lang.AssertionError.class)
    public void testArbitraryRequestFailsValidation() throws Exception {

        setupEndpointAndRegion();

        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROP_METHOD, "FETCH");

        createFlowFiles(runner);

        runner.run();
    }

    @Test
    public void testProxy() throws Exception {
        addHandler(new MyProxyHandler());
        setupEndpointAndRegion();
        URL proxyURL = new URL(url);

        runner.setVariable("proxy.host", proxyURL.getHost());
        runner.setVariable("proxy.port", String.valueOf(proxyURL.getPort()));
        runner.setVariable("proxy.username", "username");
        runner.setVariable("proxy.password", "password");

        runner.setProperty(InvokeAWSGatewayApi.PROP_AWS_GATEWAY_API_ENDPOINT, "http://nifi.apache.org/");
        runner.setProperty(InvokeAWSGatewayApi.PROP_RESOURCE_NAME, "/status/200");
        runner.setProperty(InvokeAWSGatewayApi.PROXY_HOST, "${proxy.host}");

        try {
            runner.run();
            Assert.fail();
        } catch (AssertionError e) {
            // Expect assertion error when proxy port isn't set but host is.
        }
        runner.setProperty(InvokeAWSGatewayApi.PROXY_HOST_PORT, "${proxy.port}");

        runner.setProperty(InvokeAWSGatewayApi.PROXY_USERNAME, "${proxy.username}");

        try {
            runner.run();
            Assert.fail();
        } catch (AssertionError e) {
            // Expect assertion error when proxy password isn't set but host is.
        }
        runner.setProperty(InvokeAWSGatewayApi.PROXY_PASSWORD, "${proxy.password}");

        createFlowFiles(runner);

        runner.run();

        runner.assertTransferCount(InvokeAWSGatewayApi.REL_SUCCESS_REQ, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RESPONSE, 1);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_RETRY, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_NO_RETRY, 0);
        runner.assertTransferCount(InvokeAWSGatewayApi.REL_FAILURE, 0);
        runner.assertPenalizeCount(0);

        //expected in request status.code and status.message
        //original flow file (+attributes)
        final MockFlowFile bundle = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_SUCCESS_REQ).get(0);
        bundle.assertContentEquals("Hello".getBytes("UTF-8"));
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle.assertAttributeEquals("Foo", "Bar");

        //expected in response
        //status code, status message, all headers from server response --> ff attributes
        //server response message body into payload of ff
        final MockFlowFile bundle1 = runner
            .getFlowFilesForRelationship(InvokeAWSGatewayApi.REL_RESPONSE).get(0);
        bundle1.assertContentEquals("http://nifi.apache.org/status/200".getBytes("UTF-8"));
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_CODE, "200");
        bundle1.assertAttributeEquals(InvokeAWSGatewayApi.STATUS_MESSAGE, "OK");
        bundle1.assertAttributeEquals("Foo", "Bar");
        bundle1.assertAttributeEquals("Content-Type", "text/plain;charset=iso-8859-1");
    }

    /*
        @Test
        public void testOnPropertyModified() throws Exception {
            final InvokeHTTP processor = new InvokeHTTP();
            final Field regexAttributesToSendField = InvokeHTTP.class.getDeclaredField("regexAttributesToSend");
            regexAttributesToSendField.setAccessible(true);

            assertNull(regexAttributesToSendField.get(processor));

            // Set Attributes to Send.
            processor.onPropertyModified(InvokeHTTP.PROP_ATTRIBUTES_TO_SEND, null, "uuid");
            assertNotNull(regexAttributesToSendField.get(processor));

            // Null clear Attributes to Send. NIFI-1125: Throws NullPointerException.
            processor.onPropertyModified(InvokeHTTP.PROP_ATTRIBUTES_TO_SEND, "uuid", null);
            assertNull(regexAttributesToSendField.get(processor));

            // Set Attributes to Send.
            processor.onPropertyModified(InvokeHTTP.PROP_ATTRIBUTES_TO_SEND, null, "uuid");
            assertNotNull(regexAttributesToSendField.get(processor));

            // Clear Attributes to Send with empty string.
            processor.onPropertyModified(InvokeHTTP.PROP_ATTRIBUTES_TO_SEND, "uuid", "");
            assertNull(regexAttributesToSendField.get(processor));

        }

    */
    public static void createFlowFiles(final TestRunner testRunner)
        throws UnsupportedEncodingException {
        final Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.MIME_TYPE.key(), "application/plain-text");
        attributes.put("Foo", "Bar");
        testRunner.enqueue("Hello".getBytes("UTF-8"), attributes);

    }


    protected static class DateHandler extends AbstractHandler {

        private String dateString;

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            dateString = request.getHeader("Date");

            response.setStatus(200);
            response.setContentType("text/plain");
            response.getWriter().println("Way to go!");
        }
    }

    private enum MutativeMethod {POST, PUT, PATCH}


    public static class MutativeMethodHandler extends AbstractHandler {

        private final MutativeMethod method;
        private final String expectedContentType;
        private String headerToTrack;
        private String trackedHeaderValue;

        public MutativeMethodHandler(final MutativeMethod method) {
            this(method, "application/plain-text");
        }

        public MutativeMethodHandler(final MutativeMethod method,
                                     final String expectedContentType) {
            this.method = method;
            this.expectedContentType = expectedContentType;
        }

        private void setHeaderToTrack(String headerToTrack) {
            this.headerToTrack = headerToTrack;
        }

        public String getTrackedHeaderValue() {
            return trackedHeaderValue;
        }

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {

            baseRequest.setHandled(true);

            if (method.name().equals(request.getMethod())) {
                if (this.expectedContentType.isEmpty()) {
                    // with nothing set, aws defaults to form encoded
                    Assert.assertEquals(request.getHeader("Content-Type"),
                                        "application/x-www-form-urlencoded; charset=UTF-8");
                } else {
                    assertEquals(this.expectedContentType, request.getHeader("Content-Type"));
                }

                final String body = request.getReader().readLine();
                this.trackedHeaderValue = baseRequest.getHttpFields().get(headerToTrack);

                if (this.expectedContentType.isEmpty()) {
                    Assert.assertNull(body);
                } else {
                    assertEquals("Hello", body);
                }
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(0);
            }

        }

    }

    public static class GetOrHeadHandler extends AbstractHandler {

        boolean force404 = false;

        public GetOrHeadHandler() {
        }

        public GetOrHeadHandler(boolean force404) {
            this.force404 = force404;
        }

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            final int status = Integer.valueOf(target.substring("/status".length() + 1));
            response.setStatus(status);

            if (!force404 && "GET".equalsIgnoreCase(request.getMethod())) {
                if (status == 304) {
                    // Status code 304 ("Not Modified") must not contain a message body
                    response.getWriter().write("{\"name\":\"test\"}");
                    return;
                } else if (status == 302) {
                    // this will be treated as an error
                    // the target content must be json
                    target = "{\"status\":\"moved\"}";
                }
                response.setContentType("application/json");
                target = String.format("{\"status\":\"%d\"}", status);
                response.setContentLength(target.length());
                response.setHeader("Cache-Control", "public,max-age=1");

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(target);
                    writer.flush();
                }
            } else if (force404 || !"HEAD".equalsIgnoreCase(request.getMethod())) {
                response.setStatus(404);
                response.setContentType("application/json");
                String body = "{ \"error\": \"oops\"}";
                response.setContentLength(body.length());

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(body);
                    writer.flush();
                }
            }

        }
    }

    public static class GetLargeHandler extends AbstractHandler {

        private boolean force404 = false;

        public GetLargeHandler() {
        }

        public GetLargeHandler(boolean force404) {
            this.force404 = force404;
        }

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            final int status = Integer.valueOf(target.substring("/status".length() + 1));
            response.setStatus(status);

            response.setContentType("text/plain");
            response.setContentLength(target.length());

            if (!force404 && "GET".equalsIgnoreCase(request.getMethod())) {
                try (PrintWriter writer = response.getWriter()) {
                    writer.print(target);
                    writer.flush();
                }
            } else {
                response.setStatus(404);
                response.setContentType("application/json");

                //Lorem Ipsum
                String body =
                    "{\"name\":\"Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. "
                        + "Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor "
                        + "in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, "
                        + "sunt in culpa qui officia deserunt mollit anim id est laborum.\"}";

                response.setContentLength(body.length());
                response.setContentType("application/json");

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(body);
                    writer.flush();
                }
            }

        }
    }

    public static class GetMultipleHeaderHandler extends AbstractHandler {

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            final int status = Integer.valueOf(target.substring("/status".length() + 1));
            response.setStatus(status);

            response.setContentType("text/plain");
            response.setContentLength(target.length());

            if ("GET".equalsIgnoreCase(request.getMethod())) {
                response.addHeader("double", "1");
                response.addHeader("double", "2");

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(target);
                    writer.flush();
                }
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(0);
            }

        }
    }

    public static class DeleteHandler extends AbstractHandler {

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            if ("DELETE".equalsIgnoreCase(request.getMethod())) {
                final int status = Integer.valueOf(target.substring("/status".length() + 1));
                response.setStatus(status);
                response.setContentLength(0);
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(0);
            }
        }
    }

    public static class OptionsHandler extends AbstractHandler {

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            if ("OPTIONS".equalsIgnoreCase(request.getMethod())) {
                final int status = Integer.valueOf(target.substring("/status".length() + 1));
                response.setStatus(status);
                response.setContentType("application/json");
                target = String.format("{\"status\":\"%d\"}", status);
                response.setContentLength(target.length());

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(target);
                    writer.flush();
                }
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(target.length());
            }
        }
    }

    public static class AttributesSentHandler extends AbstractHandler {

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            if ("Get".equalsIgnoreCase(request.getMethod())) {
                String headerValue = request.getHeader("Foo");
                response.setHeader("dynamicHeader", request.getHeader("dynamicHeader"));
                final int status = Integer.valueOf(target.substring("/status".length() + 1));
                response.setStatus(status);
                response.setContentLength(headerValue.length());
                response.setContentType("text/plain");

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(headerValue);
                    writer.flush();
                }
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(0);
            }
        }
    }

    public static class ReadTimeoutHandler extends AbstractHandler {

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            if ("Get".equalsIgnoreCase(request.getMethod())) {
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    return;
                }
                String headerValue = request.getHeader("Foo");
                headerValue = headerValue == null ? "" : headerValue;
                final int status = Integer.valueOf(target.substring("/status".length() + 1));
                response.setStatus(status);
                response.setContentLength(headerValue.length());
                response.setContentType("text/plain");

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(headerValue);
                    writer.flush();
                }
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(0);
            }
        }
    }

    public static class MyProxyHandler extends AbstractHandler {

        @Override
        public void handle(String target, Request baseRequest, HttpServletRequest request,
                           HttpServletResponse response) throws IOException, ServletException {
            baseRequest.setHandled(true);

            if ("Get".equalsIgnoreCase(request.getMethod())) {
                response.setStatus(200);
                String proxyPath = baseRequest.getHttpURI().toString();
                response.setContentLength(proxyPath.length());
                response.setContentType("text/plain");

                try (PrintWriter writer = response.getWriter()) {
                    writer.print(proxyPath);
                    writer.flush();
                }
            } else {
                response.setStatus(404);
                response.setContentType("text/plain");
                response.setContentLength(0);
            }
        }
    }
}