package org.apache.mesos.logstash.systemtest;

import com.containersol.minimesos.MesosCluster;
import com.containersol.minimesos.mesos.ClusterArchitecture;
import com.containersol.minimesos.mesos.DockerClientFactory;
import com.containersol.minimesos.state.Framework;
import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.github.dockerjava.api.DockerClient;
import com.mashape.unirest.http.exceptions.UnirestException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

import static com.jayway.awaitility.Awaitility.*;

/**
 * Tests whether the framework is deployed correctly
 */
public class DeploymentSystemTest {

    private static DockerClient dockerClient = DockerClientFactory.build();

    private static MesosCluster cluster = new MesosCluster(new ClusterArchitecture.Builder()
            .withZooKeeper()
            .withMaster()
            .withSlave(zooKeeper -> new LogstashMesosSlave(dockerClient, zooKeeper))
            .build());

    @Before
    public void before() {
        cluster.start();
    }

    @After
    public void after() {
        cluster.stop();
    }

    @Test
    public void testDeployment() throws JsonParseException, UnirestException, JsonMappingException {
        String zookeeperIpAddress = cluster.getZkContainer().getIpAddress();
        LogstashSchedulerContainer container = new LogstashSchedulerContainer(dockerClient, zookeeperIpAddress);
        cluster.addAndStartContainer(container);

        await().atMost(1, TimeUnit.MINUTES).pollInterval(1, TimeUnit.SECONDS).until(() -> {
            Framework framework = cluster.getStateInfo().getFramework("logstash");
            return
                framework != null && framework.getTasks().size() > 0 &&
                framework.getTasks().get(0).getState().equals("TASK_RUNNING");
        });
    }

    @Test
    public void willAddExecutorOnNewNodes() throws JsonParseException, UnirestException, JsonMappingException {
        String zookeeperIpAddress = cluster.getZkContainer().getIpAddress();
        LogstashSchedulerContainer container = new LogstashSchedulerContainer(dockerClient, zookeeperIpAddress);
        cluster.addAndStartContainer(container);

        await().atMost(1, TimeUnit.MINUTES).pollInterval(1, TimeUnit.SECONDS).until(() -> {
            Framework framework = cluster.getStateInfo().getFramework("logstash");
            return
                framework != null && framework.getTasks().size() > 0 &&
                framework.getTasks().get(0).getState().equals("TASK_RUNNING");
        });

        IntStream.range(0, 2).forEach(value -> cluster.addAndStartContainer(new LogstashMesosSlave(dockerClient, cluster.getZkContainer())));

        await().atMost(1, TimeUnit.MINUTES).pollInterval(1, TimeUnit.SECONDS).until(() -> cluster.getStateInfo().getFramework("logstash").getTasks().stream().filter(task -> task.getState().equals("TASK_RUNNING")).count() == 3);
    }

}

@@ -479,24 +479,51 @@
 tException;%0A
+import org.json.JSONArray;%0A
 import org.j
@@ -530,24 +530,24 @@
 unit.After;%0A
-
 import org.j
@@ -579,24 +579,46 @@
 unit.Test;%0A%0A
+import java.util.Set;%0A
 import java.
@@ -631,16 +631,42 @@
 reeMap;%0A
+import java.util.TreeSet;%0A
 import j
@@ -731,16 +731,16 @@
 tream;%0A%0A
-
 import s
@@ -780,16 +780,61 @@
 ility.*;
+%0Aimport static org.junit.Assert.assertEquals;
 %0A%0A/**%0A *
@@ -3056,24 +3056,41 @@
 ONDS).until(
+%0A                
 () -%3E cluste
@@ -3218,16 +3218,524 @@
 t() == 3
+%0A        );%0A%0A        // TODO use com.containersol.minimesos.state.Task when it exposes the slave_id property https://github.com/ContainerSolutions/minimesos/issues/168%0A        JSONArray tasks = cluster.getStateInfoJSON().getJSONArray(%22frameworks%22).getJSONObject(0).getJSONArray(%22tasks%22);%0A        Set%3CString%3E slaveIds = new TreeSet%3C%3E();%0A        for (int i = 0; i %3C tasks.length(); i++) %7B%0A            slaveIds.add(tasks.getJSONObject(i).getString(%22slave_id%22));%0A        %7D%0A        assertEquals(3, slaveIds.size()
 );%0A    %7D

Test that all tasks are running on different slaves