public static <InputT, OutputT> PTransformRunnerFactory<?> forValueMapFnFactory(
      ValueMapFnFactory<InputT, OutputT> fnFactory) {
    return new Factory<>(new CompressedValueOnlyMapperFactory<>(fnFactory));
  }
@Test
  public void testValueOnlyMapping() throws Exception {
    PTransformRunnerFactoryTestContext context =
        PTransformRunnerFactoryTestContext.builder(EXPECTED_ID, EXPECTED_PTRANSFORM)
            .processBundleInstructionId("57")
            .pCollections(Collections.singletonMap("inputPC", INPUT_PCOLLECTION))
            .coders(Collections.singletonMap("coder-id", valueCoder))
            .build();
    List<WindowedValue<?>> outputConsumer = new ArrayList<>();
    context.addPCollectionConsumer("outputPC", outputConsumer::add);

    ValueMapFnFactory<String, String> factory = (ptId, pt) -> String::toUpperCase;
    MapFnRunners.forValueMapFnFactory(factory).createRunnerForPTransform(context);

    assertThat(context.getStartBundleFunctions(), empty());
    assertThat(context.getFinishBundleFunctions(), empty());
    assertThat(context.getTearDownFunctions(), empty());

    assertThat(
        context.getPCollectionConsumers().keySet(), containsInAnyOrder("inputPC", "outputPC"));

    context.getPCollectionConsumer("inputPC").accept(valueInGlobalWindow("abc"));

    assertThat(outputConsumer, contains(valueInGlobalWindow("ABC")));
  }