ptest

Documentation

Overview

Package ptest contains utilities for pipeline unit testing.

Index

• Variables
• func BuildAndRun(t *testing.T, build func(s beam.Scope)) beam.PipelineResult
• func Create(values []any) (*beam.Pipeline, beam.Scope, beam.PCollection)
• func Create2(a, b []any) (*beam.Pipeline, beam.Scope, beam.PCollection, beam.PCollection)
• func CreateList(values any) (*beam.Pipeline, beam.Scope, beam.PCollection)
• func CreateList2(a, b any) (*beam.Pipeline, beam.Scope, beam.PCollection, beam.PCollection)
• func DefaultRunner() string
• func Main(m *testing.M)
• func MainCalled() bool
• func MainRet(m *testing.M) int
• func MainRetWithDefault(m *testing.M, runner string) int
• func MainWithDefault(m *testing.M, runner string)
• func Run(p *beam.Pipeline) error
• func RunAndValidate(t *testing.T, p *beam.Pipeline) beam.PipelineResult
• func RunWithMetrics(p *beam.Pipeline) (beam.PipelineResult, error)

Variables

var (
	Runner = runners.Runner
)

Runner is a flag that sets which runner pipelines under test will use.

The test file must have a TestMain that calls Main or MainWithDefault to function.

Functions

func BuildAndRun

func BuildAndRun(t *testing.T, build func(s beam.Scope)) beam.PipelineResult

BuildAndRun calls the provided pipeline building function, and then executes the resulting pipeline, failing the test if the pipeline fails.

func Create

func Create(values []any) (*beam.Pipeline, beam.Scope, beam.PCollection)

Create creates a pipeline and a PCollection with the given values.

func Create2

func Create2(a, b []any) (*beam.Pipeline, beam.Scope, beam.PCollection, beam.PCollection)

Create2 creates a pipeline and 2 PCollections with the given values.

func CreateList

func CreateList(values any) (*beam.Pipeline, beam.Scope, beam.PCollection)

CreateList creates a pipeline and a PCollection with the given values.

func CreateList2

func CreateList2(a, b any) (*beam.Pipeline, beam.Scope, beam.PCollection, beam.PCollection)

CreateList2 creates a pipeline and 2 PCollections with the given values.

func DefaultRunner

func DefaultRunner() string

DefaultRunner returns the default runner name for the test file.

func Main

func Main(m *testing.M)

Main is an implementation of testing's TestMain to permit testing pipelines on runners other than the direct runner.

To enable this behavior, _ import the desired runner, and set the flag accordingly. For example:

import _ "github.com/apache/beam/sdks/v2/go/pkg/runners/flink"

func TestMain(m *testing.M) {
	ptest.Main(m)
}

func MainCalled

func MainCalled() bool

MainCalled returns true iff Main or MainRet has been called.

func MainRet

func MainRet(m *testing.M) int

MainRet is equivalent to Main, but returns an exit code to pass to os.Exit().

Example:

func TestMain(m *testing.M) {
	os.Exit(ptest.Main(m))
}

func MainRetWithDefault

func MainRetWithDefault(m *testing.M, runner string) int

MainRetWithDefault is equivalent to MainWithDefault but returns an exit code to pass to os.Exit().

func MainWithDefault

func MainWithDefault(m *testing.M, runner string)

MainWithDefault is an implementation of testing's TestMain to permit testing pipelines on runners other than the direct runner, while setting the default runner to use.

func Run

func Run(p *beam.Pipeline) error

Run runs a pipeline for testing. The semantics of the pipeline is expected to be verified through passert.

func RunAndValidate

func RunAndValidate(t *testing.T, p *beam.Pipeline) beam.PipelineResult

RunAndValidate runs a pipeline for testing and validates the result, failing the test if the pipeline fails.

func RunWithMetrics

func RunWithMetrics(p *beam.Pipeline) (beam.PipelineResult, error)

RunWithMetrics runs a pipeline for testing with that returns metrics.Results in the form of Pipeline Result