workflow: fail on multiple operators in a task

digdag-core/src/main/java/io/digdag/core/config/DuplicateKeyYAMLException.java

@@ -0,0 +1,11 @@
+package io.digdag.core.config;
+
+import org.yaml.snakeyaml.error.YAMLException;
+
+class DuplicateKeyYAMLException extends YAMLException
+{
+    public DuplicateKeyYAMLException()
+    {
+        super("duplicate key");
+    }
+}

digdag-core/src/main/java/io/digdag/core/config/StrictSafeConstructor.java

@@ -0,0 +1,61 @@
+package io.digdag.core.config;
+
+import io.digdag.client.config.ConfigException;
+import org.yaml.snakeyaml.constructor.BaseConstructor;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+import org.yaml.snakeyaml.error.MarkedYAMLException;
+import org.yaml.snakeyaml.error.YAMLException;
+import org.yaml.snakeyaml.nodes.MappingNode;
+import org.yaml.snakeyaml.nodes.Node;
+import org.yaml.snakeyaml.nodes.NodeTuple;
+import org.yaml.snakeyaml.nodes.ScalarNode;
+import org.yaml.snakeyaml.nodes.Tag;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link SafeConstructor} that disallows duplicate keys.
+ */
+class StrictSafeConstructor
+        extends SafeConstructor
+{
+    StrictSafeConstructor()
+    {
+        this.yamlConstructors.put(Tag.MAP, new StrictConstructYamlMap());
+    }
+
+    private class StrictConstructYamlMap
+            extends StrictSafeConstructor.ConstructYamlMap
+    {
+        @Override
+        public Object construct(Node node)
+        {
+            if (node instanceof MappingNode) {
+                validateKeys((MappingNode) node);
+            }
+            return super.construct(node);
+        }
+
+        @Override
+        public void construct2ndStep(Node node, Object object)
+        {
+            if (node instanceof MappingNode) {
+                validateKeys((MappingNode) node);
+            }
+            super.construct2ndStep(node, object);
+        }
+
+        private void validateKeys(MappingNode node)
+        {
+            List<String> keys = node.getValue().stream()
+                    .map(NodeTuple::getKeyNode)
+                    .filter(n -> n instanceof ScalarNode)
+                    .map(n -> ((ScalarNode)n).getValue())
+                    .collect(Collectors.toList());
+            if (keys.stream().distinct().count() != keys.stream().count()) {
+                throw new DuplicateKeyYAMLException();
+            }
+        }
+    }
+}

digdag-core/src/main/java/io/digdag/core/config/YamlConfigLoader.java

@@ -14,6 +14,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.CharStreams;
 import com.google.inject.Inject;
+import io.digdag.client.config.ConfigException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import io.digdag.client.config.Config;
@@ -22,6 +23,7 @@
 import org.yaml.snakeyaml.Yaml;
 import org.yaml.snakeyaml.DumperOptions;
 import org.yaml.snakeyaml.constructor.SafeConstructor;
+import org.yaml.snakeyaml.error.YAMLException;
 import org.yaml.snakeyaml.representer.Representer;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -56,8 +58,15 @@ public ConfigElement loadString(String content)
     {
         // here doesn't use jackson-dataformat-yaml so that snakeyaml calls Resolver
         // and Composer. See also YamlTagResolver.
-        Yaml yaml = new Yaml(new SafeConstructor(), new Representer(), new DumperOptions(), new YamlTagResolver());
-        ObjectNode object = normalizeValidateObjectNode(yaml.load(content));
+        Yaml yaml = new Yaml(new StrictSafeConstructor(), new Representer(), new DumperOptions(), new YamlTagResolver());
+        Object raw;
+        try {
+            raw = yaml.load(content);
+        }
+        catch (YAMLException e) {
+            throw new ConfigException("Invalid workflow definition: " + e.getMessage());
+        }
+        ObjectNode object = normalizeValidateObjectNode(raw);
         return ConfigElement.of(object);
     }
 

digdag-core/src/main/java/io/digdag/core/workflow/WorkflowCompiler.java

@@ -159,6 +159,9 @@ else if (config.has("_type") || config.getKeys().stream().anyMatch(key -> key.en
                 if (!subtaskConfigs.isEmpty()) {
                     throw new ConfigException("A task can't have subtasks: " + config);
                 }
+                if (config.getKeys().stream().filter(key -> key.endsWith(">")).count() > 1) {
+                    throw new ConfigException("A task can't have more than one operator: " + config);
+                }
                 return addTask(parent, name, fullName, false, config);
             }
             else {

digdag-core/src/test/java/io/digdag/core/config/YamlConfigLoaderTest.java

@@ -0,0 +1,29 @@
+package io.digdag.core.config;
+
+import com.google.common.io.Resources;
+import io.digdag.client.config.ConfigException;
+import org.junit.Before;
+import org.junit.Test;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.*;
+
+public class YamlConfigLoaderTest
+{
+
+    YamlConfigLoader loader;
+
+    @Before
+    public void setUp()
+            throws Exception
+    {
+        loader = new YamlConfigLoader();
+    }
+
+    @Test(expected = ConfigException.class)
+    public void verifyDuplicateKeysDisallowed()
+            throws Exception
+    {
+        loader.loadString("{\"a\":1, \"a\":2}");
+    }
+}

digdag-core/src/test/java/io/digdag/core/workflow/WorkflowCompilerTest.java

@@ -0,0 +1,63 @@
+package io.digdag.core.workflow;
+
+import com.google.common.base.Throwables;
+import com.google.common.io.Resources;
+import io.digdag.client.config.Config;
+import io.digdag.client.config.ConfigException;
+import io.digdag.client.config.ConfigFactory;
+import io.digdag.core.DigdagEmbed;
+import io.digdag.core.config.YamlConfigLoader;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.IOException;
+
+import static io.digdag.core.workflow.WorkflowTestingUtils.setupEmbed;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+public class WorkflowCompilerTest
+{
+
+    @Rule public ExpectedException exception = ExpectedException.none();
+
+    private DigdagEmbed embed;
+
+    private WorkflowCompiler compiler;
+
+    @Before
+    public void setUp()
+            throws Exception
+    {
+        embed = setupEmbed();
+        compiler = new WorkflowCompiler();
+    }
+
+    private Config loadYamlResource(String name)
+    {
+        try {
+            String content = Resources.toString(getClass().getResource(name), UTF_8);
+            return embed.getInjector().getInstance(YamlConfigLoader.class)
+                    .loadString(content)
+                    .toConfig(embed.getInjector().getInstance(ConfigFactory.class));
+        }
+        catch (IOException ex) {
+            throw Throwables.propagate(ex);
+        }
+    }
+
+    @Test
+    public void verifySingleOperatorPasses() {
+        Config config = loadYamlResource("/digdag/workflow/cases/single_operator.yml");
+        compiler.compile("+foo", config);
+    }
+
+    @Test
+    public void verifyMultipleOperatorsFail()
+    {
+        Config config = loadYamlResource("/digdag/workflow/cases/multiple_operators.yml");
+        exception.expect(ConfigException.class);
+        compiler.compile("+foo", config);
+    }
+}

digdag-core/src/test/resources/digdag/workflow/cases/multiple_operators.yml

@@ -0,0 +1,5 @@
+run: +foo
+
++foo:
+  sh>: echo this should fail
+  py>: bar

digdag-core/src/test/resources/digdag/workflow/cases/single_operator.yml

@@ -0,0 +1,4 @@
+run: +foo
+
++foo:
+  sh>: echo hello world