Source code for tox.venv

import ast
import codecs
import os
import pipes
import re
import sys
import warnings

import py

import tox

from .config import DepConfig

class CreationConfig:
    def __init__(self, md5, python, version, sitepackages, usedevelop, deps, alwayscopy):
        self.md5 = md5
        self.python = python
        self.version = version
        self.sitepackages = sitepackages
        self.usedevelop = usedevelop
        self.alwayscopy = alwayscopy
        self.deps = deps

    def writeconfig(self, path):
        lines = [
            "{} {}".format(self.md5, self.python),
            "{} {:d} {:d} {:d}".format(
                self.version, self.sitepackages, self.usedevelop, self.alwayscopy
        for dep in self.deps:
            lines.append("{} {}".format(*dep))

    def readconfig(cls, path):
            lines = path.readlines(cr=0)
            value = lines.pop(0).split(None, 1)
            md5, python = value
            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:
                md5, depstring = line.split(None, 1)
                deps.append((md5, depstring))
            return CreationConfig(md5, python, version, sitepackages, usedevelop, deps, alwayscopy)
        except Exception:
            return None

    def matches(self, other):
        return (
            and self.md5 == other.md5
            and self.python == other.python
            and self.version == other.version
            and self.sitepackages == other.sitepackages
            and self.usedevelop == other.usedevelop
            and self.alwayscopy == other.alwayscopy
            and self.deps == other.deps

[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 not self.envconfig.recreate and rconfig and rconfig.matches(self._getliveconfig()):"reusing", self.envconfig.envdir) return 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: return sys.exc_info()[1] try: self.hook.tox_testenv_install_deps(action=action, venv=self) except tox.exception.InvocationError: v = sys.exc_info()[1] return "could not install deps {}; v = {!r}".format(self.envconfig.deps, v)
def _getliveconfig(self): python = self.envconfig.python_info.executable md5 = getdigest(python) version = tox.__version__ sitepackages = self.envconfig.sitepackages develop = self.envconfig.usedevelop alwayscopy = self.envconfig.alwayscopy deps = [] for dep in self.get_resolved_dependencies(): raw_dep = md5 = getdigest(raw_dep) deps.append((md5, raw_dep)) return CreationConfig(md5, python, 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): self._getliveconfig().writeconfig(self.path_config) 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._getenv() output = action.popen(args, cwd=setupdir, redirect=False, returnout=True, env=env) name = output.strip() args = [self.envconfig.envpython, "-c", "import sys; print(sys.path)"] out = action.popen(args, redirect=False, returnout=True, env=env) try: sys_path = ast.literal_eval(out.strip()) except SyntaxError: sys_path = [] egg_info_fname = ".".join((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 return any( conf_file.check() and conf_file.mtime() > egg_info.mtime() for conf_file in (setup_py, setup_cfg) ) def developpkg(self, setupdir, action): assert action is not None if getattr(self, "just_created", False): action.setactivity("develop-inst", setupdir) self.finish() extraopts = [] else: if not self._needs_reinstall(setupdir, action): action.setactivity("develop-inst-noop", setupdir) return action.setactivity("develop-inst-nodeps", setupdir) extraopts = ["--no-deps"] if action.venv.envconfig.extras: setupdir += "[{}]".format(",".join(action.venv.envconfig.extras)) self._install(["-e", setupdir], extraopts=extraopts, action=action) def installpkg(self, sdistpath, action): assert action is not None if getattr(self, "just_created", False): action.setactivity("inst", sdistpath) self.finish() extraopts = [] else: action.setactivity("inst-nodeps", sdistpath) extraopts = ["-U", "--no-deps"] if action.venv.envconfig.extras: sdistpath += "[{}]".format(",".join(action.venv.envconfig.extras)) self._install([sdistpath], extraopts=extraopts, action=action) 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=()): argv = self.envconfig.install_command[:] i = argv.index("{packages}") argv[i : i + 1] = packages if "{opts}" in argv: i = argv.index("{opts}") argv[i : i + 1] = list(options) for x in ("PIP_RESPECT_VIRTUALENV", "PIP_REQUIRE_VIRTUALENV", "__PYVENV_LAUNCHER__"): os.environ.pop(x, 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") old_stdout = sys.stdout sys.stdout = codecs.getwriter("utf8")(sys.stdout) try: self._pcall( argv, cwd=self.envconfig.config.toxinidir, action=action, < 2, ) finally: sys.stdout = old_stdout 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 _getenv(self, testcommand=False): if testcommand: # for executing tests we construct a clean environment env = {} for envname in self.envconfig.passenv: if envname in os.environ: env[envname] = os.environ[envname] 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): with self.session.newaction(self, "runtests") as action: self.status = 0 self.session.make_emptydir(self.envconfig.envtmpdir) self.envconfig.envtmpdir.ensure(dir=1) cwd = self.envconfig.changedir env = self._getenv(testcommand=True) # Display PYTHONHASHSEED to assist with reproducibility. action.setactivity("runtests", "PYTHONHASHSEED={!r}".format(env.get("PYTHONHASHSEED"))) for i, argv in enumerate(self.envconfig.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("runtests", 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, testcommand=True, ) except tox.exception.InvocationError as err: if self.envconfig.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 self.envconfig.ignore_errors: break # Don't process remaining commands except KeyboardInterrupt: self.status = "keyboardinterrupt" raise def _pcall( self, args, cwd, venv=True, testcommand=False, action=None, redirect=True, ignore_ret=False ): os.environ.pop("VIRTUALENV_PYTHON", None) cwd.ensure(dir=1) args[0] = self.getcommandpath(args[0], venv, cwd) if sys.platform != "win32" and "TOX_LIMITED_SHEBANG" in os.environ: args = prepend_shebang_interpreter(args) env = self._getenv(testcommand=testcommand) bindir = str(self.envconfig.envbindir) env["PATH"] = p = os.pathsep.join([bindir, os.environ["PATH"]])"setting PATH={}".format(p)) return action.popen(args, cwd=cwd, env=env, redirect=redirect, ignore_ret=ignore_ret)
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() interp_args = interp.split(None, 1)[:2] return interp_args + args except 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_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