apache/beam · error · ValueError

f'Unmocked output of .If any used output is mocked all used…

Error message

f'Unmocked output {tag} of {name}.If any used output is mocked all used outputs must be mocked.'

What it means

Raised during test pipeline construction when a transform output is referenced but not covered by mocks. The mock-resolution helper _require_output found the transform in mocked_outputs_by_id but the specific output tag was absent from its mock set, and the rule enforced here is all-or-nothing: if any used output of a transform is mocked, every used output must be mocked. The input at fault is the unmocked output tag of the named transform.

Solutions

1. Add a mock_outputs entry for the missing output tag of that transform
2. If outputs are referenced by tag, name the mock entry 'transformName.tag' so it matches the unmocked tag
3. Alternatively remove the mock on the sibling output so the transform is treated as unmocked

When it happens

Trigger: Thrown at sdks/python/apache_beam/yaml/yaml_testing.py:250 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

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

      return {
          tag: require_output_or_outputs(input_ref)
          for tag, input_ref in yaml_transform.empty_if_explicitly_empty(
              input_spec).items()
      }

  def require_output(name: str) -> str:
    # The same output may be referenced under different names.
    # Normalize before we cache.
    transform_id, tag = scope.get_transform_id_and_output_name(name)
    return _require_output(transform_id, tag) or name

  @functools.cache
  def _require_output(transform_id: str, tag: str) -> Optional[str]:
    if transform_id in mocked_outputs_by_id:
      if tag not in mocked_outputs_by_id[transform_id]:
        name = next(iter(
            mocked_outputs_by_id[transform_id].values()))['name'].split('.')[0]
        raise ValueError(
            f'Unmocked output {tag} of {name}.'
            'If any used output is mocked all used outputs must be mocked.')
      return create_mocked_output(transform_id, tag)
    else:
      _use_transform(transform_id)
      return None  # Use original name.

  @functools.cache
  def _use_transform(transform_id: str) -> None:
    transform_spec = dict(scope.get_transform_spec(transform_id))
    transform_spec['input'] = create_inputs(transform_id)
    transforms.append(transform_spec)

  @functools.cache
  def create_mocked_input(transform_id: str) -> str:
    transform = create_create(
        f'MockInput[{mocked_inputs_by_id[transform_id]["name"]}]',
        mocked_inputs_by_id[transform_id]['elements'],

View on GitHub (pinned to 12126d8942)