Throw appropriate error when input is empty

When the watch `input` is an empty object, we need to throw an appropriate error indicating the problem (instead of the NPE that is currently thrown)

Closes elastic/elasticsearch#427

Original commit: elastic/x-pack-elasticsearch@c8b27e6653

src/main/java/org/elasticsearch/watcher/input/InputRegistry.java

@@ -34,19 +34,32 @@ public class InputRegistry {
     public ExecutableInput parse(String watchId, XContentParser parser) throws IOException {
         String type = null;
 
+        if (parser.currentToken() != XContentParser.Token.START_OBJECT) {
+            throw new InputException("could not parse input for watch [{}]. expected an object representing the input, but found [{}] instead", watchId, parser.currentToken());
+        }
+
         XContentParser.Token token;
         ExecutableInput input = null;
         while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
             if (token == XContentParser.Token.FIELD_NAME) {
                 type = parser.currentName();
-            } else if (token == XContentParser.Token.START_OBJECT && type != null) {
+            } else if (type == null) {
+                throw new InputException("could not parse input for watch [{}]. expected field indicating the input type, but found [{}] instead", watchId, token);
+            } else if (token == XContentParser.Token.START_OBJECT) {
                 InputFactory factory = factories.get(type);
                 if (factory == null) {
                     throw new InputException("could not parse input for watch [{}]. unknown input type [{}]", watchId, type);
                 }
                 input = factory.parseExecutable(watchId, parser);
+            } else {
+                throw new InputException("could not parse input for watch [{}]. expected an object representing input [{}], but found [{}] instead", watchId, type, token);
             }
         }
+
+        if (input == null) {
+            throw new InputException("could not parse input for watch [{}]. expected field indicating the input type, but found an empty object instead", watchId, token);
+        }
+
         return input;
     }
 

src/test/java/org/elasticsearch/watcher/input/InputRegistryTests.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.input;
+
+import org.elasticsearch.common.collect.ImmutableMap;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Test;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+
+/**
+ *
+ */
+public class InputRegistryTests extends ElasticsearchTestCase {
+
+    @Test(expected = InputException.class)
+    public void testParse_EmptyInput() throws Exception {
+        InputRegistry registry = new InputRegistry(ImmutableMap.<String, InputFactory>of());
+        XContentParser parser = JsonXContent.jsonXContent.createParser(
+                jsonBuilder().startObject().endObject().bytes());
+        parser.nextToken();
+        registry.parse("_id", parser);
+        fail("expecting an exception when trying to parse an empty input");
+    }
+
+    @Test(expected = InputException.class)
+    public void testParse_ArrayInput() throws Exception {
+        InputRegistry registry = new InputRegistry(ImmutableMap.<String, InputFactory>of());
+        XContentParser parser = JsonXContent.jsonXContent.createParser(
+                jsonBuilder().startArray().endArray().bytes());
+        parser.nextToken();
+        registry.parse("_id", parser);
+        fail("expecting an exception when trying to parse an input that is not an object");
+    }
+}