def assert_matches_stdout(actual, expected_stdout, normalize_fn=(lambda elem: elem), label=''):
    'Asserts a PCollection of strings matches the expected stdout elements.\n\n  Args:\n    actual (beam.PCollection): A PCollection.\n    expected (List[str]): A list of stdout elements, one line per element.\n    normalize_fn (Function[any]): A function to normalize elements before\n        comparing them. Can be used to sort lists before comparing.\n    label (str): [optional] Label to make transform names unique.\n  '

    def stdout_to_python_object(elem_str):
        try:
            elem = ast.literal_eval(elem_str)
        except (SyntaxError, ValueError):
            elem = elem_str
        return normalize_fn(elem)
    actual = (actual | (label >> beam.Map(stdout_to_python_object)))
    expected = list(map(stdout_to_python_object, expected_stdout))
    assert_that(actual, equal_to(expected), ('assert ' + label))
Asserts a PCollection of strings matches the expected stdout elements.

Args:
  actual (beam.PCollection): A PCollection.
  expected (List[str]): A list of stdout elements, one line per element.
  normalize_fn (Function[any]): A function to normalize elements before
      comparing them. Can be used to sort lists before comparing.
  label (str): [optional] Label to make transform names unique.
def assert_matches_stdout(actual, expected_stdout, normalize_fn=(lambda elem: elem), label=):
    'Asserts a PCollection of strings matches the expected stdout elements.\n\n  Args:\n    actual (beam.PCollection): A PCollection.\n    expected (List[str]): A list of stdout elements, one line per element.\n    normalize_fn (Function[any]): A function to normalize elements before\n        comparing them. Can be used to sort lists before comparing.\n    label (str): [optional] Label to make transform names unique.\n  '

    def stdout_to_python_object(elem_str):
        try:
            elem = ast.literal_eval(elem_str)
        except (SyntaxError, ValueError):
            elem = elem_str
        return normalize_fn(elem)
    actual = (actual | (label >> beam.Map(stdout_to_python_object)))
    expected = list(map(stdout_to_python_object, expected_stdout))
    assert_that(actual, equal_to(expected), ('assert ' + label))