Add CIPD pin reporting to swarming.

appengine/swarming/cipd.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.
 
 """CIPD-specific code is concentrated here."""
 
 import re
 
 # Regular expressions below are copied from
 # https://chromium.googlesource.com/infra/infra/+/468bb43/appengine/chrome_infra_packages/cipd/impl.py
@@ skipping 15 matching lines @@
 # os can be "linux", "mac" or "windows" and arch can be "386", "amd64" or
 # "armv6l".
 PARAM_PLATFORM = '${platform}'
 # OS version parameter defines major and minor version of the OS distribution.
 # It is useful if package depends on .dll/.so libraries provided by the OS.
 # Example values: "ubuntu14_04", "mac10_9", "win6_1".
 PARAM_OS_VER = '${os_ver}'
 ALL_PARAMS = (PARAM_PLATFORM, PARAM_OS_VER)
 
 
+class PinChecker(object):
+  PARAM_PLATFORM_ESC = re.escape(PARAM_PLATFORM)
+  PARAM_OS_VER_ESC = re.escape(PARAM_OS_VER)
+
+  def __init__(self):
+    self.platform = None
+    self.os_ver = None
+
+  def check(self, original, expanded):

[M-A Ruel, 2016/08/30 02:18:07]

optional nit: I think I'd prefer if this function returned a new instance of
PinChecker instead of mutating.

[iannucci, 2016/08/30 09:13:29]

Done.

edit: Actually I changed it to a contextmanager. I tried the 'return an new
instance' thing, but it was super-awkward. I think the new version represents
the semantics correctly; doing checks within some lexical context where all
template parameters must be consistent.

+    """Raises ValueError if expanded is not pluasibly a pinned version of

[M-A Ruel, 2016/08/30 02:18:07]

plausibly

[iannucci, 2016/08/30 09:13:29]

Done.

+    original.
+
+    Args:
+      original - a CipdPackage which may contain templates like ${platform}
+      expanded - a CipdPackage which is nominally an expansion of original.
+
+    CipdPackage is duck-typed to have three string properties 'package_name',
+    'path' and 'version'.
+
+    Side effects:
+      If this PinChecker has not encountered a platform or os_ver value yet,
+      the appropriate property will be set on this PinChecker. Subsequent
+      executions of check will replace ${platform} with the previously-scanned
+      value.
+
+    Raises:
+      ValueError if expanded is not a valid derivation of original.
+    """
+    if original.path != expanded.path:
+      raise ValueError('Mismatched path')
+
+    def sub_param(regex, param_esc, param_re, param_const):
+      # The 1 allows each substitution to only show up in the pattern once.
+      # If it shows up more than once, then this is a very strange package name,
+      # and should be subject to further scrutiny. If it turns out that for some
+      # reason the substitutions SHOULD be allowed more than once per name, we
+      # will need to assert that the matched-values of them are also all the
+      # same (e.g. ${platform} cannot resolve to more than one value)
+      ret = False
+      if param_const is None:
+        ret = param_esc in regex
+        if ret:
+          regex = regex.replace(param_esc, param_re, 1)
+      else:
+        regex = regex.replace(param_esc, param_const, 1)
+      return regex, ret
+
+    name_regex = re.escape(original.package_name)
+    name_regex, scan_plat = sub_param(
+        name_regex, self.PARAM_PLATFORM_ESC, '(?P<platform>\w+-[a-z0-9]+)',

[M-A Ruel, 2016/08/30 02:18:07]

Use r'' otherwise your \ may not do what you think it'll do.

[iannucci, 2016/08/30 09:13:29]

Done

+        self.platform)
+    name_regex, scan_os_ver = sub_param(
+        name_regex, self.PARAM_OS_VER_ESC, '(?P<os_ver>[_a-z0-9]+)',

[M-A Ruel, 2016/08/30 02:18:07]

r'' as per habit.

[iannucci, 2016/08/30 09:13:29]

Done

+        self.os_ver)
+
+    match = re.match(name_regex, expanded.package_name)
+    if not match:
+      raise ValueError('Mismatched package_name')
+
+    if is_pinned_version(original.version):
+      if original.version != expanded.version:
+        raise ValueError('Mismatched pins')
+    else:
+      if not is_pinned_version(expanded.version):
+        raise ValueError('Pin value is not a pin')
+
+    if scan_plat:
+      self.platform = re.escape(match.group('platform'))
+    if scan_os_ver:
+      self.os_ver = re.escape(match.group('os_ver'))
+
+
 def is_valid_package_name(package_name):
   """Returns True if |package_name| is a valid CIPD package name."""
   return bool(PACKAGE_NAME_RE.match(package_name))
 
 
 def is_valid_package_name_template(template):
   """Returns True if |package_name| is a valid CIPD package name template."""
   # Render known parameters first.
   for p in ALL_PARAMS:
     template = template.replace(p, 'x')
@@ skipping 13 matching lines @@
   """True if string looks like a valid package instance tag."""
   if not tag or ':' not in tag or len(tag) > TAG_MAX_LEN:
     return False
   # Care only about the key. Value can be anything (including empty string).
   return bool(TAG_KEY_RE.match(tag.split(':', 1)[0]))
 
 
 def is_pinned_version(version):
   """Returns True if |version| is pinned."""
   return bool(INSTANCE_ID_RE.match(version)) or is_valid_tag(version)