<commit_before>#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""Utilities for testing Beam pipelines."""

from __future__ import absolute_import

import collections
import glob
import io
import tempfile
from builtins import object

from hamcrest.core import assert_that as hamcrest_assert
from hamcrest.library.collection import contains_inanyorder

from apache_beam import pvalue
from apache_beam.transforms import window
from apache_beam.transforms.core import Create
from apache_beam.transforms.core import DoFn
from apache_beam.transforms.core import Map
from apache_beam.transforms.core import ParDo
from apache_beam.transforms.core import WindowInto
from apache_beam.transforms.ptransform import PTransform
from apache_beam.transforms.util import CoGroupByKey
from apache_beam.utils.annotations import experimental

__all__ = [
    'assert_that',
    'equal_to',
    'is_empty',
    'is_not_empty',
    'matches_all',
    # open_shards is internal and has no backwards compatibility guarantees.
    'open_shards',
    'TestWindowedValue',
    ]


class BeamAssertException(Exception):
  """Exception raised by matcher classes used by assert_that transform."""

  pass


# Used for reifying timestamps and windows for assert_that matchers.
TestWindowedValue = collections.namedtuple(
    'TestWindowedValue', 'value timestamp windows')


def contains_in_any_order(iterable):
  """Creates an object that matches another iterable if they both have the
  same count of items.

  Arguments:
    iterable: An iterable of hashable objects.
  """
  class InAnyOrder(object):
    def __init__(self, iterable):
      self._counter = collections.Counter(iterable)

    def __eq__(self, other):
      return self._counter == collections.Counter(other)

    def __ne__(self, other):
      # TODO(BEAM-5949): Needed for Python 2 compatibility.
      return not self == other

    def __hash__(self):
      return hash(self._counter)

    def __repr__(self):
      return "InAnyOrder(%s)" % self._counter

  return InAnyOrder(iterable)


def equal_to_per_window(expected_window_to_elements):
  """Matcher used by assert_that to check on values for specific windows.

  Arguments:
    expected_window_to_elements: A dictionary where the keys are the windows
      to check and the values are the elements associated with each window.
  """
  def matcher(elements):
    actual_elements_in_window, window = elements
    if window in expected_window_to_elements:
      expected_elements_in_window = list(
          expected_window_to_elements[window])
      sorted_expected = sorted(expected_elements_in_window)
      sorted_actual = sorted(actual_elements_in_window)
      if sorted_expected != sorted_actual:
        # Results for the same window don't necessarily come all
        # at once. Hence the same actual window may contain only
        # subsets of the expected elements for the window.
        # For example, in the presence of early triggers.
        if all(elem in sorted_expected for elem in sorted_actual) is False:
          raise BeamAssertException(
              'Failed assert: %r not in %r' % (sorted_actual, sorted_expected))
  return matcher


# Note that equal_to checks if expected and actual are permutations of each
# other. However, only permutations of the top level are checked. Therefore
# [1,2] and [2,1] are considered equal and [[1,2]] and [[2,1]] are not.
def equal_to(expected):

  def _equal(actual):
    expected_list = list(expected)

    # Try to compare actual and expected by sorting. This fails with a
    # TypeError in Python 3 if different types are present in the same
    # collection.
    try:
      sorted_expected = sorted(expected)
      sorted_actual = sorted(actual)
      if sorted_expected != sorted_actual:
        raise BeamAssertException(
            'Failed assert: %r == %r' % (sorted_expected, sorted_actual))
    # Fall back to slower method which works for different types on Python 3.
    except TypeError:
      for element in actual:
        try:
          expected_list.remove(element)
        except ValueError:
          raise BeamAssertException(
              'Failed assert: %r == %r' % (expected, actual))
      if expected_list:
        raise BeamAssertException(
            'Failed assert: %r == %r' % (expected, actual))

  return _equal


def matches_all(expected):
  """Matcher used by assert_that to check a set of matchers.

  Args:
    expected: A list of elements or hamcrest matchers to be used to match
      the elements of a single PCollection.
  """
  def _matches(actual):
    expected_list = list(expected)

    hamcrest_assert(actual, contains_inanyorder(*expected_list))

  return _matches


def is_empty():
  def _empty(actual):
    actual = list(actual)
    if actual:
      raise BeamAssertException(
          'Failed assert: [] == %r' % actual)<commit_msg>Importing hamcrest dependency at matcher runtime<commit_after>#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""Utilities for testing Beam pipelines."""

from __future__ import absolute_import

import collections
import glob
import io
import tempfile
from builtins import object

from apache_beam import pvalue
from apache_beam.transforms import window
from apache_beam.transforms.core import Create
from apache_beam.transforms.core import DoFn
from apache_beam.transforms.core import Map
from apache_beam.transforms.core import ParDo
from apache_beam.transforms.core import WindowInto
from apache_beam.transforms.ptransform import PTransform
from apache_beam.transforms.util import CoGroupByKey
from apache_beam.utils.annotations import experimental

__all__ = [
    'assert_that',
    'equal_to',
    'is_empty',
    'is_not_empty',
    'matches_all',
    # open_shards is internal and has no backwards compatibility guarantees.
    'open_shards',
    'TestWindowedValue',
    ]


class BeamAssertException(Exception):
  """Exception raised by matcher classes used by assert_that transform."""

  pass


# Used for reifying timestamps and windows for assert_that matchers.
TestWindowedValue = collections.namedtuple(
    'TestWindowedValue', 'value timestamp windows')


def contains_in_any_order(iterable):
  """Creates an object that matches another iterable if they both have the
  same count of items.

  Arguments:
    iterable: An iterable of hashable objects.
  """
  class InAnyOrder(object):
    def __init__(self, iterable):
      self._counter = collections.Counter(iterable)

    def __eq__(self, other):
      return self._counter == collections.Counter(other)

    def __ne__(self, other):
      # TODO(BEAM-5949): Needed for Python 2 compatibility.
      return not self == other

    def __hash__(self):
      return hash(self._counter)

    def __repr__(self):
      return "InAnyOrder(%s)" % self._counter

  return InAnyOrder(iterable)


def equal_to_per_window(expected_window_to_elements):
  """Matcher used by assert_that to check on values for specific windows.

  Arguments:
    expected_window_to_elements: A dictionary where the keys are the windows
      to check and the values are the elements associated with each window.
  """
  def matcher(elements):
    actual_elements_in_window, window = elements
    if window in expected_window_to_elements:
      expected_elements_in_window = list(
          expected_window_to_elements[window])
      sorted_expected = sorted(expected_elements_in_window)
      sorted_actual = sorted(actual_elements_in_window)
      if sorted_expected != sorted_actual:
        # Results for the same window don't necessarily come all
        # at once. Hence the same actual window may contain only
        # subsets of the expected elements for the window.
        # For example, in the presence of early triggers.
        if all(elem in sorted_expected for elem in sorted_actual) is False:
          raise BeamAssertException(
              'Failed assert: %r not in %r' % (sorted_actual, sorted_expected))
  return matcher


# Note that equal_to checks if expected and actual are permutations of each
# other. However, only permutations of the top level are checked. Therefore
# [1,2] and [2,1] are considered equal and [[1,2]] and [[2,1]] are not.
def equal_to(expected):

  def _equal(actual):
    expected_list = list(expected)

    # Try to compare actual and expected by sorting. This fails with a
    # TypeError in Python 3 if different types are present in the same
    # collection.
    try:
      sorted_expected = sorted(expected)
      sorted_actual = sorted(actual)
      if sorted_expected != sorted_actual:
        raise BeamAssertException(
            'Failed assert: %r == %r' % (sorted_expected, sorted_actual))
    # Fall back to slower method which works for different types on Python 3.
    except TypeError:
      for element in actual:
        try:
          expected_list.remove(element)
        except ValueError:
          raise BeamAssertException(
              'Failed assert: %r == %r' % (expected, actual))
      if expected_list:
        raise BeamAssertException(
            'Failed assert: %r == %r' % (expected, actual))

  return _equal


def matches_all(expected):
  """Matcher used by assert_that to check a set of matchers.

  Args:
    expected: A list of elements or hamcrest matchers to be used to match
      the elements of a single PCollection.
  """
  def _matches(actual):
    from hamcrest.core import assert_that as hamcrest_assert
    from hamcrest.library.collection import contains_inanyorder
    expected_list = list(expected)

    hamcrest_assert(actual, contains_inanyorder(*expected_list))

  return _matches


def is_empty():
  def _empty(actual):
    actual = list(actual)
    if actual:
      raise BeamAssertException(
          'Failed assert: [] == %r' % actual)
{'fraction_non_alphanumeric': 0.06851025869759143, 'fraction_numerical': 0.003389830508474576, 'mean_word_length': 3.6849143334726286, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7544067', 'n_tokens_mistral': 3089, 'n_tokens_neox': 2983, 'n_words': 1164}