@Test
public void testAppDataPush() throws Exception
{

  if (StramTestSupport.isInTravis()) {
    // disable this test in travis because of an intermittent problem similar to this:
    // http://stackoverflow.com/questions/32172925/travis-ci-sporadic-timeouts-to-localhost
    // We should remove this when we find a solution to this.
    LOG.info("Test testAppDataPush is disabled in Travis");
    return;
  }
  final String topic = "xyz";
  final List<String> messages = new ArrayList<>();
  EmbeddedWebSocketServer server = new EmbeddedWebSocketServer(0);
  server.setWebSocket(new WebSocket.OnTextMessage()
  {
    @Override
    public void onMessage(String data)
    {
      messages.add(data);
    }
    @Override
    public void onOpen(WebSocket.Connection connection)
    {
    }
    @Override
    public void onClose(int closeCode, String message)
    {
    }
  });
  try {
    server.start();
    int port = server.getPort();
    TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
    GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
    dag.addStream("o1.outport", o1.outport, o2.inport1);
    dag.setAttribute(LogicalPlan.METRICS_TRANSPORT, new AutoMetricBuiltInTransport(topic));
    dag.setAttribute(LogicalPlan.GATEWAY_CONNECT_ADDRESS, "localhost:" + port);
    dag.setAttribute(LogicalPlan.PUBSUB_CONNECT_TIMEOUT_MILLIS, 2000);
    StramLocalCluster lc = new StramLocalCluster(dag);
    StreamingContainerManager dnmgr = lc.dnmgr;
    StramAppContext appContext = new StramTestSupport.TestAppContext(dag.getAttributes());
    AppDataPushAgent pushAgent = new AppDataPushAgent(dnmgr, appContext);
    pushAgent.init();
    pushAgent.pushData();
    Thread.sleep(1000);
    Assert.assertTrue(messages.size() > 0);
    pushAgent.close();
    JSONObject message = new JSONObject(messages.get(0));
    Assert.assertEquals(topic, message.getString("topic"));
    Assert.assertEquals("publish", message.getString("type"));
    JSONObject data = message.getJSONObject("data");
    Assert.assertTrue(StringUtils.isNotBlank(data.getString("appId")));
    Assert.assertTrue(StringUtils.isNotBlank(data.getString("appUser")));
    Assert.assertTrue(StringUtils.isNotBlank(data.getString("appName")));
    JSONObject logicalOperators = data.getJSONObject("logicalOperators");
    for (String opName : new String[]{"o1", "o2"}) {
      JSONObject opObj = logicalOperators.getJSONObject(opName);
      Assert.assertTrue(opObj.has("totalTuplesProcessed"));
      Assert.assertTrue(opObj.has("totalTuplesEmitted"));
      Assert.assertTrue(opObj.has("tuplesProcessedPSMA"));
      Assert.assertTrue(opObj.has("tuplesEmittedPSMA"));
      Assert.assertTrue(opObj.has("latencyMA"));
    }
  } finally {
    server.stop();
  }
}