passert

Documentation

Overview

Package passert contains verification transformations for testing pipelines. The transformations are not tied to any particular runner, i.e., they can notably be used for remote execution runners, such as Dataflow.

Index

• func AllWithinBounds(s beam.Scope, col beam.PCollection, lo, hi float64)
• func Count(s beam.Scope, col beam.PCollection, name string, count int)
• func Diff(s beam.Scope, a, b beam.PCollection) (left, both, right beam.PCollection)
• func Empty(s beam.Scope, col beam.PCollection) beam.PCollection
• func Equals(s beam.Scope, col beam.PCollection, values ...any) beam.PCollection
• func EqualsFloat(s beam.Scope, observed, expected beam.PCollection, threshold float64)
• func EqualsList(s beam.Scope, col beam.PCollection, list any) beam.PCollection
• func False(s beam.Scope, col beam.PCollection, fn any) beam.PCollection
• func Hash(s beam.Scope, col beam.PCollection, name, hash string, size int)
• func NonEmpty(s beam.Scope, col beam.PCollection) beam.PCollection
• func Sum(s beam.Scope, col beam.PCollection, name string, size, value int)
• func True(s beam.Scope, col beam.PCollection, fn any) beam.PCollection
• func TryEqualsFloat(s beam.Scope, observed, expected beam.PCollection, threshold float64) error

Examples

• Equals
• Equals (Pcollection)
• EqualsList
• EqualsList (Mismatch)

Functions

func AllWithinBounds

func AllWithinBounds(s beam.Scope, col beam.PCollection, lo, hi float64)

AllWithinBounds checks that a PCollection of numeric types is within the bounds [lo, high]. Checks for case where bounds are flipped and swaps them so the bounds passed to the doFn are always lo <= hi.

func Count

func Count(s beam.Scope, col beam.PCollection, name string, count int)

Count verifies the given PCollection<T> has the specified number of elements.

func Diff

func Diff(s beam.Scope, a, b beam.PCollection) (left, both, right beam.PCollection)

Diff splits 2 incoming PCollections into 3: left only, both, right only. Duplicates are preserved, so a value may appear multiple times and in multiple collections. Coder equality is used to determine equality. Should only be used for small collections, because all values are held in memory at the same time.

func Empty

func Empty(s beam.Scope, col beam.PCollection) beam.PCollection

Empty asserts that col is empty.

func Equals

func Equals(s beam.Scope, col beam.PCollection, values ...any) beam.PCollection

Equals verifies the given collection has the same values as the given values, under coder equality. The values can be provided as single PCollection.

Example

p, s := beam.NewPipelineWithRoot()
col := beam.Create(s, "some", "example", "strings")

Equals(s, col, "example", "some", "strings")
err := ptest.Run(p)
fmt.Println(err == nil)

Output:

true

Example (Pcollection)

p, s := beam.NewPipelineWithRoot()
col := beam.Create(s, "some", "example", "strings")
exp := beam.Create(s, "example", "some", "strings")

Equals(s, col, exp)
err := ptest.Run(p)
fmt.Println(err == nil)

Output:

true

func EqualsFloat

func EqualsFloat(s beam.Scope, observed, expected beam.PCollection, threshold float64)

EqualsFloat calls into TryEqualsFloat, checkong that two PCollections of non-complex numeric types are equal, with each element being within a provided threshold of an expected value. Panics if TryEqualsFloat returns an error.

func EqualsList

func EqualsList(s beam.Scope, col beam.PCollection, list any) beam.PCollection

EqualsList verifies that the given collection has the same values as a given list, under coder equality. The values must be provided as an array or slice. This is equivalent to passing a beam.CreateList PCollection to Equals.

Example

p, s := beam.NewPipelineWithRoot()
col := beam.Create(s, "example", "inputs", "here")
list := [3]string{"here", "example", "inputs"}

EqualsList(s, col, list)
err := ptest.Run(p)
fmt.Println(err == nil)

Output:

true

Example (Mismatch)

p, s := beam.NewPipelineWithRoot()
col := beam.Create(s, "example", "inputs", "here")
list := [3]string{"wrong", "inputs", "here"}

EqualsList(s, col, list)
err := ptest.Run(p)
err = unwrapError(err)

// Process error for cleaner example output, demonstrating the diff.
processedErr := strings.SplitAfter(err.Error(), ".failIfBadEntries] failed:")
fmt.Println(processedErr[1])

Output:

actual PCollection does not match expected values
=========
2 correct entries (present in both)
=========
1 unexpected entries (present in actual, missing in expected)
+++
example
=========
1 missing entries (missing in actual, present in expected)
---
wrong

func False

func False(s beam.Scope, col beam.PCollection, fn any) beam.PCollection

False asserts that the given predicate does not satisfy any element in the condition.

func Hash

func Hash(s beam.Scope, col beam.PCollection, name, hash string, size int)

Hash validates that the incoming PCollection<string> has the given size and base64-encoded MD5 hash code. It buffers the entire PCollection in memory and sorts it for determinism.

func NonEmpty

func NonEmpty(s beam.Scope, col beam.PCollection) beam.PCollection

NonEmpty asserts that the given PCollection has at least one element.

func Sum

func Sum(s beam.Scope, col beam.PCollection, name string, size, value int)

Sum validates that the sum and count of elements in the incoming PCollection<int> is the same as the given sum and count, under coder equality. Sum is a specialized version of Equals that avoids a lot of machinery for testing.

func True

func True(s beam.Scope, col beam.PCollection, fn any) beam.PCollection

True asserts that all elements satisfy the given predicate.

func TryEqualsFloat

func TryEqualsFloat(s beam.Scope, observed, expected beam.PCollection, threshold float64) error

TryEqualsFloat checks that two PCollections of floats are equal, with each element being within a specified threshold of its corresponding element. Both PCollections are loaded into memory, sorted, and compared element by element. Returns an error if the PCollection types are complex or non-numeric.