Enable subsequences and/or regexes checking in a Steps's command.
Bug: 939117
Change-Id: I874be95e2ec7ea6c00494b53fde48d68f42ef4b2
Reviewed-on: https://chromium-review.googlesource.com/c/infra/luci/recipes-py/+/1601463
Commit-Queue: Garrett Beaty <gbeaty@chromium.org>
Reviewed-by: Robbie Iannucci <iannucci@chromium.org>
diff --git a/recipe_engine/internal/test/magic_check_fn.py b/recipe_engine/internal/test/magic_check_fn.py
index 33498ff..267875c 100644
--- a/recipe_engine/internal/test/magic_check_fn.py
+++ b/recipe_engine/internal/test/magic_check_fn.py
@@ -15,7 +15,7 @@
import itertools
import weakref
-from collections import OrderedDict, deque, defaultdict, namedtuple
+from collections import Iterable, OrderedDict, deque, defaultdict, namedtuple
import astunparse
@@ -624,6 +624,55 @@
return attr.ib(metadata={'parser': parser, 'unparser': unparser}, **kwargs)
+class Command(list):
+ """Specialized list enabling enhanced searching in command arguments.
+
+ Command is a list of strings that supports searching for individual strings
+ or subsequences of strings comparing by either string equality or regular
+ expression. Regular expression elements are compared against strings using the
+ search method of the regular expression object.
+
+ e.g. the following all evaluate as True:
+ 'foo' in Command(['foo', 'bar', 'baz'])
+ re.compile('a') in Command(['foo', 'bar', 'baz'])
+ ['foo', 'bar'] in Command(['foo', 'bar', 'baz'])
+ [re.compile('o$'), 'bar', re.compile('^b')] in Command(['foo', 'bar', 'baz'])
+ """
+
+ def __contains__(self, item):
+ # Get a function that can be used for matching against an element
+ # Command's elements will always be strings, so we'll only try to match
+ # against strings or regexes
+ def get_matcher(obj):
+ if isinstance(obj, basestring):
+ return lambda other: obj == other
+ if isinstance(obj, re._pattern_type):
+ return obj.search
+ return None
+
+ if isinstance(item, Iterable) and not isinstance(item, basestring):
+ matchers = [get_matcher(e) for e in item]
+ else:
+ matchers = [get_matcher(item)]
+
+ # If None is present in matchers, then that means item is/contains an object
+ # of a type that we won't use for matching
+ if any(m is None for m in matchers):
+ return False
+
+ # At this point, matchers is a list of functions that we can apply against
+ # the elements of each subsequence in the list; if each matcher matches the
+ # corresponding element of the subsequence then we say that the sequence of
+ # strings/regexes is contained in the command
+ for i in xrange(len(self) - len(matchers) + 1):
+ for j, matcher in enumerate(matchers):
+ if not matcher(self[i + j]):
+ break
+ else:
+ return True
+ return False
+
+
@attr.s
class Step(object):
"""The representation of a step provided to post-process hooks."""
@@ -729,6 +778,8 @@
step = Step(**{k: v for k, v in step_dict.iteritems()
if k != '~followup_annotations'})
+ if step.cmd is not None:
+ step.cmd = Command(step.cmd)
parsers = [f.metadata['parser'] for f in cls._get_annotation_fields()]
for annotation in step_dict.get('~followup_annotations', []):
for field in cls._get_annotation_fields():
@@ -746,6 +797,8 @@
prototype = Step('')._as_dict()
step_dict = {k: v for k, v in self._as_dict().iteritems()
if k == 'name' or v != prototype[k]}
+ if step_dict.get('cmd', None) is not None:
+ step_dict['cmd'] = list(step_dict['cmd'])
annotations = []
for field in self._get_annotation_fields():
value = step_dict.pop(field.name, MISSING)
@@ -900,6 +953,10 @@
for k, v in rslt.iteritems():
if isinstance(v, Step):
rslt[k] = v.to_step_dict()
+ else:
+ cmd = rslt[k].get('cmd', None)
+ if cmd is not None:
+ rslt[k]['cmd'] = list(cmd)
msg = VerifySubset(rslt, raw_expectations)
if msg:
raise PostProcessError('post process: steps' + msg)
diff --git a/unittests/checker_test.py b/unittests/checker_test.py
index 0a8e624..083718e 100755
--- a/unittests/checker_test.py
+++ b/unittests/checker_test.py
@@ -6,6 +6,7 @@
import sys
import copy
import datetime
+import re
from collections import OrderedDict
@@ -13,7 +14,8 @@
from recipe_engine.recipe_test_api import PostprocessHookContext, RecipeTestApi
from recipe_engine.internal.test.magic_check_fn import \
- Checker, CheckFrame, PostProcessError, Step, VerifySubset, post_process
+ Checker, CheckFrame, Command, PostProcessError, Step, VerifySubset, \
+ post_process
HOOK_CONTEXT = PostprocessHookContext(lambda: None, (), {}, '<filename>', 0)
@@ -498,6 +500,40 @@
self.assertEqual(s.to_step_dict(), {'name': 'foo'})
+class CommandTest(test_env.RecipeEngineUnitTest):
+ def test_contains_single_non_matcher(self):
+ c = Command(['foo', 'bar', 'baz'])
+ self.assertFalse(0 in c)
+
+ def test_contains_single_string(self):
+ c = Command(['foo', 'bar', 'baz'])
+ self.assertTrue('foo' in c)
+ self.assertTrue('bar' in c)
+ self.assertTrue('baz' in c)
+ self.assertFalse('quux' in c)
+
+ def test_contains_single_regex(self):
+ c = Command(['foo', 'bar', 'baz'])
+ self.assertTrue(re.compile('ba.') in c)
+ self.assertTrue(re.compile('a') in c)
+ self.assertTrue(re.compile('z$') in c)
+ self.assertTrue(re.compile('^bar$') in c)
+ self.assertFalse(re.compile('^a$') in c)
+
+ def test_contains_string_sequence(self):
+ c = Command(['foo', 'bar', 'baz'])
+ self.assertTrue(['bar'] in c)
+ self.assertTrue(['foo', 'bar'] in c)
+ self.assertTrue(['bar', 'baz'] in c)
+ self.assertTrue(['foo', 'bar', 'baz'] in c)
+ self.assertFalse(['foo', 'baz'] in c)
+
+ def test_contains_matcher_sequence(self):
+ c = Command(['foo', 'bar', 'baz'])
+ self.assertTrue([re.compile('z')] in c)
+ self.assertTrue([re.compile('.o.'), 'bar', re.compile('z')] in c)
+ self.assertFalse([re.compile('f'), re.compile('z'), re.compile('r')] in c)
+
class TestVerifySubset(test_env.RecipeEngineUnitTest):
@staticmethod
def mkData(*steps):