<commit_before>package com.continuuity.metrics2.collector.server;

import akka.dispatch.Await;
import akka.util.Duration;
import com.continuuity.common.conf.CConfiguration;
import com.continuuity.common.conf.Constants;
import com.continuuity.metrics2.collector.MetricRequest;
import com.continuuity.metrics2.collector.MetricResponse;
import com.continuuity.metrics2.collector.server.plugins.FlowMetricsProcessor;
import com.continuuity.metrics2.collector.server.plugins.MetricsProcessor;
import org.junit.Assert;
import org.junit.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.concurrent.TimeUnit;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;

/**
 * Tests Flow Metric Processor.
 */
public class FlowMetricsProcessorTest {
  private static String connectionUrl;
  private static CConfiguration configuration;

  // Static section executed when class is created.
  {
    connectionUrl = "jdbc:hsqldb:mem:metrictest?user=sa";
    configuration = CConfiguration.create();
    configuration.setStrings(Constants.CFG_METRICS_CONNECTION_URL,
                             connectionUrl);
  }

  @Test
  public void testAddingMetric() throws Exception {
    MetricsProcessor processor
      = new FlowMetricsProcessor(configuration);

    for(int i = 1; i <= 10; ++i) {
      MetricRequest request = new MetricRequest.Builder(true)
        .setRequestType("put")
        .setMetricName("accountId.applicationId.flowId.runId.flowletId.1.processed")
        .setTimestamp(System.currentTimeMillis()/1000)
        .setValue(i)
        .setMetricType("FlowSystem")
        .create();
        // Makes a blocking call.
        MetricResponse.Status status = Await.result(processor.process(request),
                                          Duration.create(1, TimeUnit.SECONDS));
        Assert.assertTrue(status == MetricResponse.Status.SUCCESS);
    }

    // Now try reading the metric add back from sql.
    Connection connection = DriverManager.getConnection(connectionUrl);

    assertNotNull(connection);
    String readSQL = "SELECT * FROM metrics";
    Statement stmt = null;
    try {
       stmt = connection.createStatement();
       assertNotNull(stmt);
       if(stmt.execute(readSQL)) {
         ResultSet rs = stmt.getResultSet();
         assertNotNull(rs);
         // We expect only one row here.
         while(rs.next()) {
           assertTrue(10.0 == rs.getFloat("value"));<commit_msg>ENG-646 First cut of time series metrics -- still lot more work to do<commit_after>package com.continuuity.metrics2.collector.server;

import akka.dispatch.Await;
import akka.util.Duration;
import com.continuuity.common.conf.CConfiguration;
import com.continuuity.common.conf.Constants;
import com.continuuity.metrics2.collector.MetricRequest;
import com.continuuity.metrics2.collector.MetricResponse;
import com.continuuity.metrics2.collector.server.plugins.FlowMetricsProcessor;
import com.continuuity.metrics2.collector.server.plugins.MetricsProcessor;
import org.junit.Assert;
import org.junit.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.concurrent.TimeUnit;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;

/**
 * Tests Flow Metric Processor.
 */
public class FlowMetricsProcessorTest {
  private static String connectionUrl;
  private static CConfiguration configuration;

  // Static section executed when class is created.
  {
    connectionUrl = "jdbc:hsqldb:mem:metrictest?user=sa";
    configuration = CConfiguration.create();
    configuration.setStrings(Constants.CFG_METRICS_CONNECTION_URL,
                             connectionUrl);
  }

  @Test
  public void testAddingMetric() throws Exception {
    MetricsProcessor processor
      = new FlowMetricsProcessor(configuration);

    for(int i = 1; i <= 10; ++i) {
      MetricRequest request = new MetricRequest.Builder(true)
        .setRequestType("put")
        .setMetricName("accountId.applicationId.flowId.runId.flowletId.1.processed")
        .setTimestamp(System.currentTimeMillis()/1000)
        .setValue(i)
        .setMetricType("FlowSystem")
        .create();
        // Makes a blocking call.
        MetricResponse.Status status = Await.result(processor.process(request),
                                          Duration.create(5, TimeUnit.SECONDS));
        Assert.assertTrue(status == MetricResponse.Status.SUCCESS);
    }

    // Now try reading the metric add back from sql.
    Connection connection = DriverManager.getConnection(connectionUrl);

    assertNotNull(connection);
    String readSQL = "SELECT * FROM metrics";
    Statement stmt = null;
    try {
       stmt = connection.createStatement();
       assertNotNull(stmt);
       if(stmt.execute(readSQL)) {
         ResultSet rs = stmt.getResultSet();
         assertNotNull(rs);
         // We expect only one row here.
         while(rs.next()) {
           assertTrue(10.0 == rs.getFloat("value"));
{'fraction_non_alphanumeric': 0.10074211502782932, 'fraction_numerical': 0.006864564007421151, 'mean_word_length': 3.4962468723936615, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3024902', 'n_tokens_mistral': 1515, 'n_tokens_neox': 1477, 'n_words': 313}