package io.digdag.standards.operator;

import io.digdag.core.database.TransactionManager;
import io.digdag.core.workflow.WorkflowTestingUtils;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
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;

public class ForEachOperatorFactoryTest
{
    public static DigdagEmbed embed;

    @BeforeClass
    public static void createDigdagEmbed()
    {
        embed = WorkflowTestingUtils.setupEmbed();
    }

    @AfterClass
    public static void destroyDigdagEmbed()
            throws Exception
    {
        embed.close();
    }

    @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)));

        assertTrue(WorkflowTestingUtils.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,136 +94,24 @@
 ort 
-io.digdag.core.workflow.WorkflowTestingUtils;%0Aimport org.junit.AfterClass;%0Aimport org.junit.Before;%0Aimport org.junit.BeforeClass
+org.junit.Before
 ;%0Aim
@@ -1046,69 +1046,22 @@
 ce;%0A
-%0Apublic class ForEachOperatorFactoryTest%0A%7B%0A    public
+import
  static 
 Digd
@@ -1060,276 +1060,179 @@
 tic 
-D
+io.d
 igdag
-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
+.core.workflow.WorkflowTestingUtils.setupEmbed;%0Aimport static io.digdag.core.workflow.WorkflowTestingUtils.runWorkflow;%0A%0Apublic class ForEachOperatorFactoryTest%0A%7B
 %0A   
@@ -2079,40 +2079,93 @@
     
-assertTrue(WorkflowTestingUtils.
+try (DigdagEmbed embed = setupEmbed()) %7B%0A            assertTrue(%0A                    
 runW
@@ -2223,16 +2223,28 @@
         
+     
+       
 .getStat
@@ -2268,16 +2268,28 @@
         
+          
+  
 .isSucce
@@ -2301,18 +2301,32 @@
         
-);
+    );%0A        %7D
 %0A    %7D%0A%0A

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