Source code for apache_beam.testing.metric_result_matchers

# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

"""MetricResult matchers for validating metrics in PipelineResults.

example usage:

    result =
    all_metrics = result.metrics().all_metrics()

    matchers = [
              'pcollection': 'myCollection',
              'myCustomKey': 'myCustomValue'
    errors = metric_result_matchers.verify_all(all_metrics, matchers)
    self.assertFalse(errors, errors)


# pytype: skip-file

from hamcrest import equal_to
from hamcrest.core import string_description
from hamcrest.core.base_matcher import BaseMatcher
from hamcrest.core.matcher import Matcher

from apache_beam.metrics.cells import DistributionResult

def _matcher_or_equal_to(value_or_matcher):
  """Pass-thru for matchers, and wraps value inputs in an equal_to matcher."""
  if value_or_matcher is None:
    return None
  if isinstance(value_or_matcher, Matcher):
    return value_or_matcher
  return equal_to(value_or_matcher)

[docs]class MetricResultMatcher(BaseMatcher): """A PyHamcrest matcher that validates counter MetricResults.""" def __init__( self, namespace=None, name=None, step=None, labels=None, attempted=None, committed=None, sum_value=None, count_value=None, min_value=None, max_value=None, ): self.namespace = _matcher_or_equal_to(namespace) = _matcher_or_equal_to(name) self.step = _matcher_or_equal_to(step) self.attempted = _matcher_or_equal_to(attempted) self.committed = _matcher_or_equal_to(committed) labels = labels or {} self.label_matchers = {} for (k, v) in labels.items(): self.label_matchers[_matcher_or_equal_to(k)] = _matcher_or_equal_to(v) def _matches(self, metric_result): if self.namespace is not None and not self.namespace.matches( metric_result.key.metric.namespace): return False if and not return False if self.step and not self.step.matches(metric_result.key.step): return False if (self.attempted is not None and not self.attempted.matches(metric_result.attempted)): return False if (self.committed is not None and not self.committed.matches(metric_result.committed)): return False for (k_matcher, v_matcher) in self.label_matchers.items(): matched_keys = [ key for key in metric_result.key.labels.keys() if k_matcher.matches(key) ] matched_key = matched_keys[0] if matched_keys else None if not matched_key: return False label_value = metric_result.key.labels[matched_key] if not v_matcher.matches(label_value): return False return True
[docs] def describe_to(self, description): if self.namespace: description.append_text(' namespace: ') self.namespace.describe_to(description) if description.append_text(' name: ') if self.step: description.append_text(' step: ') self.step.describe_to(description) for (k_matcher, v_matcher) in self.label_matchers.items(): description.append_text(' (label_key: ') k_matcher.describe_to(description) description.append_text(' label_value: ') v_matcher.describe_to(description) description.append_text('). ') if self.attempted is not None: description.append_text(' attempted: ') self.attempted.describe_to(description) if self.committed is not None: description.append_text(' committed: ') self.committed.describe_to(description)
[docs] def describe_mismatch(self, metric_result, mismatch_description): mismatch_description.append_text("was").append_value(metric_result)
[docs]class DistributionMatcher(BaseMatcher): """A PyHamcrest matcher that validates counter distributions.""" def __init__( self, sum_value=None, count_value=None, min_value=None, max_value=None): self.sum_value = _matcher_or_equal_to(sum_value) self.count_value = _matcher_or_equal_to(count_value) self.min_value = _matcher_or_equal_to(min_value) self.max_value = _matcher_or_equal_to(max_value) def _matches(self, distribution_result): if not isinstance(distribution_result, DistributionResult): return False if self.sum_value and not self.sum_value.matches(distribution_result.sum): return False if self.count_value and not self.count_value.matches( distribution_result.count): return False if self.min_value and not self.min_value.matches(distribution_result.min): return False if self.max_value and not self.max_value.matches(distribution_result.max): return False return True
[docs] def describe_to(self, description): if self.sum_value: description.append_text(' sum_value: ') self.sum_value.describe_to(description) if self.count_value: description.append_text(' count_value: ') self.count_value.describe_to(description) if self.min_value: description.append_text(' min_value: ') self.min_value.describe_to(description) if self.max_value: description.append_text(' max_value: ') self.max_value.describe_to(description)
[docs] def describe_mismatch(self, distribution_result, mismatch_description): mismatch_description.append_text('was').append_value(distribution_result)
[docs]def verify_all(all_metrics, matchers): """Verified that every matcher matches a metric result in all_metrics.""" errors = [] matched_metrics = [] for matcher in matchers: matched_metrics = [mr for mr in all_metrics if matcher.matches(mr)] if not matched_metrics: errors.append( 'Unable to match metrics for matcher %s' % (string_description.tostring(matcher))) if errors: errors.append( '\nActual MetricResults:\n' + '\n'.join([str(mr) for mr in all_metrics])) return ''.join(errors)