public DirectGraph getGraph() {
    checkState(finalized, "Can't get a graph before the Pipeline has been completely traversed");
    return DirectGraph.create(
        producers, viewWriters, perElementConsumers, rootTransforms, stepNames);
  }
@Test
  public void getStepNamesContainsAllTransforms() {
    PCollection<String> created = p.apply(Create.of("1", "2", "3"));
    PCollection<String> transformed =
        created.apply(
            ParDo.of(
                new DoFn<String, String>() {
                  @ProcessElement
                  public void processElement(DoFn<String, String>.ProcessContext c)
                      throws Exception {
                    c.output(Integer.toString(c.element().length()));
                  }
                }));
    transformed.apply(
        new PTransform<PInput, PDone>() {
          @Override
          public PDone expand(PInput input) {
            return PDone.in(input.getPipeline());
          }
        });

    p.traverseTopologically(visitor);
    DirectGraph graph = visitor.getGraph();
    // Step names are of the format "s#" such as "s0", "s1", ...
    int createdStepIndex =
        Integer.parseInt(graph.getStepName(graph.getProducer(created)).substring(1));
    assertThat(
        graph.getStepName(graph.getProducer(transformed)), equalTo("s" + (createdStepIndex + 1)));
    // finished doesn't have a producer, because it's not a PValue.
    // TODO: Demonstrate that PCollectionList/Tuple and other composite PValues are either safe to
    // use, or make them so.
  }