public static FusedPipeline fuse(Pipeline p) {
    return new GreedyPipelineFuser(p).fusedPipeline;
  }
@Test
  public void multipleEnvironmentsBecomesMultipleStages() {
    Components components =
        partialComponents
            .toBuilder()
            .putTransforms(
                "read",
                PTransform.newBuilder()
                    .setUniqueName("Read")
                    .putInputs("input", "impulse.out")
                    .putOutputs("output", "read.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("read.out", pc("read.out"))
            .putTransforms(
                "goTransform",
                PTransform.newBuilder()
                    .setUniqueName("GoTransform")
                    .putInputs("input", "read.out")
                    .putOutputs("output", "go.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("go")
                    .build())
            .putPcollections("go.out", pc("go.out"))
            .putTransforms(
                "pyTransform",
                PTransform.newBuilder()
                    .setUniqueName("PyTransform")
                    .putInputs("input", "read.out")
                    .putOutputs("output", "py.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.ASSIGN_WINDOWS_TRANSFORM_URN)
                            .setPayload(
                                WindowIntoPayload.newBuilder()
                                    .setWindowFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("py.out", pc("py.out"))
            .build();
    FusedPipeline fused =
        GreedyPipelineFuser.fuse(Pipeline.newBuilder().setComponents(components).build());

    // Impulse is the runner transform
    assertThat(fused.getRunnerExecutedTransforms(), hasSize(1));
    assertThat(fused.getFusedStages(), hasSize(3));
    assertThat(
        fused.getFusedStages(),
        containsInAnyOrder(
            ExecutableStageMatcher.withInput("impulse.out")
                .withOutputs("read.out")
                .withTransforms("read"),
            ExecutableStageMatcher.withInput("read.out")
                .withNoOutputs()
                .withTransforms("pyTransform"),
            ExecutableStageMatcher.withInput("read.out")
                .withNoOutputs()
                .withTransforms("goTransform")));
  }