/*
 * Copyright © 2014 Cask Data, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package co.cask.cdap.examples.sentiment;

import co.cask.cdap.test.ApplicationManager;
import co.cask.cdap.test.FlowManager;
import co.cask.cdap.test.ProcedureClient;
import co.cask.cdap.test.ProcedureManager;
import co.cask.cdap.test.RuntimeMetrics;
import co.cask.cdap.test.RuntimeStats;
import co.cask.cdap.test.StreamWriter;
import co.cask.cdap.test.TestBase;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import com.google.common.reflect.TypeToken;
import com.google.gson.Gson;
import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 *
 */
public class SentimentAnalysisTest extends TestBase {

  @Test
  public void test() throws Exception {
    try {
      ApplicationManager appManager = deployApplication(SentimentAnalysisApp.class);

      Map<String, String> args = Maps.newHashMap();
      args.put("disable.public", "true");

      // Starts a Flow
      FlowManager flowManager = appManager.startFlow("analysis", args);

      try {
        // Write a message to Stream
        StreamWriter streamWriter = appManager.getStreamWriter("sentence");
        streamWriter.send("i love movie");
        streamWriter.send("i hate movie");
        streamWriter.send("i am neutral to movie");
        streamWriter.send("i am happy today that I got this working.");

        // Wait for the last Flowlet processed all tokens
        RuntimeMetrics countMetrics = RuntimeStats.getFlowletMetrics("sentiment", "analysis", "update");
        countMetrics.waitForProcessed(4, 15, TimeUnit.SECONDS);
      } finally {
        flowManager.stop();
      }

      // Start procedure and verify
      ProcedureManager procedureManager = appManager.startProcedure("sentiment-query");
      try {
        String response = procedureManager.getClient().query("aggregates", Collections.<String, String>emptyMap());

        // Verify the aggregates
        Map<String, Long> result = new Gson().fromJson(response, new TypeToken<Map<String, Long>>() { }.getType());
        Assert.assertEquals(2, result.get("positive").intValue());
        Assert.assertEquals(1, result.get("negative").intValue());
        Assert.assertEquals(1, result.get("neutral").intValue());

        // Verify retrieval of sentiments
        response = procedureManager.getClient().query("sentiments", ImmutableMap.of("sentiment", "positive"));
        result = new Gson().fromJson(response, new TypeToken<Map<String, Long>>() { }.getType());
        Assert.assertEquals(ImmutableSet.of("i love movie", "i am happy today that I got this working."),
                            result.keySet());

        response = procedureManager.getClient().query("sentiments", ImmutableMap.of("sentiment", "negative"));
        result = new Gson().fromJson(response, new TypeToken<Map<String, Long>>() { }.getType());
        Assert.assertEquals(ImmutableSet.of("i hate movie"), result.keySet());

        response = procedureManager.getClient().query("sentiments", ImmutableMap.of("sentiment", "neutral"));
        result = new Gson().fromJson(response, new TypeToken<Map<String, Long>>() { }.getType());
        Assert.assertEquals(ImmutableSet.of("i am neutral to movie"), result.keySet());
      } finally {
        procedureManager.stop();
      }
    } finally {
      TimeUnit.SECONDS.sleep(1);
      clear();
    }
  }

  @Test
  public void test2() throws InterruptedException, IOException {
    ApplicationManager appManager = deployApplication(SentimentAnalysisApp.class);
    Map<String, String> args = Maps.newHashMap();
//    args.put("oauth.consumerKey", "guzpQsLtdKs0jlap64nY1nX4N");
//    args.put("oauth.consumerSecret", "z7Ux5TPQyeOXd8xlXmm87V3qxi1vABSV9NFVelDaCus8m39tIe");
//    args.put("oauth.accessToken", "366210197-mpzoVZgENXzrEnVXvgdOqoDkCv55m2M5IYSp4ouv");
//    args.put("oauth.accessTokenSecret", "Q9ST3W4d68KgBnIKmQYWgFaBdCBrVsQXtE54ol8UhudoL");
    FlowManager flowManager = appManager.startFlow("analysis", args);

    ProcedureManager procedureManager = appManager.startProcedure("sentiment-query");
    ProcedureClient client = procedureManager.getClient();

    for(int i = 0 ; i < 30; i++) {
      String response = client.query("aggregates", Collections.<String, String>emptyMap());
      TimeUnit.SECONDS.sleep(1);
    }

    flowManager.stop();

    // use procedure and verify
    try {
      String response = client.query("aggregates", Collections.<String, String>emptyMap());

      // Verify the aggregates
      Map<String, Long> result = new Gson().fromJson(response, new TypeToken<Map<String, Long>>() { }.getType());
      Assert.assertTrue(1 < result.get("positive").intValue());
      Assert.assertTrue(1 < result.get("negative").intValue());
      Assert.assertTrue(1 < result.get("neutral").intValue());

    } finally {
      procedureManager.stop();
    }
  }


}

@@ -4077,1495 +4077,7 @@
  %7D%0A%0A
-  @Test%0A  public void test2() throws InterruptedException, IOException %7B%0A    ApplicationManager appManager = deployApplication(SentimentAnalysisApp.class);%0A    Map%3CString, String%3E args = Maps.newHashMap();%0A//    args.put(%22oauth.consumerKey%22, %22guzpQsLtdKs0jlap64nY1nX4N%22);%0A//    args.put(%22oauth.consumerSecret%22, %22z7Ux5TPQyeOXd8xlXmm87V3qxi1vABSV9NFVelDaCus8m39tIe%22);%0A//    args.put(%22oauth.accessToken%22, %22366210197-mpzoVZgENXzrEnVXvgdOqoDkCv55m2M5IYSp4ouv%22);%0A//    args.put(%22oauth.accessTokenSecret%22, %22Q9ST3W4d68KgBnIKmQYWgFaBdCBrVsQXtE54ol8UhudoL%22);%0A    FlowManager flowManager = appManager.startFlow(%22analysis%22, args);%0A%0A    ProcedureManager procedureManager = appManager.startProcedure(%22sentiment-query%22);%0A    ProcedureClient client = procedureManager.getClient();%0A%0A    for(int i = 0 ; i %3C 30; i++) %7B%0A      String response = client.query(%22aggregates%22, Collections.%3CString, String%3EemptyMap());%0A      TimeUnit.SECONDS.sleep(1);%0A    %7D%0A%0A    flowManager.stop();%0A%0A    // use procedure and verify%0A    try %7B%0A      String response = client.query(%22aggregates%22, Collections.%3CString, String%3EemptyMap());%0A%0A      // Verify the aggregates%0A      Map%3CString, Long%3E result = new Gson().fromJson(response, new TypeToken%3CMap%3CString, Long%3E%3E() %7B %7D.getType());%0A      Assert.assertTrue(1 %3C result.get(%22positive%22).intValue());%0A      Assert.assertTrue(1 %3C result.get(%22negative%22).intValue());%0A      Assert.assertTrue(1 %3C result.get(%22neutral%22).intValue());%0A%0A    %7D finally %7B%0A      procedureManager.stop();%0A    %7D%0A  %7D%0A%0A
 %0A%7D%0A

remove unstable test (which relies on twitter sending particular tweets).