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.

[martiniss, 2016/10/03 19:14:19]

Add a file doc string and tests?

+
+import re
+
+from collections import defaultdict, OrderedDict
+
+
+class _filterObject(object):
+  def __init__(self, data, re_data):
+    self.data = data
+    self.re_data = re_data
+
+  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 not field_set:

[dnj, 2016/10/04 16:42:33]

Can we have an error if both specific and regex match the same step? I think
this is an error case.

[dnj, 2016/10/04 16:42:34]

nit: You're using "is None" below, but "not" here. I prefer "is None".

[iannucci, 2016/10/06 20:28:08]

I don't think that's an error case, I've documented the behavior.

[iannucci, 2016/10/06 20:28:09]

Done.

+        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)

[dnj, 2016/10/04 16:42:34]

If I say, "at most 3", I would expect 3 to be valid?

[iannucci, 2016/10/06 20:28:09]

Done.

+
+    return to_ret
+
+  def include(self, step_name, *fields):
+    new_data = self.data.copy()
+    new_data[step_name] = set(fields)
+    return _filterObject(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)] = (at_least, at_most, set(fields))

[dnj, 2016/10/04 16:42:34]

Let's use a namedtuple here. Simple to create one inside the _filterObject
class.

[iannucci, 2016/10/06 20:28:08]

Done.

+    return _filterObject(self.data, new_re_data)
+
+
+def NewFilter(*steps):

[dnj, 2016/10/04 16:42:34]

docstrings!

[iannucci, 2016/10/06 20:28:09]

Done.

+  return _filterObject({name: set() for name in steps}, {})
+
+
+def DoesntRun(check, step_odict, *steps):

[dnj, 2016/10/04 16:42:34]

This looks awkward. How about DoesNotRun?

[iannucci, 2016/10/06 20:28:09]

Done.

+  banSet = set(steps)
+  for step_name in step_odict:
+    check(step_name not in banSet)
+
+
+def DoesntRunRE(check, step_odict, *step_regexes):

[dnj, 2016/10/04 16:42:34]

(same)

[iannucci, 2016/10/06 20:28:09]

Done.

+  step_regexes = [re.compile(r) for r in step_regexes]

[dnj, 2016/10/04 16:42:34]

Assert that "step_odict" is an OrderedDict? Here and at other public function
boundaries.

[iannucci, 2016/10/06 20:28:09]

That would be a terrible abstraction violation :D. post_process is guaranteed to
call this with an OrderedDict. Otherwise it's broken. If I put checks here, it
would mean that users also have to put the same checks in their own assertion
functions.

+  for step_name in step_odict:
+    for r in step_regexes:
+      check(not r.match(step_name))
+
+
+def MustRun(check, step_odict, *steps):
+  step_names = step_odict.keys()
+  for step_name in steps:
+    check(step_name in step_names)
+
+
+def MustRunRE(check, step_odict, step_regex, at_least=1, at_most=None):

[martiniss, 2016/10/03 19:14:18]

Maybe have examples of these in other engine tests?

[iannucci, 2016/10/06 20:28:09]

there are docs now, I don't think an example of every single function is
necessary.

+  step_regex = re.compile(step_regex)

[dnj, 2016/10/04 16:42:34]

WDYT about allowing "step_regex" to actually be a compiled regex? And above.

_RE_TYPE = typeof(re.compile(r''))

def _get_re(value):
  if isinstance(value, _RE_TYPE):
    return value
  elif isinstance(value, basestring):
    return re.compile(value)
  else:
    raise TypeError("Don't know how to make a regex from %s" %
(type(value).__name__,)

[iannucci, 2016/10/06 20:28:08]

It's already the case.  

f = re.compile('foo')
F = re.compile(f)

f == F

+  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):
+  return []