apache/beam · error · RuntimeError

assert_that must be used within a beam.Pipeline context…

Error message

assert_that must be used within a beam.Pipeline context. Prior to Beam 2.60.0, asserts outside of the context of a pipeline were silently ignored, starting with Beam 2.60.0 this is no longer allowed. To fix, move your assert_that call into your pipeline context so that it is added before the pipeline is run. For more information, see https://github.com/apache/beam/pull/30771

What it means

RuntimeError raised by the public `assert_that` when the pipeline attached to the PCollection has already been run (`actual.pipeline.result` is set). Since Beam 2.60.0, asserts added after the pipeline finished executing were no longer silently ignored — they are rejected so tests can't pass vacuously.

Solutions

1. Move every `assert_that` call inside the pipeline context so it is applied before `pipeline.run()` (or before the `with` block exits).
2. If using explicit run, restructure: build transforms and asserts first, then call `pipeline.run().wait_until_finish()`.
3. On Beam < 2.60.0 this was a silent no-op — audit tests for asserts that never executed.

Example fix

// before
with beam.Pipeline() as p:
  result = p | beam.Map(str)
assert_that(result, equal_to(['1']))
// after
with beam.Pipeline() as p:
  result = p | beam.Map(str)
  assert_that(result, equal_to(['1']))

# inside test: ensure asserts happen before run
with beam.Pipeline() as p:
    result = p | beam.Map(str)
    assert_that(result, equal_to(expected))  # keep inside the with-block

try:
    assert_that(pcoll, matcher)
except RuntimeError as e:
    if 'beam.Pipeline context' in str(e):
        # move assert inside pipeline context
        ...

When it happens

Trigger: Calling `assert_that(pcoll, matcher)` after `pipeline.run()` has been invoked (or after a `with beam.Pipeline()` block exits), e.g. collecting asserts outside the `with` scope.

Common situations: Upgrading to Beam >= 2.60.0 and old tests that built asserts after the run; restructuring test code so `assert_that` lines were moved below the pipeline execution; using a variable that references a pipeline whose context already exited.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Failed assert: nothing matches the criterion
• Failed assert: pcol is empty
• passert.Sum( ) = , want
• %s
• The pipeline contains abandoned PAssert(s).

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/6730795aee19c417. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/testing/util.py:348

    matcher: A matcher function taking as argument the actual value of a
      materialized PCollection. The matcher validates this actual value against
      expectations and raises BeamAssertException if they are not met.
    label: Optional string label. This is needed in case several assert_that
      transforms are introduced in the same pipeline.
    reify_windows: If True, matcher is passed a list of TestWindowedValue.
    use_global_window: If False, matcher is passed a dictionary of
      (k, v) = (window, elements in the window).

  Returns:
    Ignored.
  """
  assert isinstance(actual, pvalue.PCollection), (
      '%s is not a supported type for Beam assert' % type(actual))
  pipeline = actual.pipeline
  if getattr(actual.pipeline, 'result', None):
    # The pipeline was already run. The user most likely called assert_that
    # after the pipeleline context.
    raise RuntimeError(
        'assert_that must be used within a beam.Pipeline context. ' +
        'Prior to Beam 2.60.0, asserts outside of the context of a pipeline ' +
        'were silently ignored, starting with Beam 2.60.0 this is no longer ' +
        'allowed. To fix, move your assert_that call into your pipeline ' +
        'context so that it is added before the pipeline is run. For more ' +
        'information, see https://github.com/apache/beam/pull/30771')

  # Usually, the uniqueness of the label is left to the pipeline
  # writer to guarantee. Since we're in a testing context, we'll
  # just automatically append a number to the label if it's
  # already in use, as tests don't typically have to worry about
  # long-term update compatibility stability of stage names.
  if label in pipeline.applied_labels:
    label_idx = 2
    while f"{label}_{label_idx}" in pipeline.applied_labels:
      label_idx += 1
    label = f"{label}_{label_idx}"

View on GitHub (pinned to 12126d8942)