blob: 83cffb5530e5e76e47d3d44c2a2799d4a1a7880d [file] [log] [blame]
# Copyright 2017 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import collections
import re
from . import util
Package = collections.namedtuple('Package', (
'name', 'tags', 'install_mode', 'compress_level'))
# "compress_level" value for "no compression".
RE_DISALLOWED_PACKAGE_CHARS = re.compile(r'^[a-zA-Z-_]')
def normalize_package_name(name):
return RE_DISALLOWED_PACKAGE_CHARS.sub(name.lower(), '-')
class Cipd(object):
def __init__(self, system, cache_dir):
self._system = system
self._cache_dir = cache_dir
self._exists_cache = set()
def _run_cipd(self, run_fn, args, dryrun=False):
cmd = [] + list(args)
if dryrun:
# While joining with ' ' here isn't fully correct, none of the commands
# we run seem to need quotes, so this is easy enough. Plus this is only
# for debugging purposes, and devs are smrat.'[dryrun] would have run: %s', ' '.join(cmd))
return 0, ''
return run_fn(cmd, env={'CIPD_CACHE_DIR': self._cache_dir})
def run(self, *args, **kwargs):
return self._run_cipd(, args, **kwargs)
def check_run(self, *args, **kwargs):
return self._run_cipd(self._system.check_run, args, **kwargs)
def exists(self, name, *versions):
if any((name, v) in self._exists_cache for v in versions):
return True
for v in versions:
rc, cmd_output ='resolve', name, '-version', v)
if not rc:
self._exists_cache.add((name, v))
return True
if 'ambiguity when resolving the tag' in cmd_output:
# Don't continue if we hit this unexpected condition, as we'll just
# keep trying to upload more instances of the package.
raise ValueError('Multiple instances matching tag %s for package %s.' %
(v, name))
return False
def create(self, package, root):
cmd = [
'-compression-level', str(package.compress_level),
'-in', root,
'-install-mode', package.install_mode,
for tag in package.tags or ():
cmd += ['-tag', tag]
def init(self, root):
"""Initializes a CIPD site root at the given location."""
self.check_run('init', '-force', root)
def install(self, name, version, root):
self.check_run('install', '-root', root, name, version)
def register_package(self, path, tags, dryrun=False):
cmd = [
for tag in tags:
cmd += ['-tag', tag]
self.check_run(*cmd, dryrun=dryrun)
def create_package(self, package, src_dir, pkg_path):
'-compression-level', str(package.compress_level),
'-in', src_dir,
'-install-mode', package.install_mode,
'-out', pkg_path,
def fetch_package(self, name, version, path):
'-version', version,
'-out', path,
def deploy_package(self, path, root):
self.check_run('pkg-deploy', '-root', root, path)
def set_refs(self, package, version, refs, dryrun=False):
cmd = ['set-ref', package, '-version', version]
for ref in refs:
cmd += ['-ref', ref]
self.check_run(*cmd, dryrun=dryrun)