apache/beam · error · ValueError

f'Non-mocked source {name_or_type} at line {yaml_transform.S

Error message

f'Non-mocked source {name_or_type} at line {yaml_transform.SafeLineLoader.get_line(transform)}'

What it means

run_test in yaml_testing enforces that all sources in a test pipeline are mocked or explicitly allowed. If a transform reads from a real source (its type or name not in allowed_sources and it has non-empty input handling), a ValueError naming the transform and its YAML line is raised, preventing accidental reads from production systems.

Source

Thrown at sdks/python/apache_beam/yaml/yaml_testing.py:92

    pipeline_spec_dict = yaml.load(
        pipeline_spec, Loader=yaml_utils.SafeLineLoader)
  else:
    pipeline_spec_dict = pipeline_spec

  processed_pipeline_spec = _preprocess_for_testing(pipeline_spec_dict)

  transform_spec, recording_ids = inject_test_tranforms(
      processed_pipeline_spec,
      test_spec,
      fix_failures)

  allowed_sources = set(test_spec.get('allowed_sources', []) + ['Create'])
  for transform in transform_spec['transforms']:
    name_or_type = transform.get('name', transform['type'])
    if (not yaml_transform.empty_if_explicitly_empty(transform.get('input', []))
        and not transform.get('name') in allowed_sources and
        not transform['type'] in allowed_sources):
      raise ValueError(
          f'Non-mocked source {name_or_type} '
          f'at line {yaml_transform.SafeLineLoader.get_line(transform)}')

  if options is None:
    options = beam.options.pipeline_options.PipelineOptions(
        pickle_library='cloudpickle',
        **yaml_transform.SafeLineLoader.strip_metadata(
            pipeline_spec_dict.get('options', {})))

  providers = yaml_provider.merge_providers(
      yaml_provider.parse_providers(
          '', pipeline_spec_dict.get('providers', [])),
      {
          'AssertEqualAndRecord': yaml_provider.as_provider_list(
              'AssertEqualAndRecord', AssertEqualAndRecord)
      })

  with beam.Pipeline(options=options) as p:

View on GitHub (pinned to 12126d8942)

Solutions

1. Add the source's type or transform name to the test spec's allowed_sources list.
2. Replace the real source with mock_inputs in the test spec.
3. Remove the unmocked source if the test should not read external data.

Example fix

# before
- tests:
  - name: my_test
    expected_outputs: {...}

# after
- tests:
  - name: my_test
    allowed_sources: [ReadFromBigQuery]
    expected_outputs: {...}

for t in transform_spec['transforms']:
    if t.get('type') not in allowed_sources and t.get('name') not in allowed_sources:
        raise ValueError(f'unmocked source: {t.get("name", t["type"])}')

try:
    run_test(test_spec, ...)
except ValueError as e:
    if 'Non-mocked source' in str(e):
        add_source_to_allowed_sources_or_mock(e)
    raise

When it happens

Trigger: Running a Beam YAML test whose pipeline contains a source transform (e.g. ReadFromBigQuery, Kafka Read) whose type or name is not listed in the test spec's allowed_sources and is not mocked via mock_inputs.

Common situations: Adding a new source to a pipeline without updating the test spec, forgetting to mock a database/file read, or a renamed transform no longer matching allowed_sources entries.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• At most one of --create_test and --fix_tests may be specifie
• tests attribute must be a list of test specifications.
• f'Test specification must be an object, got {type(test_spec)
• f'allowed_sources of test specification {identifier} must be
• f'test specification {identifier} must have at least one exp

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