Add initial postprocess unit test thingy.

recipe_engine/post_process.py

+# Copyright 2016 The LUCI Authors. All rights reserved.
+# Use of this source code is governed under the Apache License, Version 2.0
+# that can be found in the LICENSE file.
+
+"""This file contains post process filters for use with the
+RecipeTestApi.post_process method in GenTests.
+"""
+
+import re
+
+from collections import defaultdict, OrderedDict, namedtuple
+
+
+class Filter(object):
+  """Filter is the interface of the object returned by NewFilter."""
+
+  def include(self, step_name, *fields):
+    """Include adds a step to the included steps set.
+
+    Additionally, if any specified fields are provided, they will be the total
+    set of fields in the filtered step. The 'name' field is always included. If
+    fields is omitted, the entire step will be included.
+
+    Args:
+      step_name (str) - The name of the step to include
+      fields (str) - The field(s) to include. Omit to include all fields.
+
+    Returns the new filter.
+    """
+    raise NotImplementedError()
+
+  def include_re(self, step_name_re, at_least=1, at_most=None, *fields):
+    """This includes all steps which match the given regular expression.
+
+    If a step matches both an include() directive as well as include_re(), the
+    include() directive will take precedence.
+
+    Args:
+      step_name_re (str or regex) - the regular expression of step names to
+        match.
+      at_least (int) - the number of steps that this regular expression MUST
+        match.
+      at_most (int) - the maximum number of steps that this regular expression
+        MUST NOT exceed.
+      fields (str) - the field(s) to include in the matched steps. Omit to
+        include all fields.
+
+    Returns the new filter.
+    """
+    raise NotImplementedError()
+
+
+class _filterImpl(Filter):
+  _reEntry = namedtuple('_reEntry', 'at_most at_least fields')
+
+  def __init__(self, data, re_data):
+    self.data = data  # {step_name: frozenset(fields)}
+    self.re_data = re_data  # {regex: _reEntry}
+
+  def __call__(self, check, step_odict):
+    includes = self.data.copy()
+    re_data = self.re_data.copy()
+
+    re_usage_count = defaultdict(int)
+
+    to_ret = OrderedDict()
+    for name, step in step_odict.iteritems():
+      field_set = includes.pop(name, None)
+      if field_set is None:
+        for exp, (_, _, fset) in re_data.iteritems():
+          if exp.match(name):
+            re_usage_count[exp] += 1
+            field_set = fset
+            break
+      if field_set is None:
+        continue
+      if len(field_set) == 0:
+        to_ret[name] = step
+      else:
+        to_ret[name] = {
+          k: v for k, v in step.iteritems()
+          if k in field_set or k == 'name'
+        }
+
+    check('all includes were used', len(includes) == 0)
+
+    for regex, (at_least, at_most, _) in re_data.iteritems():
+      check(re_usage_count[regex] >= at_least)
+      if at_most is not None:
+        check(re_usage_count[regex] <= at_most)
+
+    return to_ret
+
+  def include(self, step_name, *fields):
+    new_data = self.data.copy()

[Michael Achenbach, 2016/10/07 16:51:30]

Why not mutate self.data as well (by replacing it with a modified copy)? And
then return self?

Users might forget to reassign and use:
f = NewFilter()
f.include(...)

Which functional use case do you cover by not permitting permutation here?

[Michael Achenbach, 2016/10/07 16:54:40]

More self thinking:
Agreed that if we mutate we couldn't use it in the following way:
f = NewFilter(...lots of cool initialization)

Test + api.post_process(f.include(...more stuff))

Test + api.post_process(f.include(...other stuff))

etc.

If users use it that way and f is mutated, the effects might be worse than the
case I sketched above. So, maybe just ignore this comment.

+    new_data[step_name] = frozenset(fields)
+    return _filterImpl(new_data, self.re_data)
+
+  def include_re(self, step_name_re, at_least=1, at_most=None, *fields):
+    new_re_data = self.re_data.copy()
+    new_re_data[re.compile(step_name_re)] = _filterImpl._reEntry(
+      (at_least, at_most, frozenset(fields)))
+    return _filterImpl(self.data, new_re_data)
+
+
+def NewFilter(*steps):
+  """NewFilter returns a new Filter object. It may be optionally prepopulated by
+  specifying steps.
+
+  Usage:
+    f = NewFilter('step_a', 'step_b')
+    yield TEST + api.post_process(f)
+
+    f = f.include('other_step')
+    yield TEST + api.post_process(f)
+
+    yield TEST + api.post_process(NewFilter, 'step_a', 'step_b', 'other_step')
+  """
+  return _filterImpl({name: () for name in steps}, {})
+
+
+def DoesNotRun(check, step_odict, *steps):
+  """Asserts that the given steps to not run.
+
+  Usage:
+    yield TEST + api.post_process(DoesNotRun, 'step_a', 'step_b')
+
+  """
+  banSet = set(steps)
+  for step_name in step_odict:
+    check(step_name not in banSet)
+
+
+def DoesNotRunRE(check, step_odict, *step_regexes):
+  """Asserts that no steps matching any of the regexes have run.
+
+  Args:
+    step_regexes (str) - The step name regexes to ban.
+
+  Usage:
+    yield TEST + api.post_process(DoesNotRunRE, '.*with_patch.*', '.*compile.*')
+
+  """
+  step_regexes = [re.compile(r) for r in step_regexes]
+  for step_name in step_odict:
+    for r in step_regexes:
+      check(not r.match(step_name))

[Michael Achenbach, 2016/10/07 16:56:41]

nit: Should these canned checks not get some more human readable hint? Or will
the hint be human readable enough?

+
+
+def MustRun(check, step_odict, *steps):
+  """Asserts that steps with the given names are in the expectations.
+
+  Args:
+    steps (str) - The steps that must have run.
+
+  Usage:
+    yield TEST + api.post_process(MustRun, 'step_a', 'step_b')
+  """
+  for step_name in steps:
+    check(step_name in step_odict)
+
+
+def MustRunRE(check, step_odict, step_regex, at_least=1, at_most=None):
+  """Assert that steps matching the given regex completely are in the
+  exepectations.
+
+  Args:
+    step_regex (str, compiled regex) - The regular expression to match.
+    at_least (int) - Match at least this many steps. Matching fewer than this
+      is a CHECK failure.
+    at_most (int) - Optional upper bound on the number of matches. Matching
+      more than this is a CHECK failure.
+
+  Usage:
+    yield TEST + api.post_process(MustRunRE, r'.*with_patch.*', at_most=2)
+  """
+  step_regex = re.compile(step_regex)
+  matches = 0
+  for step_name in step_odict:
+    if step_regex.match(step_name):
+      matches += 1
+  check(matches >= at_least)
+  if at_most is not None:
+    check(matches <= at_most)
+
+
+def DropExpectation(_check, _step_odict):
+  """Using this post-process hook will drop the expectations for this test
+  completely.
+
+  Usage:
+    yield TEST + api.post_process(DropExpectation)
+
+  """
+  return {}