Class TestingTest

storm-server/src/test/java/org/apache/storm/TestingTest.java:37–124

Test that the testing class does what it should do.

 * Test that the testing class does what it should do.
 */
@IntegrationTest
public class TestingTest {

    private static final TestJob COMPLETE_TOPOLOGY_TESTJOB = (cluster) -> {
        TopologyBuilder tb = new TopologyBuilder();
        tb.setSpout("spout", new TestWordSpout(true), 3);
        tb.setBolt("2", new TestWordCounter(), 4)
          .fieldsGrouping("spout", new Fields("word"));
        tb.setBolt("3", new TestGlobalCount())
          .globalGrouping("spout");
        tb.setBolt("4", new TestAggregatesCounter())
          .globalGrouping("2");

        MockedSources mocked = new MockedSources();
        mocked.addMockData("spout",
                           new Values("nathan"),
                           new Values("bob"),
                           new Values("joey"),
                           new Values("nathan"));

        Config topoConf = new Config();
        topoConf.setNumWorkers(2);

        CompleteTopologyParam ctp = new CompleteTopologyParam();
        ctp.setMockedSources(mocked);
        ctp.setStormConf(topoConf);

        Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, tb.createTopology(), ctp);
        List<List<Object>> spoutTuples = Testing.readTuples(results, "spout");
        List<List<Object>> expectedSpoutTuples = Arrays.asList(Arrays.asList("nathan"), Arrays.asList("bob"), Arrays.asList("joey"),
                                                               Arrays.asList("nathan"));
        assertTrue(Testing.multiseteq(expectedSpoutTuples, spoutTuples), expectedSpoutTuples + " expected, but found " + spoutTuples);

        List<List<Object>> twoTuples = Testing.readTuples(results, "2");
        List<List<Object>> expectedTwoTuples = Arrays.asList(Arrays.asList("nathan", 1), Arrays.asList("nathan", 2),
                                                             Arrays.asList("bob", 1), Arrays.asList("joey", 1));
        assertTrue(Testing.multiseteq(expectedTwoTuples, twoTuples), expectedTwoTuples + " expected, but found " + twoTuples);

        List<List<Object>> threeTuples = Testing.readTuples(results, "3");
        List<List<Object>> expectedThreeTuples = Arrays.asList(Arrays.asList(1), Arrays.asList(2),
                                                               Arrays.asList(3), Arrays.asList(4));
        assertTrue(Testing.multiseteq(expectedThreeTuples, threeTuples), expectedThreeTuples + " expected, but found " + threeTuples);

        List<List<Object>> fourTuples = Testing.readTuples(results, "4");
        List<List<Object>> expectedFourTuples = Arrays.asList(Arrays.asList(1), Arrays.asList(2),
                                                              Arrays.asList(3), Arrays.asList(4));
        assertTrue(Testing.multiseteq(expectedFourTuples, fourTuples), expectedFourTuples + " expected, but found " + fourTuples);
    };

    @Test
    public void testCompleteTopologyNettySimulated() throws Exception {
        Config daemonConf = new Config();
        daemonConf.put(Config.STORM_LOCAL_MODE_ZMQ, true);
        MkClusterParam param = new MkClusterParam();
        param.setSupervisors(4);
        param.setDaemonConf(daemonConf);

        Testing.withSimulatedTimeLocalCluster(param, COMPLETE_TOPOLOGY_TESTJOB);