@Override
  public StageBundleFactory forStage(ExecutableStage executableStage) {
    return new SimpleStageBundleFactory(executableStage);
  }
@Test
  public void createsMultipleEnvironmentOfSingleType() throws Exception {
    ServerFactory serverFactory = ServerFactory.createDefault();

    Environment environmentA =
        Environment.newBuilder()
            .setUrn("env:urn:a")
            .setPayload(ByteString.copyFrom(new byte[1]))
            .build();
    Environment environmentAA =
        Environment.newBuilder()
            .setUrn("env:urn:a")
            .setPayload(ByteString.copyFrom(new byte[2]))
            .build();
    EnvironmentFactory envFactoryA = mock(EnvironmentFactory.class);
    when(envFactoryA.createEnvironment(eq(environmentA), any())).thenReturn(remoteEnvironment);
    when(envFactoryA.createEnvironment(eq(environmentAA), any())).thenReturn(remoteEnvironment);
    EnvironmentFactory.Provider environmentProviderFactoryA =
        mock(EnvironmentFactory.Provider.class);
    when(environmentProviderFactoryA.createEnvironmentFactory(
            any(), any(), any(), any(), any(), any()))
        .thenReturn(envFactoryA);
    when(environmentProviderFactoryA.getServerFactory()).thenReturn(serverFactory);

    Environment environmentB = Environment.newBuilder().setUrn("env:urn:b").build();
    EnvironmentFactory envFactoryB = mock(EnvironmentFactory.class);
    when(envFactoryB.createEnvironment(eq(environmentB), any())).thenReturn(remoteEnvironment);
    EnvironmentFactory.Provider environmentProviderFactoryB =
        mock(EnvironmentFactory.Provider.class);
    when(environmentProviderFactoryB.createEnvironmentFactory(
            any(), any(), any(), any(), any(), any()))
        .thenReturn(envFactoryB);
    when(environmentProviderFactoryB.getServerFactory()).thenReturn(serverFactory);

    Map<String, Provider> environmentFactoryProviderMap =
        ImmutableMap.of(
            environmentA.getUrn(), environmentProviderFactoryA,
            environmentB.getUrn(), environmentProviderFactoryB);
    try (DefaultJobBundleFactory bundleFactory =
        createDefaultJobBundleFactory(environmentFactoryProviderMap)) {
      bundleFactory.forStage(getExecutableStage(environmentA));
      verify(environmentProviderFactoryA, Mockito.times(1))
          .createEnvironmentFactory(any(), any(), any(), any(), any(), any());
      verify(environmentProviderFactoryB, Mockito.times(0))
          .createEnvironmentFactory(any(), any(), any(), any(), any(), any());
      verify(envFactoryA, Mockito.times(1)).createEnvironment(eq(environmentA), any());
      verify(envFactoryA, Mockito.times(0)).createEnvironment(eq(environmentAA), any());

      bundleFactory.forStage(getExecutableStage(environmentAA));
      verify(environmentProviderFactoryA, Mockito.times(2))
          .createEnvironmentFactory(any(), any(), any(), any(), any(), any());
      verify(environmentProviderFactoryB, Mockito.times(0))
          .createEnvironmentFactory(any(), any(), any(), any(), any(), any());
      verify(envFactoryA, Mockito.times(1)).createEnvironment(eq(environmentA), any());
      verify(envFactoryA, Mockito.times(1)).createEnvironment(eq(environmentAA), any());
    }
  }