package io.digdag.standards.operator;

import io.digdag.core.database.TransactionManager;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;

import java.nio.file.Path;
import java.nio.file.Paths;

import io.digdag.client.config.Config;
import io.digdag.core.DigdagEmbed;
import io.digdag.spi.Operator;
import io.digdag.spi.TaskResult;
import io.digdag.standards.operator.ForEachOperatorFactory.ForEachOperator;

import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static io.digdag.client.config.ConfigUtils.newConfig;
import static io.digdag.core.workflow.OperatorTestingUtils.newOperatorFactory;
import static io.digdag.core.workflow.OperatorTestingUtils.newTaskRequest;
import static io.digdag.core.workflow.OperatorTestingUtils.newContext;
import static io.digdag.core.workflow.WorkflowTestingUtils.loadYamlResource;
import static io.digdag.core.workflow.WorkflowTestingUtils.setupEmbed;
import static io.digdag.core.workflow.WorkflowTestingUtils.runWorkflow;

public class ForEachOperatorFactoryTest
{
    @Rule
    public final ExpectedException exception = ExpectedException.none();

    @Rule
    public TemporaryFolder folder = new TemporaryFolder();

    private Path tempPath;
    private ForEachOperatorFactory factory;

    @Before
    public void createInstance()
    {
        this.factory = newOperatorFactory(ForEachOperatorFactory.class);
        this.tempPath = folder.getRoot().toPath();
    }

    private void assertByResource(String configResource, String expectedResource)
        throws Exception
    {
        ForEachOperator op = factory.newOperator(newContext(
                    tempPath, newTaskRequest().withConfig(loadYamlResource(configResource))));
        TaskResult result = op.run();
        Config subtasks = result.getSubtaskConfig();
        assertThat(subtasks, is(loadYamlResource(expectedResource)));

        try (DigdagEmbed embed = setupEmbed()) {
            assertTrue(
                    runWorkflow(embed, tempPath, "test", subtasks)
                            .getStateFlags()
                            .isSuccess()
            );
        }
    }

    @Test
    public void testBasic()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/basic.yml",
                "/io/digdag/standards/operator/for_each/basic_expected.yml");
    }

    @Test
    public void testDuplicateValues()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/duplicate_values.yml",
                "/io/digdag/standards/operator/for_each/duplicate_values_expected.yml");
    }

    @Test
    public void parallelComplex()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/parallel_complex.yml",
                "/io/digdag/standards/operator/for_each/parallel_complex_expected.yml");
    }

    @Test
    public void parseNestedMap()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/parse_nested_map.yml",
                "/io/digdag/standards/operator/for_each/parse_nested_map_expected.yml");
    }

    @Test
    public void parseNestedArrays()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/parse_nested_arrays.yml",
                "/io/digdag/standards/operator/for_each/parse_nested_arrays_expected.yml");
    }

    @Test
    public void escapeValues()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/escape_values.yml",
                "/io/digdag/standards/operator/for_each/escape_values_expected.yml");
    }

    @Test
    public void escapeKeys()
        throws Exception
    {
        assertByResource(
                "/io/digdag/standards/operator/for_each/escape_keys.yml",
                "/io/digdag/standards/operator/for_each/escape_keys_expected.yml");
    }
}

@@ -94,24 +94,136 @@
 ort 
-org.junit.Before
+io.digdag.core.workflow.WorkflowTestingUtils;%0Aimport org.junit.AfterClass;%0Aimport org.junit.Before;%0Aimport org.junit.BeforeClass
 ;%0Aim
@@ -1154,30 +1154,77 @@
 source;%0A
-import
+%0Apublic class ForEachOperatorFactoryTest%0A%7B%0A    public
  static 
 io.digda
@@ -1219,179 +1219,276 @@
 tic 
-io.d
+D
 igdag
-.core.workflow.WorkflowTestingUtils.setupEmbed;%0Aimport static io.digdag.core.workflow.WorkflowTestingUtils.runWorkflow;%0A%0Apublic class ForEachOperatorFactoryTest%0A%7B
+Embed embed;%0A%0A    @BeforeClass%0A    public static void createDigdagEmbed()%0A    %7B%0A        embed = WorkflowTestingUtils.setupEmbed();%0A    %7D%0A%0A    @AfterClass%0A    public static void destroyDigdagEmbed()%0A            throws Exception%0A    %7B%0A        embed.close();%0A    %7D%0A
 %0A   
@@ -2335,93 +2335,40 @@
     
-try (DigdagEmbed embed = setupEmbed()) %7B%0A            assertTrue(%0A                    
+assertTrue(WorkflowTestingUtils.
 runW
@@ -2426,28 +2426,16 @@
         
-     
-       
 .getStat
@@ -2459,28 +2459,16 @@
         
-          
-  
 .isSucce
@@ -2484,24 +2484,10 @@
     
-    );%0A        %7D
+);
 %0A   

Fix ForEachOperatorFactoryTest not to recreate DigdagEmbed instance for each test method to make tests faster