/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.job.workflow.composite;

import static org.junit.Assert.assertEquals;

import alluxio.job.JobIntegrationTest;
import alluxio.job.SleepJobConfig;
import alluxio.job.plan.NoopPlanConfig;
import alluxio.job.wire.JobInfo;
import alluxio.job.wire.Status;

import com.google.common.collect.Lists;
import org.junit.Test;

import java.util.List;

/**
 * Integration tests for {@link CompositeExecution}.
 */
public class CompositeIntegrationTest extends JobIntegrationTest {

  @Test
  public void simpleTest() throws Exception {
    CompositeConfig config = new CompositeConfig(
        Lists.newArrayList(new SleepJobConfig(1), new SleepJobConfig(1)), false);

    waitForJobToFinish(mJobMaster.run(config));
  }

  @Test
  public void nestedTest() throws Exception {
    CompositeConfig config = new CompositeConfig(
        Lists.newArrayList(
            new CompositeConfig(
                Lists.newArrayList(new SleepJobConfig(2), new SleepJobConfig(3)), false),
            new SleepJobConfig(1)
        ),
        true
    );

    long jobId = mJobMaster.run(config);
    waitForJobToFinish(jobId);

    JobInfo status = mJobMaster.getStatus(jobId);

    assertEquals(Status.COMPLETED, status.getStatus());
    List<JobInfo> children = status.getChildren();
    assertEquals(2, children.size());

    JobInfo child0 = children.get(0);
    assertEquals(child0, mJobMaster.getStatus(children.get(0).getId()));
    assertEquals(Status.COMPLETED, child0.getStatus());
    assertEquals(2, child0.getChildren().size());

    JobInfo child1 = children.get(1);
    assertEquals(child1, mJobMaster.getStatus(children.get(1).getId()));
    assertEquals(Status.COMPLETED, child1.getStatus());
    assertEquals(1, child1.getChildren().size());
  }

  @Test
  public void waitSequential() throws Exception {
    CompositeConfig config = new CompositeConfig(
        Lists.newArrayList(new SleepJobConfig(1000000), new SleepJobConfig(1)), true);

    long jobId = mJobMaster.run(config);

    JobInfo status = mJobMaster.getStatus(jobId);

    assertEquals(Status.RUNNING, status.getStatus());
    assertEquals(1, status.getChildren().size());
  }

  @Test
  public void waitParallel() throws Exception {
    CompositeConfig config = new CompositeConfig(
        Lists.newArrayList(new SleepJobConfig(1000000), new SleepJobConfig(1)), false);

    long jobId = mJobMaster.run(config);

    JobInfo status = mJobMaster.getStatus(jobId);

    assertEquals(Status.RUNNING, status.getStatus());
    assertEquals(2, status.getChildren().size());
  }

  @Test
  public void testCompositeDoNothing() throws Exception {
    NoopPlanConfig jobConfig = new NoopPlanConfig();

    long jobId = mJobMaster.run(new CompositeConfig(Lists.newArrayList(jobConfig), true));

    waitForJobToFinish(jobId);
  }
}
