apache/beam · error · TypeError

f'allowed_sources of test specification {identifier} must be

Error message

f'allowed_sources of test specification {identifier} must be a list, got {type(test_spec["allowed_sources"])}'

What it means

validate_test_spec requires the test spec's allowed_sources field, if present, to be a list of source names/types. Any other type raises a TypeError that includes the test's identifier (name and line).

Source

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

      yaml_transform.preprocess_chain,
      yaml_transform.tag_explicit_inputs,
      yaml_transform.normalize_inputs_outputs,
  ]:
    spec = yaml_transform.apply_phase(phase, spec)

  return spec


def validate_test_spec(test_spec):
  if not isinstance(test_spec, dict):
    raise TypeError(
        f'Test specification must be an object, got {type(test_spec)}')
  identifier = (
      test_spec.get('name', 'unknown') +
      f' at line {yaml_transform.SafeLineLoader.get_line(test_spec)}')

  if not isinstance(test_spec.get('allowed_sources', []), list):
    raise TypeError(
        f'allowed_sources of test specification {identifier} '
        f'must be a list, got {type(test_spec["allowed_sources"])}')

  if (not test_spec.get('expected_outputs', []) and
      not test_spec.get('expected_inputs', [])):
    raise ValueError(
        f'test specification {identifier} '
        f'must have at least one expected_outputs or expected_inputs')

  unknown_attrs = set(
      yaml_transform.SafeLineLoader.strip_metadata(test_spec).keys()) - set([
          'name',
          'mock_inputs',
          'mock_outputs',
          'expected_outputs',
          'expected_inputs',
          'allowed_sources',
      ])

View on GitHub (pinned to 12126d8942)

Solutions

Write allowed_sources as a YAML list: allowed_sources: [Create].
Use block-list syntax (- Create) for multiple entries.
Coerce to a list in code: allowed_sources if isinstance(allowed_sources, list) else [allowed_sources].

Example fix

# before
allowed_sources: Create

# after
allowed_sources:
  - Create

Defensive patterns

Strategy: type-guard

Validate before calling

als = test_spec.get('allowed_sources', [])
if not isinstance(als, list):
    test_spec['allowed_sources'] = [als]

Type guard

def has_list_allowed_sources(spec: dict) -> bool:
    return isinstance(spec.get('allowed_sources', []), list)

Try / catch

try:
    validate_test_spec(test_spec)
except TypeError as e:
    logging.error('allowed_sources must be a list: %s', e)
    raise

Prevention

Always use YAML list syntax for allowed_sources
Use flow style [Create, Read] or block style consistently
Add a schema check in CI for test specs

When it happens

Trigger: Specifying allowed_sources as a string (e.g. allowed_sources: Create), a mapping, or a comma-separated scalar instead of a YAML list.

Common situations: Users write a single source without list syntax; YAML stringifies it instead of producing a sequence.

Related errors

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