Source code for tox.session

Automatically package and test a Python project against configurable
Python2 and Python3 based virtual environments. Environments are
setup by using virtualenv. Configuration is generally done through an
INI-style "tox.ini" file.
from __future__ import with_statement

import tox
import py
import os
import sys
import subprocess
from tox._verlib import NormalizedVersion, IrrationalVersionError
from tox.venv import VirtualEnv
from tox.config import parseconfig
from tox.result import ResultLog
from subprocess import STDOUT

def now():
    return py.std.time.time()

def prepare(args):
    config = parseconfig(args)
        raise SystemExit(0)
    elif config.option.helpini:
        raise SystemExit(0)
    return config

def main(args=None):
        config = prepare(args)
        retcode = Session(config).runcommand()
        raise SystemExit(retcode)
    except KeyboardInterrupt:
        raise SystemExit(2)
    except tox.exception.MinVersionError as e:
        r = Reporter(None)
        raise SystemExit(1)

def show_help(config):
    tw =
    tw.line("Environment variables", bold=True)
    tw.line("TOXENV: comma separated list of environments "
            "(overridable by '-e')")
    tw.line("TOX_TESTENV_PASSENV: space-separated list of extra "
            "environment variables to be passed into test command "

def show_help_ini(config):
    tw =
    tw.sep("-", "per-testenv attributes")
    for env_attr in config._testenv_attr:
        tw.line("%-15s %-8s default: %s" %
                (, "<" + env_attr.type + ">", env_attr.default), bold=True)

class Action(object):
    def __init__(self, session, venv, msg, args):
        self.venv = venv
        self.msg = msg
        self.activity = msg.split(" ", 1)[0]
        self.session = session =
        self.args = args = venv and venv.envconfig.envname or "tox"
        self._popenlist = []
        if self.venv:
            self.venvname =
            self.venvname = "GLOB"
        if msg == "runtests":
            cat = "test"
            cat = "setup"
        envlog = session.resultlog.get_envlog(self.venvname)
        self.commandlog = envlog.get_commandlog(cat)

    def __enter__(self):

    def __exit__(self, *args):

    def setactivity(self, name, msg):
        self.activity = name"%s %s: %s" % (self.venvname, name, msg), bold=True)

    def info(self, name, msg):"%s %s: %s" % (self.venvname, name, msg), bold=True)

    def _initlogpath(self, actionid):
        if self.venv:
            logdir = self.venv.envconfig.envlogdir
            logdir = self.session.config.logdir
            l = logdir.listdir("%s-*" % actionid)
        except py.error.ENOENT:
            l = []
        num = len(l)
        path = logdir.join("%s-%s.log" % (actionid, num))
        f ='w')
        return f

    def popen(self, args, cwd=None, env=None, redirect=True, returnout=False, ignore_ret=False):
        stdout = outpath = None
        resultjson = self.session.config.option.resultjson
        if resultjson or redirect:
            fout = self._initlogpath(
            fout.write("actionid: %s\nmsg: %s\ncmdargs: %r\nenv: %s\n\n" % (
      , self.msg, args, env))
            self.popen_outpath = outpath = py.path.local(
            fin =
    # read the header, so it won't be written to stdout
            stdout = fout
        elif returnout:
            stdout = subprocess.PIPE
        if cwd is None:
            # XXX cwd = self.session.config.cwd
            cwd = py.path.local()
            popen = self._popen(args, cwd, env=env,
                                stdout=stdout, stderr=STDOUT)
        except OSError as e:
  "invocation failed (errno %d), args: %s, cwd: %s" %
                              (e.errno, args, cwd))
        popen.outpath = outpath
        popen.args = [str(x) for x in args]
        popen.cwd = cwd
        popen.action = self
  , env=env)
                if resultjson and not redirect:
                    assert popen.stderr is None  # prevent deadlock
                    out = None
                    last_time = now()
                    while 1:
                        fin_pos = fin.tell()
                        # we have to read one byte at a time, otherwise there
                        # might be no output for a long time with slow tests
                        data =
                        if data:
                            if '\n' in data or (now() - last_time) > 1:
                                # we flush on newlines or after 1 second to
                                # provide quick enough feedback to the user
                                # when printing a dot per test
                                last_time = now()
                        elif popen.poll() is not None:
                            if popen.stdout is not None:
                    out, err = popen.communicate()
            except KeyboardInterrupt:
                raise KeyboardInterrupt()
            ret = popen.wait()
        if ret and not ignore_ret:
            invoked = " ".join(map(str, popen.args))
            if outpath:
      "invocation failed (exit code %d), logfile: %s" %
                                  (ret, outpath))
                out =
                if hasattr(self, "commandlog"):
                    self.commandlog.add_command(popen.args, out, ret)
                raise tox.exception.InvocationError(
                    "%s (see %s)" % (invoked, outpath), ret)
                raise tox.exception.InvocationError("%r" % (invoked, ), ret)
        if not out and outpath:
            out =
        if hasattr(self, "commandlog"):
            self.commandlog.add_command(popen.args, out, ret)
        return out

    def _rewriteargs(self, cwd, args):
        newargs = []
        for arg in args:
            if sys.platform != "win32" and isinstance(arg, py.path.local):
                arg = cwd.bestrelpath(arg)

        # subprocess does not always take kindly to .py scripts
        # so adding the interpreter here.
        if sys.platform == "win32":
            ext = os.path.splitext(str(newargs[0]))[1].lower()
            if ext == '.py' and self.venv:
                newargs = [str(self.venv.envconfig.envpython)] + newargs

        return newargs

    def _popen(self, args, cwd, stdout, stderr, env=None):
        args = self._rewriteargs(cwd, args)
        if env is None:
            env = os.environ.copy()
        return self.session.popen(args, shell=False, cwd=str(cwd),
                                  stdout=stdout, stderr=stderr, env=env)

class Reporter(object):
    actionchar = "-"

    def __init__(self, session): =
        self.session = session
        self._reportedlines = []
        # self.cumulated_time = 0.0

    def _get_verbosity(self):
        if self.session:
            return self.session.config.option.verbosity
            return 2

    def logpopen(self, popen, env):
        """ log information about the action.popen() created process. """
        cmd = " ".join(map(str, popen.args))
        if popen.outpath:
            self.verbosity1("  %s$ %s >%s" % (popen.cwd, cmd, popen.outpath,))
            self.verbosity1("  %s$ %s " % (popen.cwd, cmd))

    def logaction_start(self, action):
        msg = action.msg + " " + " ".join(map(str, action.args))
        self.verbosity2("%s start: %s" % (action.venvname, msg), bold=True)
        assert not hasattr(action, "_starttime")
        action._starttime = now()

    def logaction_finish(self, action):
        duration = now() - action._starttime
        # self.cumulated_time += duration
        self.verbosity2("%s finish: %s after %.2f seconds" % (
            action.venvname, action.msg, duration), bold=True)
        delattr(action, '_starttime')

    def startsummary(self):"_", "summary")

    def info(self, msg):
        if self._get_verbosity() >= 2:

    def using(self, msg):
        if self._get_verbosity() >= 1:
            self.logline("using %s" % (msg,), bold=True)

    def keyboard_interrupt(self):

#    def venv_installproject(self, venv, pkg):
#        self.logline("installing to %s: %s" % (venv.envconfig.envname, pkg))

    def keyvalue(self, name, value):
        if name.endswith(":"):
            name += " ", bold=True)

    def line(self, msg, **opts):
        self.logline(msg, **opts)

    def good(self, msg):
        self.logline(msg, green=True)

    def warning(self, msg):
        self.logline("WARNING:" + msg, red=True)

    def error(self, msg):
        self.logline("ERROR: " + msg, red=True)

    def skip(self, msg):
        self.logline("SKIPPED:" + msg, yellow=True)

    def logline(self, msg, **opts):
        self._reportedlines.append(msg)"%s" % msg, **opts)

    def verbosity0(self, msg, **opts):
        if self._get_verbosity() >= 0:
            self.logline("%s" % msg, **opts)

    def verbosity1(self, msg, **opts):
        if self._get_verbosity() >= 1:
            self.logline("%s" % msg, **opts)

    def verbosity2(self, msg, **opts):
        if self._get_verbosity() >= 2:
            self.logline("%s" % msg, **opts)

    # def log(self, msg):
    #    py.builtin.print_(msg, file=sys.stderr)

[docs]class Session: """ (unstable API). the session object that ties together configuration, reporting, venv creation, testing. """ def __init__(self, config, popen=subprocess.Popen, Report=Reporter): self.config = config self.popen = popen self.resultlog = ResultLog() = Report(self) self.make_emptydir(config.logdir) config.logdir.ensure(dir=1) #"logdir %s" %(self.config.logdir,))"tox.ini: %s" % (self.config.toxinipath,)) self._spec2pkg = {} self._name2venv = {} try: self.venvlist = [ self.getvenv(x) for x in self.config.envlist ] except LookupError: raise SystemExit(1) except tox.exception.ConfigError as e: raise SystemExit(1) self._actions = [] @property def hook(self): return self.config.pluginmanager.hook def _makevenv(self, name): envconfig = self.config.envconfigs.get(name, None) if envconfig is None:"unknown environment %r" % name) raise LookupError(name) elif envconfig.envdir == self.config.toxinidir: "venv %r in %s would delete project" % (name, envconfig.envdir)) raise tox.exception.ConfigError('envdir must not equal toxinidir') venv = VirtualEnv(envconfig=envconfig, session=self) self._name2venv[name] = venv return venv
[docs] def getvenv(self, name): """ return a VirtualEnv controler object for the 'name' env. """ try: return self._name2venv[name] except KeyError: return self._makevenv(name)
def newaction(self, venv, msg, *args): action = Action(self, venv, msg, args) self._actions.append(action) return action def runcommand(self):"tox-%s from %s" % (tox.__version__, tox.__file__)) if self.config.option.showconfig: self.showconfig() elif self.config.option.listenvs: self.showenvs() else: return self.subcommand_test() def _copyfiles(self, srcdir, pathlist, destdir): for relpath in pathlist: src = srcdir.join(relpath) if not src.check():"missing source file: %s" % (src,)) raise SystemExit(1) target = destdir.join(relpath) target.dirpath().ensure(dir=1) src.copy(target) def _makesdist(self): setup = self.config.setupdir.join("") if not setup.check(): raise tox.exception.MissingFile(setup) action = self.newaction(None, "packaging") with action: action.setactivity("sdist-make", setup) self.make_emptydir(self.config.distdir) action.popen([sys.executable, setup, "sdist", "--formats=zip", "--dist-dir", self.config.distdir, ], cwd=self.config.setupdir) try: return self.config.distdir.listdir()[0] except py.error.ENOENT: # check if empty or comment only data = [] with open(str(setup)) as fp: for line in fp: if line and line[0] == '#': continue data.append(line) if not ''.join(data).strip(): ' is empty' ) raise SystemExit(1) 'No dist directory found. Please check, e.g with:\n' ' python sdist' ) raise SystemExit(1) def make_emptydir(self, path): if path.check():" removing %s" % path) py.std.shutil.rmtree(str(path), ignore_errors=True) path.ensure(dir=1) def setupenv(self, venv): if not venv.matching_platform(): venv.status = "platform mismatch" return # we simply omit non-matching platforms action = self.newaction(venv, "getenv", venv.envconfig.envdir) with action: venv.status = 0 envlog = self.resultlog.get_envlog( try: status = venv.update(action=action) except tox.exception.InvocationError: status = sys.exc_info()[1] if status: commandlog = envlog.get_commandlog("setup") commandlog.add_command(["setup virtualenv"], str(status), 1) venv.status = status return False commandpath = venv.getcommandpath("python") envlog.set_python_info(commandpath) return True def finishvenv(self, venv): action = self.newaction(venv, "finishvenv") with action: venv.finish() return True def developpkg(self, venv, setupdir): action = self.newaction(venv, "developpkg", setupdir) with action: try: venv.developpkg(setupdir, action) return True except tox.exception.InvocationError: venv.status = sys.exc_info()[1] return False
[docs] def installpkg(self, venv, path): """Install package in the specified virtual environment. :param :class:`tox.config.VenvConfig`: Destination environment :param str path: Path to the distribution package. :return: True if package installed otherwise False. :rtype: bool """ self.resultlog.set_header(installpkg=py.path.local(path)) action = self.newaction(venv, "installpkg", path) with action: try: venv.installpkg(path, action) return True except tox.exception.InvocationError: venv.status = sys.exc_info()[1] return False
[docs] def get_installpkg_path(self): """ :return: Path to the distribution :rtype: py.path.local """ if not self.config.option.sdistonly and (self.config.sdistsrc or self.config.option.installpkg): path = self.config.option.installpkg if not path: path = self.config.sdistsrc path = self._resolve_pkg(path)"using package %r, skipping 'sdist' activity " % str(path)) else: try: path = self._makesdist() except tox.exception.InvocationError: v = sys.exc_info()[1]"FAIL could not package project - v = %r" % v) return sdistfile = self.config.distshare.join(path.basename) if sdistfile != path:"copying new sdistfile to %r" % str(sdistfile)) try: sdistfile.dirpath().ensure(dir=1) except py.error.Error:"could not copy distfile to %s" % sdistfile.dirpath()) else: path.copy(sdistfile) return path
def subcommand_test(self): if self.config.skipsdist:"skipping sdist step") path = None else: path = self.get_installpkg_path() if not path: return 2 if self.config.option.sdistonly: return for venv in self.venvlist: if self.setupenv(venv): if venv.envconfig.usedevelop: self.developpkg(venv, self.config.setupdir) elif self.config.skipsdist or venv.envconfig.skip_install: self.finishvenv(venv) else: self.installpkg(venv, path) # write out version dependency information action = self.newaction(venv, "envreport") with action: args = venv.envconfig.list_dependencies_command output = venv._pcall(args, cwd=self.config.toxinidir, action=action) # the output contains a mime-header, skip it output = output.split("\n\n")[-1] packages = output.strip().split("\n") action.setactivity("installed", ",".join(packages)) envlog = self.resultlog.get_envlog( envlog.set_installed(packages) self.runtestenv(venv) retcode = self._summary() return retcode def runtestenv(self, venv, redirect=False): if not self.config.option.notest: if venv.status: return self.hook.tox_runtest_pre(venv=venv) venv.test(redirect=redirect) self.hook.tox_runtest_post(venv=venv) else: venv.status = "skipped tests" def _summary(self): retcode = 0 for venv in self.venvlist: status = venv.status if isinstance(status, tox.exception.InterpreterNotFound): msg = " %s: %s" % (venv.envconfig.envname, str(status)) if self.config.option.skip_missing_interpreters: else: retcode = 1 elif status == "platform mismatch": msg = " %s: %s" % (venv.envconfig.envname, str(status)) elif status and status == "ignored failed command": msg = " %s: %s" % (venv.envconfig.envname, str(status)) elif status and status != "skipped tests": msg = " %s: %s" % (venv.envconfig.envname, str(status)) retcode = 1 else: if not status: status = "commands succeeded"" %s: %s" % (venv.envconfig.envname, status)) if not retcode:" congratulations :)") path = self.config.option.resultjson if path: path = py.path.local(path) path.write(self.resultlog.dumps_json())"wrote json report at: %s" % path) return retcode def showconfig(self): self.info_versions()"config-file:", self.config.option.configfile)"toxinipath: ", self.config.toxinipath)"toxinidir: ", self.config.toxinidir)"toxworkdir: ", self.config.toxworkdir)"setupdir: ", self.config.setupdir)"distshare: ", self.config.distshare)"skipsdist: ", self.config.skipsdist) for envconfig in self.config.envconfigs.values():"[testenv:%s]" % envconfig.envname, bold=True) for attr in self.config._parser._testenv_attr:" %-15s = %s" % (, getattr(envconfig, def showenvs(self): for env in self.config.envlist:"%s" % env) def info_versions(self): versions = ['tox-%s' % tox.__version__] try: version = py.process.cmdexec("virtualenv --version") except py.process.cmdexec.Error: versions.append("virtualenv-1.9.1 (vendored)") else: versions.append("virtualenv-%s" % version.strip())"tool-versions:", " ".join(versions)) def _resolve_pkg(self, pkgspec): try: return self._spec2pkg[pkgspec] except KeyError: self._spec2pkg[pkgspec] = x = self._resolvepkg(pkgspec) return x def _resolvepkg(self, pkgspec): if not os.path.isabs(str(pkgspec)): return pkgspec p = py.path.local(pkgspec) if p.check(): return p if not p.dirpath().check(dir=1): raise tox.exception.MissingDirectory(p.dirpath())"determining %s" % p) candidates = p.dirpath().listdir(p.basename) if len(candidates) == 0: raise tox.exception.MissingDependency(pkgspec) if len(candidates) > 1: items = [] for x in candidates: ver = getversion(x.basename) if ver is not None: items.append((ver, x)) else:"could not determine version of: %s" % str(x)) items.sort() if not items: raise tox.exception.MissingDependency(pkgspec) return items[-1][1] else: return candidates[0]
_rex_getversion ="[\w_\-\+\.]+-(.*)(\.zip|\.tar.gz)") def getversion(basename): m = _rex_getversion.match(basename) if m is None: return None version = try: return NormalizedVersion(version) except IrrationalVersionError: return None