blob: 0464bb20c99433a0e929dd5a63a77ffd695b362d [file] [log] [blame]
# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Virtualenv management"""
from __future__ import absolute_import
from __future__ import print_function
from __future__ import unicode_literals
import collections
import distutils.util
import functools
import hashlib
import itertools
import json
import os
import platform
import re
import shutil
import subprocess
import sys
import sysconfig
import tempfile
import warnings
from cros_venv import constants
from cros_venv import flock
_PACKAGE_DIR = os.path.join(constants.REPO_DIR, 'pip_packages')
_VIRTUALENV_COMMAND = 'virtualenv'
# BASE_DEPENDENCIES are pip requirements automatically included in every
# virtualenv so we have control over which versions of bootstrap
# packages are installed.
_BASE_DEPENDENCIES = ('setuptools==44.0.0', 'pip==20.0.2')
class _VenvPaths(object):
"""Wrapper defining paths inside a versioned virtualenv."""
def __init__(self, venvdir):
"""Initialize instance.
venvdir is the absolute path to a virtualenv.
self.venvdir = venvdir
def __repr__(self):
return '{cls}({this.venvdir!r})'.format(
def python(self):
"""Path to the virtualenv's Python binary."""
return os.path.join(self.venvdir, 'bin', 'python')
def lockfile(self):
"""Path to lock file for changing virtualenv."""
return os.path.join(self.venvdir, 'change.lock')
def logfile(self):
"""Path to log file for creating virtualenv."""
return os.path.join(self.venvdir, 'create.log')
def spec(self):
"""Path to spec file inside virtualenv directory."""
return os.path.join(self.venvdir, 'spec.json')
class VersionedVenv(object):
"""Versioned virtualenv, specified by a _VenvSpec.
This class provides a method for ensuring the versioned virtualenv
is created.
def __init__(self, spec):
"""Initialize instance.
spec is a _VenvSpec.
self._spec = spec
self._paths = _VenvPaths(_get_venvdir(spec))
def __repr__(self):
return '{cls}({this._spec!r})'.format(
def ensure(self):
"""Ensure that the virtualenv exists."""
with flock.FileLock(self._paths.lockfile):
return self._paths.venvdir
def logfile(self):
"""Show path to internal log file."""
return self._paths.logfile
def logdata(self):
"""Get any internal logged data."""
with open(self._paths.logfile) as fp:
def _log_env(self, logfile):
"""Log details about the active runtime for debugging."""
def get_var(var):
return sysconfig.get_config_var(var)
'venv spec: %s\n' % (self._spec,),
'Distutils platform tag: %s\n' % (distutils.util.get_platform(),),
'Python implementation: %s\n' % (platform.python_implementation(),),
for var in ('py_version_nodot', 'SOABI', 'Py_DEBUG', 'WITH_PYMALLOC',
logfile.write('sysconfig %s: %s\n' % (var, get_var(var)))
def _check_or_create(self):
"""Check virtualenv, creating it if it is not created."""
existing_spec = self._load_spec()
except IOError:
def _create(self):
"""Create virtualenv."""
with open(self._paths.logfile, 'w') as logfile, \
_make_reqs_file(self._spec) as reqs_file:
def _check(self, spec):
"""Check if the given spec matches our spec.
Raise SpecMismatchError if check fails.
if spec != self._spec:
raise SpecMismatchError
def _dump_spec(self):
"""Save the _VenvSpec to the virtualenv on disk."""
with open(self._paths.spec, 'w') as f:
return _dump_spec(self._spec, f)
def _load_spec(self):
"""Return the _VenvSpec for the virtualenv on disk."""
with open(self._paths.spec, 'r') as f:
return _load_spec(f)
class SpecMismatchError(Exception):
"""Versioned virtualenv specs do not match."""
_VenvSpec = collections.namedtuple('_VenvSpec', 'py_version,reqs')
def make_spec(f):
"""Make _VenvSpec from a requirements file object."""
return _VenvSpec(_get_python_version(),
def _get_reqs_hash(spec):
"""Return hash string for _VenvSpec requirements.
Make sure to check for collisions.
hasher = hashlib.md5()
return hasher.hexdigest()
def _get_venvdir(spec):
"""Return the virtualenv directory to use for the _VenvSpec.
Returns absolute path.
cache_dir = _get_cache_dir()
return os.path.join(
cache_dir, 'venv-%s-%s' % (spec.py_version, _get_reqs_hash(spec)))
def _dump_spec(spec, f):
"""Dump _VenvSpec to a file."""
json.dump(spec, f)
def _load_spec(f):
"""Load _VenvSpec from a file."""
return _VenvSpec._make(json.load(f))
def _make_reqs_file(spec):
"""Return a temporary reqs file for the virtualenv spec.
The return value is a tempfile.NamedTemporaryFile, which cleans
up on close. The filename is accessible via the name attribute.
f = tempfile.NamedTemporaryFile('w')
f.writelines(req + '\n' for req in _BASE_DEPENDENCIES)
return f
def _virtualenv_version():
"""Return a version tuple for virtualenv."""
# Versions before 20.0.0 returned just the version.
# The 20.0.0 series switched to:
# virtualenv 20.0.13 from .../virtualenv/
result = subprocess.check_output([_VIRTUALENV_COMMAND, '--version'],
except OSError:
raise EnvironmentError(
'Error executing virtualenv, make sure that you install '
'virtualenv first (see '
# This strips off "rc" and related tags, but we don't care for our use.
m = re.match(br'^(virtualenv *)?([0-9.]+)', result)
assert m, 'Could not find version in "%s"' % (result,)
return tuple(int(x) for x in'.'))
def _create_venv(venvdir, logfile):
"""Create a virtualenv at the given path."""
ver = _virtualenv_version()
MIN_VER = (1, 10)
assert ver >= MIN_VER, ('virtualenv %s is required, but %s was found' %
(ver, MIN_VER))
env = os.environ.copy()
if ver < (20, 0):
# TODO( upstream virtualenv has a bug when this is
# set. See also
command = [_VIRTUALENV_COMMAND, venvdir, '-p', sys.executable,
'--extra-search-dir', _PACKAGE_DIR, '--clear',
'--never-download', '-vvvvv']
_log_check_call(command, logfile=logfile, env=env)
except OSError as e:
raise VirtualenvMissingError(e)
class VirtualenvMissingError(Exception):
"""Virtualenv is missing."""
def __init__(self, cause):
"""Initialize instance.
cause is an object that describes the underlying cause of the
error. This is usually an exception, but can be any object.
The object's string representation is used in this exception's
string representation.
super(VirtualenvMissingError, self).__init__(cause)
self.cause = cause
def __str__(self):
return 'virtualenv is not installed (caused by %s)' % (self.cause)
def __repr__(self):
return '%s(%r)' % (type(self).__name__, self.cause)
def _install_reqs_file(python_path, reqs_path, logfile):
"""Install reqs file using pip."""
command = [python_path, '-m', 'pip', 'install', '-vvv',
'--no-index', '-f', 'file://' + _PACKAGE_DIR, '-r', reqs_path]
_log_check_call(command, logfile=logfile)
def _add_call_logging(call_func):
"""Wrap a subprocess-style call with logging."""
def wrapped_command(args, logfile, **kwargs):
"""Logging-wrapped call.
Arguments are similar to subprocess.Popen, depending on the
underlying call. There is an extra keyword-only parameter
logfile, which takes a file object.
logfile.write('Running %r\n' % (args,))
call_func(args, stdout=logfile, **kwargs)
return wrapped_command
_log_check_call = _add_call_logging(subprocess.check_call)
def _get_python_version():
"""Return the version string for the current Python."""
return '.'.join(str(part) for part in sys.version_info[:3])
def _get_cache_dir():
"""Get cache dir to use for cros_venv.
Returns absolute path.
return os.environ.get('CROS_VENV_CACHE',
def _makedirs_exist_ok(path):
"""Make directories recursively, ignoring if directory already exists."""
except OSError:
if not os.path.isdir(path): # pragma: no cover