import org.apache.storm.Testing; //导入依赖的package包/类
@Test
public void shouldSuccessfulSendDatapoint() throws Exception {
    Datapoint datapoint = new Datapoint("metric", timestamp, Collections.emptyMap(), 123);

    MockedSources sources = new MockedSources();
    // TODO: rather than use Topic.OTSDB, grab it from the TopologyConfig object (which does
    // not exist at this point in the code.
    sources.addMockData(Topic.OTSDB+"-spout",
            new Values(MAPPER.writeValueAsString(datapoint)));
    completeTopologyParam.setMockedSources(sources);

    Testing.withTrackedCluster(clusterParam, (cluster) ->  {
        OpenTSDBTopology topology = new TestingTargetTopology(new TestingKafkaBolt());
        StormTopology stormTopology = topology.createTopology();

        Map result = Testing.completeTopology(cluster, stormTopology, completeTopologyParam);
    });

    //verify that request is sent to OpenTSDB server
    mockServer.verify(HttpRequest.request(), VerificationTimes.exactly(1));
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
@Test
public void shouldSendDatapointRequestsOnlyOnce() throws Exception {
    Datapoint datapoint = new Datapoint("metric", timestamp, Collections.emptyMap(), 123);
    String jsonDatapoint = MAPPER.writeValueAsString(datapoint);

    MockedSources sources = new MockedSources();
    sources.addMockData(Topic.OTSDB+"-spout",
            new Values(jsonDatapoint), new Values(jsonDatapoint));
    completeTopologyParam.setMockedSources(sources);

    Testing.withTrackedCluster(clusterParam, (cluster) ->  {
        OpenTSDBTopology topology = new TestingTargetTopology(new TestingKafkaBolt());
        StormTopology stormTopology = topology.createTopology();

        Testing.completeTopology(cluster, stormTopology, completeTopologyParam);
    });
    //verify that request is sent to OpenTSDB server once
    mockServer.verify(HttpRequest.request(), VerificationTimes.exactly(1));
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private void runTopology(File responderFile) throws Exception
{
  MkClusterParam mkClusterParam = new MkClusterParam();
  // The test sometimes fails because of timing issues when more than 1 supervisor set.
  mkClusterParam.setSupervisors(1);

  // Maybe using "withSimulatedTimeLocalCluster" would be better to avoid worrying about timing.
  Config conf = PirkTopology.createStormConf();
  conf.put(StormConstants.OUTPUT_FILE_KEY, responderFile.getAbsolutePath());
  conf.put(StormConstants.N_SQUARED_KEY, nSquared.toString());
  conf.put(StormConstants.QUERY_INFO_KEY, queryInfo.toMap());
  // conf.setDebug(true);
  mkClusterParam.setDaemonConf(conf);

  TestJob testJob = createPirkTestJob(conf);
  Testing.withLocalCluster(mkClusterParam, testJob);
  // Testing.withSimulatedTimeLocalCluster(mkClusterParam, testJob);
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
public void shouldSendOKOnStartupForDeterministic() {
  long t1 = 14000;
  bolt.setCurrentTime(t1);
  sendSubAlarmCreated(metricDef4, subAlarm4);

  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setStream(MetricAggregationBolt.METRIC_AGGREGATION_CONTROL_STREAM);
  final Tuple lagTuple =
      Testing.testTuple(Arrays.asList(MetricAggregationBolt.METRICS_BEHIND), tupleParam);
  bolt.execute(lagTuple);
  verify(collector, times(1)).ack(lagTuple);

  // Won't send OK on this Tick Tuple because of the METRIC_BEHIND message sent above
  t1 += 60000;
  bolt.setCurrentTime(t1);
  sendTickTuple();
  verify(collector, never()).emit(new Values(subAlarm4.getAlarmId(), subAlarm4));

  t1 += 60000;
  bolt.setCurrentTime(t1);
  sendTickTuple();
  assertEquals(subAlarm4.getState(), AlarmState.OK);

  verify(collector, times(1)).emit(new Values(subAlarm4.getAlarmId(), subAlarm4));
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
public void validateMetricDefDeleted() {
  MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.METRIC_ALARM_EVENT_STREAM_FIELDS);
  tupleParam.setStream(EventProcessingBolt.METRIC_ALARM_EVENT_STREAM_ID);
  MetricDefinitionAndTenantId metricDefinitionAndTenantId =
      new MetricDefinitionAndTenantId(metricDef1, TENANT_ID);
  bolt.getOrCreateSubAlarmStatsRepo(metricDefinitionAndTenantId);

  sendSubAlarmCreated(metricDef1, subAlarm1);

  assertNotNull(bolt.metricDefToSubAlarmStatsRepos.get(metricDefinitionAndTenantId).get(ALARM_ID_1));

  // We don't have an AlarmDefinition so no id, but the MetricAggregationBolt doesn't use this
  // field anyways
  final String alarmDefinitionId = "";
  bolt.execute(Testing.testTuple(Arrays.asList(EventProcessingBolt.DELETED,
      new TenantIdAndMetricName(metricDefinitionAndTenantId), metricDefinitionAndTenantId,
      alarmDefinitionId, ALARM_ID_1), tupleParam));

  assertNull(bolt.metricDefToSubAlarmStatsRepos.get(metricDefinitionAndTenantId));
  assertTrue(bolt.subAlarmRemoved(ALARM_ID_1, metricDefinitionAndTenantId));
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private void sendAlarmDefinitionDeleted(final AlarmDefinition alarmDefinition) {
  final Map<String, MetricDefinition> subAlarmMetricDefinitions = new HashMap<>();
  for (final AlarmSubExpression subExpr : alarmDefinition.getAlarmExpression().getSubExpressions()) {
    subAlarmMetricDefinitions.put(getNextId(), subExpr.getMetricDefinition());
  }
  // Delete the Alarm Definition
  final AlarmDefinitionDeletedEvent event =
      new AlarmDefinitionDeletedEvent(alarmDefinition.getId(), subAlarmMetricDefinitions);
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.ALARM_DEFINITION_EVENT_FIELDS);
  tupleParam.setStream(EventProcessingBolt.ALARM_DEFINITION_EVENT_STREAM_ID);
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.DELETED, event), tupleParam);

  bolt.execute(tuple);
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
@Test
public void portStatsTest() throws Exception {
    final String switchId = "00:00:00:00:00:00:00:01";
    final List<PortStatsEntry> entries = IntStream.range(1, 53).boxed().map(port -> {
        int baseCount = port * 20;
        return new PortStatsEntry(port, baseCount, baseCount + 1, baseCount + 2, baseCount + 3,
                baseCount + 4, baseCount + 5, baseCount + 6, baseCount + 7,
                baseCount + 8, baseCount + 9, baseCount + 10, baseCount + 11);
    }).collect(toList());
    final List<PortStatsReply> replies = Collections.singletonList(new PortStatsReply(1, entries));
    InfoMessage message = new InfoMessage(new PortStatsData(switchId, replies), timestamp, CORRELATION_ID,
            Destination.WFM_STATS);

    //mock kafka spout
    MockedSources sources = new MockedSources();
    sources.addMockData(StatsComponentType.STATS_OFS_KAFKA_SPOUT.toString(),
            new Values(MAPPER.writeValueAsString(message)));
    completeTopologyParam.setMockedSources(sources);

    //execute topology
    Testing.withTrackedCluster(clusterParam, (cluster) ->  {
        StatsTopology topology = new TestingTargetTopology(new TestingKafkaBolt());
        StormTopology stormTopology = topology.createTopology();

        //verify results
        Map result = Testing.completeTopology(cluster, stormTopology, completeTopologyParam);
        ArrayList<FixedTuple> tuples =
                (ArrayList<FixedTuple>) result.get(StatsComponentType.PORT_STATS_METRIC_GEN.name());
        assertThat(tuples.size(), is(728));
        tuples.stream()
                .map(this::readFromJson)
                .forEach(datapoint -> {
                    assertThat(datapoint.getTags().get("switchId"), is(switchId.replaceAll(":", "")));
                    assertThat(datapoint.getTime(), is(timestamp));
                    assertThat(datapoint.getMetric(), startsWith("pen.switch"));
                });
    });
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
@Test
public void meterConfigStatsTest() throws Exception {
    final String switchId = "00:00:00:00:00:00:00:01";
    final List<MeterConfigReply> stats = Collections.singletonList(new MeterConfigReply(2, Arrays.asList(1L, 2L, 3L)));
    InfoMessage message = new InfoMessage(new MeterConfigStatsData(switchId, stats), timestamp, CORRELATION_ID,
            Destination.WFM_STATS);

    //mock kafka spout
    MockedSources sources = new MockedSources();
    sources.addMockData(StatsComponentType.STATS_OFS_KAFKA_SPOUT.toString(),
            new Values(MAPPER.writeValueAsString(message)));
    completeTopologyParam.setMockedSources(sources);

    //execute topology
    Testing.withTrackedCluster(clusterParam, (cluster) ->  {
        StatsTopology topology = new TestingTargetTopology(new TestingKafkaBolt());
        StormTopology stormTopology = topology.createTopology();

        //verify results
        Map result = Testing.completeTopology(cluster, stormTopology, completeTopologyParam);
        ArrayList<FixedTuple> tuples =
                (ArrayList<FixedTuple>) result.get(StatsComponentType.METER_CFG_STATS_METRIC_GEN.name());
        assertThat(tuples.size(), is(3));
        tuples.stream()
                .map(this::readFromJson)
                .forEach(datapoint -> {
                    assertThat(datapoint.getTags().get("switchid"), is(switchId.replaceAll(":", "")));
                    assertThat(datapoint.getTime(), is(timestamp));
                    assertThat(datapoint.getMetric(), is("pen.switch.meters"));
                });
    });
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
@Test
public void flowStatsTest() throws Exception {
    final String switchId = "00:00:00:00:00:00:00:01";

    List<FlowStatsEntry> entries = Collections.singletonList(new FlowStatsEntry((short) 1, 0x1FFFFFFFFL, 1500L, 3000L));
    final List<FlowStatsReply> stats = Collections.singletonList(new FlowStatsReply(3, entries));
    InfoMessage message = new InfoMessage(new FlowStatsData(switchId, stats),
            timestamp, CORRELATION_ID, Destination.WFM_STATS);

    //mock kafka spout
    MockedSources sources = new MockedSources();
    sources.addMockData(StatsComponentType.STATS_OFS_KAFKA_SPOUT.toString(),
            new Values(MAPPER.writeValueAsString(message)));
    completeTopologyParam.setMockedSources(sources);

    //execute topology
    Testing.withTrackedCluster(clusterParam, (cluster) ->  {
        StatsTopology topology = new TestingTargetTopology(new TestingKafkaBolt());
        StormTopology stormTopology = topology.createTopology();

        Map result = Testing.completeTopology(cluster, stormTopology, completeTopologyParam);

        //verify results which were sent to Kafka bold
        ArrayList<FixedTuple> tuples =
                (ArrayList<FixedTuple>) result.get(StatsComponentType.FLOW_STATS_METRIC_GEN.name());
        assertThat(tuples.size(), is(4));
        tuples.stream()
                .map(this::readFromJson)
                .forEach(datapoint -> {
                    assertThat(datapoint.getTags().get("switchid"), is(switchId.replaceAll(":", "")));
                    assertThat(datapoint.getTime(), is(timestamp));
                });
    });
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createTuple(final Object event) {
  MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields("event");
  tupleParam.setStream(Streams.DEFAULT_STREAM_ID);
  final Tuple tuple = Testing.testTuple(Arrays.asList(event), tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createSubExpressionUpdated(final SubExpression newExpr,
                                         final String alarmDefinitionId) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.METRIC_SUB_ALARM_EVENT_STREAM_FIELDS);
  tupleParam.setStream(EventProcessingBolt.METRIC_SUB_ALARM_EVENT_STREAM_ID);
  return Testing.testTuple(
      Arrays.asList(EventProcessingBolt.UPDATED, newExpr, alarmDefinitionId), tupleParam);
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createAlarmUpdateTuple(AlarmUpdatedEvent event) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.ALARM_EVENT_STREAM_FIELDS);
  tupleParam.setStream(EventProcessingBolt.ALARM_EVENT_STREAM_ID);
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.UPDATED, event.alarmId, event),
          tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createAlarmDefinitionUpdateTuple(AlarmDefinitionUpdatedEvent event) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.ALARM_DEFINITION_EVENT_FIELDS);
  tupleParam.setStream(EventProcessingBolt.ALARM_DEFINITION_EVENT_STREAM_ID);
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.UPDATED, event), tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createSubAlarmStateChangeTuple(String alarmId, final SubAlarm subAlarm) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields("alarmId", "subAlarm");
  tupleParam.setStream(Streams.DEFAULT_STREAM_ID);
  final Tuple tuple = Testing.testTuple(Arrays.asList(alarmId, subAlarm), tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createNewAlarmDefinitionTuple(final AlarmDefinition alarmDef) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.ALARM_DEFINITION_EVENT_FIELDS);
  tupleParam.setStream(EventProcessingBolt.ALARM_DEFINITION_EVENT_STREAM_ID);
  final AlarmDefinitionCreatedEvent event =
      new AlarmDefinitionCreatedEvent(alarmDef.getTenantId(), alarmDef.getId(),
          alarmDef.getName(), alarmDef.getDescription(), alarmDef.getAlarmExpression()
              .getExpression(), createSubExpressionMap(alarmDef), alarmDef.getMatchBy());
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.CREATED, event), tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createDeleteAlarmDefinitionTuple(final AlarmDefinition alarmDef) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.ALARM_DEFINITION_EVENT_FIELDS);
  tupleParam.setStream(EventProcessingBolt.ALARM_DEFINITION_EVENT_STREAM_ID);
  final Map<String, MetricDefinition> subAlarmMetricDefinitions = new HashMap<>();
  for (final AlarmSubExpression subExpr : alarmDef.getAlarmExpression().getSubExpressions()) {
    subAlarmMetricDefinitions.put(getNextId(), subExpr.getMetricDefinition());
  }
  final AlarmDefinitionDeletedEvent event =
      new AlarmDefinitionDeletedEvent(alarmDef.getId(), subAlarmMetricDefinitions);
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.DELETED, event), tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createMetricDefinitionDeletionTuple(final MetricDefinitionAndTenantId mtid,
    final String alarmDefinitionId) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.METRIC_ALARM_EVENT_STREAM_FIELDS);
  tupleParam.setStream(EventProcessingBolt.METRIC_ALARM_EVENT_STREAM_ID);
  // This code doesn't know the sub alarm id and the metric filtering bolt doesn't care
  final String subAlarmId = "";
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.DELETED,
          new TenantIdAndMetricName(mtid), mtid, alarmDefinitionId, subAlarmId), tupleParam);

  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private Tuple createMetricTuple(final MetricDefinition metricDefinition, final long timestamp,
    final Metric metric) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(MetricSpout.FIELDS);
  tupleParam.setStream(Streams.DEFAULT_STREAM_ID);
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(new TenantIdAndMetricName(TEST_TENANT_ID,
          metricDefinition.name), timestamp, metric), tupleParam);
  return tuple;
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private void sendSubAlarmCreated(MetricDefinition metricDef, SubAlarm subAlarm) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(AlarmCreationBolt.ALARM_CREATION_FIELDS);
  tupleParam.setStream(AlarmCreationBolt.ALARM_CREATION_STREAM);
  final String alarmDefinitionString = ""; // TODO - Figure out what this needs to be
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(EventProcessingBolt.CREATED, new TenantIdAndMetricName(
          TENANT_ID, metricDef.name), new MetricDefinitionAndTenantId(metricDef, TENANT_ID),
          alarmDefinitionString, subAlarm), tupleParam);
  bolt.execute(tuple);
}
 

import org.apache.storm.Testing; //导入依赖的package包/类
private void sendSubAlarmMsg(String command, MetricDefinition metricDef, SubAlarm subAlarm) {
  final MkTupleParam tupleParam = new MkTupleParam();
  tupleParam.setFields(EventProcessingBolt.METRIC_ALARM_EVENT_STREAM_FIELDS);
  tupleParam.setStream(EventProcessingBolt.METRIC_ALARM_EVENT_STREAM_ID);
  final String alarmDefinitionId = "";
  final MetricDefinitionAndTenantId mdtid = new MetricDefinitionAndTenantId(metricDef, TENANT_ID);
  final Tuple tuple =
      Testing.testTuple(Arrays.asList(command, new TenantIdAndMetricName(mdtid), mdtid,
          alarmDefinitionId, subAlarm.getId()), tupleParam);
  bolt.execute(tuple);
}