Source code for tox.venv

import codecs
import json
import os
import pipes
import re
import sys
import warnings
from itertools import chain

import py
from pkg_resources import to_filename

import tox

from .config import DepConfig

class CreationConfig:
    def __init__(
        self.base_resolved_python_md5 = base_resolved_python_md5
        self.base_resolved_python_path = base_resolved_python_path
        self.tox_version = tox_version
        self.sitepackages = sitepackages
        self.usedevelop = usedevelop
        self.alwayscopy = alwayscopy
        self.deps = deps

    def writeconfig(self, path):
        lines = [
            "{} {}".format(self.base_resolved_python_md5, self.base_resolved_python_path),
            "{} {:d} {:d} {:d}".format(
                self.tox_version, self.sitepackages, self.usedevelop, self.alwayscopy
        for dep in self.deps:
            lines.append("{} {}".format(*dep))
        content = "\n".join(lines)
        return content

    def readconfig(cls, path):
            lines = path.readlines(cr=0)
            base_resolved_python_info = lines.pop(0).split(None, 1)
            tox_version, sitepackages, usedevelop, alwayscopy = lines.pop(0).split(None, 4)
            sitepackages = bool(int(sitepackages))
            usedevelop = bool(int(usedevelop))
            alwayscopy = bool(int(alwayscopy))
            deps = []
            for line in lines:
                base_resolved_python_md5, depstring = line.split(None, 1)
                deps.append((base_resolved_python_md5, depstring))
            base_resolved_python_md5, base_resolved_python_path = base_resolved_python_info
            return CreationConfig(
        except Exception:
            return None

    def matches_with_reason(self, other, deps_matches_subset=False):
        for attr in (
            left = getattr(self, attr)
            right = getattr(other, attr)
            if left != right:
                return False, "attr {} {!r}!={!r}".format(attr, left, right)
        self_deps = set(self.deps)
        other_deps = set(other.deps)
        if self_deps != other_deps:
            if deps_matches_subset:
                diff = other_deps - self_deps
                if diff:
                    return False, "missing in previous {!r}".format(diff)
                return False, "{!r}!={!r}".format(self_deps, other_deps)
        return True, None

    def matches(self, other, deps_matches_subset=False):
        outcome, _ = self.matches_with_reason(other, deps_matches_subset)
        return outcome

[docs]class VirtualEnv(object): def __init__(self, envconfig=None, session=None): self.envconfig = envconfig self.session = session @property def hook(self): return self.envconfig.config.pluginmanager.hook @property def path(self): """ Path to environment base dir. """ return self.envconfig.envdir @property def path_config(self): return self.path.join(".tox-config1") @property def name(self): """ test environment name. """ return self.envconfig.envname def __repr__(self): return "<VirtualEnv at {!r}>".format(self.path)
[docs] def getcommandpath(self, name, venv=True, cwd=None): """ Return absolute path (str or localpath) for specified command name. - If it's a local path we will rewrite it as as a relative path. - If venv is True we will check if the command is coming from the venv or is whitelisted to come from external. """ name = str(name) if os.path.isabs(name): return name if os.path.split(name)[0] == ".": path = cwd.join(name) if path.check(): return str(path) if venv: path = self._venv_lookup_and_check_external_whitelist(name) else: path = self._normal_lookup(name) if path is None: raise tox.exception.InvocationError("could not find executable {!r}".format(name)) return str(path) # will not be rewritten for reporting
def _venv_lookup_and_check_external_whitelist(self, name): path = self._venv_lookup(name) if path is None: path = self._normal_lookup(name) if path is not None: self._check_external_allowed_and_warn(path) return path def _venv_lookup(self, name): return py.path.local.sysfind(name, paths=[self.envconfig.envbindir]) def _normal_lookup(self, name): return py.path.local.sysfind(name) def _check_external_allowed_and_warn(self, path): if not self.is_allowed_external(path): "test command found but not installed in testenv\n" " cmd: {}\n" " env: {}\n" "Maybe you forgot to specify a dependency? " "See also the whitelist_externals envconfig setting.".format( path, self.envconfig.envdir ) ) def is_allowed_external(self, p): tryadd = [""] if tox.INFO.IS_WIN: tryadd += [os.path.normcase(x) for x in os.environ["PATHEXT"].split(os.pathsep)] p = py.path.local(os.path.normcase(str(p))) for x in self.envconfig.whitelist_externals: for add in tryadd: if p.fnmatch(x + add): return True return False
[docs] def update(self, action): """ return status string for updating actual venv to match configuration. if status string is empty, all is ok. """ rconfig = CreationConfig.readconfig(self.path_config) if self.envconfig.recreate: reason = "-r flag" else: if rconfig is None: reason = "no previous config {}".format(self.path_config) else: live_config = self._getliveconfig() deps_subset_match = getattr(self.envconfig, "deps_matches_subset", False) outcome, reason = rconfig.matches_with_reason(live_config, deps_subset_match) if reason is None:"reusing", self.envconfig.envdir) return"cannot reuse", reason) if rconfig is None: action.setactivity("create", self.envconfig.envdir) else: action.setactivity("recreate", self.envconfig.envdir) try: self.hook.tox_testenv_create(action=action, venv=self) self.just_created = True except tox.exception.UnsupportedInterpreter as exception: return exception try: self.hook.tox_testenv_install_deps(action=action, venv=self) except tox.exception.InvocationError as exception: return "could not install deps {}; v = {!r}".format(self.envconfig.deps, exception)
def _getliveconfig(self): base_resolved_python_path = self.envconfig.python_info.executable version = tox.__version__ sitepackages = self.envconfig.sitepackages develop = self.envconfig.usedevelop alwayscopy = self.envconfig.alwayscopy deps = [] for dep in self.get_resolved_dependencies(): dep_name_md5 = getdigest( deps.append((dep_name_md5, base_resolved_python_md5 = getdigest(base_resolved_python_path) return CreationConfig( base_resolved_python_md5, base_resolved_python_path, version, sitepackages, develop, deps, alwayscopy, ) def _getresolvedeps(self): warnings.warn( "that's a private function there, use get_resolved_dependencies," "this will be removed in 3.2", category=DeprecationWarning, ) return self.get_resolved_dependencies() def get_resolved_dependencies(self): dependencies = [] for dependency in self.envconfig.deps: if dependency.indexserver is None: package = self.session._resolve_package( if package != dependency = dependency.__class__(package) dependencies.append(dependency) return dependencies def getsupportedinterpreter(self): return self.envconfig.getsupportedinterpreter() def matching_platform(self): return re.match(self.envconfig.platform, sys.platform) def finish(self): previous_config = CreationConfig.readconfig(self.path_config) live_config = self._getliveconfig() if previous_config is None or not previous_config.matches(live_config): content = live_config.writeconfig(self.path_config) "write config to {} as {!r}".format(self.path_config, content) ) def _needs_reinstall(self, setupdir, action): setup_py = setupdir.join("") setup_cfg = setupdir.join("setup.cfg") args = [self.envconfig.envpython, str(setup_py), "--name"] env = self._get_os_environ() output = action.popen( args, cwd=setupdir, redirect=False, returnout=True, env=env, capture_err=False ) name = next( (i for i in output.split("\n") if i and not i.startswith("pydev debugger:")), "" ) args = [ self.envconfig.envpython, "-c", "import sys; import json; print(json.dumps(sys.path))", ] out = action.popen(args, redirect=False, returnout=True, env=env) try: sys_path = json.loads(out) except ValueError: sys_path = [] egg_info_fname = ".".join((to_filename(name), "egg-info")) for d in reversed(sys_path): egg_info = py.path.local(d).join(egg_info_fname) if egg_info.check(): break else: return True needs_reinstall = any( conf_file.check() and conf_file.mtime() > egg_info.mtime() for conf_file in (setup_py, setup_cfg) ) # Ensure the modification time of the egg-info folder is updated so we # won't need to do this again. # TODO(stephenfin): Remove once the minimum version of setuptools is # high enough to include if needs_reinstall: egg_info.setmtime() return needs_reinstall def install_pkg(self, dir, action, name, is_develop=False): assert action is not None if getattr(self, "just_created", False): action.setactivity(name, dir) self.finish() pip_flags = ["--exists-action", "w"] else: if is_develop and not self._needs_reinstall(dir, action): action.setactivity("{}-noop".format(name), dir) return action.setactivity("{}-nodeps".format(name), dir) pip_flags = ["--no-deps"] + ([] if is_develop else ["-U"]) pip_flags.extend(["-v"] * min(3, - 2)) if action.venv.envconfig.extras: dir += "[{}]".format(",".join(action.venv.envconfig.extras)) target = [dir] if is_develop: target.insert(0, "-e") self._install(target, extraopts=pip_flags, action=action) def developpkg(self, setupdir, action): self.install_pkg(setupdir, action, "develop-inst", is_develop=True) def installpkg(self, sdistpath, action): self.install_pkg(sdistpath, action, "inst") def _installopts(self, indexserver): options = [] if indexserver: options += ["-i", indexserver] if self.envconfig.pip_pre: options.append("--pre") return options def run_install_command(self, packages, action, options=()): def expand(val): # expand an install command if val == "{packages}": for package in packages: yield package elif val == "{opts}": for opt in options: yield opt else: yield val cmd = list(chain.from_iterable(expand(val) for val in self.envconfig.install_command)) self.ensure_pip_os_environ_ok() old_stdout = sys.stdout sys.stdout = codecs.getwriter("utf8")(sys.stdout) try: self._pcall( cmd, cwd=self.envconfig.config.toxinidir, action=action, < 2, ) finally: sys.stdout = old_stdout def ensure_pip_os_environ_ok(self): for key in ("PIP_RESPECT_VIRTUALENV", "PIP_REQUIRE_VIRTUALENV", "__PYVENV_LAUNCHER__"): os.environ.pop(key, None) if "PYTHONPATH" not in self.envconfig.passenv: # If PYTHONPATH not explicitly asked for, remove it. if "PYTHONPATH" in os.environ: "Discarding $PYTHONPATH from environment, to override " "specify PYTHONPATH in 'passenv' in your configuration." ) os.environ.pop("PYTHONPATH") # installing packages at user level may mean we're not installing inside the venv os.environ["PIP_USER"] = "0" # installing without dependencies may lead to broken packages os.environ["PIP_NO_DEPS"] = "0" def _install(self, deps, extraopts=None, action=None): if not deps: return d = {} ixservers = [] for dep in deps: if isinstance(dep, (str, py.path.local)): dep = DepConfig(str(dep), None) assert isinstance(dep, DepConfig), dep if dep.indexserver is None: ixserver = self.envconfig.config.indexserver["default"] else: ixserver = dep.indexserver d.setdefault(ixserver, []).append( if ixserver not in ixservers: ixservers.append(ixserver) assert ixserver.url is None or isinstance(ixserver.url, str) for ixserver in ixservers: packages = d[ixserver] options = self._installopts(ixserver.url) if extraopts: options.extend(extraopts) self.run_install_command(packages=packages, options=options, action=action) def _get_os_environ(self, is_test_command=False): if is_test_command: # for executing tests we construct a clean environment env = {} for env_key in self.envconfig.passenv: if env_key in os.environ: env[env_key] = os.environ[env_key] else: # for executing non-test commands we use the full # invocation environment env = os.environ.copy() # in any case we honor per-testenv setenv configuration env.update(self.envconfig.setenv) env["VIRTUAL_ENV"] = str(self.path) return env def test( self, redirect=False, name="runtests", commands=None, ignore_outcome=None, ignore_errors=None, display_hash_seed=False, ): if commands is None: commands = self.envconfig.commands if ignore_outcome is None: ignore_outcome = self.envconfig.ignore_outcome if ignore_errors is None: ignore_errors = self.envconfig.ignore_errors with self.session.newaction(self, name) as action: cwd = self.envconfig.changedir if display_hash_seed: env = self._get_os_environ(is_test_command=True) # Display PYTHONHASHSEED to assist with reproducibility. action.setactivity(name, "PYTHONHASHSEED={!r}".format(env.get("PYTHONHASHSEED"))) for i, argv in enumerate(commands): # have to make strings as _pcall changes argv[0] to a local() # happens if the same environment is invoked twice message = "commands[{}] | {}".format( i, " ".join([pipes.quote(str(x)) for x in argv]) ) action.setactivity(name, message) # check to see if we need to ignore the return code # if so, we need to alter the command line arguments if argv[0].startswith("-"): ignore_ret = True if argv[0] == "-": del argv[0] else: argv[0] = argv[0].lstrip("-") else: ignore_ret = False try: self._pcall( argv, cwd=cwd, action=action, redirect=redirect, ignore_ret=ignore_ret, is_test_command=True, ) except tox.exception.InvocationError as err: if ignore_outcome: msg = "command failed but result from testenv is ignored\ncmd:""{} {}".format(msg, err)) self.status = "ignored failed command" continue # keep processing commands self.status = "commands failed" if not ignore_errors: break # Don't process remaining commands except KeyboardInterrupt: self.status = "keyboardinterrupt" raise def _pcall( self, args, cwd, venv=True, is_test_command=False, action=None, redirect=True, ignore_ret=False, returnout=False, ): # construct environment variables os.environ.pop("VIRTUALENV_PYTHON", None) env = self._get_os_environ(is_test_command=is_test_command) bin_dir = str(self.envconfig.envbindir) env["PATH"] = os.pathsep.join([bin_dir, os.environ["PATH"]])"setting PATH={}".format(env["PATH"])) # get command args[0] = self.getcommandpath(args[0], venv, cwd) if sys.platform != "win32" and "TOX_LIMITED_SHEBANG" in os.environ: args = prepend_shebang_interpreter(args) cwd.ensure(dir=1) # ensure the cwd exists return action.popen( args, cwd=cwd, env=env, redirect=redirect, ignore_ret=ignore_ret, returnout=returnout )
def getdigest(path): path = py.path.local(path) if not path.check(file=1): return "0" * 32 return path.computehash() def prepend_shebang_interpreter(args): # prepend interpreter directive (if any) to argument list # # When preparing virtual environments in a file container which has large # length, the system might not be able to invoke shebang scripts which # define interpreters beyond system limits (e.x. Linux as a limit of 128; # BINPRM_BUF_SIZE). This method can be used to check if the executable is # a script containing a shebang line. If so, extract the interpreter (and # possible optional argument) and prepend the values to the provided # argument list. tox will only attempt to read an interpreter directive of # a maximum size of 2048 bytes to limit excessive reading and support UNIX # systems which may support a longer interpret length. try: with open(args[0], "rb") as f: if == b"#" and == b"!": MAXINTERP = 2048 interp = f.readline(MAXINTERP).rstrip().decode("UTF-8") interp_args = interp.split(None, 1)[:2] return interp_args + args except (UnicodeDecodeError, IOError): pass return args @tox.hookimpl def tox_testenv_create(venv, action): config_interpreter = venv.getsupportedinterpreter() args = [sys.executable, "-m", "virtualenv"] if venv.envconfig.sitepackages: args.append("--system-site-packages") if venv.envconfig.alwayscopy: args.append("--always-copy") # add interpreter explicitly, to prevent using default (virtualenv.ini) args.extend(["--python", str(config_interpreter)]) venv.session.make_emptydir(venv.path) basepath = venv.path.dirpath() basepath.ensure(dir=1) args.append(venv.path.basename) venv._pcall(args, venv=False, action=action, cwd=basepath) return True # Return non-None to indicate plugin has completed @tox.hookimpl def tox_testenv_install_deps(venv, action): deps = venv.get_resolved_dependencies() if deps: depinfo = ", ".join(map(str, deps)) action.setactivity("installdeps", depinfo) venv._install(deps, action=action) return True # Return non-None to indicate plugin has completed @tox.hookimpl def tox_runtest(venv, redirect): venv.test(redirect=redirect) return True # Return non-None to indicate plugin has completed @tox.hookimpl def tox_runtest_pre(venv): venv.status = 0 venv.session.make_emptydir(venv.envconfig.envtmpdir) venv.envconfig.envtmpdir.ensure(dir=1) venv.test( name="run-test-pre", commands=venv.envconfig.commands_pre, redirect=False, ignore_outcome=False, ignore_errors=False, display_hash_seed=True, ) @tox.hookimpl def tox_runtest_post(venv): venv.test( name="run-test-post", commands=venv.envconfig.commands_post, redirect=False, ignore_outcome=False, ignore_errors=False, ) @tox.hookimpl def tox_runenvreport(venv, action): # write out version dependency information args = venv.envconfig.list_dependencies_command output = venv._pcall(args, cwd=venv.envconfig.config.toxinidir, action=action) # the output contains a mime-header, skip it output = output.split("\n\n")[-1] packages = output.strip().split("\n") return packages # Return non-None to indicate plugin has completed