{"record":{"id":"6730795aee19c417","repo":"apache/beam","slug":"assert-that-must-be-used-within-a-beam-pipeline-context","errorCode":null,"errorMessage":"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","messagePattern":"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","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/testing/util.py","lineNumber":348,"sourceCode":"    matcher: A matcher function taking as argument the actual value of a\n      materialized PCollection. The matcher validates this actual value against\n      expectations and raises BeamAssertException if they are not met.\n    label: Optional string label. This is needed in case several assert_that\n      transforms are introduced in the same pipeline.\n    reify_windows: If True, matcher is passed a list of TestWindowedValue.\n    use_global_window: If False, matcher is passed a dictionary of\n      (k, v) = (window, elements in the window).\n\n  Returns:\n    Ignored.\n  \"\"\"\n  assert isinstance(actual, pvalue.PCollection), (\n      '%s is not a supported type for Beam assert' % type(actual))\n  pipeline = actual.pipeline\n  if getattr(actual.pipeline, 'result', None):\n    # The pipeline was already run. The user most likely called assert_that\n    # after the pipeleline context.\n    raise RuntimeError(\n        'assert_that must be used within a beam.Pipeline context. ' +\n        'Prior to Beam 2.60.0, asserts outside of the context of a pipeline ' +\n        'were silently ignored, starting with Beam 2.60.0 this is no longer ' +\n        'allowed. To fix, move your assert_that call into your pipeline ' +\n        'context so that it is added before the pipeline is run. For more ' +\n        'information, see https://github.com/apache/beam/pull/30771')\n\n  # Usually, the uniqueness of the label is left to the pipeline\n  # writer to guarantee. Since we're in a testing context, we'll\n  # just automatically append a number to the label if it's\n  # already in use, as tests don't typically have to worry about\n  # long-term update compatibility stability of stage names.\n  if label in pipeline.applied_labels:\n    label_idx = 2\n    while f\"{label}_{label_idx}\" in pipeline.applied_labels:\n      label_idx += 1\n    label = f\"{label}_{label_idx}\"\n","sourceCodeStart":330,"sourceCodeEnd":366,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/testing/util.py#L330-L366","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Move every `assert_that` call inside the pipeline context so it is applied before `pipeline.run()` (or before the `with` block exits).","If using explicit run, restructure: build transforms and asserts first, then call `pipeline.run().wait_until_finish()`.","On Beam < 2.60.0 this was a silent no-op — audit tests for asserts that never executed."],"exampleFix":"// before\nwith beam.Pipeline() as p:\n  result = p | beam.Map(str)\nassert_that(result, equal_to(['1']))\n// after\nwith beam.Pipeline() as p:\n  result = p | beam.Map(str)\n  assert_that(result, equal_to(['1']))","handlingStrategy":"validation","validationCode":"# inside test: ensure asserts happen before run\nwith beam.Pipeline() as p:\n    result = p | beam.Map(str)\n    assert_that(result, equal_to(expected))  # keep inside the with-block","typeGuard":null,"tryCatchPattern":"try:\n    assert_that(pcoll, matcher)\nexcept RuntimeError as e:\n    if 'beam.Pipeline context' in str(e):\n        # move assert inside pipeline context\n        ...","preventionTips":["Always put assert_that calls inside the `with beam.Pipeline()` block","Run `pipeline.run()` only after all asserts are declared","After upgrading to Beam >= 2.60.0, audit tests for post-run asserts"],"tags":["apache-beam","testing","pipeline-context","assertion"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}