gt:         + TAGV_STRING + "\"}}");
context: // This file is part of OpenTSDB.
  // Copyright (C) 2013-2016  The OpenTSDB Authors.
  //
  // This program is free software: you can redistribute it and/or modify it
  // under the terms of the GNU Lesser General Public License as published by
  // the Free Software Foundation, either version 2.1 of the License, or (at your
  // option) any later version.  This program is distributed in the hope that it
  // will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  // of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
  // General Public License for more details.  You should have received a copy
  // of the GNU Lesser General Public License along with this program.  If not,
  // see <http://www.gnu.org/licenses/>.
  package net.opentsdb.tsd;
  
  import static org.mockito.Matchers.any;
  import static org.mockito.Matchers.anyLong;
  import static org.mockito.Matchers.anyString;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.never;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.when;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  
  import java.nio.charset.Charset;
  import java.util.HashMap;
  
  import org.hbase.async.HBaseException;
  import org.hbase.async.PleaseThrottleException;
  import org.hbase.async.PutRequest;
  import org.jboss.netty.channel.Channel;
  import org.jboss.netty.handler.codec.http.HttpResponseStatus;
  import org.junit.Test;
  import org.junit.runner.RunWith;
  import org.powermock.core.classloader.annotations.PowerMockIgnore;
  import org.powermock.modules.junit4.PowerMockRunner;
  
  import com.stumbleupon.async.Deferred;
  
  @RunWith(PowerMockRunner.class)
  //"Classloader hell"...  It's real.  Tell PowerMock to ignore these classes
  //because they fiddle with the class loader.  We don't test them anyway.
  @PowerMockIgnore({"javax.management.*", "javax.xml.*",
               "ch.qos.*", "org.slf4j.*",
               "com.sum.*", "org.xml.*"})
  public final class TestPutRpc extends BaseTestPutRpc {
    
    @Test
    public void constructor() {
      assertNotNull(new PutDataPointRpc(tsdb.getConfig()));
    }
    
    // Socket RPC Tests ------------------------------------
  
    @Test
    public void execute() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      verify(chan, never()).write(any());
      verify(chan, never()).isConnected();
      validateSEH(false);
    }
  
    @Test
    public void executeBadValue() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "notanum", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(false);
    }
    
    @Test
    public void executeMissingMetric() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", "", 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(false);
    }
    
    @Test
    public void executeMissingMetricNotWriteable() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      when(chan.isWritable()).thenReturn(false);
      put.execute(tsdb, chan, new String[] { "put", "", 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 1, 0, 0);
      verify(chan, never()).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(false);
    }
    
    @Test
    public void executeUnknownMetric() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", NSUN_METRIC, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(false);
    }
    
    @SuppressWarnings("unchecked")
    @Test (expected = RuntimeException.class)
    public void executeRuntimeException() throws Exception {
      when(tsdb.addPoint(anyString(), anyLong(), anyLong(), 
          (HashMap<String, String>)any()))
          .thenThrow(new RuntimeException("Fail!"));
      
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING });
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0);
      validateSEH(true);
    }
    
    @Test
    public void executeHBaseError() throws Exception {
      when(client.put(any(PutRequest.class)))
          .thenReturn(Deferred.fromError(mock(HBaseException.class)));
      
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(true);
    }
    
    @Test
    public void executeHBaseErrorNotWriteable() throws Exception {
      when(client.put(any(PutRequest.class)))
        .thenReturn(Deferred.fromError(mock(HBaseException.class)));
      
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      when(chan.isWritable()).thenReturn(false);
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0);
      verify(chan, never()).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(true);
    }
    
    @Test
    public void executeHBaseErrorHandler() throws Exception {
      when(client.put(any(PutRequest.class)))
        .thenReturn(Deferred.fromError(mock(HBaseException.class)));
      setStorageExceptionHandler();
      
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(true);
    }
  
    @Test
    public void executePleaseThrottle() throws Exception {
      when(client.put(any(PutRequest.class)))
        .thenReturn(Deferred.fromError(mock(PleaseThrottleException.class)));
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(true);
    }
    
    @Test
    public void executePleaseThrottleNotWriteable() throws Exception {
      when(client.put(any(PutRequest.class)))
        .thenReturn(Deferred.fromError(mock(PleaseThrottleException.class)));
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      when(chan.isWritable()).thenReturn(false);
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0);
      verify(chan, never()).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(true);
    }
    
    @Test
    public void executePleaseThrottleHandler() throws Exception {
      setStorageExceptionHandler();
      when(client.put(any(PutRequest.class)))
        .thenReturn(Deferred.fromError(mock(PleaseThrottleException.class)));
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(true);
    }
    
    @Test (expected = NullPointerException.class)
    public void executeNullTSDB() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(null, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING });
    }
    
    @Test
    public void executeNullChannelOK() throws Exception {
      // we can pass in a null channel but since we only write when an error occurs
      // then we won't fail.
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, null, new String[] { "put", METRIC_STRING, 
          "1365465600", "42", TAGK_STRING + "=" + TAGV_STRING })
        .joinUninterruptibly();
      assertEquals(1, telnet_requests.get());
      assertEquals(0, invalid_values.get());
      validateCounters(1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test (expected = NullPointerException.class)
    public void executeNullChannelError() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, null, new String[] { "put", METRIC_STRING, 
          "1365465600", "notanumber", TAGK_STRING + "=" + TAGV_STRING });
    }
    
    @Test (expected = NullPointerException.class)
    public void executeNullArray() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, null);
    }
    
    @Test (expected = ArrayIndexOutOfBoundsException.class)
    public void executeEmptyArray() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[0]);
    }
    
    @Test
    public void executeShortArray() throws Exception {
      final PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      final Channel chan = NettyMocks.fakeChannel();
      put.execute(tsdb, chan, new String[] { "put", METRIC_STRING, 
          "1365465600", "42" }).joinUninterruptibly();
      validateCounters(1, 0, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0);
      verify(chan, times(1)).write(any());
      verify(chan, times(1)).isConnected();
      validateSEH(false);
    }
    
    // HTTP RPC Tests --------------------------------------
    
    @Test
    public void putSingle() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put", 
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
          +":42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.NO_CONTENT, query.response().getStatus());
      validateCounters(0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
  
    @Test
    public void putTwo() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put", 
          "[{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
          + ":42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING 
          + "\"}},{\"metric\":\"" + METRIC_B_STRING + "\","
          + "\"timestamp\":1365465600,\"value\":24,\"tags\":"
          + "{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}]");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.NO_CONTENT, query.response().getStatus());
      validateCounters(0, 1, 2, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putSingleSummary() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put?summary", 
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
              +":42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.OK, query.response().getStatus());
      final String response = 
        query.response().getContent().toString(Charset.forName("UTF-8"));
      assertTrue(response.contains("\"failed\":0"));
      assertTrue(response.contains("\"success\":1"));
      validateCounters(0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putSingleDetails() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put?details", 
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
              +":42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.OK, query.response().getStatus());
      final String response = 
        query.response().getContent().toString(Charset.forName("UTF-8"));
      assertTrue(response.contains("\"failed\":0"));
      assertTrue(response.contains("\"success\":1"));
      assertTrue(response.contains("\"errors\":[]"));
      validateCounters(0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putSingleSummaryAndDetails() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put?summary&details", 
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
              +":42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.OK, query.response().getStatus());
      final String response = 
        query.response().getContent().toString(Charset.forName("UTF-8"));
      assertTrue(response.contains("\"failed\":0"));
      assertTrue(response.contains("\"success\":1"));
      assertTrue(response.contains("\"errors\":[]"));
      validateCounters(0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putTwoSummary() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put?summary", 
          "[{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
              + ":42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING 
              + "\"}},{\"metric\":\"" + METRIC_B_STRING + "\","
              + "\"timestamp\":1365465600,\"value\":24,\"tags\":"
              + "{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}]");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.OK, query.response().getStatus());
      final String response = 
        query.response().getContent().toString(Charset.forName("UTF-8"));
      assertTrue(response.contains("\"failed\":0"));
      assertTrue(response.contains("\"success\":2"));
      validateCounters(0, 1, 2, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putNegativeInt() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put", 
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
              +":-42,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.NO_CONTENT, query.response().getStatus());
      validateCounters(0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putFloat() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put", 
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
              +":42.2,\"tags\":{\"" + TAGK_STRING + "\":\"" + TAGV_STRING + "\"}}");
      PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
      put.execute(tsdb, query);
      assertEquals(HttpResponseStatus.NO_CONTENT, query.response().getStatus());
      validateCounters(0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
      validateSEH(false);
    }
    
    @Test
    public void putDoublePrecisionFloatingPoint() throws Exception {
      HttpQuery query = NettyMocks.postQuery(tsdb, "/api/put",
          "{\"metric\":\"" + METRIC_STRING + "\",\"timestamp\":1365465600,\"value\""
          + ":123456.123456789,\"tags\":{\"" + TAGK_STRING + "\":\"" 
  