You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2016/04/08 23:02:08 UTC
[3/5] incubator-beam git commit: Rename DataflowAssert to PAssert
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/DataflowAssert.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/DataflowAssert.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/DataflowAssert.java
deleted file mode 100644
index 8f4c066..0000000
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/DataflowAssert.java
+++ /dev/null
@@ -1,826 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.google.cloud.dataflow.sdk.testing;
-
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.not;
-import static org.junit.Assert.assertThat;
-
-import com.google.cloud.dataflow.sdk.Pipeline;
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.coders.CoderException;
-import com.google.cloud.dataflow.sdk.coders.IterableCoder;
-import com.google.cloud.dataflow.sdk.coders.KvCoder;
-import com.google.cloud.dataflow.sdk.coders.MapCoder;
-import com.google.cloud.dataflow.sdk.coders.VoidCoder;
-import com.google.cloud.dataflow.sdk.options.StreamingOptions;
-import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
-import com.google.cloud.dataflow.sdk.transforms.Aggregator;
-import com.google.cloud.dataflow.sdk.transforms.Create;
-import com.google.cloud.dataflow.sdk.transforms.DoFn;
-import com.google.cloud.dataflow.sdk.transforms.PTransform;
-import com.google.cloud.dataflow.sdk.transforms.ParDo;
-import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
-import com.google.cloud.dataflow.sdk.transforms.Sum;
-import com.google.cloud.dataflow.sdk.transforms.View;
-import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
-import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
-import com.google.cloud.dataflow.sdk.util.CoderUtils;
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.PBegin;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.cloud.dataflow.sdk.values.PCollectionView;
-import com.google.cloud.dataflow.sdk.values.PDone;
-import com.google.common.base.Optional;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Serializable;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-
-/**
- * An assertion on the contents of a {@link PCollection}
- * incorporated into the pipeline. Such an assertion
- * can be checked no matter what kind of {@link PipelineRunner} is
- * used.
- *
- * <p>Note that the {@code DataflowAssert} call must precede the call
- * to {@link Pipeline#run}.
- *
- * <p>Examples of use:
- * <pre>{@code
- * Pipeline p = TestPipeline.create();
- * ...
- * PCollection<String> output =
- * input
- * .apply(ParDo.of(new TestDoFn()));
- * DataflowAssert.that(output)
- * .containsInAnyOrder("out1", "out2", "out3");
- * ...
- * PCollection<Integer> ints = ...
- * PCollection<Integer> sum =
- * ints
- * .apply(Combine.globally(new SumInts()));
- * DataflowAssert.that(sum)
- * .is(42);
- * ...
- * p.run();
- * }</pre>
- *
- * <p>JUnit and Hamcrest must be linked in by any code that uses DataflowAssert.
- */
-public class DataflowAssert {
-
- private static final Logger LOG = LoggerFactory.getLogger(DataflowAssert.class);
-
- static final String SUCCESS_COUNTER = "DataflowAssertSuccess";
- static final String FAILURE_COUNTER = "DataflowAssertFailure";
-
- private static int assertCount = 0;
-
- // Do not instantiate.
- private DataflowAssert() {}
-
- /**
- * Constructs an {@link IterableAssert} for the elements of the provided
- * {@link PCollection}.
- */
- public static <T> IterableAssert<T> that(PCollection<T> actual) {
- return new IterableAssert<>(
- new CreateActual<T, Iterable<T>>(actual, View.<T>asIterable()),
- actual.getPipeline())
- .setCoder(actual.getCoder());
- }
-
- /**
- * Constructs an {@link IterableAssert} for the value of the provided
- * {@link PCollection} which must contain a single {@code Iterable<T>}
- * value.
- */
- public static <T> IterableAssert<T>
- thatSingletonIterable(PCollection<? extends Iterable<T>> actual) {
-
- List<? extends Coder<?>> maybeElementCoder = actual.getCoder().getCoderArguments();
- Coder<T> tCoder;
- try {
- @SuppressWarnings("unchecked")
- Coder<T> tCoderTmp = (Coder<T>) Iterables.getOnlyElement(maybeElementCoder);
- tCoder = tCoderTmp;
- } catch (NoSuchElementException | IllegalArgumentException exc) {
- throw new IllegalArgumentException(
- "DataflowAssert.<T>thatSingletonIterable requires a PCollection<Iterable<T>>"
- + " with a Coder<Iterable<T>> where getCoderArguments() yields a"
- + " single Coder<T> to apply to the elements.");
- }
-
- @SuppressWarnings("unchecked") // Safe covariant cast
- PCollection<Iterable<T>> actualIterables = (PCollection<Iterable<T>>) actual;
-
- return new IterableAssert<>(
- new CreateActual<Iterable<T>, Iterable<T>>(
- actualIterables, View.<Iterable<T>>asSingleton()),
- actual.getPipeline())
- .setCoder(tCoder);
- }
-
- /**
- * Constructs an {@link IterableAssert} for the value of the provided
- * {@code PCollectionView PCollectionView<Iterable<T>>}.
- */
- public static <T> IterableAssert<T> thatIterable(PCollectionView<Iterable<T>> actual) {
- return new IterableAssert<>(new PreExisting<Iterable<T>>(actual), actual.getPipeline());
- }
-
- /**
- * Constructs a {@link SingletonAssert} for the value of the provided
- * {@code PCollection PCollection<T>}, which must be a singleton.
- */
- public static <T> SingletonAssert<T> thatSingleton(PCollection<T> actual) {
- return new SingletonAssert<>(
- new CreateActual<T, T>(actual, View.<T>asSingleton()), actual.getPipeline())
- .setCoder(actual.getCoder());
- }
-
- /**
- * Constructs a {@link SingletonAssert} for the value of the provided {@link PCollection}.
- *
- * <p>Note that the actual value must be coded by a {@link KvCoder},
- * not just any {@code Coder<K, V>}.
- */
- public static <K, V> SingletonAssert<Map<K, Iterable<V>>>
- thatMultimap(PCollection<KV<K, V>> actual) {
- @SuppressWarnings("unchecked")
- KvCoder<K, V> kvCoder = (KvCoder<K, V>) actual.getCoder();
-
- return new SingletonAssert<>(
- new CreateActual<>(actual, View.<K, V>asMultimap()), actual.getPipeline())
- .setCoder(MapCoder.of(kvCoder.getKeyCoder(), IterableCoder.of(kvCoder.getValueCoder())));
- }
-
- /**
- * Constructs a {@link SingletonAssert} for the value of the provided {@link PCollection},
- * which must have at most one value per key.
- *
- * <p>Note that the actual value must be coded by a {@link KvCoder},
- * not just any {@code Coder<K, V>}.
- */
- public static <K, V> SingletonAssert<Map<K, V>> thatMap(PCollection<KV<K, V>> actual) {
- @SuppressWarnings("unchecked")
- KvCoder<K, V> kvCoder = (KvCoder<K, V>) actual.getCoder();
-
- return new SingletonAssert<>(
- new CreateActual<>(actual, View.<K, V>asMap()), actual.getPipeline())
- .setCoder(MapCoder.of(kvCoder.getKeyCoder(), kvCoder.getValueCoder()));
- }
-
- ////////////////////////////////////////////////////////////
-
- /**
- * An assertion about the contents of a {@link PCollectionView} yielding an {@code Iterable<T>}.
- */
- public static class IterableAssert<T> implements Serializable {
- private final Pipeline pipeline;
- private final PTransform<PBegin, PCollectionView<Iterable<T>>> createActual;
- private Optional<Coder<T>> coder;
-
- protected IterableAssert(
- PTransform<PBegin, PCollectionView<Iterable<T>>> createActual, Pipeline pipeline) {
- this.createActual = createActual;
- this.pipeline = pipeline;
- this.coder = Optional.absent();
- }
-
- /**
- * Sets the coder to use for elements of type {@code T}, as needed for internal purposes.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- public IterableAssert<T> setCoder(Coder<T> coderOrNull) {
- this.coder = Optional.fromNullable(coderOrNull);
- return this;
- }
-
- /**
- * Gets the coder, which may yet be absent.
- */
- public Coder<T> getCoder() {
- if (coder.isPresent()) {
- return coder.get();
- } else {
- throw new IllegalStateException(
- "Attempting to access the coder of an IterableAssert"
- + " that has not been set yet.");
- }
- }
-
- /**
- * Applies a {@link SerializableFunction} to check the elements of the {@code Iterable}.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- public IterableAssert<T> satisfies(SerializableFunction<Iterable<T>, Void> checkerFn) {
- pipeline.apply(
- "DataflowAssert$" + (assertCount++),
- new OneSideInputAssert<Iterable<T>>(createActual, checkerFn));
- return this;
- }
-
- /**
- * Applies a {@link SerializableFunction} to check the elements of the {@code Iterable}.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- public IterableAssert<T> satisfies(
- AssertRelation<Iterable<T>, Iterable<T>> relation,
- final Iterable<T> expectedElements) {
- pipeline.apply(
- "DataflowAssert$" + (assertCount++),
- new TwoSideInputAssert<Iterable<T>, Iterable<T>>(createActual,
- new CreateExpected<T, Iterable<T>>(expectedElements, coder, View.<T>asIterable()),
- relation));
-
- return this;
- }
-
- /**
- * Applies a {@link SerializableMatcher} to check the elements of the {@code Iterable}.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- IterableAssert<T> satisfies(final SerializableMatcher<Iterable<? extends T>> matcher) {
- // Safe covariant cast. Could be elided by changing a lot of this file to use
- // more flexible bounds.
- @SuppressWarnings({"rawtypes", "unchecked"})
- SerializableFunction<Iterable<T>, Void> checkerFn =
- (SerializableFunction) new MatcherCheckerFn<>(matcher);
- pipeline.apply(
- "DataflowAssert$" + (assertCount++),
- new OneSideInputAssert<Iterable<T>>(
- createActual,
- checkerFn));
- return this;
- }
-
- private static class MatcherCheckerFn<T> implements SerializableFunction<T, Void> {
- private SerializableMatcher<T> matcher;
-
- public MatcherCheckerFn(SerializableMatcher<T> matcher) {
- this.matcher = matcher;
- }
-
- @Override
- public Void apply(T actual) {
- assertThat(actual, matcher);
- return null;
- }
- }
-
- /**
- * Checks that the {@code Iterable} is empty.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- public IterableAssert<T> empty() {
- return satisfies(new AssertContainsInAnyOrderRelation<T>(), Collections.<T>emptyList());
- }
-
- /**
- * @throws UnsupportedOperationException always
- * @deprecated {@link Object#equals(Object)} is not supported on DataflowAssert objects.
- * If you meant to test object equality, use a variant of {@link #containsInAnyOrder}
- * instead.
- */
- @Deprecated
- @Override
- public boolean equals(Object o) {
- throw new UnsupportedOperationException(
- "If you meant to test object equality, use .containsInAnyOrder instead.");
- }
-
- /**
- * @throws UnsupportedOperationException always.
- * @deprecated {@link Object#hashCode()} is not supported on DataflowAssert objects.
- */
- @Deprecated
- @Override
- public int hashCode() {
- throw new UnsupportedOperationException(
- String.format("%s.hashCode() is not supported.", IterableAssert.class.getSimpleName()));
- }
-
- /**
- * Checks that the {@code Iterable} contains the expected elements, in any
- * order.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- public IterableAssert<T> containsInAnyOrder(Iterable<T> expectedElements) {
- return satisfies(new AssertContainsInAnyOrderRelation<T>(), expectedElements);
- }
-
- /**
- * Checks that the {@code Iterable} contains the expected elements, in any
- * order.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- @SafeVarargs
- public final IterableAssert<T> containsInAnyOrder(T... expectedElements) {
- return satisfies(
- new AssertContainsInAnyOrderRelation<T>(),
- Arrays.asList(expectedElements));
- }
-
- /**
- * Checks that the {@code Iterable} contains elements that match the provided matchers,
- * in any order.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- @SafeVarargs
- final IterableAssert<T> containsInAnyOrder(
- SerializableMatcher<? super T>... elementMatchers) {
- return satisfies(SerializableMatchers.<T>containsInAnyOrder(elementMatchers));
- }
- }
-
- /**
- * An assertion about the single value of type {@code T}
- * associated with a {@link PCollectionView}.
- */
- public static class SingletonAssert<T> implements Serializable {
- private final Pipeline pipeline;
- private final CreateActual<?, T> createActual;
- private Optional<Coder<T>> coder;
-
- protected SingletonAssert(
- CreateActual<?, T> createActual, Pipeline pipeline) {
- this.pipeline = pipeline;
- this.createActual = createActual;
- this.coder = Optional.absent();
- }
-
- /**
- * Always throws an {@link UnsupportedOperationException}: users are probably looking for
- * {@link #isEqualTo}.
- */
- @Deprecated
- @Override
- public boolean equals(Object o) {
- throw new UnsupportedOperationException(
- String.format(
- "tests for Java equality of the %s object, not the PCollection in question. "
- + "Call a test method, such as isEqualTo.",
- getClass().getSimpleName()));
- }
-
- /**
- * @throws UnsupportedOperationException always.
- * @deprecated {@link Object#hashCode()} is not supported on DataflowAssert objects.
- */
- @Deprecated
- @Override
- public int hashCode() {
- throw new UnsupportedOperationException(
- String.format("%s.hashCode() is not supported.", SingletonAssert.class.getSimpleName()));
- }
-
- /**
- * Sets the coder to use for elements of type {@code T}, as needed
- * for internal purposes.
- *
- * <p>Returns this {@code IterableAssert}.
- */
- public SingletonAssert<T> setCoder(Coder<T> coderOrNull) {
- this.coder = Optional.fromNullable(coderOrNull);
- return this;
- }
-
- /**
- * Gets the coder, which may yet be absent.
- */
- public Coder<T> getCoder() {
- if (coder.isPresent()) {
- return coder.get();
- } else {
- throw new IllegalStateException(
- "Attempting to access the coder of an IterableAssert that has not been set yet.");
- }
- }
-
- /**
- * Applies a {@link SerializableFunction} to check the value of this
- * {@code SingletonAssert}'s view.
- *
- * <p>Returns this {@code SingletonAssert}.
- */
- public SingletonAssert<T> satisfies(SerializableFunction<T, Void> checkerFn) {
- pipeline.apply(
- "DataflowAssert$" + (assertCount++),
- new OneSideInputAssert<T>(createActual, checkerFn));
- return this;
- }
-
- /**
- * Applies an {@link AssertRelation} to check the provided relation against the
- * value of this assert and the provided expected value.
- *
- * <p>Returns this {@code SingletonAssert}.
- */
- public SingletonAssert<T> satisfies(
- AssertRelation<T, T> relation,
- final T expectedValue) {
- pipeline.apply(
- "DataflowAssert$" + (assertCount++),
- new TwoSideInputAssert<T, T>(createActual,
- new CreateExpected<T, T>(Arrays.asList(expectedValue), coder, View.<T>asSingleton()),
- relation));
-
- return this;
- }
-
- /**
- * Checks that the value of this {@code SingletonAssert}'s view is equal
- * to the expected value.
- *
- * <p>Returns this {@code SingletonAssert}.
- */
- public SingletonAssert<T> isEqualTo(T expectedValue) {
- return satisfies(new AssertIsEqualToRelation<T>(), expectedValue);
- }
-
- /**
- * Checks that the value of this {@code SingletonAssert}'s view is not equal
- * to the expected value.
- *
- * <p>Returns this {@code SingletonAssert}.
- */
- public SingletonAssert<T> notEqualTo(T expectedValue) {
- return satisfies(new AssertNotEqualToRelation<T>(), expectedValue);
- }
-
- /**
- * Checks that the value of this {@code SingletonAssert}'s view is equal to
- * the expected value.
- *
- * @deprecated replaced by {@link #isEqualTo}
- */
- @Deprecated
- public SingletonAssert<T> is(T expectedValue) {
- return isEqualTo(expectedValue);
- }
-
- }
-
- ////////////////////////////////////////////////////////////////////////
-
- private static class CreateActual<T, ActualT>
- extends PTransform<PBegin, PCollectionView<ActualT>> {
-
- private final transient PCollection<T> actual;
- private final transient PTransform<PCollection<T>, PCollectionView<ActualT>> actualView;
-
- private CreateActual(PCollection<T> actual,
- PTransform<PCollection<T>, PCollectionView<ActualT>> actualView) {
- this.actual = actual;
- this.actualView = actualView;
- }
-
- @Override
- public PCollectionView<ActualT> apply(PBegin input) {
- final Coder<T> coder = actual.getCoder();
- return actual
- .apply(Window.<T>into(new GlobalWindows()))
- .apply(ParDo.of(new DoFn<T, T>() {
- @Override
- public void processElement(ProcessContext context) throws CoderException {
- context.output(CoderUtils.clone(coder, context.element()));
- }
- }))
- .apply(actualView);
- }
- }
-
- private static class CreateExpected<T, ExpectedT>
- extends PTransform<PBegin, PCollectionView<ExpectedT>> {
-
- private final Iterable<T> elements;
- private final Optional<Coder<T>> coder;
- private final transient PTransform<PCollection<T>, PCollectionView<ExpectedT>> view;
-
- private CreateExpected(Iterable<T> elements, Optional<Coder<T>> coder,
- PTransform<PCollection<T>, PCollectionView<ExpectedT>> view) {
- this.elements = elements;
- this.coder = coder;
- this.view = view;
- }
-
- @Override
- public PCollectionView<ExpectedT> apply(PBegin input) {
- Create.Values<T> createTransform = Create.<T>of(elements);
- if (coder.isPresent()) {
- createTransform = createTransform.withCoder(coder.get());
- }
- return input.apply(createTransform).apply(view);
- }
- }
-
- private static class PreExisting<T> extends PTransform<PBegin, PCollectionView<T>> {
-
- private final PCollectionView<T> view;
-
- private PreExisting(PCollectionView<T> view) {
- this.view = view;
- }
-
- @Override
- public PCollectionView<T> apply(PBegin input) {
- return view;
- }
- }
-
- /**
- * An assertion checker that takes a single
- * {@link PCollectionView PCollectionView<ActualT>}
- * and an assertion over {@code ActualT}, and checks it within a dataflow
- * pipeline.
- *
- * <p>Note that the entire assertion must be serializable. If
- * you need to make assertions involving multiple inputs
- * that are each not serializable, use TwoSideInputAssert.
- *
- * <p>This is generally useful for assertion functions that
- * are serializable but whose underlying data may not have a coder.
- */
- static class OneSideInputAssert<ActualT>
- extends PTransform<PBegin, PDone> implements Serializable {
- private final transient PTransform<PBegin, PCollectionView<ActualT>> createActual;
- private final SerializableFunction<ActualT, Void> checkerFn;
-
- public OneSideInputAssert(
- PTransform<PBegin, PCollectionView<ActualT>> createActual,
- SerializableFunction<ActualT, Void> checkerFn) {
- this.createActual = createActual;
- this.checkerFn = checkerFn;
- }
-
- @Override
- public PDone apply(PBegin input) {
- final PCollectionView<ActualT> actual = input.apply("CreateActual", createActual);
-
- input
- .apply(Create.<Void>of((Void) null).withCoder(VoidCoder.of()))
- .apply(ParDo.named("RunChecks").withSideInputs(actual)
- .of(new CheckerDoFn<>(checkerFn, actual)));
-
- return PDone.in(input.getPipeline());
- }
- }
-
- /**
- * A {@link DoFn} that runs a checking {@link SerializableFunction} on the contents of
- * a {@link PCollectionView}, and adjusts counters and thrown exceptions for use in testing.
- */
- private static class CheckerDoFn<ActualT> extends DoFn<Void, Void> {
- private final SerializableFunction<ActualT, Void> checkerFn;
- private final Aggregator<Integer, Integer> success =
- createAggregator(SUCCESS_COUNTER, new Sum.SumIntegerFn());
- private final Aggregator<Integer, Integer> failure =
- createAggregator(FAILURE_COUNTER, new Sum.SumIntegerFn());
- private final PCollectionView<ActualT> actual;
-
- private CheckerDoFn(
- SerializableFunction<ActualT, Void> checkerFn,
- PCollectionView<ActualT> actual) {
- this.checkerFn = checkerFn;
- this.actual = actual;
- }
-
- @Override
- public void processElement(ProcessContext c) {
- try {
- ActualT actualContents = c.sideInput(actual);
- checkerFn.apply(actualContents);
- success.addValue(1);
- } catch (Throwable t) {
- LOG.error("DataflowAssert failed expectations.", t);
- failure.addValue(1);
- // TODO: allow for metrics to propagate on failure when running a streaming pipeline
- if (!c.getPipelineOptions().as(StreamingOptions.class).isStreaming()) {
- throw t;
- }
- }
- }
- }
-
- /**
- * An assertion checker that takes a {@link PCollectionView PCollectionView<ActualT>},
- * a {@link PCollectionView PCollectionView<ExpectedT>}, a relation
- * over {@code A} and {@code B}, and checks that the relation holds
- * within a dataflow pipeline.
- *
- * <p>This is useful when either/both of {@code A} and {@code B}
- * are not serializable, but have coders (provided
- * by the underlying {@link PCollection}s).
- */
- static class TwoSideInputAssert<ActualT, ExpectedT>
- extends PTransform<PBegin, PDone> implements Serializable {
-
- private final transient PTransform<PBegin, PCollectionView<ActualT>> createActual;
- private final transient PTransform<PBegin, PCollectionView<ExpectedT>> createExpected;
- private final AssertRelation<ActualT, ExpectedT> relation;
-
- protected TwoSideInputAssert(
- PTransform<PBegin, PCollectionView<ActualT>> createActual,
- PTransform<PBegin, PCollectionView<ExpectedT>> createExpected,
- AssertRelation<ActualT, ExpectedT> relation) {
- this.createActual = createActual;
- this.createExpected = createExpected;
- this.relation = relation;
- }
-
- @Override
- public PDone apply(PBegin input) {
- final PCollectionView<ActualT> actual = input.apply("CreateActual", createActual);
- final PCollectionView<ExpectedT> expected = input.apply("CreateExpected", createExpected);
-
- input
- .apply(Create.<Void>of((Void) null).withCoder(VoidCoder.of()))
- .apply(ParDo.named("RunChecks").withSideInputs(actual, expected)
- .of(new CheckerDoFn<>(relation, actual, expected)));
-
- return PDone.in(input.getPipeline());
- }
-
- private static class CheckerDoFn<ActualT, ExpectedT> extends DoFn<Void, Void> {
- private final Aggregator<Integer, Integer> success =
- createAggregator(SUCCESS_COUNTER, new Sum.SumIntegerFn());
- private final Aggregator<Integer, Integer> failure =
- createAggregator(FAILURE_COUNTER, new Sum.SumIntegerFn());
- private final AssertRelation<ActualT, ExpectedT> relation;
- private final PCollectionView<ActualT> actual;
- private final PCollectionView<ExpectedT> expected;
-
- private CheckerDoFn(AssertRelation<ActualT, ExpectedT> relation,
- PCollectionView<ActualT> actual, PCollectionView<ExpectedT> expected) {
- this.relation = relation;
- this.actual = actual;
- this.expected = expected;
- }
-
- @Override
- public void processElement(ProcessContext c) {
- try {
- ActualT actualContents = c.sideInput(actual);
- ExpectedT expectedContents = c.sideInput(expected);
- relation.assertFor(expectedContents).apply(actualContents);
- success.addValue(1);
- } catch (Throwable t) {
- LOG.error("DataflowAssert failed expectations.", t);
- failure.addValue(1);
- // TODO: allow for metrics to propagate on failure when running a streaming pipeline
- if (!c.getPipelineOptions().as(StreamingOptions.class).isStreaming()) {
- throw t;
- }
- }
- }
- }
- }
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * A {@link SerializableFunction} that verifies that an actual value is equal to an
- * expected value.
- */
- private static class AssertIsEqualTo<T> implements SerializableFunction<T, Void> {
- private T expected;
-
- public AssertIsEqualTo(T expected) {
- this.expected = expected;
- }
-
- @Override
- public Void apply(T actual) {
- assertThat(actual, equalTo(expected));
- return null;
- }
- }
-
- /**
- * A {@link SerializableFunction} that verifies that an actual value is not equal to an
- * expected value.
- */
- private static class AssertNotEqualTo<T> implements SerializableFunction<T, Void> {
- private T expected;
-
- public AssertNotEqualTo(T expected) {
- this.expected = expected;
- }
-
- @Override
- public Void apply(T actual) {
- assertThat(actual, not(equalTo(expected)));
- return null;
- }
- }
-
- /**
- * A {@link SerializableFunction} that verifies that an {@code Iterable} contains
- * expected items in any order.
- */
- private static class AssertContainsInAnyOrder<T>
- implements SerializableFunction<Iterable<T>, Void> {
- private T[] expected;
-
- @SafeVarargs
- public AssertContainsInAnyOrder(T... expected) {
- this.expected = expected;
- }
-
- @SuppressWarnings("unchecked")
- public AssertContainsInAnyOrder(Collection<T> expected) {
- this((T[]) expected.toArray());
- }
-
- public AssertContainsInAnyOrder(Iterable<T> expected) {
- this(Lists.<T>newArrayList(expected));
- }
-
- @Override
- public Void apply(Iterable<T> actual) {
- assertThat(actual, containsInAnyOrder(expected));
- return null;
- }
- }
-
- ////////////////////////////////////////////////////////////
-
- /**
- * A binary predicate between types {@code Actual} and {@code Expected}.
- * Implemented as a method {@code assertFor(Expected)} which returns
- * a {@code SerializableFunction<Actual, Void>}
- * that should verify the assertion..
- */
- private static interface AssertRelation<ActualT, ExpectedT> extends Serializable {
- public SerializableFunction<ActualT, Void> assertFor(ExpectedT input);
- }
-
- /**
- * An {@link AssertRelation} implementing the binary predicate that two objects are equal.
- */
- private static class AssertIsEqualToRelation<T>
- implements AssertRelation<T, T> {
- @Override
- public SerializableFunction<T, Void> assertFor(T expected) {
- return new AssertIsEqualTo<T>(expected);
- }
- }
-
- /**
- * An {@link AssertRelation} implementing the binary predicate that two objects are not equal.
- */
- private static class AssertNotEqualToRelation<T>
- implements AssertRelation<T, T> {
- @Override
- public SerializableFunction<T, Void> assertFor(T expected) {
- return new AssertNotEqualTo<T>(expected);
- }
- }
-
- /**
- * An {@code AssertRelation} implementing the binary predicate that two collections are equal
- * modulo reordering.
- */
- private static class AssertContainsInAnyOrderRelation<T>
- implements AssertRelation<Iterable<T>, Iterable<T>> {
- @Override
- public SerializableFunction<Iterable<T>, Void> assertFor(Iterable<T> expectedElements) {
- return new AssertContainsInAnyOrder<T>(expectedElements);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/PAssert.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/PAssert.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/PAssert.java
new file mode 100644
index 0000000..aebce42
--- /dev/null
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/PAssert.java
@@ -0,0 +1,825 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.dataflow.sdk.testing;
+
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+import static org.junit.Assert.assertThat;
+
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.coders.Coder;
+import com.google.cloud.dataflow.sdk.coders.CoderException;
+import com.google.cloud.dataflow.sdk.coders.IterableCoder;
+import com.google.cloud.dataflow.sdk.coders.KvCoder;
+import com.google.cloud.dataflow.sdk.coders.MapCoder;
+import com.google.cloud.dataflow.sdk.coders.VoidCoder;
+import com.google.cloud.dataflow.sdk.options.StreamingOptions;
+import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
+import com.google.cloud.dataflow.sdk.transforms.Aggregator;
+import com.google.cloud.dataflow.sdk.transforms.Create;
+import com.google.cloud.dataflow.sdk.transforms.DoFn;
+import com.google.cloud.dataflow.sdk.transforms.PTransform;
+import com.google.cloud.dataflow.sdk.transforms.ParDo;
+import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
+import com.google.cloud.dataflow.sdk.transforms.Sum;
+import com.google.cloud.dataflow.sdk.transforms.View;
+import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
+import com.google.cloud.dataflow.sdk.util.CoderUtils;
+import com.google.cloud.dataflow.sdk.values.KV;
+import com.google.cloud.dataflow.sdk.values.PBegin;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+import com.google.cloud.dataflow.sdk.values.PCollectionView;
+import com.google.cloud.dataflow.sdk.values.PDone;
+import com.google.common.base.Optional;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+/**
+ * An assertion on the contents of a {@link PCollection}
+ * incorporated into the pipeline. Such an assertion
+ * can be checked no matter what kind of {@link PipelineRunner} is
+ * used.
+ *
+ * <p>Note that the {@code PAssert} call must precede the call
+ * to {@link Pipeline#run}.
+ *
+ * <p>Examples of use:
+ * <pre>{@code
+ * Pipeline p = TestPipeline.create();
+ * ...
+ * PCollection<String> output =
+ * input
+ * .apply(ParDo.of(new TestDoFn()));
+ * PAssert.that(output)
+ * .containsInAnyOrder("out1", "out2", "out3");
+ * ...
+ * PCollection<Integer> ints = ...
+ * PCollection<Integer> sum =
+ * ints
+ * .apply(Combine.globally(new SumInts()));
+ * PAssert.that(sum)
+ * .is(42);
+ * ...
+ * p.run();
+ * }</pre>
+ *
+ * <p>JUnit and Hamcrest must be linked in by any code that uses PAssert.
+ */
+public class PAssert {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PAssert.class);
+
+ static final String SUCCESS_COUNTER = "PAssertSuccess";
+ static final String FAILURE_COUNTER = "PAssertFailure";
+
+ private static int assertCount = 0;
+
+ // Do not instantiate.
+ private PAssert() {}
+
+ /**
+ * Constructs an {@link IterableAssert} for the elements of the provided
+ * {@link PCollection}.
+ */
+ public static <T> IterableAssert<T> that(PCollection<T> actual) {
+ return new IterableAssert<>(
+ new CreateActual<T, Iterable<T>>(actual, View.<T>asIterable()),
+ actual.getPipeline())
+ .setCoder(actual.getCoder());
+ }
+
+ /**
+ * Constructs an {@link IterableAssert} for the value of the provided
+ * {@link PCollection} which must contain a single {@code Iterable<T>}
+ * value.
+ */
+ public static <T> IterableAssert<T>
+ thatSingletonIterable(PCollection<? extends Iterable<T>> actual) {
+
+ List<? extends Coder<?>> maybeElementCoder = actual.getCoder().getCoderArguments();
+ Coder<T> tCoder;
+ try {
+ @SuppressWarnings("unchecked")
+ Coder<T> tCoderTmp = (Coder<T>) Iterables.getOnlyElement(maybeElementCoder);
+ tCoder = tCoderTmp;
+ } catch (NoSuchElementException | IllegalArgumentException exc) {
+ throw new IllegalArgumentException(
+ "PAssert.<T>thatSingletonIterable requires a PCollection<Iterable<T>>"
+ + " with a Coder<Iterable<T>> where getCoderArguments() yields a"
+ + " single Coder<T> to apply to the elements.");
+ }
+
+ @SuppressWarnings("unchecked") // Safe covariant cast
+ PCollection<Iterable<T>> actualIterables = (PCollection<Iterable<T>>) actual;
+
+ return new IterableAssert<>(
+ new CreateActual<Iterable<T>, Iterable<T>>(
+ actualIterables, View.<Iterable<T>>asSingleton()),
+ actual.getPipeline())
+ .setCoder(tCoder);
+ }
+
+ /**
+ * Constructs an {@link IterableAssert} for the value of the provided
+ * {@code PCollectionView PCollectionView<Iterable<T>>}.
+ */
+ public static <T> IterableAssert<T> thatIterable(PCollectionView<Iterable<T>> actual) {
+ return new IterableAssert<>(new PreExisting<Iterable<T>>(actual), actual.getPipeline());
+ }
+
+ /**
+ * Constructs a {@link SingletonAssert} for the value of the provided
+ * {@code PCollection PCollection<T>}, which must be a singleton.
+ */
+ public static <T> SingletonAssert<T> thatSingleton(PCollection<T> actual) {
+ return new SingletonAssert<>(
+ new CreateActual<T, T>(actual, View.<T>asSingleton()), actual.getPipeline())
+ .setCoder(actual.getCoder());
+ }
+
+ /**
+ * Constructs a {@link SingletonAssert} for the value of the provided {@link PCollection}.
+ *
+ * <p>Note that the actual value must be coded by a {@link KvCoder},
+ * not just any {@code Coder<K, V>}.
+ */
+ public static <K, V> SingletonAssert<Map<K, Iterable<V>>>
+ thatMultimap(PCollection<KV<K, V>> actual) {
+ @SuppressWarnings("unchecked")
+ KvCoder<K, V> kvCoder = (KvCoder<K, V>) actual.getCoder();
+
+ return new SingletonAssert<>(
+ new CreateActual<>(actual, View.<K, V>asMultimap()), actual.getPipeline())
+ .setCoder(MapCoder.of(kvCoder.getKeyCoder(), IterableCoder.of(kvCoder.getValueCoder())));
+ }
+
+ /**
+ * Constructs a {@link SingletonAssert} for the value of the provided {@link PCollection},
+ * which must have at most one value per key.
+ *
+ * <p>Note that the actual value must be coded by a {@link KvCoder},
+ * not just any {@code Coder<K, V>}.
+ */
+ public static <K, V> SingletonAssert<Map<K, V>> thatMap(PCollection<KV<K, V>> actual) {
+ @SuppressWarnings("unchecked")
+ KvCoder<K, V> kvCoder = (KvCoder<K, V>) actual.getCoder();
+
+ return new SingletonAssert<>(
+ new CreateActual<>(actual, View.<K, V>asMap()), actual.getPipeline())
+ .setCoder(MapCoder.of(kvCoder.getKeyCoder(), kvCoder.getValueCoder()));
+ }
+
+ ////////////////////////////////////////////////////////////
+
+ /**
+ * An assertion about the contents of a {@link PCollectionView} yielding an {@code Iterable<T>}.
+ */
+ public static class IterableAssert<T> implements Serializable {
+ private final Pipeline pipeline;
+ private final PTransform<PBegin, PCollectionView<Iterable<T>>> createActual;
+ private Optional<Coder<T>> coder;
+
+ protected IterableAssert(
+ PTransform<PBegin, PCollectionView<Iterable<T>>> createActual, Pipeline pipeline) {
+ this.createActual = createActual;
+ this.pipeline = pipeline;
+ this.coder = Optional.absent();
+ }
+
+ /**
+ * Sets the coder to use for elements of type {@code T}, as needed for internal purposes.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ public IterableAssert<T> setCoder(Coder<T> coderOrNull) {
+ this.coder = Optional.fromNullable(coderOrNull);
+ return this;
+ }
+
+ /**
+ * Gets the coder, which may yet be absent.
+ */
+ public Coder<T> getCoder() {
+ if (coder.isPresent()) {
+ return coder.get();
+ } else {
+ throw new IllegalStateException(
+ "Attempting to access the coder of an IterableAssert"
+ + " that has not been set yet.");
+ }
+ }
+
+ /**
+ * Applies a {@link SerializableFunction} to check the elements of the {@code Iterable}.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ public IterableAssert<T> satisfies(SerializableFunction<Iterable<T>, Void> checkerFn) {
+ pipeline.apply(
+ "PAssert$" + (assertCount++),
+ new OneSideInputAssert<Iterable<T>>(createActual, checkerFn));
+ return this;
+ }
+
+ /**
+ * Applies a {@link SerializableFunction} to check the elements of the {@code Iterable}.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ public IterableAssert<T> satisfies(
+ AssertRelation<Iterable<T>, Iterable<T>> relation,
+ final Iterable<T> expectedElements) {
+ pipeline.apply(
+ "PAssert$" + (assertCount++),
+ new TwoSideInputAssert<Iterable<T>, Iterable<T>>(
+ createActual,
+ new CreateExpected<T, Iterable<T>>(expectedElements, coder, View.<T>asIterable()),
+ relation));
+
+ return this;
+ }
+
+ /**
+ * Applies a {@link SerializableMatcher} to check the elements of the {@code Iterable}.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ IterableAssert<T> satisfies(final SerializableMatcher<Iterable<? extends T>> matcher) {
+ // Safe covariant cast. Could be elided by changing a lot of this file to use
+ // more flexible bounds.
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ SerializableFunction<Iterable<T>, Void> checkerFn =
+ (SerializableFunction) new MatcherCheckerFn<>(matcher);
+ pipeline.apply(
+ "PAssert$" + (assertCount++),
+ new OneSideInputAssert<Iterable<T>>(createActual, checkerFn));
+ return this;
+ }
+
+ private static class MatcherCheckerFn<T> implements SerializableFunction<T, Void> {
+ private SerializableMatcher<T> matcher;
+
+ public MatcherCheckerFn(SerializableMatcher<T> matcher) {
+ this.matcher = matcher;
+ }
+
+ @Override
+ public Void apply(T actual) {
+ assertThat(actual, matcher);
+ return null;
+ }
+ }
+
+ /**
+ * Checks that the {@code Iterable} is empty.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ public IterableAssert<T> empty() {
+ return satisfies(new AssertContainsInAnyOrderRelation<T>(), Collections.<T>emptyList());
+ }
+
+ /**
+ * @throws UnsupportedOperationException always
+ * @deprecated {@link Object#equals(Object)} is not supported on PAssert objects.
+ * If you meant to test object equality, use a variant of {@link #containsInAnyOrder}
+ * instead.
+ */
+ @Deprecated
+ @Override
+ public boolean equals(Object o) {
+ throw new UnsupportedOperationException(
+ "If you meant to test object equality, use .containsInAnyOrder instead.");
+ }
+
+ /**
+ * @throws UnsupportedOperationException always.
+ * @deprecated {@link Object#hashCode()} is not supported on PAssert objects.
+ */
+ @Deprecated
+ @Override
+ public int hashCode() {
+ throw new UnsupportedOperationException(
+ String.format("%s.hashCode() is not supported.", IterableAssert.class.getSimpleName()));
+ }
+
+ /**
+ * Checks that the {@code Iterable} contains the expected elements, in any
+ * order.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ public IterableAssert<T> containsInAnyOrder(Iterable<T> expectedElements) {
+ return satisfies(new AssertContainsInAnyOrderRelation<T>(), expectedElements);
+ }
+
+ /**
+ * Checks that the {@code Iterable} contains the expected elements, in any
+ * order.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ @SafeVarargs
+ public final IterableAssert<T> containsInAnyOrder(T... expectedElements) {
+ return satisfies(
+ new AssertContainsInAnyOrderRelation<T>(),
+ Arrays.asList(expectedElements));
+ }
+
+ /**
+ * Checks that the {@code Iterable} contains elements that match the provided matchers,
+ * in any order.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ @SafeVarargs
+ final IterableAssert<T> containsInAnyOrder(
+ SerializableMatcher<? super T>... elementMatchers) {
+ return satisfies(SerializableMatchers.<T>containsInAnyOrder(elementMatchers));
+ }
+ }
+
+ /**
+ * An assertion about the single value of type {@code T}
+ * associated with a {@link PCollectionView}.
+ */
+ public static class SingletonAssert<T> implements Serializable {
+ private final Pipeline pipeline;
+ private final CreateActual<?, T> createActual;
+ private Optional<Coder<T>> coder;
+
+ protected SingletonAssert(
+ CreateActual<?, T> createActual, Pipeline pipeline) {
+ this.pipeline = pipeline;
+ this.createActual = createActual;
+ this.coder = Optional.absent();
+ }
+
+ /**
+ * Always throws an {@link UnsupportedOperationException}: users are probably looking for
+ * {@link #isEqualTo}.
+ */
+ @Deprecated
+ @Override
+ public boolean equals(Object o) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "tests for Java equality of the %s object, not the PCollection in question. "
+ + "Call a test method, such as isEqualTo.",
+ getClass().getSimpleName()));
+ }
+
+ /**
+ * @throws UnsupportedOperationException always.
+ * @deprecated {@link Object#hashCode()} is not supported on PAssert objects.
+ */
+ @Deprecated
+ @Override
+ public int hashCode() {
+ throw new UnsupportedOperationException(
+ String.format("%s.hashCode() is not supported.", SingletonAssert.class.getSimpleName()));
+ }
+
+ /**
+ * Sets the coder to use for elements of type {@code T}, as needed
+ * for internal purposes.
+ *
+ * <p>Returns this {@code IterableAssert}.
+ */
+ public SingletonAssert<T> setCoder(Coder<T> coderOrNull) {
+ this.coder = Optional.fromNullable(coderOrNull);
+ return this;
+ }
+
+ /**
+ * Gets the coder, which may yet be absent.
+ */
+ public Coder<T> getCoder() {
+ if (coder.isPresent()) {
+ return coder.get();
+ } else {
+ throw new IllegalStateException(
+ "Attempting to access the coder of an IterableAssert that has not been set yet.");
+ }
+ }
+
+ /**
+ * Applies a {@link SerializableFunction} to check the value of this
+ * {@code SingletonAssert}'s view.
+ *
+ * <p>Returns this {@code SingletonAssert}.
+ */
+ public SingletonAssert<T> satisfies(SerializableFunction<T, Void> checkerFn) {
+ pipeline.apply(
+ "PAssert$" + (assertCount++), new OneSideInputAssert<T>(createActual, checkerFn));
+ return this;
+ }
+
+ /**
+ * Applies an {@link AssertRelation} to check the provided relation against the
+ * value of this assert and the provided expected value.
+ *
+ * <p>Returns this {@code SingletonAssert}.
+ */
+ public SingletonAssert<T> satisfies(
+ AssertRelation<T, T> relation,
+ final T expectedValue) {
+ pipeline.apply(
+ "PAssert$" + (assertCount++),
+ new TwoSideInputAssert<T, T>(
+ createActual,
+ new CreateExpected<T, T>(Arrays.asList(expectedValue), coder, View.<T>asSingleton()),
+ relation));
+
+ return this;
+ }
+
+ /**
+ * Checks that the value of this {@code SingletonAssert}'s view is equal
+ * to the expected value.
+ *
+ * <p>Returns this {@code SingletonAssert}.
+ */
+ public SingletonAssert<T> isEqualTo(T expectedValue) {
+ return satisfies(new AssertIsEqualToRelation<T>(), expectedValue);
+ }
+
+ /**
+ * Checks that the value of this {@code SingletonAssert}'s view is not equal
+ * to the expected value.
+ *
+ * <p>Returns this {@code SingletonAssert}.
+ */
+ public SingletonAssert<T> notEqualTo(T expectedValue) {
+ return satisfies(new AssertNotEqualToRelation<T>(), expectedValue);
+ }
+
+ /**
+ * Checks that the value of this {@code SingletonAssert}'s view is equal to
+ * the expected value.
+ *
+ * @deprecated replaced by {@link #isEqualTo}
+ */
+ @Deprecated
+ public SingletonAssert<T> is(T expectedValue) {
+ return isEqualTo(expectedValue);
+ }
+
+ }
+
+ ////////////////////////////////////////////////////////////////////////
+
+ private static class CreateActual<T, ActualT>
+ extends PTransform<PBegin, PCollectionView<ActualT>> {
+
+ private final transient PCollection<T> actual;
+ private final transient PTransform<PCollection<T>, PCollectionView<ActualT>> actualView;
+
+ private CreateActual(PCollection<T> actual,
+ PTransform<PCollection<T>, PCollectionView<ActualT>> actualView) {
+ this.actual = actual;
+ this.actualView = actualView;
+ }
+
+ @Override
+ public PCollectionView<ActualT> apply(PBegin input) {
+ final Coder<T> coder = actual.getCoder();
+ return actual
+ .apply(Window.<T>into(new GlobalWindows()))
+ .apply(ParDo.of(new DoFn<T, T>() {
+ @Override
+ public void processElement(ProcessContext context) throws CoderException {
+ context.output(CoderUtils.clone(coder, context.element()));
+ }
+ }))
+ .apply(actualView);
+ }
+ }
+
+ private static class CreateExpected<T, ExpectedT>
+ extends PTransform<PBegin, PCollectionView<ExpectedT>> {
+
+ private final Iterable<T> elements;
+ private final Optional<Coder<T>> coder;
+ private final transient PTransform<PCollection<T>, PCollectionView<ExpectedT>> view;
+
+ private CreateExpected(Iterable<T> elements, Optional<Coder<T>> coder,
+ PTransform<PCollection<T>, PCollectionView<ExpectedT>> view) {
+ this.elements = elements;
+ this.coder = coder;
+ this.view = view;
+ }
+
+ @Override
+ public PCollectionView<ExpectedT> apply(PBegin input) {
+ Create.Values<T> createTransform = Create.<T>of(elements);
+ if (coder.isPresent()) {
+ createTransform = createTransform.withCoder(coder.get());
+ }
+ return input.apply(createTransform).apply(view);
+ }
+ }
+
+ private static class PreExisting<T> extends PTransform<PBegin, PCollectionView<T>> {
+
+ private final PCollectionView<T> view;
+
+ private PreExisting(PCollectionView<T> view) {
+ this.view = view;
+ }
+
+ @Override
+ public PCollectionView<T> apply(PBegin input) {
+ return view;
+ }
+ }
+
+ /**
+ * An assertion checker that takes a single
+ * {@link PCollectionView PCollectionView<ActualT>}
+ * and an assertion over {@code ActualT}, and checks it within a dataflow
+ * pipeline.
+ *
+ * <p>Note that the entire assertion must be serializable. If
+ * you need to make assertions involving multiple inputs
+ * that are each not serializable, use TwoSideInputAssert.
+ *
+ * <p>This is generally useful for assertion functions that
+ * are serializable but whose underlying data may not have a coder.
+ */
+ static class OneSideInputAssert<ActualT>
+ extends PTransform<PBegin, PDone> implements Serializable {
+ private final transient PTransform<PBegin, PCollectionView<ActualT>> createActual;
+ private final SerializableFunction<ActualT, Void> checkerFn;
+
+ public OneSideInputAssert(
+ PTransform<PBegin, PCollectionView<ActualT>> createActual,
+ SerializableFunction<ActualT, Void> checkerFn) {
+ this.createActual = createActual;
+ this.checkerFn = checkerFn;
+ }
+
+ @Override
+ public PDone apply(PBegin input) {
+ final PCollectionView<ActualT> actual = input.apply("CreateActual", createActual);
+
+ input
+ .apply(Create.<Void>of((Void) null).withCoder(VoidCoder.of()))
+ .apply(ParDo.named("RunChecks").withSideInputs(actual)
+ .of(new CheckerDoFn<>(checkerFn, actual)));
+
+ return PDone.in(input.getPipeline());
+ }
+ }
+
+ /**
+ * A {@link DoFn} that runs a checking {@link SerializableFunction} on the contents of
+ * a {@link PCollectionView}, and adjusts counters and thrown exceptions for use in testing.
+ */
+ private static class CheckerDoFn<ActualT> extends DoFn<Void, Void> {
+ private final SerializableFunction<ActualT, Void> checkerFn;
+ private final Aggregator<Integer, Integer> success =
+ createAggregator(SUCCESS_COUNTER, new Sum.SumIntegerFn());
+ private final Aggregator<Integer, Integer> failure =
+ createAggregator(FAILURE_COUNTER, new Sum.SumIntegerFn());
+ private final PCollectionView<ActualT> actual;
+
+ private CheckerDoFn(
+ SerializableFunction<ActualT, Void> checkerFn,
+ PCollectionView<ActualT> actual) {
+ this.checkerFn = checkerFn;
+ this.actual = actual;
+ }
+
+ @Override
+ public void processElement(ProcessContext c) {
+ try {
+ ActualT actualContents = c.sideInput(actual);
+ checkerFn.apply(actualContents);
+ success.addValue(1);
+ } catch (Throwable t) {
+ LOG.error("PAssert failed expectations.", t);
+ failure.addValue(1);
+ // TODO: allow for metrics to propagate on failure when running a streaming pipeline
+ if (!c.getPipelineOptions().as(StreamingOptions.class).isStreaming()) {
+ throw t;
+ }
+ }
+ }
+ }
+
+ /**
+ * An assertion checker that takes a {@link PCollectionView PCollectionView<ActualT>},
+ * a {@link PCollectionView PCollectionView<ExpectedT>}, a relation
+ * over {@code A} and {@code B}, and checks that the relation holds
+ * within a dataflow pipeline.
+ *
+ * <p>This is useful when either/both of {@code A} and {@code B}
+ * are not serializable, but have coders (provided
+ * by the underlying {@link PCollection}s).
+ */
+ static class TwoSideInputAssert<ActualT, ExpectedT>
+ extends PTransform<PBegin, PDone> implements Serializable {
+
+ private final transient PTransform<PBegin, PCollectionView<ActualT>> createActual;
+ private final transient PTransform<PBegin, PCollectionView<ExpectedT>> createExpected;
+ private final AssertRelation<ActualT, ExpectedT> relation;
+
+ protected TwoSideInputAssert(
+ PTransform<PBegin, PCollectionView<ActualT>> createActual,
+ PTransform<PBegin, PCollectionView<ExpectedT>> createExpected,
+ AssertRelation<ActualT, ExpectedT> relation) {
+ this.createActual = createActual;
+ this.createExpected = createExpected;
+ this.relation = relation;
+ }
+
+ @Override
+ public PDone apply(PBegin input) {
+ final PCollectionView<ActualT> actual = input.apply("CreateActual", createActual);
+ final PCollectionView<ExpectedT> expected = input.apply("CreateExpected", createExpected);
+
+ input
+ .apply(Create.<Void>of((Void) null).withCoder(VoidCoder.of()))
+ .apply(ParDo.named("RunChecks").withSideInputs(actual, expected)
+ .of(new CheckerDoFn<>(relation, actual, expected)));
+
+ return PDone.in(input.getPipeline());
+ }
+
+ private static class CheckerDoFn<ActualT, ExpectedT> extends DoFn<Void, Void> {
+ private final Aggregator<Integer, Integer> success =
+ createAggregator(SUCCESS_COUNTER, new Sum.SumIntegerFn());
+ private final Aggregator<Integer, Integer> failure =
+ createAggregator(FAILURE_COUNTER, new Sum.SumIntegerFn());
+ private final AssertRelation<ActualT, ExpectedT> relation;
+ private final PCollectionView<ActualT> actual;
+ private final PCollectionView<ExpectedT> expected;
+
+ private CheckerDoFn(AssertRelation<ActualT, ExpectedT> relation,
+ PCollectionView<ActualT> actual, PCollectionView<ExpectedT> expected) {
+ this.relation = relation;
+ this.actual = actual;
+ this.expected = expected;
+ }
+
+ @Override
+ public void processElement(ProcessContext c) {
+ try {
+ ActualT actualContents = c.sideInput(actual);
+ ExpectedT expectedContents = c.sideInput(expected);
+ relation.assertFor(expectedContents).apply(actualContents);
+ success.addValue(1);
+ } catch (Throwable t) {
+ LOG.error("PAssert failed expectations.", t);
+ failure.addValue(1);
+ // TODO: allow for metrics to propagate on failure when running a streaming pipeline
+ if (!c.getPipelineOptions().as(StreamingOptions.class).isStreaming()) {
+ throw t;
+ }
+ }
+ }
+ }
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+
+ /**
+ * A {@link SerializableFunction} that verifies that an actual value is equal to an
+ * expected value.
+ */
+ private static class AssertIsEqualTo<T> implements SerializableFunction<T, Void> {
+ private T expected;
+
+ public AssertIsEqualTo(T expected) {
+ this.expected = expected;
+ }
+
+ @Override
+ public Void apply(T actual) {
+ assertThat(actual, equalTo(expected));
+ return null;
+ }
+ }
+
+ /**
+ * A {@link SerializableFunction} that verifies that an actual value is not equal to an
+ * expected value.
+ */
+ private static class AssertNotEqualTo<T> implements SerializableFunction<T, Void> {
+ private T expected;
+
+ public AssertNotEqualTo(T expected) {
+ this.expected = expected;
+ }
+
+ @Override
+ public Void apply(T actual) {
+ assertThat(actual, not(equalTo(expected)));
+ return null;
+ }
+ }
+
+ /**
+ * A {@link SerializableFunction} that verifies that an {@code Iterable} contains
+ * expected items in any order.
+ */
+ private static class AssertContainsInAnyOrder<T>
+ implements SerializableFunction<Iterable<T>, Void> {
+ private T[] expected;
+
+ @SafeVarargs
+ public AssertContainsInAnyOrder(T... expected) {
+ this.expected = expected;
+ }
+
+ @SuppressWarnings("unchecked")
+ public AssertContainsInAnyOrder(Collection<T> expected) {
+ this((T[]) expected.toArray());
+ }
+
+ public AssertContainsInAnyOrder(Iterable<T> expected) {
+ this(Lists.<T>newArrayList(expected));
+ }
+
+ @Override
+ public Void apply(Iterable<T> actual) {
+ assertThat(actual, containsInAnyOrder(expected));
+ return null;
+ }
+ }
+
+ ////////////////////////////////////////////////////////////
+
+ /**
+ * A binary predicate between types {@code Actual} and {@code Expected}.
+ * Implemented as a method {@code assertFor(Expected)} which returns
+ * a {@code SerializableFunction<Actual, Void>}
+ * that should verify the assertion..
+ */
+ private static interface AssertRelation<ActualT, ExpectedT> extends Serializable {
+ public SerializableFunction<ActualT, Void> assertFor(ExpectedT input);
+ }
+
+ /**
+ * An {@link AssertRelation} implementing the binary predicate that two objects are equal.
+ */
+ private static class AssertIsEqualToRelation<T>
+ implements AssertRelation<T, T> {
+ @Override
+ public SerializableFunction<T, Void> assertFor(T expected) {
+ return new AssertIsEqualTo<T>(expected);
+ }
+ }
+
+ /**
+ * An {@link AssertRelation} implementing the binary predicate that two objects are not equal.
+ */
+ private static class AssertNotEqualToRelation<T>
+ implements AssertRelation<T, T> {
+ @Override
+ public SerializableFunction<T, Void> assertFor(T expected) {
+ return new AssertNotEqualTo<T>(expected);
+ }
+ }
+
+ /**
+ * An {@code AssertRelation} implementing the binary predicate that two collections are equal
+ * modulo reordering.
+ */
+ private static class AssertContainsInAnyOrderRelation<T>
+ implements AssertRelation<Iterable<T>, Iterable<T>> {
+ @Override
+ public SerializableFunction<Iterable<T>, Void> assertFor(Iterable<T> expectedElements) {
+ return new AssertContainsInAnyOrder<T>(expectedElements);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SerializableMatcher.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SerializableMatcher.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SerializableMatcher.java
index 879d7e2..238e0d1 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SerializableMatcher.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SerializableMatcher.java
@@ -24,7 +24,7 @@ import java.io.Serializable;
/**
* A {@link Matcher} that is also {@link Serializable}.
*
- * <p>Such matchers can be used with {@link DataflowAssert}, which builds Dataflow pipelines
+ * <p>Such matchers can be used with {@link PAssert}, which builds Dataflow pipelines
* such that these matchers may be serialized and executed remotely.
*
* <p>To create a {@code SerializableMatcher}, extend {@link org.hamcrest.BaseMatcher}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SourceTestUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SourceTestUtils.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SourceTestUtils.java
index a0b4ffd..feff16a 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SourceTestUtils.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/SourceTestUtils.java
@@ -71,7 +71,7 @@ import java.util.concurrent.Future;
* {@link com.google.cloud.dataflow.sdk.io.AvroSource} or
* {@link com.google.cloud.dataflow.sdk.io.XmlSource}.
*
- * <p>Like {@link DataflowAssert}, requires JUnit and Hamcrest to be present in the classpath.
+ * <p>Like {@link PAssert}, requires JUnit and Hamcrest to be present in the classpath.
*/
public class SourceTestUtils {
private static final Logger LOG = LoggerFactory.getLogger(SourceTestUtils.class);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestDataflowPipelineRunner.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestDataflowPipelineRunner.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestDataflowPipelineRunner.java
index 065d808..8f7374f 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestDataflowPipelineRunner.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestDataflowPipelineRunner.java
@@ -158,8 +158,8 @@ public class TestDataflowPipelineRunner extends PipelineRunner<DataflowPipelineJ
@Override
public <OutputT extends POutput, InputT extends PInput> OutputT apply(
PTransform<InputT, OutputT> transform, InputT input) {
- if (transform instanceof DataflowAssert.OneSideInputAssert
- || transform instanceof DataflowAssert.TwoSideInputAssert) {
+ if (transform instanceof PAssert.OneSideInputAssert
+ || transform instanceof PAssert.TwoSideInputAssert) {
expectedNumberOfAssertions += 1;
}
@@ -188,9 +188,9 @@ public class TestDataflowPipelineRunner extends PipelineRunner<DataflowPipelineJ
// Don't double count using the non-tentative version of the metric.
continue;
}
- if (DataflowAssert.SUCCESS_COUNTER.equals(metric.getName().getName())) {
+ if (PAssert.SUCCESS_COUNTER.equals(metric.getName().getName())) {
successes += ((BigDecimal) metric.getScalar()).intValue();
- } else if (DataflowAssert.FAILURE_COUNTER.equals(metric.getName().getName())) {
+ } else if (PAssert.FAILURE_COUNTER.equals(metric.getName().getName())) {
failures += ((BigDecimal) metric.getScalar()).intValue();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestPipeline.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestPipeline.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestPipeline.java
index e97e90f..98d4823 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestPipeline.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/testing/TestPipeline.java
@@ -55,14 +55,14 @@ import javax.annotation.Nullable;
* <li> Jars containing the SDK and test classes must be added to the test classpath.
* </ul>
*
- * <p>Use {@link DataflowAssert} for tests, as it integrates with this test
+ * <p>Use {@link PAssert} for tests, as it integrates with this test
* harness in both direct and remote execution modes. For example:
*
* <pre>{@code
* Pipeline p = TestPipeline.create();
* PCollection<Integer> output = ...
*
- * DataflowAssert.that(output)
+ * PAssert.that(output)
* .containsInAnyOrder(1, 2, 3, 4);
* p.run();
* }</pre>
@@ -75,7 +75,7 @@ public class TestPipeline extends Pipeline {
/**
* Creates and returns a new test pipeline.
*
- * <p>Use {@link DataflowAssert} to add tests, then call
+ * <p>Use {@link PAssert} to add tests, then call
* {@link Pipeline#run} to execute the pipeline and check the tests.
*/
public static TestPipeline create() {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/PipelineTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/PipelineTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/PipelineTest.java
index 8463f8c..e6a6426 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/PipelineTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/PipelineTest.java
@@ -30,8 +30,8 @@ import com.google.cloud.dataflow.sdk.options.PipelineOptions.CheckEnabled;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
import com.google.cloud.dataflow.sdk.testing.ExpectedLogs;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Create;
@@ -137,7 +137,7 @@ public class PipelineTest {
PCollection<String> both = PCollectionList.of(left).and(right)
.apply(Flatten.<String>pCollections());
- DataflowAssert.that(both).containsInAnyOrder("a++", "b++", "a+", "b+");
+ PAssert.that(both).containsInAnyOrder("a++", "b++", "a+", "b+");
p.run();
}
@@ -205,7 +205,7 @@ public class PipelineTest {
.apply(Create.<Integer>of(1, 2, 3, 4))
.apply("IdentityTransform", new IdentityTransform<PCollection<Integer>>());
- DataflowAssert.that(output).containsInAnyOrder(1, 2, 3, 4);
+ PAssert.that(output).containsInAnyOrder(1, 2, 3, 4);
pipeline.run();
}
@@ -234,7 +234,7 @@ public class PipelineTest {
PCollection<Integer> output = tuple
.apply("ProjectTag", new TupleProjectionTransform<Integer>(tag));
- DataflowAssert.that(output).containsInAnyOrder(1, 2, 3, 4);
+ PAssert.that(output).containsInAnyOrder(1, 2, 3, 4);
pipeline.run();
}
@@ -268,7 +268,7 @@ public class PipelineTest {
PCollectionTuple output = input
.apply("ProjectTag", new TupleInjectionTransform<Integer>(tag));
- DataflowAssert.that(output.get(tag)).containsInAnyOrder(1, 2, 3, 4);
+ PAssert.that(output.get(tag)).containsInAnyOrder(1, 2, 3, 4);
pipeline.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/AvroCoderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/AvroCoderTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/AvroCoderTest.java
index 021b4a5..d938a1c 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/AvroCoderTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/AvroCoderTest.java
@@ -27,7 +27,7 @@ import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.coders.Coder.Context;
import com.google.cloud.dataflow.sdk.coders.Coder.NonDeterministicException;
import com.google.cloud.dataflow.sdk.testing.CoderProperties;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Create;
import com.google.cloud.dataflow.sdk.transforms.DoFn;
@@ -220,7 +220,7 @@ public class AvroCoderTest {
p.apply(Create.of(new Pojo("hello", 1), new Pojo("world", 2)))
.apply(ParDo.of(new GetTextFn()));
- DataflowAssert.that(output)
+ PAssert.that(output)
.containsInAnyOrder("hello", "world");
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/SerializableCoderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/SerializableCoderTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/SerializableCoderTest.java
index ac2136a..799da7c 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/SerializableCoderTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/coders/SerializableCoderTest.java
@@ -22,7 +22,7 @@ import static org.junit.Assert.assertNull;
import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.testing.CoderProperties;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Create;
import com.google.cloud.dataflow.sdk.transforms.DoFn;
@@ -139,7 +139,7 @@ public class SerializableCoderTest implements Serializable {
.apply(ParDo.of(new StringToRecord()))
.apply(ParDo.of(new RecordToString()));
- DataflowAssert.that(output)
+ PAssert.that(output)
.containsInAnyOrder("Hello", "World");
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/AvroIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/AvroIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/AvroIOTest.java
index ccc24d5..7949b9f 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/AvroIOTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/AvroIOTest.java
@@ -27,7 +27,7 @@ import com.google.cloud.dataflow.sdk.coders.AvroCoder;
import com.google.cloud.dataflow.sdk.coders.DefaultCoder;
import com.google.cloud.dataflow.sdk.io.AvroIO.Write.Bound;
import com.google.cloud.dataflow.sdk.runners.DirectPipeline;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Create;
import com.google.cloud.dataflow.sdk.util.IOChannelUtils;
@@ -130,7 +130,7 @@ public class AvroIOTest {
PCollection<GenericClass> input = p
.apply(AvroIO.Read.from(outputFile.getAbsolutePath()).withSchema(GenericClass.class));
- DataflowAssert.that(input).containsInAnyOrder(values);
+ PAssert.that(input).containsInAnyOrder(values);
p.run();
}
@@ -195,7 +195,7 @@ public class AvroIOTest {
PCollection<GenericClassV2> input = p
.apply(AvroIO.Read.from(outputFile.getAbsolutePath()).withSchema(GenericClassV2.class));
- DataflowAssert.that(input).containsInAnyOrder(expected);
+ PAssert.that(input).containsInAnyOrder(expected);
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/BoundedReadFromUnboundedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/BoundedReadFromUnboundedSourceTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/BoundedReadFromUnboundedSourceTest.java
index a98402d..24b0fca 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/BoundedReadFromUnboundedSourceTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/BoundedReadFromUnboundedSourceTest.java
@@ -25,7 +25,7 @@ import static org.junit.Assert.assertTrue;
import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.dataflow.TestCountingSource;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
@@ -126,7 +126,7 @@ public class BoundedReadFromUnboundedSourceTest {
// Because some of the NUM_RECORDS elements read are dupes, the final output
// will only have output from 0 to n where n < NUM_RECORDS.
- DataflowAssert.that(output).satisfies(new Checker(dedup, timeBound));
+ PAssert.that(output).satisfies(new Checker(dedup, timeBound));
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CompressedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CompressedSourceTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CompressedSourceTest.java
index 3de0513..331c896 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CompressedSourceTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CompressedSourceTest.java
@@ -29,7 +29,7 @@ import com.google.cloud.dataflow.sdk.io.CompressedSource.CompressionMode;
import com.google.cloud.dataflow.sdk.io.CompressedSource.DecompressingChannelFactory;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.SourceTestUtils;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.values.PCollection;
@@ -140,7 +140,7 @@ public class CompressedSourceTest {
.withDecompression(CompressionMode.GZIP);
PCollection<Byte> output = p.apply(Read.from(source));
- DataflowAssert.that(output).containsInAnyOrder(Bytes.asList(expected));
+ PAssert.that(output).containsInAnyOrder(Bytes.asList(expected));
p.run();
}
@@ -210,7 +210,7 @@ public class CompressedSourceTest {
CompressedSource.from(new ByteSource(filePattern, 1));
PCollection<Byte> output = p.apply(Read.from(source));
- DataflowAssert.that(output).containsInAnyOrder(expected);
+ PAssert.that(output).containsInAnyOrder(expected);
p.run();
}
@@ -327,7 +327,7 @@ public class CompressedSourceTest {
.withDecompression(CompressionMode.GZIP);
PCollection<Byte> output = p.apply(Read.from(source));
- DataflowAssert.that(output).containsInAnyOrder(expected);
+ PAssert.that(output).containsInAnyOrder(expected);
p.run();
}
@@ -389,7 +389,7 @@ public class CompressedSourceTest {
source = source.withDecompression(decompressionFactory);
}
PCollection<Byte> output = p.apply(Read.from(source));
- DataflowAssert.that(output).containsInAnyOrder(Bytes.asList(expected));
+ PAssert.that(output).containsInAnyOrder(Bytes.asList(expected));
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingInputTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingInputTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingInputTest.java
index 5a7c2fb..2b369ed 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingInputTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingInputTest.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertThat;
import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.io.CountingInput.UnboundedCountingInput;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Count;
@@ -49,18 +49,18 @@ import org.junit.runners.JUnit4;
public class CountingInputTest {
public static void addCountingAsserts(PCollection<Long> input, long numElements) {
// Count == numElements
- DataflowAssert.thatSingleton(input.apply("Count", Count.<Long>globally()))
+ PAssert.thatSingleton(input.apply("Count", Count.<Long>globally()))
.isEqualTo(numElements);
// Unique count == numElements
- DataflowAssert.thatSingleton(
+ PAssert.thatSingleton(
input
.apply(RemoveDuplicates.<Long>create())
.apply("UniqueCount", Count.<Long>globally()))
.isEqualTo(numElements);
// Min == 0
- DataflowAssert.thatSingleton(input.apply("Min", Min.<Long>globally())).isEqualTo(0L);
+ PAssert.thatSingleton(input.apply("Min", Min.<Long>globally())).isEqualTo(0L);
// Max == numElements-1
- DataflowAssert.thatSingleton(input.apply("Max", Max.<Long>globally()))
+ PAssert.thatSingleton(input.apply("Max", Max.<Long>globally()))
.isEqualTo(numElements - 1);
}
@@ -133,7 +133,7 @@ public class CountingInputTest {
.apply("TimestampDiff", ParDo.of(new ElementValueDiff()))
.apply("RemoveDuplicateTimestamps", RemoveDuplicates.<Long>create());
// This assert also confirms that diffs only has one unique value.
- DataflowAssert.thatSingleton(diffs).isEqualTo(0L);
+ PAssert.thatSingleton(diffs).isEqualTo(0L);
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingSourceTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingSourceTest.java
index 09c1f2a..5abd081 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingSourceTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/CountingSourceTest.java
@@ -27,7 +27,7 @@ import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.io.CountingSource.CounterMark;
import com.google.cloud.dataflow.sdk.io.CountingSource.UnboundedCountingSource;
import com.google.cloud.dataflow.sdk.io.UnboundedSource.UnboundedReader;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Count;
@@ -59,20 +59,20 @@ public class CountingSourceTest {
public static void addCountingAsserts(PCollection<Long> input, long numElements) {
// Count == numElements
- DataflowAssert
+ PAssert
.thatSingleton(input.apply("Count", Count.<Long>globally()))
.isEqualTo(numElements);
// Unique count == numElements
- DataflowAssert
+ PAssert
.thatSingleton(input.apply(RemoveDuplicates.<Long>create())
.apply("UniqueCount", Count.<Long>globally()))
.isEqualTo(numElements);
// Min == 0
- DataflowAssert
+ PAssert
.thatSingleton(input.apply("Min", Min.<Long>globally()))
.isEqualTo(0L);
// Max == numElements-1
- DataflowAssert
+ PAssert
.thatSingleton(input.apply("Max", Max.<Long>globally()))
.isEqualTo(numElements - 1);
}
@@ -150,7 +150,7 @@ public class CountingSourceTest {
.apply("TimestampDiff", ParDo.of(new ElementValueDiff()))
.apply("RemoveDuplicateTimestamps", RemoveDuplicates.<Long>create());
// This assert also confirms that diffs only has one unique value.
- DataflowAssert.thatSingleton(diffs).isEqualTo(0L);
+ PAssert.thatSingleton(diffs).isEqualTo(0L);
p.run();
}
@@ -176,7 +176,7 @@ public class CountingSourceTest {
.apply("TimestampDiff", ParDo.of(new ElementValueDiff()))
.apply("RemoveDuplicateTimestamps", RemoveDuplicates.<Long>create());
// This assert also confirms that diffs only has one unique value.
- DataflowAssert.thatSingleton(diffs).isEqualTo(0L);
+ PAssert.thatSingleton(diffs).isEqualTo(0L);
Instant started = Instant.now();
p.run();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/FileBasedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/FileBasedSourceTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/FileBasedSourceTest.java
index d7a2e09..1071a67 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/FileBasedSourceTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/FileBasedSourceTest.java
@@ -35,7 +35,7 @@ import com.google.cloud.dataflow.sdk.io.FileBasedSource.FileBasedReader;
import com.google.cloud.dataflow.sdk.io.Source.Reader;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.util.CoderUtils;
import com.google.cloud.dataflow.sdk.util.IOChannelFactory;
@@ -715,7 +715,7 @@ public class FileBasedSourceTest {
TestFileBasedSource source = new TestFileBasedSource(file.getPath(), 64, null);
PCollection<String> output = p.apply(Read.from(source).named("ReadFileData"));
- DataflowAssert.that(output).containsInAnyOrder(data);
+ PAssert.that(output).containsInAnyOrder(data);
p.run();
}
@@ -745,7 +745,7 @@ public class FileBasedSourceTest {
expectedResults.addAll(data2);
expectedResults.addAll(data3);
- DataflowAssert.that(output).containsInAnyOrder(expectedResults);
+ PAssert.that(output).containsInAnyOrder(expectedResults);
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
index 5a9a8c6..bc3a6fe 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
@@ -35,7 +35,7 @@ import com.google.cloud.dataflow.sdk.coders.VoidCoder;
import com.google.cloud.dataflow.sdk.io.TextIO.CompressionType;
import com.google.cloud.dataflow.sdk.io.TextIO.TextSource;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.SourceTestUtils;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Create;
@@ -100,7 +100,7 @@ public class TextIOTest {
PCollection<T> output = p.apply(read);
- DataflowAssert.that(output).containsInAnyOrder(expected);
+ PAssert.that(output).containsInAnyOrder(expected);
p.run();
}
@@ -340,7 +340,7 @@ public class TextIOTest {
TextIO.Read.from(filename).withCompressionType(CompressionType.GZIP);
PCollection<String> output = p.apply(read);
- DataflowAssert.that(output).containsInAnyOrder(expected);
+ PAssert.that(output).containsInAnyOrder(expected);
p.run();
}
@@ -363,7 +363,7 @@ public class TextIOTest {
TextIO.Read.from(filename).withCompressionType(CompressionType.GZIP);
PCollection<String> output = p.apply(read);
- DataflowAssert.that(output).containsInAnyOrder(expected);
+ PAssert.that(output).containsInAnyOrder(expected);
p.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d73ceab8/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/XmlSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/XmlSourceTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/XmlSourceTest.java
index 41e2887..9865615 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/XmlSourceTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/XmlSourceTest.java
@@ -31,7 +31,7 @@ import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.io.Source.Reader;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
-import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
+import com.google.cloud.dataflow.sdk.testing.PAssert;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.values.PCollection;
import com.google.common.collect.ImmutableList;
@@ -583,7 +583,7 @@ public class XmlSourceTest {
new Train("Toby", 7, "brown", null), new Train("Gordon", 4, "blue", null),
new Train("Emily", -1, "red", null), new Train("Percy", 6, "green", null));
- DataflowAssert.that(output).containsInAnyOrder(expectedResults);
+ PAssert.that(output).containsInAnyOrder(expectedResults);
p.run();
}
@@ -667,7 +667,7 @@ public class XmlSourceTest {
.withMinBundleSize(1024);
PCollection<Train> output = p.apply(Read.from(source).named("ReadFileData"));
- DataflowAssert.that(output).containsInAnyOrder(trains);
+ PAssert.that(output).containsInAnyOrder(trains);
p.run();
}
@@ -819,7 +819,7 @@ public class XmlSourceTest {
expectedResults.addAll(trains2);
expectedResults.addAll(trains3);
- DataflowAssert.that(output).containsInAnyOrder(expectedResults);
+ PAssert.that(output).containsInAnyOrder(expectedResults);
p.run();
}
}