com.google.cloud.dataflow.sdk.testing

## Class SourceTestUtils

- java.lang.Object
- - com.google.cloud.dataflow.sdk.testing.SourceTestUtils

    public class SourceTestUtils
    extends Object

Helper functions and test harnesses for checking correctness of `Source` implementations.

Contains a few lightweight utilities (e.g. reading items from a source or a reader, such as `readFromSource(com.google.cloud.dataflow.sdk.io.BoundedSource<T>, com.google.cloud.dataflow.sdk.options.PipelineOptions)` and `readFromUnstartedReader(com.google.cloud.dataflow.sdk.io.Source.Reader<T>)`), as well as heavyweight property testing and stress testing harnesses that help getting a large amount of test coverage with few code. Most notable ones are:

- `assertSourcesEqualReferenceSource(com.google.cloud.dataflow.sdk.io.BoundedSource<T>, java.util.List<? extends com.google.cloud.dataflow.sdk.io.BoundedSource<T>>, com.google.cloud.dataflow.sdk.options.PipelineOptions)` helps testing that the data read by the union of sources produced by `BoundedSource.splitIntoBundles(long, com.google.cloud.dataflow.sdk.options.PipelineOptions)` is the same as data read by the original source.
- If your source implements dynamic work rebalancing, use the `assertSplitAtFraction` family of functions - they test behavior of `BoundedSource.BoundedReader#splitAtFraction`, in particular, that various consistency properties are respected and the total set of data read by the source is preserved when splits happen. Use `assertSplitAtFractionBehavior(com.google.cloud.dataflow.sdk.io.BoundedSource<T>, int, double, com.google.cloud.dataflow.sdk.testing.SourceTestUtils.ExpectedSplitOutcome, com.google.cloud.dataflow.sdk.options.PipelineOptions)` to test individual cases of `splitAtFraction` and use `assertSplitAtFractionExhaustive(com.google.cloud.dataflow.sdk.io.BoundedSource<T>, com.google.cloud.dataflow.sdk.options.PipelineOptions)` as a heavy-weight stress test including concurrency. We strongly recommend to use both.

For example usages, see the unit tests of classes such as `AvroSource` or `XmlSource`.

Like `DataflowAssert`, requires JUnit and Hamcrest to be present in the classpath.

### Nested Class Summary

Nested Classes
`static class` `SourceTestUtils.ExpectedSplitOutcome` Expected outcome of `BoundedSource.BoundedReader.splitAtFraction(double)`.

### Constructor Summary

Constructors
Constructor and Description
`SourceTestUtils()`

### Method Summary

All Methods Static Methods Concrete Methods
`static <T> void` `assertSourcesEqualReferenceSource(BoundedSource<T> referenceSource, List<? extends BoundedSource<T>> sources, PipelineOptions options)` Given a reference `Source` and a list of `Source`s, assert that the union of the records read from the list of sources is equal to the records read from the reference source.

`static <T> com.google.cloud.dataflow.sdk.testing.SourceTestUtils.SplitAtFractionResult`

`assertSplitAtFractionBehavior(BoundedSource<T> source, int numItemsToReadBeforeSplit, double splitFraction, SourceTestUtils.ExpectedSplitOutcome expectedOutcome, PipelineOptions options)`
Asserts that the `source`'s reader either fails to `splitAtFraction(fraction)` after reading `numItemsToReadBeforeSplit` items, or succeeds in a way that is consistent according to `assertSplitAtFractionSucceedsAndConsistent(com.google.cloud.dataflow.sdk.io.BoundedSource<T>, int, double, com.google.cloud.dataflow.sdk.options.PipelineOptions)`.

`static <T> void` `assertSplitAtFractionExhaustive(BoundedSource<T> source, PipelineOptions options)` Asserts that for each possible start position, `BoundedSource.BoundedReader#splitAtFraction` at every interesting fraction (halfway between two fractions that differ by at least one item) can be called successfully and the results are consistent if a split succeeds.
`static <T> void` `assertSplitAtFractionFails(BoundedSource<T> source, int numItemsToReadBeforeSplit, double splitFraction, PipelineOptions options)` Asserts that the `source`'s reader fails to `splitAtFraction(fraction)` after reading `numItemsToReadBeforeSplit` items.
`static <T> void` `assertSplitAtFractionSucceedsAndConsistent(BoundedSource<T> source, int numItemsToReadBeforeSplit, double splitFraction, PipelineOptions options)` Verifies some consistency properties of `BoundedSource.BoundedReader#splitAtFraction` on the given source.
`static <T> void` `assertUnstartedReaderReadsSameAsItsSource(BoundedSource.BoundedReader<T> reader, PipelineOptions options)` Assert that a `Reader` returns a `Source` that, when read from, produces the same records as the reader.
`static <T> List<com.google.cloud.dataflow.sdk.testing.SourceTestUtils.ReadableStructuralValue<T>>` `createStructuralValues(Coder<T> coder, List<T> list)` Testing utilities below depend on standard assertions and matchers to compare elements read by sources.
`static <T> List<T>` `readFromSource(BoundedSource<T> source, PipelineOptions options)` Reads all elements from the given `BoundedSource`.
`static <T> List<T>` `readFromStartedReader(Source.Reader<T> reader)` Reads all elements from the given started `Source.Reader`.
`static <T> List<T>` `readFromUnstartedReader(Source.Reader<T> reader)` Reads all elements from the given unstarted `Source.Reader`.
`static <T> List<T>` `readNItemsFromStartedReader(Source.Reader<T> reader, int n)` Read elements from a `Source.Reader` that has already had `Source.Reader#start` called on it, until n elements are read.
`static <T> List<T>` `readNItemsFromUnstartedReader(Source.Reader<T> reader, int n)` Read elements from a `Source.Reader` until n elements are read.
`static <T> List<T>` `readRemainingFromReader(Source.Reader<T> reader, boolean started)` Read all remaining elements from a `Source.Reader`.
`static <T> BoundedSource<T>` `toUnsplittableSource(BoundedSource<T> boundedSource)` Returns an equivalent unsplittable `BoundedSource<T>`.

### Methods inherited from class java.lang.Object

`clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

### Constructor Detail

#### SourceTestUtils

    public SourceTestUtils()

### Method Detail

#### createStructuralValues

    public static <T> List<com.google.cloud.dataflow.sdk.testing.SourceTestUtils.ReadableStructuralValue<T>> createStructuralValues(Coder<T> coder,
                                                                                                                                    List<T> list)
                                                                                                                             throws Exception

Testing utilities below depend on standard assertions and matchers to compare elements read by sources. In general the elements may not implement `equals`/`hashCode` properly, however every source has a `Coder` and every `Coder` can produce a `Coder.structuralValue(T)` whose `equals`/`hashCode` is consistent with equality of encoded format. So we use this `Coder.structuralValue(T)` to compare elements read by sources.

Throws:
  `Exception`

#### readFromSource

    public static <T> List<T> readFromSource(BoundedSource<T> source,
                                             PipelineOptions options)
                                      throws IOException

Reads all elements from the given `BoundedSource`.

Throws:
  `IOException`

#### readFromUnstartedReader

    public static <T> List<T> readFromUnstartedReader(Source.Reader<T> reader)
                                               throws IOException

Reads all elements from the given unstarted `Source.Reader`.

Throws:
  `IOException`

#### readFromStartedReader

    public static <T> List<T> readFromStartedReader(Source.Reader<T> reader)
                                             throws IOException

Reads all elements from the given started `Source.Reader`.

Throws:
  `IOException`

#### readNItemsFromUnstartedReader

    public static <T> List<T> readNItemsFromUnstartedReader(Source.Reader<T> reader,
                                                            int n)
                                                     throws IOException

Read elements from a `Source.Reader` until n elements are read.

Throws:
  `IOException`

#### readNItemsFromStartedReader

    public static <T> List<T> readNItemsFromStartedReader(Source.Reader<T> reader,
                                                          int n)
                                                   throws IOException

Read elements from a `Source.Reader` that has already had `Source.Reader#start` called on it, until n elements are read.

Throws:
  `IOException`

#### readRemainingFromReader

    public static <T> List<T> readRemainingFromReader(Source.Reader<T> reader,
                                                      boolean started)
                                               throws IOException

Read all remaining elements from a `Source.Reader`.

Throws:
  `IOException`

#### assertSourcesEqualReferenceSource

    public static <T> void assertSourcesEqualReferenceSource(BoundedSource<T> referenceSource,
                                                             List<? extends BoundedSource<T>> sources,
                                                             PipelineOptions options)
                                                      throws Exception

Given a reference `Source` and a list of `Source`s, assert that the union of the records read from the list of sources is equal to the records read from the reference source.

Throws:
  `Exception`

#### assertUnstartedReaderReadsSameAsItsSource

    public static <T> void assertUnstartedReaderReadsSameAsItsSource(BoundedSource.BoundedReader<T> reader,
                                                                     PipelineOptions options)
                                                              throws Exception

Assert that a `Reader` returns a `Source` that, when read from, produces the same records as the reader.

Throws:
  `Exception`

#### assertSplitAtFractionBehavior

    public static <T> com.google.cloud.dataflow.sdk.testing.SourceTestUtils.SplitAtFractionResult assertSplitAtFractionBehavior(BoundedSource<T> source,
                                                                                                                                int numItemsToReadBeforeSplit,
                                                                                                                                double splitFraction,
                                                                                                                                SourceTestUtils.ExpectedSplitOutcome expectedOutcome,
                                                                                                                                PipelineOptions options)
                                                                                                                         throws Exception

Asserts that the `source`'s reader either fails to `splitAtFraction(fraction)` after reading `numItemsToReadBeforeSplit` items, or succeeds in a way that is consistent according to `assertSplitAtFractionSucceedsAndConsistent(com.google.cloud.dataflow.sdk.io.BoundedSource<T>, int, double, com.google.cloud.dataflow.sdk.options.PipelineOptions)`.

Returns SplitAtFractionResult.

Throws:
  `Exception`

#### assertSplitAtFractionSucceedsAndConsistent

    public static <T> void assertSplitAtFractionSucceedsAndConsistent(BoundedSource<T> source,
                                                                      int numItemsToReadBeforeSplit,
                                                                      double splitFraction,
                                                                      PipelineOptions options)
                                                               throws Exception

Verifies some consistency properties of `BoundedSource.BoundedReader#splitAtFraction` on the given source. Equivalent to the following pseudocode:

       Reader reader = source.createReader();
       read N items from reader;
       Source residual = reader.splitAtFraction(splitFraction);
       Source primary = reader.getCurrentSource();
       assert: items in primary == items we read so far
                                   + items we'll get by continuing to read from reader;
       assert: items in original source == items in primary + items in residual

Throws:
  `Exception`

#### assertSplitAtFractionFails

    public static <T> void assertSplitAtFractionFails(BoundedSource<T> source,
                                                      int numItemsToReadBeforeSplit,
                                                      double splitFraction,
                                                      PipelineOptions options)
                                               throws Exception

Asserts that the `source`'s reader fails to `splitAtFraction(fraction)` after reading `numItemsToReadBeforeSplit` items.

Throws:
  `Exception`

#### assertSplitAtFractionExhaustive

    public static <T> void assertSplitAtFractionExhaustive(BoundedSource<T> source,
                                                           PipelineOptions options)
                                                    throws Exception

Asserts that for each possible start position, `BoundedSource.BoundedReader#splitAtFraction` at every interesting fraction (halfway between two fractions that differ by at least one item) can be called successfully and the results are consistent if a split succeeds. Verifies multithreaded splitting as well.

Throws:
  `Exception`

#### toUnsplittableSource

    public static <T> BoundedSource<T> toUnsplittableSource(BoundedSource<T> boundedSource)

Returns an equivalent unsplittable `BoundedSource<T>`.

It forwards most methods to the given `boundedSource`, except:

1. `BoundedSource.splitIntoBundles(long, com.google.cloud.dataflow.sdk.options.PipelineOptions)` rejects initial splitting by returning itself in a list.
2. `BoundedReader#splitAtFraction` rejects dynamic splitting by returning null.