public Input saveInput(AWSInputCreateRequest request, User user) throws Exception {

        // Transpose the SaveAWSInputRequest to the needed InputCreateRequest
        final HashMap<String, Object> configuration = new HashMap<>();
        configuration.put(AWSCodec.CK_AWS_MESSAGE_TYPE, request.awsMessageType());
        configuration.put(ThrottleableTransport.CK_THROTTLING_ALLOWED, request.throttlingAllowed());
        configuration.put(AWSCodec.CK_FLOW_LOG_PREFIX, request.addFlowLogPrefix());
        configuration.put(AWSInput.CK_AWS_REGION, request.region());
        configuration.put(AWSInput.CK_ACCESS_KEY, request.awsAccessKeyId());
        configuration.put(AWSInput.CK_SECRET_KEY, request.awsSecretAccessKey());
        configuration.put(AWSInput.CK_ASSUME_ROLE_ARN, request.assumeRoleArn());
        configuration.put(AWSInput.CK_CLOUDWATCH_ENDPOINT, request.cloudwatchEndpoint());
        configuration.put(AWSInput.CK_DYNAMODB_ENDPOINT, request.dynamodbEndpoint());
        configuration.put(AWSInput.CK_IAM_ENDPOINT, request.iamEndpoint());
        configuration.put(AWSInput.CK_KINESIS_ENDPOINT, request.kinesisEndpoint());

        AWSMessageType inputType = AWSMessageType.valueOf(request.awsMessageType());
        if (inputType.isKinesis()) {
            configuration.put(KinesisTransport.CK_KINESIS_STREAM_NAME, request.streamName());
            configuration.put(KinesisTransport.CK_KINESIS_RECORD_BATCH_SIZE, request.batchSize());
        } else {
            throw new Exception("The specified input type is not supported.");
        }

        // Create and save the input.
        final InputCreateRequest inputCreateRequest = InputCreateRequest.create(request.name(),
                AWSInput.TYPE,
                true,
                configuration,
                nodeId.getNodeId());
        try {
            final MessageInput messageInput = messageInputFactory.create(inputCreateRequest, user.getName(), nodeId.getNodeId());
            messageInput.checkConfiguration();
            final Input input = this.inputService.create(messageInput.asMap());
            final String newInputId = inputService.save(input);
            LOG.debug("New AWS input created. id [{}] request [{}]", newInputId, request);
            return input;
        } catch (NoSuchInputTypeException e) {
            LOG.error("There is no such input type registered.", e);
            throw new NotFoundException("There is no such input type registered.", e);
        } catch (ConfigurationException e) {
            LOG.error("Missing or invalid input configuration.", e);
            throw new BadRequestException("Missing or invalid input configuration.", e);
        }
    }
@Test
    public void testSaveInput() throws Exception {
        when(inputService.create(isA(HashMap.class))).thenCallRealMethod();
        when(inputService.save(isA(Input.class))).thenReturn("input-id");
        when(user.getName()).thenReturn("a-user-name");
        when(messageInputFactory.create(isA(InputCreateRequest.class), isA(String.class), isA(String.class))).thenReturn(messageInput);

        AWSInputCreateRequest request =
                AWSInputCreateRequest.builder().region(Region.US_EAST_1.id())
                        .awsAccessKeyId("a-key")
                        .awsSecretAccessKey(encryptedValue)
                                     .name("AWS Input")
                                     .awsMessageType(AWSMessageType.KINESIS_CLOUDWATCH_FLOW_LOGS.toString())
                                     .streamName("a-stream")
                                     .batchSize(10000)
                                     .addFlowLogPrefix(true)
                                     .throttlingAllowed(true)
                                     .build();
        awsService.saveInput(request, user);

        // Verify that inputService received a valid input to save.
        final ArgumentCaptor<InputCreateRequest> argumentCaptor = ArgumentCaptor.forClass(InputCreateRequest.class);
        verify(messageInputFactory, times(1)).create(argumentCaptor.capture(), eq("a-user-name"), eq("5ca1ab1e-0000-4000-a000-000000000000"));

        // Just verify that the input create request was prepared correctly. This verifies the important argument
        // transposition logic.
        // It's too hard to mock the full inputService.save process, so we are not going to check the final resulting input.
        InputCreateRequest input = argumentCaptor.getValue();
        assertEquals("AWS Input", input.title());
        assertEquals(AWSInput.TYPE, input.type());
        assertTrue(input.global());
        assertEquals("us-east-1", input.configuration().get(AWSInput.CK_AWS_REGION));
        assertEquals("KINESIS_CLOUDWATCH_FLOW_LOGS", input.configuration().get(AWSCodec.CK_AWS_MESSAGE_TYPE));
        assertEquals("a-key", input.configuration().get(AWSInput.CK_ACCESS_KEY));
        assertEquals(encryptedValue, input.configuration().get(AWSInput.CK_SECRET_KEY));
        assertEquals("us-east-1", input.configuration().get(AWSInput.CK_AWS_REGION));
        assertEquals("a-stream", input.configuration().get(KinesisTransport.CK_KINESIS_STREAM_NAME));
        assertEquals(10000, input.configuration().get(KinesisTransport.CK_KINESIS_RECORD_BATCH_SIZE));
    }