@Test
  public void testAddSource() {
    StreamApplicationDescriptorImpl appDesc = mock(StreamApplicationDescriptorImpl.class);
    JobGraph graph = new JobGraph(new MapConfig(), appDesc);

    /**
     * s1 -> 1
     * s2 ->|
     *
     * s3 -> 2
     *   |-> 3
     */
    JobNode n1 = graph.getOrCreateJobNode("1", "1");
    JobNode n2 = graph.getOrCreateJobNode("2", "1");
    JobNode n3 = graph.getOrCreateJobNode("3", "1");
    StreamSpec s1 = genStream();
    StreamSpec s2 = genStream();
    StreamSpec s3 = genStream();
    graph.addInputStream(s1, n1);
    graph.addInputStream(s2, n1);
    graph.addInputStream(s3, n2);
    graph.addInputStream(s3, n3);

    assertTrue(graph.getInputStreams().size() == 3);

    assertTrue(graph.getOrCreateJobNode("1", "1").getInEdges().size() == 2);
    assertTrue(graph.getOrCreateJobNode("2", "1").getInEdges().size() == 1);
    assertTrue(graph.getOrCreateJobNode("3", "1").getInEdges().size() == 1);

    assertTrue(graph.getOrCreateStreamEdge(s1).getSourceNodes().size() == 0);
    assertTrue(graph.getOrCreateStreamEdge(s1).getTargetNodes().size() == 1);
    assertTrue(graph.getOrCreateStreamEdge(s2).getSourceNodes().size() == 0);
    assertTrue(graph.getOrCreateStreamEdge(s2).getTargetNodes().size() == 1);
    assertTrue(graph.getOrCreateStreamEdge(s3).getSourceNodes().size() == 0);
    assertTrue(graph.getOrCreateStreamEdge(s3).getTargetNodes().size() == 2);
  }
graph4 is a graph of single-loop node
1<->1
samza-core/src/test/java/org/apache/samza/execution/TestJobGraph.java