ImmutabilityEnforcementFactory implements ModelEnforcementFactory { @Override public <T> ModelEnforcement<T> forBundle( CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer) { if (isReadTransform(consumer)) { return NoopReadEnforcement.INSTANCE; } return new ImmutabilityCheckingEnforcement<>(input, consumer); }  static ModelEnforcementFactory create(); @Override ModelEnforcement<T> forBundle(
      CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer);  }
@Test public void unchangedSucceeds() { WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes(UTF_8)); CommittedBundle<byte[]> elements = bundleFactory.createBundle(pcollection).add(element).commit(Instant.now()); ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer); enforcement.beforeElement(element); enforcement.afterElement(element); enforcement.afterFinish( elements, StepTransformResult.<byte[]>withoutHold(consumer).build(), Collections.emptyList()); }
@Test public void mutatedDuringProcessElementThrows() { WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes(UTF_8)); CommittedBundle<byte[]> elements = bundleFactory.createBundle(pcollection).add(element).commit(Instant.now()); ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer); enforcement.beforeElement(element); element.getValue()[0] = 'f'; thrown.expect(IllegalMutationException.class); thrown.expectMessage(consumer.getFullName()); thrown.expectMessage("illegaly mutated"); thrown.expectMessage("Input values must not be mutated"); enforcement.afterElement(element); enforcement.afterFinish( elements, StepTransformResult.<byte[]>withoutHold(consumer).build(), Collections.emptyList()); }
@Test public void mutatedAfterProcessElementFails() { WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes(UTF_8)); CommittedBundle<byte[]> elements = bundleFactory.createBundle(pcollection).add(element).commit(Instant.now()); ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer); enforcement.beforeElement(element); enforcement.afterElement(element); element.getValue()[0] = 'f'; thrown.expect(IllegalMutationException.class); thrown.expectMessage(consumer.getFullName()); thrown.expectMessage("illegaly mutated"); thrown.expectMessage("Input values must not be mutated"); enforcement.afterFinish( elements, StepTransformResult.<byte[]>withoutHold(consumer).build(), Collections.emptyList()); }