<commit_before>/*
 * Copyright © 2014 <NAME>, 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.sparkkmeans;

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.SparkManager;
import co.cask.cdap.test.StreamWriter;
import co.cask.cdap.test.TestBase;
import com.google.common.collect.ImmutableMap;
import com.google.gson.Gson;
import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Test SparkKMeansApp
 */
public class SparkKMeansAppTest extends TestBase {

  private static final Gson GSON = new Gson();

  @Test
  public void test() throws Exception {
    // Deploy an Application
    ApplicationManager appManager = deployApplication(SparkKMeansApp.class);
    // Start a Flow
    FlowManager flowManager = appManager.startFlow("PointsFlow");
    try {
      // Inject data
      sendData(appManager);
      // Wait for the last Flowlet processing 3 events, or at most 5 seconds
      RuntimeMetrics metrics = RuntimeStats.getFlowletMetrics("SparkKMeans", "PointsFlow", "reader");
      metrics.waitForProcessed(3, 5, TimeUnit.SECONDS);
      // Start a Spark Program
      SparkManager sparkManager = appManager.startSpark("SparkKMeansProgram");
      sparkManager.waitForFinish(60, TimeUnit.SECONDS);
    } finally {
      flowManager.stop();
    }
    // Verify data processed well
    verifyCentersProcedure(appManager);
  }

  /**
   * verify execution of the CentersProcedure
   */
  private void verifyCentersProcedure(ApplicationManager appManager) throws IOException, ParseException {
    ProcedureManager procedureManager = appManager.startProcedure(
      SparkKMeansApp.CentersProcedure.class.getSimpleName());
    try {
      ProcedureClient client = procedureManager.getClient();
      Map<String, String> params = new HashMap<String, String>();
      params.put("index", "1");
      String response = client.query("centers", params);
      String centerCoordinates = GSON.fromJson(response, String.class);
      Assert.assertNotNull(centerCoordinates);
      String[] coordinates = centerCoordinates.split(",");
      Assert.assertTrue(coordinates.length == 3);
      for (String coordinate : coordinates) {
        double value = Double.parseDouble(coordinate);
        Assert.assertTrue(value > 0);
      }
      // test wrong argument
      Throwable exception = null;
      try {
        client.query("centers", ImmutableMap.of("position", "0"));
      } catch (Throwable ex) {
        exception = ex;
      }
      Assert.assertTrue(exception instanceof IOException);
    } finally {
      procedureManager.stop();
    }
  }

  /**
   * Send a few points to the stream
   */
  private void sendData(ApplicationManager appManager) throws IOException {
    StreamWriter streamWriter = appManager.getStreamWriter("pointsStream");
    streamWriter.send("0.0 0.0 0.0");
    streamWriter.send("0.1 0.1 0.1");
    streamWriter.send("0.2 0.2 0.2");
    streamWriter.send("9.0 9.0 9.0");<commit_msg>added wrong index value test<commit_after>/*
 * Copyright © 2014 <NAME>, 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.sparkkmeans;

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.SparkManager;
import co.cask.cdap.test.StreamWriter;
import co.cask.cdap.test.TestBase;
import com.google.common.collect.ImmutableMap;
import com.google.gson.Gson;
import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Test SparkKMeansApp
 */
public class SparkKMeansAppTest extends TestBase {

  private static final Gson GSON = new Gson();

  @Test
  public void test() throws Exception {
    // Deploy an Application
    ApplicationManager appManager = deployApplication(SparkKMeansApp.class);
    // Start a Flow
    FlowManager flowManager = appManager.startFlow("PointsFlow");
    try {
      // Inject data
      sendData(appManager);
      // Wait for the last Flowlet processing 3 events, or at most 5 seconds
      RuntimeMetrics metrics = RuntimeStats.getFlowletMetrics("SparkKMeans", "PointsFlow", "reader");
      metrics.waitForProcessed(3, 5, TimeUnit.SECONDS);
      // Start a Spark Program
      SparkManager sparkManager = appManager.startSpark("SparkKMeansProgram");
      sparkManager.waitForFinish(60, TimeUnit.SECONDS);
    } finally {
      flowManager.stop();
    }
    // Verify data processed well
    verifyCentersProcedure(appManager);
  }

  /**
   * verify execution of the CentersProcedure
   */
  private void verifyCentersProcedure(ApplicationManager appManager) throws IOException, ParseException {
    ProcedureManager procedureManager = appManager.startProcedure(
      SparkKMeansApp.CentersProcedure.class.getSimpleName());
    try {
      ProcedureClient client = procedureManager.getClient();
      Map<String, String> params = new HashMap<String, String>();
      params.put("index", "1");
      String response = client.query("centers", params);
      String centerCoordinates = GSON.fromJson(response, String.class);
      Assert.assertNotNull(centerCoordinates);
      String[] coordinates = centerCoordinates.split(",");
      Assert.assertTrue(coordinates.length == 3);
      for (String coordinate : coordinates) {
        double value = Double.parseDouble(coordinate);
        Assert.assertTrue(value > 0);
      }
      // test wrong argument
      Throwable exception = null;
      try {
        client.query("centers", ImmutableMap.of("position", "0"));
      } catch (Throwable ex) {
        exception = ex;
      }
      Assert.assertTrue(exception instanceof IOException);
      // test with invalid index
      exception = null;
      try {
        client.query("centers", ImmutableMap.of("index", "5"));
      } catch (Throwable ex) {
        exception = ex;
      }
      Assert.assertTrue(exception instanceof IOException);
    } finally {
      procedureManager.stop();
    }
  }

  /**
   * Send a few points to the stream
   */
  private void sendData(ApplicationManager appManager) throws IOException {
    StreamWriter streamWriter = appManager.getStreamWriter("pointsStream");
    streamWriter.send("0.0 0.0 0.0");
    streamWriter.send("0.1 0.1 0.1");
    streamWriter.send("0.2 0.2 0.2");
    streamWriter.send("9.0 9.0 9.0");
{'fraction_non_alphanumeric': 0.10616006059075991, 'fraction_numerical': 0.01072961373390558, 'mean_word_length': 3.5718407386035778, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2659135', 'n_tokens_mistral': 2303, 'n_tokens_neox': 2251, 'n_words': 609}