From 63e565d660867ee8794d0e9d06f45240108910df Mon Sep 17 00:00:00 2001 From: holger krekel Date: Tue, 12 May 2015 14:20:46 +0200 Subject: rename internal files -- in any case tox offers no external API except for the experimental plugin hooks, use tox internals at your own risk. --- CHANGELOG | 4 + setup.py | 2 +- tests/test_config.py | 12 +- tests/test_interpreters.py | 2 +- tests/test_venv.py | 16 +- tests/test_z_cmdline.py | 4 +- tox/__init__.py | 4 +- tox/__main__.py | 5 +- tox/_cmdline.py | 673 --------------------------- tox/_config.py | 1072 -------------------------------------------- tox/_exception.py | 28 -- tox/_pytestplugin.py | 10 +- tox/_venv.py | 422 ----------------- tox/config.py | 1072 ++++++++++++++++++++++++++++++++++++++++++++ tox/session.py | 678 ++++++++++++++++++++++++++++ tox/venv.py | 422 +++++++++++++++++ 16 files changed, 2204 insertions(+), 2222 deletions(-) delete mode 100644 tox/_cmdline.py delete mode 100644 tox/_config.py delete mode 100644 tox/_exception.py delete mode 100644 tox/_venv.py create mode 100644 tox/config.py create mode 100644 tox/session.py create mode 100644 tox/venv.py diff --git a/CHANGELOG b/CHANGELOG index ca59f7c..0df3656 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -48,6 +48,10 @@ for testenv sections. Can be used from plugins through the tox_add_option hook. +- rename internal files -- tox offers no external API except for the + experimental plugin hooks, use tox internals at your own risk. + + 1.9.2 ----------- diff --git a/setup.py b/setup.py index 8ea49c8..8edec31 100644 --- a/setup.py +++ b/setup.py @@ -26,7 +26,7 @@ def main(): description='virtualenv-based automation of test activities', long_description=open("README.rst").read(), url='http://tox.testrun.org/', - version='2.0.0.dev4', + version='2.0.0.dev5', license='http://opensource.org/licenses/MIT', platforms=['unix', 'linux', 'osx', 'cygwin', 'win32'], author='holger krekel', diff --git a/tests/test_config.py b/tests/test_config.py index e9c941a..cce293e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -4,9 +4,9 @@ from textwrap import dedent import py import pytest import tox -import tox._config -from tox._config import * # noqa -from tox._venv import VirtualEnv +import tox.config +from tox.config import * # noqa +from tox.venv import VirtualEnv class TestVenvConfig: @@ -1321,12 +1321,12 @@ class TestHashseedOption: """ if make_hashseed is None: make_hashseed = lambda: '123456789' - original_make_hashseed = tox._config.make_hashseed - tox._config.make_hashseed = make_hashseed + original_make_hashseed = tox.config.make_hashseed + tox.config.make_hashseed = make_hashseed try: config = newconfig(args, tox_ini) finally: - tox._config.make_hashseed = original_make_hashseed + tox.config.make_hashseed = original_make_hashseed return config.envconfigs def _get_envconfig(self, newconfig, args=None, tox_ini=None): diff --git a/tests/test_interpreters.py b/tests/test_interpreters.py index 01d6508..d2cfa27 100644 --- a/tests/test_interpreters.py +++ b/tests/test_interpreters.py @@ -3,7 +3,7 @@ import os import pytest from tox.interpreters import * # noqa -from tox._config import get_plugin_manager +from tox.config import get_plugin_manager @pytest.fixture diff --git a/tests/test_venv.py b/tests/test_venv.py index b32affd..d00b655 100644 --- a/tests/test_venv.py +++ b/tests/test_venv.py @@ -3,8 +3,8 @@ import tox import pytest import os import sys -import tox._config -from tox._venv import * # noqa +import tox.config +from tox.venv import * # noqa from tox.interpreters import NoInterpreterInfo # def test_global_virtualenv(capfd): @@ -253,14 +253,14 @@ def test_install_recreate(newmocksession, tmpdir): def test_test_hashseed_is_in_output(newmocksession): - original_make_hashseed = tox._config.make_hashseed - tox._config.make_hashseed = lambda: '123456789' + original_make_hashseed = tox.config.make_hashseed + tox.config.make_hashseed = lambda: '123456789' try: mocksession = newmocksession([], ''' [testenv] ''') finally: - tox._config.make_hashseed = original_make_hashseed + tox.config.make_hashseed = original_make_hashseed venv = mocksession.getenv('python') venv.update() venv.test() @@ -620,7 +620,7 @@ def test_sethome_only_on_option(newmocksession, monkeypatch): mocksession = newmocksession([], "") venv = mocksession.getenv('python') action = mocksession.newaction(venv, "qwe", []) - monkeypatch.setattr(tox._venv, "hack_home_env", None) + monkeypatch.setattr(tox.venv, "hack_home_env", None) venv._install(["x"], action=action) @@ -636,7 +636,7 @@ def test_sethome_works_on_option(newmocksession, monkeypatch): def test_hack_home_env(tmpdir): - from tox._venv import hack_home_env + from tox.venv import hack_home_env env = hack_home_env(tmpdir, "http://index") assert env["HOME"] == str(tmpdir) assert env["PIP_INDEX_URL"] == "http://index" @@ -650,7 +650,7 @@ def test_hack_home_env(tmpdir): def test_hack_home_env_passthrough(tmpdir, monkeypatch): - from tox._venv import hack_home_env + from tox.venv import hack_home_env env = hack_home_env(tmpdir, "http://index") monkeypatch.setattr(os, "environ", env) diff --git a/tests/test_z_cmdline.py b/tests/test_z_cmdline.py index b5cbc4e..2461790 100644 --- a/tests/test_z_cmdline.py +++ b/tests/test_z_cmdline.py @@ -9,8 +9,8 @@ except ImportError: pytest_plugins = "pytester" -from tox._cmdline import Session -from tox._config import parseconfig +from tox.session import Session +from tox.config import parseconfig def test_report_protocol(newconfig): diff --git a/tox/__init__.py b/tox/__init__.py index 7e71062..a23225b 100644 --- a/tox/__init__.py +++ b/tox/__init__.py @@ -1,5 +1,5 @@ # -__version__ = '2.0.0.dev4' +__version__ = '2.0.0.dev5' from .hookspecs import hookspec, hookimpl # noqa @@ -24,4 +24,4 @@ class exception: class MissingDependency(Error): """ a dependency could not be found or determined. """ -from tox._cmdline import main as cmdline # noqa +from tox.session import main as cmdline # noqa diff --git a/tox/__main__.py b/tox/__main__.py index d60aab2..e28d746 100644 --- a/tox/__main__.py +++ b/tox/__main__.py @@ -1,3 +1,4 @@ -from tox._cmdline import main +from tox.session import main -main() +if __name__ == "__main__": + main() diff --git a/tox/_cmdline.py b/tox/_cmdline.py deleted file mode 100644 index 0c2f9d6..0000000 --- a/tox/_cmdline.py +++ /dev/null @@ -1,673 +0,0 @@ -""" -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 main(args=None): - try: - config = parseconfig(args) - if config.option.help: - show_help(config) - raise SystemExit(0) - elif config.option.helpini: - show_help_ini(config) - raise SystemExit(0) - retcode = Session(config).runcommand() - raise SystemExit(retcode) - except KeyboardInterrupt: - raise SystemExit(2) - - -def show_help(config): - tw = py.io.TerminalWriter() - tw.write(config._parser.format_help()) - tw.line() - - -def show_help_ini(config): - tw = py.io.TerminalWriter() - tw.sep("-", "per-testenv attributes") - for env_attr in config._testenv_attr: - tw.line("%-15s %-8s default: %s" % - (env_attr.name, "<" + env_attr.type + ">", env_attr.default), bold=True) - tw.line(env_attr.help) - tw.line() - - -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.report = session.report - self.args = args - self.id = venv and venv.envconfig.envname or "tox" - self._popenlist = [] - if self.venv: - self.venvname = self.venv.name - else: - self.venvname = "GLOB" - if msg == "runtests": - cat = "test" - else: - cat = "setup" - envlog = session.resultlog.get_envlog(self.venvname) - self.commandlog = envlog.get_commandlog(cat) - - def __enter__(self): - self.report.logaction_start(self) - - def __exit__(self, *args): - self.report.logaction_finish(self) - - def setactivity(self, name, msg): - self.activity = name - self.report.verbosity0("%s %s: %s" % (self.venvname, name, msg), bold=True) - - def info(self, name, msg): - self.report.verbosity1("%s %s: %s" % (self.venvname, name, msg), bold=True) - - def _initlogpath(self, actionid): - if self.venv: - logdir = self.venv.envconfig.envlogdir - else: - logdir = self.session.config.logdir - try: - l = logdir.listdir("%s-*" % actionid) - except py.error.ENOENT: - logdir.ensure(dir=1) - l = [] - num = len(l) - path = logdir.join("%s-%s.log" % (actionid, num)) - f = path.open('w') - f.flush() - 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(self.id) - fout.write("actionid: %s\nmsg: %s\ncmdargs: %r\nenv: %s\n\n" % ( - self.id, self.msg, args, env)) - fout.flush() - self.popen_outpath = outpath = py.path.local(fout.name) - fin = outpath.open() - fin.read() # 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() - try: - popen = self._popen(args, cwd, env=env, - stdout=stdout, stderr=STDOUT) - except OSError as e: - self.report.error("invocation failed (errno %d), args: %s, cwd: %s" % - (e.errno, args, cwd)) - raise - popen.outpath = outpath - popen.args = [str(x) for x in args] - popen.cwd = cwd - popen.action = self - self._popenlist.append(popen) - try: - self.report.logpopen(popen, env=env) - try: - 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 = fin.read(1) - if data: - sys.stdout.write(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 - sys.stdout.flush() - last_time = now() - elif popen.poll() is not None: - if popen.stdout is not None: - popen.stdout.close() - break - else: - py.std.time.sleep(0.1) - fin.seek(fin_pos) - fin.close() - else: - out, err = popen.communicate() - except KeyboardInterrupt: - self.report.keyboard_interrupt() - popen.wait() - raise KeyboardInterrupt() - ret = popen.wait() - finally: - self._popenlist.remove(popen) - if ret and not ignore_ret: - invoked = " ".join(map(str, popen.args)) - if outpath: - self.report.error("invocation failed (exit code %d), logfile: %s" % - (ret, outpath)) - out = outpath.read() - self.report.error(out) - if hasattr(self, "commandlog"): - self.commandlog.add_command(popen.args, out, ret) - raise tox.exception.InvocationError( - "%s (see %s)" % (invoked, outpath), ret) - else: - raise tox.exception.InvocationError("%r" % (invoked, ), ret) - if not out and outpath: - out = outpath.read() - 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) - newargs.append(str(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.getcommandpath())] + 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), - universal_newlines=True, - stdout=stdout, stderr=stderr, env=env) - - -class Reporter(object): - actionchar = "-" - - def __init__(self, session): - self.tw = py.io.TerminalWriter() - self.session = session - self._reportedlines = [] - # self.cumulated_time = 0.0 - - 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,)) - else: - 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) - - def startsummary(self): - self.tw.sep("_", "summary") - - def info(self, msg): - if self.session.config.option.verbosity >= 2: - self.logline(msg) - - def using(self, msg): - if self.session.config.option.verbosity >= 1: - self.logline("using %s" % (msg,), bold=True) - - def keyboard_interrupt(self): - self.error("KEYBOARDINTERRUPT") - -# 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 += " " - self.tw.write(name, bold=True) - self.tw.write(value) - self.tw.line() - - 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) - self.tw.line("%s" % msg, **opts) - - def verbosity0(self, msg, **opts): - if self.session.config.option.verbosity >= 0: - self.logline("%s" % msg, **opts) - - def verbosity1(self, msg, **opts): - if self.session.config.option.verbosity >= 1: - self.logline("%s" % msg, **opts) - - def verbosity2(self, msg, **opts): - if self.session.config.option.verbosity >= 2: - self.logline("%s" % msg, **opts) - - # def log(self, msg): - # py.builtin.print_(msg, file=sys.stderr) - - -class Session: - - def __init__(self, config, popen=subprocess.Popen, Report=Reporter): - self.config = config - self.popen = popen - self.resultlog = ResultLog() - self.report = Report(self) - self.make_emptydir(config.logdir) - config.logdir.ensure(dir=1) - # self.report.using("logdir %s" %(self.config.logdir,)) - self.report.using("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) - self._actions = [] - - def _makevenv(self, name): - envconfig = self.config.envconfigs.get(name, None) - if envconfig is None: - self.report.error("unknown environment %r" % name) - raise LookupError(name) - venv = VirtualEnv(envconfig=envconfig, session=self) - self._name2venv[name] = venv - return venv - - 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): - self.report.using("tox-%s from %s" % (tox.__version__, tox.__file__)) - if self.config.minversion: - minversion = NormalizedVersion(self.config.minversion) - toxversion = NormalizedVersion(tox.__version__) - if toxversion < minversion: - self.report.error( - "tox version is %s, required is at least %s" % ( - toxversion, minversion)) - raise SystemExit(1) - 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(): - self.report.error("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("setup.py") - 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(): - self.report.error( - 'setup.py is empty' - ) - raise SystemExit(1) - self.report.error( - 'No dist directory found. Please check setup.py, e.g with:\n' - ' python setup.py sdist' - ) - raise SystemExit(1) - - def make_emptydir(self, path): - if path.check(): - self.report.info(" removing %s" % path) - py.std.shutil.rmtree(str(path), ignore_errors=True) - path.ensure(dir=1) - - def setupenv(self, venv): - action = self.newaction(venv, "getenv", venv.envconfig.envdir) - with action: - venv.status = 0 - envlog = self.resultlog.get_envlog(venv.name) - 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 - self.report.error(str(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 - - 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 - - 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) - self.report.info("using package %r, skipping 'sdist' activity " % - str(path)) - else: - try: - path = self._makesdist() - except tox.exception.InvocationError: - v = sys.exc_info()[1] - self.report.error("FAIL could not package project - v = %r" % - v) - return - sdistfile = self.config.distshare.join(path.basename) - if sdistfile != path: - self.report.info("copying new sdistfile to %r" % - str(sdistfile)) - try: - sdistfile.dirpath().ensure(dir=1) - except py.error.Error: - self.report.warning("could not copy distfile to %s" % - sdistfile.dirpath()) - else: - path.copy(sdistfile) - return path - - def subcommand_test(self): - if self.config.skipsdist: - self.report.info("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 not venv.matching_platform(): - venv.status = "platform mismatch" - continue # we simply omit non-matching platforms - 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: - pip = venv.getcommandpath("pip") - # we can't really call internal helpers here easily :/ - # output = venv._pcall([str(pip), "freeze"], - # cwd=self.config.toxinidir, - # action=action) - output = py.process.cmdexec("%s freeze" % (pip)) - packages = output.strip().split("\n") - action.setactivity("installed", ",".join(packages)) - envlog = self.resultlog.get_envlog(venv.name) - 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 - venv.test(redirect=redirect) - else: - venv.status = "skipped tests" - - def _summary(self): - self.report.startsummary() - 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: - self.report.skip(msg) - else: - retcode = 1 - self.report.error(msg) - elif status == "platform mismatch": - msg = " %s: %s" % (venv.envconfig.envname, str(status)) - self.report.verbosity1(msg) - elif status and status != "skipped tests": - msg = " %s: %s" % (venv.envconfig.envname, str(status)) - self.report.error(msg) - retcode = 1 - else: - if not status: - status = "commands succeeded" - self.report.good(" %s: %s" % (venv.envconfig.envname, status)) - if not retcode: - self.report.good(" congratulations :)") - - path = self.config.option.resultjson - if path: - path = py.path.local(path) - path.write(self.resultlog.dumps_json()) - self.report.line("wrote json report at: %s" % path) - return retcode - - def showconfig(self): - self.info_versions() - self.report.keyvalue("config-file:", self.config.option.configfile) - self.report.keyvalue("toxinipath: ", self.config.toxinipath) - self.report.keyvalue("toxinidir: ", self.config.toxinidir) - self.report.keyvalue("toxworkdir: ", self.config.toxworkdir) - self.report.keyvalue("setupdir: ", self.config.setupdir) - self.report.keyvalue("distshare: ", self.config.distshare) - self.report.keyvalue("skipsdist: ", self.config.skipsdist) - self.report.tw.line() - for envconfig in self.config.envconfigs.values(): - self.report.line("[testenv:%s]" % envconfig.envname, bold=True) - self.report.line(" basepython=%s" % envconfig.basepython) - self.report.line(" pythoninfo=%s" % (envconfig.python_info,)) - self.report.line(" envpython=%s" % envconfig.envpython) - self.report.line(" envtmpdir=%s" % envconfig.envtmpdir) - self.report.line(" envbindir=%s" % envconfig.envbindir) - self.report.line(" envlogdir=%s" % envconfig.envlogdir) - self.report.line(" changedir=%s" % envconfig.changedir) - self.report.line(" args_are_path=%s" % envconfig.args_are_paths) - self.report.line(" install_command=%s" % - envconfig.install_command) - self.report.line(" commands=") - for command in envconfig.commands: - self.report.line(" %s" % command) - self.report.line(" deps=%s" % envconfig.deps) - self.report.line(" envdir= %s" % envconfig.envdir) - self.report.line(" downloadcache=%s" % envconfig.downloadcache) - self.report.line(" usedevelop=%s" % envconfig.usedevelop) - self.report.line(" setenv=%s" % envconfig.setenv) - self.report.line(" passenv=%s" % envconfig.passenv) - - def showenvs(self): - for env in self.config.envlist: - self.report.line("%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()) - self.report.keyvalue("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()) - self.report.info("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: - self.report.warning("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 = py.std.re.compile("[\w_\-\+\.]+-(.*)(\.zip|\.tar.gz)") - - -def getversion(basename): - m = _rex_getversion.match(basename) - if m is None: - return None - version = m.group(1) - try: - return NormalizedVersion(version) - except IrrationalVersionError: - return None diff --git a/tox/_config.py b/tox/_config.py deleted file mode 100644 index 85ff822..0000000 --- a/tox/_config.py +++ /dev/null @@ -1,1072 +0,0 @@ -import argparse -import os -import random -from fnmatch import fnmatchcase -import sys -import re -import shlex -import string -import pkg_resources -import itertools -import pluggy - -import tox.interpreters -from tox import hookspecs - -import py - -import tox - -iswin32 = sys.platform == "win32" - -default_factors = {'jython': 'jython', 'pypy': 'pypy', 'pypy3': 'pypy3', - 'py': sys.executable} -for version in '26,27,32,33,34,35,36'.split(','): - default_factors['py' + version] = 'python%s.%s' % tuple(version) - -hookimpl = pluggy.HookimplMarker("tox") - - -def get_plugin_manager(): - # initialize plugin manager - pm = pluggy.PluginManager("tox") - pm.add_hookspecs(hookspecs) - pm.register(tox._config) - pm.register(tox.interpreters) - pm.load_setuptools_entrypoints("tox") - pm.check_pending() - return pm - - -class MyParser: - def __init__(self): - self.argparser = argparse.ArgumentParser( - description="tox options", add_help=False) - self._testenv_attr = [] - - def add_argument(self, *args, **kwargs): - return self.argparser.add_argument(*args, **kwargs) - - def add_testenv_attribute(self, name, type, help, default=None, postprocess=None): - self._testenv_attr.append(VenvAttribute(name, type, default, help, postprocess)) - - def add_testenv_attribute_obj(self, obj): - assert hasattr(obj, "name") - assert hasattr(obj, "type") - assert hasattr(obj, "help") - assert hasattr(obj, "postprocess") - self._testenv_attr.append(obj) - - def parse_args(self, args): - return self.argparser.parse_args(args) - - def format_help(self): - return self.argparser.format_help() - - -class VenvAttribute: - def __init__(self, name, type, default, help, postprocess): - self.name = name - self.type = type - self.default = default - self.help = help - self.postprocess = postprocess - - -class DepOption: - name = "deps" - type = "line-list" - help = "each line specifies a dependency in pip/setuptools format." - default = () - - def postprocess(self, config, reader, section_val): - deps = [] - for depline in section_val: - m = re.match(r":(\w+):\s*(\S+)", depline) - if m: - iname, name = m.groups() - ixserver = config.indexserver[iname] - else: - name = depline.strip() - ixserver = None - name = self._replace_forced_dep(name, config) - deps.append(DepConfig(name, ixserver)) - return deps - - def _replace_forced_dep(self, name, config): - """ - Override the given dependency config name taking --force-dep-version - option into account. - - :param name: dep config, for example ["pkg==1.0", "other==2.0"]. - :param config: Config instance - :return: the new dependency that should be used for virtual environments - """ - if not config.option.force_dep: - return name - for forced_dep in config.option.force_dep: - if self._is_same_dep(forced_dep, name): - return forced_dep - return name - - @classmethod - def _is_same_dep(cls, dep1, dep2): - """ - Returns True if both dependency definitions refer to the - same package, even if versions differ. - """ - dep1_name = pkg_resources.Requirement.parse(dep1).project_name - dep2_name = pkg_resources.Requirement.parse(dep2).project_name - return dep1_name == dep2_name - - -class PosargsOption: - name = "args_are_paths" - type = "bool" - default = True - help = "treat positional args in commands as paths" - - def postprocess(self, config, reader, section_val): - args = config.option.args - if args: - if section_val: - args = [] - for arg in config.option.args: - if arg: - origpath = config.invocationcwd.join(arg, abs=True) - if origpath.check(): - arg = reader.getpath("changedir", ".").bestrelpath(origpath) - args.append(arg) - reader.addsubstitutions(args) - return section_val - - -class InstallcmdOption: - name = "install_command" - type = "argv" - default = "pip install {opts} {packages}" - help = "install command for dependencies and package under test." - - def postprocess(self, config, reader, section_val): - if '{packages}' not in section_val: - raise tox.exception.ConfigError( - "'install_command' must contain '{packages}' substitution") - return section_val - - -def parseconfig(args=None): - """ - :param list[str] args: Optional list of arguments. - :type pkg: str - :rtype: :class:`Config` - :raise SystemExit: toxinit file is not found - """ - - pm = get_plugin_manager() - - if args is None: - args = sys.argv[1:] - - # prepare command line options - parser = MyParser() - pm.hook.tox_addoption(parser=parser) - - # parse command line options - option = parser.parse_args(args) - interpreters = tox.interpreters.Interpreters(hook=pm.hook) - config = Config(pluginmanager=pm, option=option, interpreters=interpreters) - config._parser = parser - config._testenv_attr = parser._testenv_attr - - # parse ini file - basename = config.option.configfile - if os.path.isabs(basename): - inipath = py.path.local(basename) - else: - for path in py.path.local().parts(reverse=True): - inipath = path.join(basename) - if inipath.check(): - break - else: - feedback("toxini file %r not found" % (basename), sysexit=True) - try: - parseini(config, inipath) - except tox.exception.InterpreterNotFound: - exn = sys.exc_info()[1] - # Use stdout to match test expectations - py.builtin.print_("ERROR: " + str(exn)) - - # post process config object - pm.hook.tox_configure(config=config) - - return config - - -def feedback(msg, sysexit=False): - py.builtin.print_("ERROR: " + msg, file=sys.stderr) - if sysexit: - raise SystemExit(1) - - -class VersionAction(argparse.Action): - def __call__(self, argparser, *args, **kwargs): - version = tox.__version__ - py.builtin.print_("%s imported from %s" % (version, tox.__file__)) - raise SystemExit(0) - - -class CountAction(argparse.Action): - def __call__(self, parser, namespace, values, option_string=None): - if hasattr(namespace, self.dest): - setattr(namespace, self.dest, int(getattr(namespace, self.dest)) + 1) - else: - setattr(namespace, self.dest, 0) - - -@hookimpl -def tox_addoption(parser): - # formatter_class=argparse.ArgumentDefaultsHelpFormatter) - parser.add_argument("--version", nargs=0, action=VersionAction, - dest="version", - help="report version information to stdout.") - parser.add_argument("-h", "--help", action="store_true", dest="help", - help="show help about options") - parser.add_argument("--help-ini", "--hi", action="store_true", dest="helpini", - help="show help about ini-names") - parser.add_argument("-v", nargs=0, action=CountAction, default=0, - dest="verbosity", - help="increase verbosity of reporting output.") - parser.add_argument("--showconfig", action="store_true", - help="show configuration information for all environments. ") - parser.add_argument("-l", "--listenvs", action="store_true", - dest="listenvs", help="show list of test environments") - parser.add_argument("-c", action="store", default="tox.ini", - dest="configfile", - help="use the specified config file name.") - parser.add_argument("-e", action="append", dest="env", - metavar="envlist", - help="work against specified environments (ALL selects all).") - parser.add_argument("--notest", action="store_true", dest="notest", - help="skip invoking test commands.") - parser.add_argument("--sdistonly", action="store_true", dest="sdistonly", - help="only perform the sdist packaging activity.") - parser.add_argument("--installpkg", action="store", default=None, - metavar="PATH", - help="use specified package for installation into venv, instead of " - "creating an sdist.") - parser.add_argument("--develop", action="store_true", dest="develop", - help="install package in the venv using 'setup.py develop' via " - "'pip -e .'") - parser.add_argument("--set-home", action="store_true", dest="sethome", - help="(experimental) force creating a new $HOME for each test " - "environment and create .pydistutils.cfg|pip.conf files " - "if index servers are specified with tox. ") - parser.add_argument('-i', action="append", - dest="indexurl", metavar="URL", - help="set indexserver url (if URL is of form name=url set the " - "url for the 'name' indexserver, specifically)") - parser.add_argument("--pre", action="store_true", dest="pre", - help="install pre-releases and development versions of dependencies. " - "This will pass the --pre option to install_command " - "(pip by default).") - parser.add_argument("-r", "--recreate", action="store_true", - dest="recreate", - help="force recreation of virtual environments") - parser.add_argument("--result-json", action="store", - dest="resultjson", metavar="PATH", - help="write a json file with detailed information about " - "all commands and results involved. This will turn off " - "pass-through output from running test commands which is " - "instead captured into the json result file.") - - # We choose 1 to 4294967295 because it is the range of PYTHONHASHSEED. - parser.add_argument("--hashseed", action="store", - metavar="SEED", default=None, - help="set PYTHONHASHSEED to SEED before running commands. " - "Defaults to a random integer in the range [1, 4294967295] " - "([1, 1024] on Windows). " - "Passing 'noset' suppresses this behavior.") - parser.add_argument("--force-dep", action="append", - metavar="REQ", default=None, - help="Forces a certain version of one of the dependencies " - "when configuring the virtual environment. REQ Examples " - "'pytest<2.7' or 'django>=1.6'.") - parser.add_argument("--sitepackages", action="store_true", - help="override sitepackages setting to True in all envs") - parser.add_argument("--skip-missing-interpreters", action="store_true", - help="don't fail tests for missing interpreters") - - parser.add_argument("args", nargs="*", - help="additional arguments available to command positional substitution") - - # add various core venv interpreter attributes - - parser.add_testenv_attribute( - name="envdir", type="path", default="{toxworkdir}/{envname}", - help="venv directory") - - parser.add_testenv_attribute( - name="envtmpdir", type="path", default="{envdir}/tmp", - help="venv temporary directory") - - parser.add_testenv_attribute( - name="envlogdir", type="path", default="{envdir}/log", - help="venv log directory") - - def downloadcache(config, reader, section_val): - if section_val: - # env var, if present, takes precedence - downloadcache = os.environ.get("PIP_DOWNLOAD_CACHE", section_val) - return py.path.local(downloadcache) - - parser.add_testenv_attribute( - name="downloadcache", type="string", default=None, postprocess=downloadcache, - help="(deprecated) set PIP_DOWNLOAD_CACHE.") - - parser.add_testenv_attribute( - name="changedir", type="path", default="{toxinidir}", - help="directory to change to when running commands") - - parser.add_testenv_attribute_obj(PosargsOption()) - - parser.add_testenv_attribute( - name="skip_install", type="bool", default=False, - help="Do not install the current package. This can be used when " - "you need the virtualenv management but do not want to install " - "the current package") - - parser.add_testenv_attribute( - name="ignore_errors", type="bool", default=False, - help="if set to True all commands will be executed irrespective of their " - "result error status.") - - def recreate(config, reader, section_val): - if config.option.recreate: - return True - return section_val - - parser.add_testenv_attribute( - name="recreate", type="bool", default=False, postprocess=recreate, - help="always recreate this test environment.") - - def setenv(config, reader, section_val): - setenv = section_val - if "PYTHONHASHSEED" not in setenv and config.hashseed is not None: - setenv['PYTHONHASHSEED'] = config.hashseed - return setenv - - parser.add_testenv_attribute( - name="setenv", type="dict", postprocess=setenv, - help="list of X=Y lines with environment variable settings") - - def passenv(config, reader, section_val): - passenv = set(["PATH"]) - - # we ensure that tmp directory settings are passed on - # we could also set it to the per-venv "envtmpdir" - # but this leads to very long paths when run with jenkins - # so we just pass it on by default for now. - if sys.platform == "win32": - passenv.add("SYSTEMROOT") # needed for python's crypto module - passenv.add("PATHEXT") # needed for discovering executables - passenv.add("TEMPDIR") - passenv.add("TMP") - else: - passenv.add("TMPDIR") - for spec in section_val: - for name in os.environ: - if fnmatchcase(name.upper(), spec.upper()): - passenv.add(name) - return passenv - - parser.add_testenv_attribute( - name="passenv", type="space-separated-list", postprocess=passenv, - help="environment variables names which shall be passed " - "from tox invocation to test environment when executing commands.") - - parser.add_testenv_attribute( - name="whitelist_externals", type="line-list", - help="each lines specifies a path or basename for which tox will not warn " - "about it coming from outside the test environment.") - - parser.add_testenv_attribute( - name="platform", type="string", default=".*", - help="regular expression which must match against ``sys.platform``. " - "otherwise testenv will be skipped.") - - def sitepackages(config, reader, section_val): - return config.option.sitepackages or section_val - - parser.add_testenv_attribute( - name="sitepackages", type="bool", default=False, postprocess=sitepackages, - help="Set to ``True`` if you want to create virtual environments that also " - "have access to globally installed packages.") - - def pip_pre(config, reader, section_val): - return config.option.pre or section_val - - parser.add_testenv_attribute( - name="pip_pre", type="bool", default=False, postprocess=pip_pre, - help="If ``True``, adds ``--pre`` to the ``opts`` passed to " - "the install command. ") - - def develop(config, reader, section_val): - return not config.option.installpkg and (section_val or config.option.develop) - - parser.add_testenv_attribute( - name="usedevelop", type="bool", postprocess=develop, default=False, - help="install package in develop/editable mode") - - def basepython_default(config, reader, section_val): - if section_val is None: - for f in reader.factors: - if f in default_factors: - return default_factors[f] - return sys.executable - return str(section_val) - - parser.add_testenv_attribute( - name="basepython", type="string", default=None, postprocess=basepython_default, - help="executable name or path of interpreter used to create a " - "virtual test environment.") - - parser.add_testenv_attribute_obj(InstallcmdOption()) - parser.add_testenv_attribute_obj(DepOption()) - - parser.add_testenv_attribute( - name="commands", type="argvlist", default="", - help="each line specifies a test command and can use substitution.") - - -class Config(object): - def __init__(self, pluginmanager, option, interpreters): - self.envconfigs = {} - self.invocationcwd = py.path.local() - self.interpreters = interpreters - self.pluginmanager = pluginmanager - self.option = option - - @property - def homedir(self): - homedir = get_homedir() - if homedir is None: - homedir = self.toxinidir # XXX good idea? - return homedir - - -class VenvConfig: - def __init__(self, envname, config): - self.envname = envname - self.config = config - - @property - def envbindir(self): - if (sys.platform == "win32" - and "jython" not in self.basepython - and "pypy" not in self.basepython): - return self.envdir.join("Scripts") - else: - return self.envdir.join("bin") - - @property - def envpython(self): - if "jython" in str(self.basepython): - name = "jython" - else: - name = "python" - return self.envbindir.join(name) - - # no @property to avoid early calling (see callable(subst[key]) checks) - def envsitepackagesdir(self): - self.getsupportedinterpreter() # for throwing exceptions - x = self.config.interpreters.get_sitepackagesdir( - info=self.python_info, - envdir=self.envdir) - return x - - @property - def python_info(self): - return self.config.interpreters.get_info(envconfig=self) - - def getsupportedinterpreter(self): - if sys.platform == "win32" and self.basepython and \ - "jython" in self.basepython: - raise tox.exception.UnsupportedInterpreter( - "Jython/Windows does not support installing scripts") - info = self.config.interpreters.get_info(envconfig=self) - if not info.executable: - raise tox.exception.InterpreterNotFound(self.basepython) - if not info.version_info: - raise tox.exception.InvocationError( - 'Failed to get version_info for %s: %s' % (info.name, info.err)) - if info.version_info < (2, 6): - raise tox.exception.UnsupportedInterpreter( - "python2.5 is not supported anymore, sorry") - return info.executable - - -testenvprefix = "testenv:" - - -def get_homedir(): - try: - return py.path.local._gethomedir() - except Exception: - return None - - -def make_hashseed(): - max_seed = 4294967295 - if sys.platform == 'win32': - max_seed = 1024 - return str(random.randint(1, max_seed)) - - -class parseini: - def __init__(self, config, inipath): - config.toxinipath = inipath - config.toxinidir = config.toxinipath.dirpath() - - self._cfg = py.iniconfig.IniConfig(config.toxinipath) - config._cfg = self._cfg - self.config = config - ctxname = getcontextname() - if ctxname == "jenkins": - reader = SectionReader("tox:jenkins", self._cfg, fallbacksections=['tox']) - distshare_default = "{toxworkdir}/distshare" - elif not ctxname: - reader = SectionReader("tox", self._cfg) - distshare_default = "{homedir}/.tox/distshare" - else: - raise ValueError("invalid context") - - if config.option.hashseed is None: - hashseed = make_hashseed() - elif config.option.hashseed == 'noset': - hashseed = None - else: - hashseed = config.option.hashseed - config.hashseed = hashseed - - reader.addsubstitutions(toxinidir=config.toxinidir, - homedir=config.homedir) - config.toxworkdir = reader.getpath("toxworkdir", "{toxinidir}/.tox") - config.minversion = reader.getstring("minversion", None) - - if not config.option.skip_missing_interpreters: - config.option.skip_missing_interpreters = \ - reader.getbool("skip_missing_interpreters", False) - - # determine indexserver dictionary - config.indexserver = {'default': IndexServerConfig('default')} - prefix = "indexserver" - for line in reader.getlist(prefix): - name, url = map(lambda x: x.strip(), line.split("=", 1)) - config.indexserver[name] = IndexServerConfig(name, url) - - override = False - if config.option.indexurl: - for urldef in config.option.indexurl: - m = re.match(r"\W*(\w+)=(\S+)", urldef) - if m is None: - url = urldef - name = "default" - else: - name, url = m.groups() - if not url: - url = None - if name != "ALL": - config.indexserver[name].url = url - else: - override = url - # let ALL override all existing entries - if override: - for name in config.indexserver: - config.indexserver[name] = IndexServerConfig(name, override) - - reader.addsubstitutions(toxworkdir=config.toxworkdir) - config.distdir = reader.getpath("distdir", "{toxworkdir}/dist") - reader.addsubstitutions(distdir=config.distdir) - config.distshare = reader.getpath("distshare", distshare_default) - reader.addsubstitutions(distshare=config.distshare) - config.sdistsrc = reader.getpath("sdistsrc", None) - config.setupdir = reader.getpath("setupdir", "{toxinidir}") - config.logdir = config.toxworkdir.join("log") - - config.envlist, all_envs = self._getenvdata(reader) - - # factors used in config or predefined - known_factors = self._list_section_factors("testenv") - known_factors.update(default_factors) - known_factors.add("python") - - # factors stated in config envlist - stated_envlist = reader.getstring("envlist", replace=False) - if stated_envlist: - for env in _split_env(stated_envlist): - known_factors.update(env.split('-')) - - # configure testenvs - for name in all_envs: - section = testenvprefix + name - factors = set(name.split('-')) - if section in self._cfg or factors <= known_factors: - config.envconfigs[name] = \ - self.make_envconfig(name, section, reader._subs, config) - - all_develop = all(name in config.envconfigs - and config.envconfigs[name].usedevelop - for name in config.envlist) - - config.skipsdist = reader.getbool("skipsdist", all_develop) - - def _list_section_factors(self, section): - factors = set() - if section in self._cfg: - for _, value in self._cfg[section].items(): - exprs = re.findall(r'^([\w{}\.,-]+)\:\s+', value, re.M) - factors.update(*mapcat(_split_factor_expr, exprs)) - return factors - - def make_envconfig(self, name, section, subs, config): - vc = VenvConfig(config=config, envname=name) - factors = set(name.split('-')) - reader = SectionReader(section, self._cfg, fallbacksections=["testenv"], - factors=factors) - reader.addsubstitutions(**subs) - reader.addsubstitutions(envname=name) - reader.vc = vc - - for env_attr in config._testenv_attr: - atype = env_attr.type - if atype in ("bool", "path", "string", "dict", "argv", "argvlist"): - meth = getattr(reader, "get" + atype) - res = meth(env_attr.name, env_attr.default) - elif atype == "space-separated-list": - res = reader.getlist(env_attr.name, sep=" ") - elif atype == "line-list": - res = reader.getlist(env_attr.name, sep="\n") - else: - raise ValueError("unknown type %r" % (atype,)) - - if env_attr.postprocess: - res = env_attr.postprocess(config, reader, res) - setattr(vc, env_attr.name, res) - - if atype == "path": - reader.addsubstitutions(**{env_attr.name: res}) - - if env_attr.name == "install_command": - reader.addsubstitutions(envbindir=vc.envbindir, envpython=vc.envpython, - envsitepackagesdir=vc.envsitepackagesdir) - return vc - - def _getenvdata(self, reader): - envstr = self.config.option.env \ - or os.environ.get("TOXENV") \ - or reader.getstring("envlist", replace=False) \ - or [] - envlist = _split_env(envstr) - - # collect section envs - all_envs = set(envlist) - set(["ALL"]) - for section in self._cfg: - if section.name.startswith(testenvprefix): - all_envs.add(section.name[len(testenvprefix):]) - if not all_envs: - all_envs.add("python") - - if not envlist or "ALL" in envlist: - envlist = sorted(all_envs) - - return envlist, all_envs - - -def _split_env(env): - """if handed a list, action="append" was used for -e """ - if not isinstance(env, list): - env = [env] - return mapcat(_expand_envstr, env) - - -def _split_factor_expr(expr): - partial_envs = _expand_envstr(expr) - return [set(e.split('-')) for e in partial_envs] - - -def _expand_envstr(envstr): - # split by commas not in groups - tokens = re.split(r'((?:\{[^}]+\})+)|,', envstr) - envlist = [''.join(g).strip() - for k, g in itertools.groupby(tokens, key=bool) if k] - - def expand(env): - tokens = re.split(r'\{([^}]+)\}', env) - parts = [token.split(',') for token in tokens] - return [''.join(variant) for variant in itertools.product(*parts)] - - return mapcat(expand, envlist) - - -def mapcat(f, seq): - return list(itertools.chain.from_iterable(map(f, seq))) - - -class DepConfig: - def __init__(self, name, indexserver=None): - self.name = name - self.indexserver = indexserver - - def __str__(self): - if self.indexserver: - if self.indexserver.name == "default": - return self.name - return ":%s:%s" % (self.indexserver.name, self.name) - return str(self.name) - __repr__ = __str__ - - -class IndexServerConfig: - def __init__(self, name, url=None): - self.name = name - self.url = url - - -#: Check value matches substitution form -#: of referencing value from other section. E.g. {[base]commands} -is_section_substitution = re.compile("{\[[^{}\s]+\]\S+?}").match - - -RE_ITEM_REF = re.compile( - r''' - (?[^[:{}]+):)? # optional sub_type for special rules - (?P[^{}]*) # substitution key - [}] - ''', - re.VERBOSE) - - -class SectionReader: - def __init__(self, section_name, cfgparser, fallbacksections=None, factors=()): - self.section_name = section_name - self._cfg = cfgparser - self.fallbacksections = fallbacksections or [] - self.factors = factors - self._subs = {} - self._subststack = [] - - def addsubstitutions(self, _posargs=None, **kw): - self._subs.update(kw) - if _posargs: - self.posargs = _posargs - - def getpath(self, name, defaultpath): - toxinidir = self._subs['toxinidir'] - path = self.getstring(name, defaultpath) - if path is None: - return path - return toxinidir.join(path, abs=True) - - def getlist(self, name, sep="\n"): - s = self.getstring(name, None) - if s is None: - return [] - return [x.strip() for x in s.split(sep) if x.strip()] - - def getdict(self, name, default=None, sep="\n"): - s = self.getstring(name, None) - if s is None: - return default or {} - - value = {} - for line in s.split(sep): - if line.strip(): - name, rest = line.split('=', 1) - value[name.strip()] = rest.strip() - - return value - - def getbool(self, name, default=None): - s = self.getstring(name, default) - if not s: - s = default - if s is None: - raise KeyError("no config value [%s] %s found" % ( - self.section_name, name)) - - if not isinstance(s, bool): - if s.lower() == "true": - s = True - elif s.lower() == "false": - s = False - else: - raise tox.exception.ConfigError( - "boolean value %r needs to be 'True' or 'False'") - return s - - def getargvlist(self, name, default=""): - s = self.getstring(name, default, replace=False) - return _ArgvlistReader.getargvlist(self, s) - - def getargv(self, name, default=""): - return self.getargvlist(name, default)[0] - - def getstring(self, name, default=None, replace=True): - x = None - for s in [self.section_name] + self.fallbacksections: - try: - x = self._cfg[s][name] - break - except KeyError: - continue - - if x is None: - x = default - else: - x = self._apply_factors(x) - - if replace and x and hasattr(x, 'replace'): - self._subststack.append((self.section_name, name)) - try: - x = self._replace(x) - finally: - assert self._subststack.pop() == (self.section_name, name) - # print "getstring", self.section_name, name, "returned", repr(x) - return x - - def _apply_factors(self, s): - def factor_line(line): - m = re.search(r'^([\w{}\.,-]+)\:\s+(.+)', line) - if not m: - return line - - expr, line = m.groups() - if any(fs <= self.factors for fs in _split_factor_expr(expr)): - return line - - lines = s.strip().splitlines() - return '\n'.join(filter(None, map(factor_line, lines))) - - def _replace_env(self, match): - match_value = match.group('substitution_value') - if not match_value: - raise tox.exception.ConfigError( - 'env: requires an environment variable name') - - default = None - envkey_split = match_value.split(':', 1) - - if len(envkey_split) is 2: - envkey, default = envkey_split - else: - envkey = match_value - - if envkey not in os.environ and default is None: - raise tox.exception.ConfigError( - "substitution env:%r: unkown environment variable %r" % - (envkey, envkey)) - - return os.environ.get(envkey, default) - - def _substitute_from_other_section(self, key): - if key.startswith("[") and "]" in key: - i = key.find("]") - section, item = key[1:i], key[i + 1:] - if section in self._cfg and item in self._cfg[section]: - if (section, item) in self._subststack: - raise ValueError('%s already in %s' % ( - (section, item), self._subststack)) - x = str(self._cfg[section][item]) - self._subststack.append((section, item)) - try: - return self._replace(x) - finally: - self._subststack.pop() - - raise tox.exception.ConfigError( - "substitution key %r not found" % key) - - def _replace_substitution(self, match): - sub_key = match.group('substitution_value') - val = self._subs.get(sub_key, None) - if val is None: - val = self._substitute_from_other_section(sub_key) - if py.builtin.callable(val): - val = val() - return str(val) - - def _replace_match(self, match): - g = match.groupdict() - - # special case: opts and packages. Leave {opts} and - # {packages} intact, they are replaced manually in - # _venv.VirtualEnv.run_install_command. - sub_value = g['substitution_value'] - if sub_value in ('opts', 'packages'): - return '{%s}' % sub_value - - handlers = { - 'env': self._replace_env, - None: self._replace_substitution, - } - try: - sub_type = g['sub_type'] - except KeyError: - raise tox.exception.ConfigError( - "Malformed substitution; no substitution type provided") - - try: - handler = handlers[sub_type] - except KeyError: - raise tox.exception.ConfigError("No support for the %s substitution type" % sub_type) - - return handler(match) - - def _replace(self, x): - if '{' in x: - return RE_ITEM_REF.sub(self._replace_match, x) - return x - - -class _ArgvlistReader: - @classmethod - def getargvlist(cls, reader, section_val): - """Parse ``commands`` argvlist multiline string. - - :param str name: Key name in a section. - :param str section_val: Content stored by key. - - :rtype: list[list[str]] - :raise :class:`tox.exception.ConfigError`: - line-continuation ends nowhere while resolving for specified section - """ - commands = [] - current_command = "" - for line in section_val.splitlines(): - line = line.rstrip() - i = line.find("#") - if i != -1: - line = line[:i].rstrip() - if not line: - continue - if line.endswith("\\"): - current_command += " " + line[:-1] - continue - current_command += line - - if is_section_substitution(current_command): - replaced = reader._replace(current_command) - commands.extend(cls.getargvlist(reader, replaced)) - else: - commands.append(cls.processcommand(reader, current_command)) - current_command = "" - else: - if current_command: - raise tox.exception.ConfigError( - "line-continuation ends nowhere while resolving for [%s] %s" % - (reader.section_name, "commands")) - return commands - - @classmethod - def processcommand(cls, reader, command): - posargs = getattr(reader, "posargs", None) - - # Iterate through each word of the command substituting as - # appropriate to construct the new command string. This - # string is then broken up into exec argv components using - # shlex. - newcommand = "" - for word in CommandParser(command).words(): - if word == "{posargs}" or word == "[]": - if posargs: - newcommand += " ".join(posargs) - continue - elif word.startswith("{posargs:") and word.endswith("}"): - if posargs: - newcommand += " ".join(posargs) - continue - else: - word = word[9:-1] - new_arg = "" - new_word = reader._replace(word) - new_word = reader._replace(new_word) - new_arg += new_word - newcommand += new_arg - - # Construct shlex object that will not escape any values, - # use all values as is in argv. - shlexer = shlex.shlex(newcommand, posix=True) - shlexer.whitespace_split = True - shlexer.escape = '' - shlexer.commenters = '' - argv = list(shlexer) - return argv - - -class CommandParser(object): - - class State(object): - def __init__(self): - self.word = '' - self.depth = 0 - self.yield_words = [] - - def __init__(self, command): - self.command = command - - def words(self): - ps = CommandParser.State() - - def word_has_ended(): - return ((cur_char in string.whitespace and ps.word and - ps.word[-1] not in string.whitespace) or - (cur_char == '{' and ps.depth == 0 and not ps.word.endswith('\\')) or - (ps.depth == 0 and ps.word and ps.word[-1] == '}') or - (cur_char not in string.whitespace and ps.word and - ps.word.strip() == '')) - - def yield_this_word(): - yieldword = ps.word - ps.word = '' - if yieldword: - ps.yield_words.append(yieldword) - - def yield_if_word_ended(): - if word_has_ended(): - yield_this_word() - - def accumulate(): - ps.word += cur_char - - def push_substitution(): - ps.depth += 1 - - def pop_substitution(): - ps.depth -= 1 - - for cur_char in self.command: - if cur_char in string.whitespace: - if ps.depth == 0: - yield_if_word_ended() - accumulate() - elif cur_char == '{': - yield_if_word_ended() - accumulate() - push_substitution() - elif cur_char == '}': - accumulate() - pop_substitution() - else: - yield_if_word_ended() - accumulate() - - if ps.word.strip(): - yield_this_word() - return ps.yield_words - - -def getcontextname(): - if any(env in os.environ for env in ['JENKINS_URL', 'HUDSON_URL']): - return 'jenkins' - return None diff --git a/tox/_exception.py b/tox/_exception.py deleted file mode 100644 index 9917f09..0000000 --- a/tox/_exception.py +++ /dev/null @@ -1,28 +0,0 @@ - -class Error(Exception): - def __str__(self): - return "%s: %s" % (self.__class__.__name__, self.args[0]) - - -class UnsupportedInterpreter(Error): - "signals an unsupported Interpreter" - - -class InterpreterNotFound(Error): - "signals that an interpreter could not be found" - - -class InvocationError(Error): - """ an error while invoking a script. """ - - -class MissingFile(Error): - """ an error while invoking a script. """ - - -class MissingDirectory(Error): - """ a directory did not exist. """ - - -class MissingDependency(Error): - """ a dependency could not be found or determined. """ diff --git a/tox/_pytestplugin.py b/tox/_pytestplugin.py index 9d7488a..ee85acc 100644 --- a/tox/_pytestplugin.py +++ b/tox/_pytestplugin.py @@ -6,10 +6,10 @@ import sys from py.builtin import _isbytes, _istext, print_ from fnmatch import fnmatch import time -from tox._config import parseconfig -from tox._venv import VirtualEnv -from tox._cmdline import Action -from tox.result import ResultLog +from .config import parseconfig +from .venv import VirtualEnv +from .session import Action +from .result import ResultLog def pytest_configure(): @@ -134,7 +134,7 @@ class pcallMock: @pytest.fixture def mocksession(request): - from tox._cmdline import Session + from tox.session import Session class MockSession(Session): def __init__(self): diff --git a/tox/_venv.py b/tox/_venv.py deleted file mode 100644 index d52a6e5..0000000 --- a/tox/_venv.py +++ /dev/null @@ -1,422 +0,0 @@ -from __future__ import with_statement -import os -import sys -import re -import codecs -import py -import tox -from tox._config import DepConfig - - -class CreationConfig: - def __init__(self, md5, python, version, sitepackages, - usedevelop, deps): - self.md5 = md5 - self.python = python - self.version = version - self.sitepackages = sitepackages - self.usedevelop = usedevelop - self.deps = deps - - def writeconfig(self, path): - lines = ["%s %s" % (self.md5, self.python)] - lines.append("%s %d %d" % (self.version, self.sitepackages, self.usedevelop)) - for dep in self.deps: - lines.append("%s %s" % dep) - path.ensure() - path.write("\n".join(lines)) - - @classmethod - def readconfig(cls, path): - try: - lines = path.readlines(cr=0) - value = lines.pop(0).split(None, 1) - md5, python = value - version, sitepackages, usedevelop = lines.pop(0).split(None, 3) - sitepackages = bool(int(sitepackages)) - usedevelop = bool(int(usedevelop)) - deps = [] - for line in lines: - md5, depstring = line.split(None, 1) - deps.append((md5, depstring)) - return CreationConfig(md5, python, version, sitepackages, usedevelop, deps) - except Exception: - return None - - def matches(self, other): - return (other 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.deps == other.deps) - - -class VirtualEnv(object): - def __init__(self, envconfig=None, session=None): - self.envconfig = envconfig - self.session = session - self.path = envconfig.envdir - self.path_config = self.path.join(".tox-config1") - - @property - def name(self): - return self.envconfig.envname - - def __repr__(self): - return "" % (self.path) - - def getcommandpath(self, name=None, venv=True, cwd=None): - if name is None: - return self.envconfig.envpython - name = str(name) - if os.path.isabs(name): - return name - if os.path.split(name)[0] == ".": - p = cwd.join(name) - if p.check(): - return str(p) - p = None - if venv: - p = py.path.local.sysfind(name, paths=[self.envconfig.envbindir]) - if p is not None: - return p - p = py.path.local.sysfind(name) - if p is None: - raise tox.exception.InvocationError( - "could not find executable %r" % (name,)) - # p is not found in virtualenv script/bin dir - if venv: - if not self.is_allowed_external(p): - self.session.report.warning( - "test command found but not installed in testenv\n" - " cmd: %s\n" - " env: %s\n" - "Maybe you forgot to specify a dependency? " - "See also the whitelist_externals envconfig setting." % ( - p, self.envconfig.envdir)) - return str(p) # will not be rewritten for reporting - - def is_allowed_external(self, p): - tryadd = [""] - if sys.platform == "win32": - 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 - - def _ispython3(self): - return "python3" in str(self.envconfig.basepython) - - def update(self, action=None): - """ return status string for updating actual venv to match configuration. - if status string is empty, all is ok. - """ - if action is None: - action = self.session.newaction(self, "update") - rconfig = CreationConfig.readconfig(self.path_config) - if not self.envconfig.recreate and rconfig and \ - rconfig.matches(self._getliveconfig()): - action.info("reusing", self.envconfig.envdir) - return - if rconfig is None: - action.setactivity("create", self.envconfig.envdir) - else: - action.setactivity("recreate", self.envconfig.envdir) - try: - self.create(action) - except tox.exception.UnsupportedInterpreter: - return sys.exc_info()[1] - except tox.exception.InterpreterNotFound: - return sys.exc_info()[1] - try: - self.install_deps(action) - except tox.exception.InvocationError: - v = sys.exc_info()[1] - return "could not install deps %s; v = %r" % ( - 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 - deps = [] - for dep in self._getresolvedeps(): - raw_dep = dep.name - md5 = getdigest(raw_dep) - deps.append((md5, raw_dep)) - return CreationConfig(md5, python, version, - sitepackages, develop, deps) - - def _getresolvedeps(self): - l = [] - for dep in self.envconfig.deps: - if dep.indexserver is None: - res = self.session._resolve_pkg(dep.name) - if res != dep.name: - dep = dep.__class__(res) - l.append(dep) - return l - - def getsupportedinterpreter(self): - return self.envconfig.getsupportedinterpreter() - - def matching_platform(self): - return re.match(self.envconfig.platform, sys.platform) - - def create(self, action=None): - # if self.getcommandpath("activate").dirpath().check(): - # return - if action is None: - action = self.session.newaction(self, "create") - - config_interpreter = self.getsupportedinterpreter() - args = [sys.executable, '-m', 'virtualenv'] - if self.envconfig.sitepackages: - args.append('--system-site-packages') - # add interpreter explicitly, to prevent using - # default (virtualenv.ini) - args.extend(['--python', str(config_interpreter)]) - # if sys.platform == "win32": - # f, path, _ = py.std.imp.find_module("virtualenv") - # f.close() - # args[:1] = [str(config_interpreter), str(path)] - # else: - self.session.make_emptydir(self.path) - basepath = self.path.dirpath() - basepath.ensure(dir=1) - args.append(self.path.basename) - self._pcall(args, venv=False, action=action, cwd=basepath) - self.just_created = True - - def finish(self): - self._getliveconfig().writeconfig(self.path_config) - - def _needs_reinstall(self, setupdir, action): - setup_py = setupdir.join('setup.py') - setup_cfg = setupdir.join('setup.cfg') - args = [self.envconfig.envpython, str(setup_py), '--name'] - output = action.popen(args, cwd=setupdir, redirect=False, - returnout=True) - name = output.strip() - egg_info = setupdir.join('.'.join((name, 'egg-info'))) - for conf_file in (setup_py, setup_cfg): - if (not egg_info.check() - or (conf_file.check() and conf_file.mtime() > egg_info.mtime())): - return True - return False - - 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'] - 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'] - self._install([sdistpath], extraopts=extraopts, action=action) - - def install_deps(self, action=None): - if action is None: - action = self.session.newaction(self, "install_deps") - deps = self._getresolvedeps() - if deps: - depinfo = ", ".join(map(str, deps)) - action.setactivity("installdeps", "%s" % depinfo) - self._install(deps, action=action) - - def _installopts(self, indexserver): - l = [] - if indexserver: - l += ["-i", indexserver] - if self.envconfig.downloadcache: - self.envconfig.downloadcache.ensure(dir=1) - l.append("--download-cache=%s" % self.envconfig.downloadcache) - if self.envconfig.pip_pre: - l.append("--pre") - return l - - def run_install_command(self, packages, options=(), - indexserver=None, action=None, - extraenv=None): - argv = self.envconfig.install_command[:] - # use pip-script on win32 to avoid the executable locking - 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__'): - try: - del os.environ[x] - except KeyError: - pass - old_stdout = sys.stdout - sys.stdout = codecs.getwriter('utf8')(sys.stdout) - if extraenv is None: - extraenv = {} - self._pcall(argv, cwd=self.envconfig.config.toxinidir, - extraenv=extraenv, action=action) - sys.stdout = old_stdout - - def _install(self, deps, extraopts=None, action=None): - if not deps: - return - d = {} - l = [] - 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(dep.name) - if ixserver not in l: - l.append(ixserver) - assert ixserver.url is None or isinstance(ixserver.url, str) - - for ixserver in l: - if self.envconfig.config.option.sethome: - extraenv = hack_home_env( - homedir=self.envconfig.envtmpdir.join("pseudo-home"), - index_url=ixserver.url) - else: - extraenv = {} - - packages = d[ixserver] - options = self._installopts(ixserver.url) - if extraopts: - options.extend(extraopts) - self.run_install_command(packages=packages, options=options, - action=action, extraenv=extraenv) - - def _getenv(self, extraenv={}): - env = {} - for envname in self.envconfig.passenv: - if envname in os.environ: - env[envname] = os.environ[envname] - - env.update(self.envconfig.setenv) - - env['VIRTUAL_ENV'] = str(self.path) - - env.update(extraenv) - return env - - def test(self, redirect=False): - action = self.session.newaction(self, "runtests") - with action: - self.status = 0 - self.session.make_emptydir(self.envconfig.envtmpdir) - cwd = self.envconfig.changedir - env = self._getenv() - # Display PYTHONHASHSEED to assist with reproducibility. - action.setactivity("runtests", "PYTHONHASHSEED=%r" % 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[%s] | %s" % (i, ' '.join( - [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) - except tox.exception.InvocationError as err: - self.session.report.error(str(err)) - self.status = "commands failed" - if not self.envconfig.ignore_errors: - break # Don't process remaining commands - except KeyboardInterrupt: - self.status = "keyboardinterrupt" - self.session.report.error(self.status) - raise - - def _pcall(self, args, venv=True, cwd=None, extraenv={}, - action=None, redirect=True, ignore_ret=False): - for name in ("VIRTUALENV_PYTHON", "PYTHONDONTWRITEBYTECODE"): - try: - del os.environ[name] - except KeyError: - pass - assert cwd - cwd.ensure(dir=1) - old = self.patchPATH() - try: - args[0] = self.getcommandpath(args[0], venv, cwd) - env = self._getenv(extraenv) - return action.popen(args, cwd=cwd, env=env, redirect=redirect, ignore_ret=ignore_ret) - finally: - os.environ['PATH'] = old - - def patchPATH(self): - oldPATH = os.environ['PATH'] - bindir = str(self.envconfig.envbindir) - os.environ['PATH'] = os.pathsep.join([bindir, oldPATH]) - self.session.report.verbosity2("setting PATH=%s" % os.environ["PATH"]) - return oldPATH - - -def getdigest(path): - path = py.path.local(path) - if not path.check(file=1): - return "0" * 32 - return path.computehash() - - -def hack_home_env(homedir, index_url=None): - # XXX HACK (this could also live with tox itself, consider) - # if tox uses pip on a package that requires setup_requires - # the index url set with pip is usually not recognized - # because it is setuptools executing very early. - # We therefore run the tox command in an artifical home - # directory and set .pydistutils.cfg and pip.conf files - # accordingly. - if not homedir.check(): - homedir.ensure(dir=1) - d = dict(HOME=str(homedir)) - if not index_url: - index_url = os.environ.get("TOX_INDEX_URL") - if index_url: - homedir.join(".pydistutils.cfg").write( - "[easy_install]\n" - "index_url = %s\n" % index_url) - d["PIP_INDEX_URL"] = index_url - d["TOX_INDEX_URL"] = index_url - return d diff --git a/tox/config.py b/tox/config.py new file mode 100644 index 0000000..8675286 --- /dev/null +++ b/tox/config.py @@ -0,0 +1,1072 @@ +import argparse +import os +import random +from fnmatch import fnmatchcase +import sys +import re +import shlex +import string +import pkg_resources +import itertools +import pluggy + +import tox.interpreters +from tox import hookspecs + +import py + +import tox + +iswin32 = sys.platform == "win32" + +default_factors = {'jython': 'jython', 'pypy': 'pypy', 'pypy3': 'pypy3', + 'py': sys.executable} +for version in '26,27,32,33,34,35,36'.split(','): + default_factors['py' + version] = 'python%s.%s' % tuple(version) + +hookimpl = pluggy.HookimplMarker("tox") + + +def get_plugin_manager(): + # initialize plugin manager + pm = pluggy.PluginManager("tox") + pm.add_hookspecs(hookspecs) + pm.register(tox.config) + pm.register(tox.interpreters) + pm.load_setuptools_entrypoints("tox") + pm.check_pending() + return pm + + +class MyParser: + def __init__(self): + self.argparser = argparse.ArgumentParser( + description="tox options", add_help=False) + self._testenv_attr = [] + + def add_argument(self, *args, **kwargs): + return self.argparser.add_argument(*args, **kwargs) + + def add_testenv_attribute(self, name, type, help, default=None, postprocess=None): + self._testenv_attr.append(VenvAttribute(name, type, default, help, postprocess)) + + def add_testenv_attribute_obj(self, obj): + assert hasattr(obj, "name") + assert hasattr(obj, "type") + assert hasattr(obj, "help") + assert hasattr(obj, "postprocess") + self._testenv_attr.append(obj) + + def parse_args(self, args): + return self.argparser.parse_args(args) + + def format_help(self): + return self.argparser.format_help() + + +class VenvAttribute: + def __init__(self, name, type, default, help, postprocess): + self.name = name + self.type = type + self.default = default + self.help = help + self.postprocess = postprocess + + +class DepOption: + name = "deps" + type = "line-list" + help = "each line specifies a dependency in pip/setuptools format." + default = () + + def postprocess(self, config, reader, section_val): + deps = [] + for depline in section_val: + m = re.match(r":(\w+):\s*(\S+)", depline) + if m: + iname, name = m.groups() + ixserver = config.indexserver[iname] + else: + name = depline.strip() + ixserver = None + name = self._replace_forced_dep(name, config) + deps.append(DepConfig(name, ixserver)) + return deps + + def _replace_forced_dep(self, name, config): + """ + Override the given dependency config name taking --force-dep-version + option into account. + + :param name: dep config, for example ["pkg==1.0", "other==2.0"]. + :param config: Config instance + :return: the new dependency that should be used for virtual environments + """ + if not config.option.force_dep: + return name + for forced_dep in config.option.force_dep: + if self._is_same_dep(forced_dep, name): + return forced_dep + return name + + @classmethod + def _is_same_dep(cls, dep1, dep2): + """ + Returns True if both dependency definitions refer to the + same package, even if versions differ. + """ + dep1_name = pkg_resources.Requirement.parse(dep1).project_name + dep2_name = pkg_resources.Requirement.parse(dep2).project_name + return dep1_name == dep2_name + + +class PosargsOption: + name = "args_are_paths" + type = "bool" + default = True + help = "treat positional args in commands as paths" + + def postprocess(self, config, reader, section_val): + args = config.option.args + if args: + if section_val: + args = [] + for arg in config.option.args: + if arg: + origpath = config.invocationcwd.join(arg, abs=True) + if origpath.check(): + arg = reader.getpath("changedir", ".").bestrelpath(origpath) + args.append(arg) + reader.addsubstitutions(args) + return section_val + + +class InstallcmdOption: + name = "install_command" + type = "argv" + default = "pip install {opts} {packages}" + help = "install command for dependencies and package under test." + + def postprocess(self, config, reader, section_val): + if '{packages}' not in section_val: + raise tox.exception.ConfigError( + "'install_command' must contain '{packages}' substitution") + return section_val + + +def parseconfig(args=None): + """ + :param list[str] args: Optional list of arguments. + :type pkg: str + :rtype: :class:`Config` + :raise SystemExit: toxinit file is not found + """ + + pm = get_plugin_manager() + + if args is None: + args = sys.argv[1:] + + # prepare command line options + parser = MyParser() + pm.hook.tox_addoption(parser=parser) + + # parse command line options + option = parser.parse_args(args) + interpreters = tox.interpreters.Interpreters(hook=pm.hook) + config = Config(pluginmanager=pm, option=option, interpreters=interpreters) + config._parser = parser + config._testenv_attr = parser._testenv_attr + + # parse ini file + basename = config.option.configfile + if os.path.isabs(basename): + inipath = py.path.local(basename) + else: + for path in py.path.local().parts(reverse=True): + inipath = path.join(basename) + if inipath.check(): + break + else: + feedback("toxini file %r not found" % (basename), sysexit=True) + try: + parseini(config, inipath) + except tox.exception.InterpreterNotFound: + exn = sys.exc_info()[1] + # Use stdout to match test expectations + py.builtin.print_("ERROR: " + str(exn)) + + # post process config object + pm.hook.tox_configure(config=config) + + return config + + +def feedback(msg, sysexit=False): + py.builtin.print_("ERROR: " + msg, file=sys.stderr) + if sysexit: + raise SystemExit(1) + + +class VersionAction(argparse.Action): + def __call__(self, argparser, *args, **kwargs): + version = tox.__version__ + py.builtin.print_("%s imported from %s" % (version, tox.__file__)) + raise SystemExit(0) + + +class CountAction(argparse.Action): + def __call__(self, parser, namespace, values, option_string=None): + if hasattr(namespace, self.dest): + setattr(namespace, self.dest, int(getattr(namespace, self.dest)) + 1) + else: + setattr(namespace, self.dest, 0) + + +@hookimpl +def tox_addoption(parser): + # formatter_class=argparse.ArgumentDefaultsHelpFormatter) + parser.add_argument("--version", nargs=0, action=VersionAction, + dest="version", + help="report version information to stdout.") + parser.add_argument("-h", "--help", action="store_true", dest="help", + help="show help about options") + parser.add_argument("--help-ini", "--hi", action="store_true", dest="helpini", + help="show help about ini-names") + parser.add_argument("-v", nargs=0, action=CountAction, default=0, + dest="verbosity", + help="increase verbosity of reporting output.") + parser.add_argument("--showconfig", action="store_true", + help="show configuration information for all environments. ") + parser.add_argument("-l", "--listenvs", action="store_true", + dest="listenvs", help="show list of test environments") + parser.add_argument("-c", action="store", default="tox.ini", + dest="configfile", + help="use the specified config file name.") + parser.add_argument("-e", action="append", dest="env", + metavar="envlist", + help="work against specified environments (ALL selects all).") + parser.add_argument("--notest", action="store_true", dest="notest", + help="skip invoking test commands.") + parser.add_argument("--sdistonly", action="store_true", dest="sdistonly", + help="only perform the sdist packaging activity.") + parser.add_argument("--installpkg", action="store", default=None, + metavar="PATH", + help="use specified package for installation into venv, instead of " + "creating an sdist.") + parser.add_argument("--develop", action="store_true", dest="develop", + help="install package in the venv using 'setup.py develop' via " + "'pip -e .'") + parser.add_argument("--set-home", action="store_true", dest="sethome", + help="(experimental) force creating a new $HOME for each test " + "environment and create .pydistutils.cfg|pip.conf files " + "if index servers are specified with tox. ") + parser.add_argument('-i', action="append", + dest="indexurl", metavar="URL", + help="set indexserver url (if URL is of form name=url set the " + "url for the 'name' indexserver, specifically)") + parser.add_argument("--pre", action="store_true", dest="pre", + help="install pre-releases and development versions of dependencies. " + "This will pass the --pre option to install_command " + "(pip by default).") + parser.add_argument("-r", "--recreate", action="store_true", + dest="recreate", + help="force recreation of virtual environments") + parser.add_argument("--result-json", action="store", + dest="resultjson", metavar="PATH", + help="write a json file with detailed information about " + "all commands and results involved. This will turn off " + "pass-through output from running test commands which is " + "instead captured into the json result file.") + + # We choose 1 to 4294967295 because it is the range of PYTHONHASHSEED. + parser.add_argument("--hashseed", action="store", + metavar="SEED", default=None, + help="set PYTHONHASHSEED to SEED before running commands. " + "Defaults to a random integer in the range [1, 4294967295] " + "([1, 1024] on Windows). " + "Passing 'noset' suppresses this behavior.") + parser.add_argument("--force-dep", action="append", + metavar="REQ", default=None, + help="Forces a certain version of one of the dependencies " + "when configuring the virtual environment. REQ Examples " + "'pytest<2.7' or 'django>=1.6'.") + parser.add_argument("--sitepackages", action="store_true", + help="override sitepackages setting to True in all envs") + parser.add_argument("--skip-missing-interpreters", action="store_true", + help="don't fail tests for missing interpreters") + + parser.add_argument("args", nargs="*", + help="additional arguments available to command positional substitution") + + # add various core venv interpreter attributes + + parser.add_testenv_attribute( + name="envdir", type="path", default="{toxworkdir}/{envname}", + help="venv directory") + + parser.add_testenv_attribute( + name="envtmpdir", type="path", default="{envdir}/tmp", + help="venv temporary directory") + + parser.add_testenv_attribute( + name="envlogdir", type="path", default="{envdir}/log", + help="venv log directory") + + def downloadcache(config, reader, section_val): + if section_val: + # env var, if present, takes precedence + downloadcache = os.environ.get("PIP_DOWNLOAD_CACHE", section_val) + return py.path.local(downloadcache) + + parser.add_testenv_attribute( + name="downloadcache", type="string", default=None, postprocess=downloadcache, + help="(deprecated) set PIP_DOWNLOAD_CACHE.") + + parser.add_testenv_attribute( + name="changedir", type="path", default="{toxinidir}", + help="directory to change to when running commands") + + parser.add_testenv_attribute_obj(PosargsOption()) + + parser.add_testenv_attribute( + name="skip_install", type="bool", default=False, + help="Do not install the current package. This can be used when " + "you need the virtualenv management but do not want to install " + "the current package") + + parser.add_testenv_attribute( + name="ignore_errors", type="bool", default=False, + help="if set to True all commands will be executed irrespective of their " + "result error status.") + + def recreate(config, reader, section_val): + if config.option.recreate: + return True + return section_val + + parser.add_testenv_attribute( + name="recreate", type="bool", default=False, postprocess=recreate, + help="always recreate this test environment.") + + def setenv(config, reader, section_val): + setenv = section_val + if "PYTHONHASHSEED" not in setenv and config.hashseed is not None: + setenv['PYTHONHASHSEED'] = config.hashseed + return setenv + + parser.add_testenv_attribute( + name="setenv", type="dict", postprocess=setenv, + help="list of X=Y lines with environment variable settings") + + def passenv(config, reader, section_val): + passenv = set(["PATH"]) + + # we ensure that tmp directory settings are passed on + # we could also set it to the per-venv "envtmpdir" + # but this leads to very long paths when run with jenkins + # so we just pass it on by default for now. + if sys.platform == "win32": + passenv.add("SYSTEMROOT") # needed for python's crypto module + passenv.add("PATHEXT") # needed for discovering executables + passenv.add("TEMPDIR") + passenv.add("TMP") + else: + passenv.add("TMPDIR") + for spec in section_val: + for name in os.environ: + if fnmatchcase(name.upper(), spec.upper()): + passenv.add(name) + return passenv + + parser.add_testenv_attribute( + name="passenv", type="space-separated-list", postprocess=passenv, + help="environment variables names which shall be passed " + "from tox invocation to test environment when executing commands.") + + parser.add_testenv_attribute( + name="whitelist_externals", type="line-list", + help="each lines specifies a path or basename for which tox will not warn " + "about it coming from outside the test environment.") + + parser.add_testenv_attribute( + name="platform", type="string", default=".*", + help="regular expression which must match against ``sys.platform``. " + "otherwise testenv will be skipped.") + + def sitepackages(config, reader, section_val): + return config.option.sitepackages or section_val + + parser.add_testenv_attribute( + name="sitepackages", type="bool", default=False, postprocess=sitepackages, + help="Set to ``True`` if you want to create virtual environments that also " + "have access to globally installed packages.") + + def pip_pre(config, reader, section_val): + return config.option.pre or section_val + + parser.add_testenv_attribute( + name="pip_pre", type="bool", default=False, postprocess=pip_pre, + help="If ``True``, adds ``--pre`` to the ``opts`` passed to " + "the install command. ") + + def develop(config, reader, section_val): + return not config.option.installpkg and (section_val or config.option.develop) + + parser.add_testenv_attribute( + name="usedevelop", type="bool", postprocess=develop, default=False, + help="install package in develop/editable mode") + + def basepython_default(config, reader, section_val): + if section_val is None: + for f in reader.factors: + if f in default_factors: + return default_factors[f] + return sys.executable + return str(section_val) + + parser.add_testenv_attribute( + name="basepython", type="string", default=None, postprocess=basepython_default, + help="executable name or path of interpreter used to create a " + "virtual test environment.") + + parser.add_testenv_attribute_obj(InstallcmdOption()) + parser.add_testenv_attribute_obj(DepOption()) + + parser.add_testenv_attribute( + name="commands", type="argvlist", default="", + help="each line specifies a test command and can use substitution.") + + +class Config(object): + def __init__(self, pluginmanager, option, interpreters): + self.envconfigs = {} + self.invocationcwd = py.path.local() + self.interpreters = interpreters + self.pluginmanager = pluginmanager + self.option = option + + @property + def homedir(self): + homedir = get_homedir() + if homedir is None: + homedir = self.toxinidir # XXX good idea? + return homedir + + +class VenvConfig: + def __init__(self, envname, config): + self.envname = envname + self.config = config + + @property + def envbindir(self): + if (sys.platform == "win32" + and "jython" not in self.basepython + and "pypy" not in self.basepython): + return self.envdir.join("Scripts") + else: + return self.envdir.join("bin") + + @property + def envpython(self): + if "jython" in str(self.basepython): + name = "jython" + else: + name = "python" + return self.envbindir.join(name) + + # no @property to avoid early calling (see callable(subst[key]) checks) + def envsitepackagesdir(self): + self.getsupportedinterpreter() # for throwing exceptions + x = self.config.interpreters.get_sitepackagesdir( + info=self.python_info, + envdir=self.envdir) + return x + + @property + def python_info(self): + return self.config.interpreters.get_info(envconfig=self) + + def getsupportedinterpreter(self): + if sys.platform == "win32" and self.basepython and \ + "jython" in self.basepython: + raise tox.exception.UnsupportedInterpreter( + "Jython/Windows does not support installing scripts") + info = self.config.interpreters.get_info(envconfig=self) + if not info.executable: + raise tox.exception.InterpreterNotFound(self.basepython) + if not info.version_info: + raise tox.exception.InvocationError( + 'Failed to get version_info for %s: %s' % (info.name, info.err)) + if info.version_info < (2, 6): + raise tox.exception.UnsupportedInterpreter( + "python2.5 is not supported anymore, sorry") + return info.executable + + +testenvprefix = "testenv:" + + +def get_homedir(): + try: + return py.path.local._gethomedir() + except Exception: + return None + + +def make_hashseed(): + max_seed = 4294967295 + if sys.platform == 'win32': + max_seed = 1024 + return str(random.randint(1, max_seed)) + + +class parseini: + def __init__(self, config, inipath): + config.toxinipath = inipath + config.toxinidir = config.toxinipath.dirpath() + + self._cfg = py.iniconfig.IniConfig(config.toxinipath) + config._cfg = self._cfg + self.config = config + ctxname = getcontextname() + if ctxname == "jenkins": + reader = SectionReader("tox:jenkins", self._cfg, fallbacksections=['tox']) + distshare_default = "{toxworkdir}/distshare" + elif not ctxname: + reader = SectionReader("tox", self._cfg) + distshare_default = "{homedir}/.tox/distshare" + else: + raise ValueError("invalid context") + + if config.option.hashseed is None: + hashseed = make_hashseed() + elif config.option.hashseed == 'noset': + hashseed = None + else: + hashseed = config.option.hashseed + config.hashseed = hashseed + + reader.addsubstitutions(toxinidir=config.toxinidir, + homedir=config.homedir) + config.toxworkdir = reader.getpath("toxworkdir", "{toxinidir}/.tox") + config.minversion = reader.getstring("minversion", None) + + if not config.option.skip_missing_interpreters: + config.option.skip_missing_interpreters = \ + reader.getbool("skip_missing_interpreters", False) + + # determine indexserver dictionary + config.indexserver = {'default': IndexServerConfig('default')} + prefix = "indexserver" + for line in reader.getlist(prefix): + name, url = map(lambda x: x.strip(), line.split("=", 1)) + config.indexserver[name] = IndexServerConfig(name, url) + + override = False + if config.option.indexurl: + for urldef in config.option.indexurl: + m = re.match(r"\W*(\w+)=(\S+)", urldef) + if m is None: + url = urldef + name = "default" + else: + name, url = m.groups() + if not url: + url = None + if name != "ALL": + config.indexserver[name].url = url + else: + override = url + # let ALL override all existing entries + if override: + for name in config.indexserver: + config.indexserver[name] = IndexServerConfig(name, override) + + reader.addsubstitutions(toxworkdir=config.toxworkdir) + config.distdir = reader.getpath("distdir", "{toxworkdir}/dist") + reader.addsubstitutions(distdir=config.distdir) + config.distshare = reader.getpath("distshare", distshare_default) + reader.addsubstitutions(distshare=config.distshare) + config.sdistsrc = reader.getpath("sdistsrc", None) + config.setupdir = reader.getpath("setupdir", "{toxinidir}") + config.logdir = config.toxworkdir.join("log") + + config.envlist, all_envs = self._getenvdata(reader) + + # factors used in config or predefined + known_factors = self._list_section_factors("testenv") + known_factors.update(default_factors) + known_factors.add("python") + + # factors stated in config envlist + stated_envlist = reader.getstring("envlist", replace=False) + if stated_envlist: + for env in _split_env(stated_envlist): + known_factors.update(env.split('-')) + + # configure testenvs + for name in all_envs: + section = testenvprefix + name + factors = set(name.split('-')) + if section in self._cfg or factors <= known_factors: + config.envconfigs[name] = \ + self.make_envconfig(name, section, reader._subs, config) + + all_develop = all(name in config.envconfigs + and config.envconfigs[name].usedevelop + for name in config.envlist) + + config.skipsdist = reader.getbool("skipsdist", all_develop) + + def _list_section_factors(self, section): + factors = set() + if section in self._cfg: + for _, value in self._cfg[section].items(): + exprs = re.findall(r'^([\w{}\.,-]+)\:\s+', value, re.M) + factors.update(*mapcat(_split_factor_expr, exprs)) + return factors + + def make_envconfig(self, name, section, subs, config): + vc = VenvConfig(config=config, envname=name) + factors = set(name.split('-')) + reader = SectionReader(section, self._cfg, fallbacksections=["testenv"], + factors=factors) + reader.addsubstitutions(**subs) + reader.addsubstitutions(envname=name) + reader.vc = vc + + for env_attr in config._testenv_attr: + atype = env_attr.type + if atype in ("bool", "path", "string", "dict", "argv", "argvlist"): + meth = getattr(reader, "get" + atype) + res = meth(env_attr.name, env_attr.default) + elif atype == "space-separated-list": + res = reader.getlist(env_attr.name, sep=" ") + elif atype == "line-list": + res = reader.getlist(env_attr.name, sep="\n") + else: + raise ValueError("unknown type %r" % (atype,)) + + if env_attr.postprocess: + res = env_attr.postprocess(config, reader, res) + setattr(vc, env_attr.name, res) + + if atype == "path": + reader.addsubstitutions(**{env_attr.name: res}) + + if env_attr.name == "install_command": + reader.addsubstitutions(envbindir=vc.envbindir, envpython=vc.envpython, + envsitepackagesdir=vc.envsitepackagesdir) + return vc + + def _getenvdata(self, reader): + envstr = self.config.option.env \ + or os.environ.get("TOXENV") \ + or reader.getstring("envlist", replace=False) \ + or [] + envlist = _split_env(envstr) + + # collect section envs + all_envs = set(envlist) - set(["ALL"]) + for section in self._cfg: + if section.name.startswith(testenvprefix): + all_envs.add(section.name[len(testenvprefix):]) + if not all_envs: + all_envs.add("python") + + if not envlist or "ALL" in envlist: + envlist = sorted(all_envs) + + return envlist, all_envs + + +def _split_env(env): + """if handed a list, action="append" was used for -e """ + if not isinstance(env, list): + env = [env] + return mapcat(_expand_envstr, env) + + +def _split_factor_expr(expr): + partial_envs = _expand_envstr(expr) + return [set(e.split('-')) for e in partial_envs] + + +def _expand_envstr(envstr): + # split by commas not in groups + tokens = re.split(r'((?:\{[^}]+\})+)|,', envstr) + envlist = [''.join(g).strip() + for k, g in itertools.groupby(tokens, key=bool) if k] + + def expand(env): + tokens = re.split(r'\{([^}]+)\}', env) + parts = [token.split(',') for token in tokens] + return [''.join(variant) for variant in itertools.product(*parts)] + + return mapcat(expand, envlist) + + +def mapcat(f, seq): + return list(itertools.chain.from_iterable(map(f, seq))) + + +class DepConfig: + def __init__(self, name, indexserver=None): + self.name = name + self.indexserver = indexserver + + def __str__(self): + if self.indexserver: + if self.indexserver.name == "default": + return self.name + return ":%s:%s" % (self.indexserver.name, self.name) + return str(self.name) + __repr__ = __str__ + + +class IndexServerConfig: + def __init__(self, name, url=None): + self.name = name + self.url = url + + +#: Check value matches substitution form +#: of referencing value from other section. E.g. {[base]commands} +is_section_substitution = re.compile("{\[[^{}\s]+\]\S+?}").match + + +RE_ITEM_REF = re.compile( + r''' + (?[^[:{}]+):)? # optional sub_type for special rules + (?P[^{}]*) # substitution key + [}] + ''', + re.VERBOSE) + + +class SectionReader: + def __init__(self, section_name, cfgparser, fallbacksections=None, factors=()): + self.section_name = section_name + self._cfg = cfgparser + self.fallbacksections = fallbacksections or [] + self.factors = factors + self._subs = {} + self._subststack = [] + + def addsubstitutions(self, _posargs=None, **kw): + self._subs.update(kw) + if _posargs: + self.posargs = _posargs + + def getpath(self, name, defaultpath): + toxinidir = self._subs['toxinidir'] + path = self.getstring(name, defaultpath) + if path is None: + return path + return toxinidir.join(path, abs=True) + + def getlist(self, name, sep="\n"): + s = self.getstring(name, None) + if s is None: + return [] + return [x.strip() for x in s.split(sep) if x.strip()] + + def getdict(self, name, default=None, sep="\n"): + s = self.getstring(name, None) + if s is None: + return default or {} + + value = {} + for line in s.split(sep): + if line.strip(): + name, rest = line.split('=', 1) + value[name.strip()] = rest.strip() + + return value + + def getbool(self, name, default=None): + s = self.getstring(name, default) + if not s: + s = default + if s is None: + raise KeyError("no config value [%s] %s found" % ( + self.section_name, name)) + + if not isinstance(s, bool): + if s.lower() == "true": + s = True + elif s.lower() == "false": + s = False + else: + raise tox.exception.ConfigError( + "boolean value %r needs to be 'True' or 'False'") + return s + + def getargvlist(self, name, default=""): + s = self.getstring(name, default, replace=False) + return _ArgvlistReader.getargvlist(self, s) + + def getargv(self, name, default=""): + return self.getargvlist(name, default)[0] + + def getstring(self, name, default=None, replace=True): + x = None + for s in [self.section_name] + self.fallbacksections: + try: + x = self._cfg[s][name] + break + except KeyError: + continue + + if x is None: + x = default + else: + x = self._apply_factors(x) + + if replace and x and hasattr(x, 'replace'): + self._subststack.append((self.section_name, name)) + try: + x = self._replace(x) + finally: + assert self._subststack.pop() == (self.section_name, name) + # print "getstring", self.section_name, name, "returned", repr(x) + return x + + def _apply_factors(self, s): + def factor_line(line): + m = re.search(r'^([\w{}\.,-]+)\:\s+(.+)', line) + if not m: + return line + + expr, line = m.groups() + if any(fs <= self.factors for fs in _split_factor_expr(expr)): + return line + + lines = s.strip().splitlines() + return '\n'.join(filter(None, map(factor_line, lines))) + + def _replace_env(self, match): + match_value = match.group('substitution_value') + if not match_value: + raise tox.exception.ConfigError( + 'env: requires an environment variable name') + + default = None + envkey_split = match_value.split(':', 1) + + if len(envkey_split) is 2: + envkey, default = envkey_split + else: + envkey = match_value + + if envkey not in os.environ and default is None: + raise tox.exception.ConfigError( + "substitution env:%r: unkown environment variable %r" % + (envkey, envkey)) + + return os.environ.get(envkey, default) + + def _substitute_from_other_section(self, key): + if key.startswith("[") and "]" in key: + i = key.find("]") + section, item = key[1:i], key[i + 1:] + if section in self._cfg and item in self._cfg[section]: + if (section, item) in self._subststack: + raise ValueError('%s already in %s' % ( + (section, item), self._subststack)) + x = str(self._cfg[section][item]) + self._subststack.append((section, item)) + try: + return self._replace(x) + finally: + self._subststack.pop() + + raise tox.exception.ConfigError( + "substitution key %r not found" % key) + + def _replace_substitution(self, match): + sub_key = match.group('substitution_value') + val = self._subs.get(sub_key, None) + if val is None: + val = self._substitute_from_other_section(sub_key) + if py.builtin.callable(val): + val = val() + return str(val) + + def _replace_match(self, match): + g = match.groupdict() + + # special case: opts and packages. Leave {opts} and + # {packages} intact, they are replaced manually in + # _venv.VirtualEnv.run_install_command. + sub_value = g['substitution_value'] + if sub_value in ('opts', 'packages'): + return '{%s}' % sub_value + + handlers = { + 'env': self._replace_env, + None: self._replace_substitution, + } + try: + sub_type = g['sub_type'] + except KeyError: + raise tox.exception.ConfigError( + "Malformed substitution; no substitution type provided") + + try: + handler = handlers[sub_type] + except KeyError: + raise tox.exception.ConfigError("No support for the %s substitution type" % sub_type) + + return handler(match) + + def _replace(self, x): + if '{' in x: + return RE_ITEM_REF.sub(self._replace_match, x) + return x + + +class _ArgvlistReader: + @classmethod + def getargvlist(cls, reader, section_val): + """Parse ``commands`` argvlist multiline string. + + :param str name: Key name in a section. + :param str section_val: Content stored by key. + + :rtype: list[list[str]] + :raise :class:`tox.exception.ConfigError`: + line-continuation ends nowhere while resolving for specified section + """ + commands = [] + current_command = "" + for line in section_val.splitlines(): + line = line.rstrip() + i = line.find("#") + if i != -1: + line = line[:i].rstrip() + if not line: + continue + if line.endswith("\\"): + current_command += " " + line[:-1] + continue + current_command += line + + if is_section_substitution(current_command): + replaced = reader._replace(current_command) + commands.extend(cls.getargvlist(reader, replaced)) + else: + commands.append(cls.processcommand(reader, current_command)) + current_command = "" + else: + if current_command: + raise tox.exception.ConfigError( + "line-continuation ends nowhere while resolving for [%s] %s" % + (reader.section_name, "commands")) + return commands + + @classmethod + def processcommand(cls, reader, command): + posargs = getattr(reader, "posargs", None) + + # Iterate through each word of the command substituting as + # appropriate to construct the new command string. This + # string is then broken up into exec argv components using + # shlex. + newcommand = "" + for word in CommandParser(command).words(): + if word == "{posargs}" or word == "[]": + if posargs: + newcommand += " ".join(posargs) + continue + elif word.startswith("{posargs:") and word.endswith("}"): + if posargs: + newcommand += " ".join(posargs) + continue + else: + word = word[9:-1] + new_arg = "" + new_word = reader._replace(word) + new_word = reader._replace(new_word) + new_arg += new_word + newcommand += new_arg + + # Construct shlex object that will not escape any values, + # use all values as is in argv. + shlexer = shlex.shlex(newcommand, posix=True) + shlexer.whitespace_split = True + shlexer.escape = '' + shlexer.commenters = '' + argv = list(shlexer) + return argv + + +class CommandParser(object): + + class State(object): + def __init__(self): + self.word = '' + self.depth = 0 + self.yield_words = [] + + def __init__(self, command): + self.command = command + + def words(self): + ps = CommandParser.State() + + def word_has_ended(): + return ((cur_char in string.whitespace and ps.word and + ps.word[-1] not in string.whitespace) or + (cur_char == '{' and ps.depth == 0 and not ps.word.endswith('\\')) or + (ps.depth == 0 and ps.word and ps.word[-1] == '}') or + (cur_char not in string.whitespace and ps.word and + ps.word.strip() == '')) + + def yield_this_word(): + yieldword = ps.word + ps.word = '' + if yieldword: + ps.yield_words.append(yieldword) + + def yield_if_word_ended(): + if word_has_ended(): + yield_this_word() + + def accumulate(): + ps.word += cur_char + + def push_substitution(): + ps.depth += 1 + + def pop_substitution(): + ps.depth -= 1 + + for cur_char in self.command: + if cur_char in string.whitespace: + if ps.depth == 0: + yield_if_word_ended() + accumulate() + elif cur_char == '{': + yield_if_word_ended() + accumulate() + push_substitution() + elif cur_char == '}': + accumulate() + pop_substitution() + else: + yield_if_word_ended() + accumulate() + + if ps.word.strip(): + yield_this_word() + return ps.yield_words + + +def getcontextname(): + if any(env in os.environ for env in ['JENKINS_URL', 'HUDSON_URL']): + return 'jenkins' + return None diff --git a/tox/session.py b/tox/session.py new file mode 100644 index 0000000..e886274 --- /dev/null +++ b/tox/session.py @@ -0,0 +1,678 @@ +""" +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) + if config.option.help: + show_help(config) + raise SystemExit(0) + elif config.option.helpini: + show_help_ini(config) + raise SystemExit(0) + return config + + +def main(args=None): + try: + config = prepare(args) + retcode = Session(config).runcommand() + raise SystemExit(retcode) + except KeyboardInterrupt: + raise SystemExit(2) + + +def show_help(config): + tw = py.io.TerminalWriter() + tw.write(config._parser.format_help()) + tw.line() + + +def show_help_ini(config): + tw = py.io.TerminalWriter() + tw.sep("-", "per-testenv attributes") + for env_attr in config._testenv_attr: + tw.line("%-15s %-8s default: %s" % + (env_attr.name, "<" + env_attr.type + ">", env_attr.default), bold=True) + tw.line(env_attr.help) + tw.line() + + +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.report = session.report + self.args = args + self.id = venv and venv.envconfig.envname or "tox" + self._popenlist = [] + if self.venv: + self.venvname = self.venv.name + else: + self.venvname = "GLOB" + if msg == "runtests": + cat = "test" + else: + cat = "setup" + envlog = session.resultlog.get_envlog(self.venvname) + self.commandlog = envlog.get_commandlog(cat) + + def __enter__(self): + self.report.logaction_start(self) + + def __exit__(self, *args): + self.report.logaction_finish(self) + + def setactivity(self, name, msg): + self.activity = name + self.report.verbosity0("%s %s: %s" % (self.venvname, name, msg), bold=True) + + def info(self, name, msg): + self.report.verbosity1("%s %s: %s" % (self.venvname, name, msg), bold=True) + + def _initlogpath(self, actionid): + if self.venv: + logdir = self.venv.envconfig.envlogdir + else: + logdir = self.session.config.logdir + try: + l = logdir.listdir("%s-*" % actionid) + except py.error.ENOENT: + logdir.ensure(dir=1) + l = [] + num = len(l) + path = logdir.join("%s-%s.log" % (actionid, num)) + f = path.open('w') + f.flush() + 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(self.id) + fout.write("actionid: %s\nmsg: %s\ncmdargs: %r\nenv: %s\n\n" % ( + self.id, self.msg, args, env)) + fout.flush() + self.popen_outpath = outpath = py.path.local(fout.name) + fin = outpath.open() + fin.read() # 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() + try: + popen = self._popen(args, cwd, env=env, + stdout=stdout, stderr=STDOUT) + except OSError as e: + self.report.error("invocation failed (errno %d), args: %s, cwd: %s" % + (e.errno, args, cwd)) + raise + popen.outpath = outpath + popen.args = [str(x) for x in args] + popen.cwd = cwd + popen.action = self + self._popenlist.append(popen) + try: + self.report.logpopen(popen, env=env) + try: + 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 = fin.read(1) + if data: + sys.stdout.write(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 + sys.stdout.flush() + last_time = now() + elif popen.poll() is not None: + if popen.stdout is not None: + popen.stdout.close() + break + else: + py.std.time.sleep(0.1) + fin.seek(fin_pos) + fin.close() + else: + out, err = popen.communicate() + except KeyboardInterrupt: + self.report.keyboard_interrupt() + popen.wait() + raise KeyboardInterrupt() + ret = popen.wait() + finally: + self._popenlist.remove(popen) + if ret and not ignore_ret: + invoked = " ".join(map(str, popen.args)) + if outpath: + self.report.error("invocation failed (exit code %d), logfile: %s" % + (ret, outpath)) + out = outpath.read() + self.report.error(out) + if hasattr(self, "commandlog"): + self.commandlog.add_command(popen.args, out, ret) + raise tox.exception.InvocationError( + "%s (see %s)" % (invoked, outpath), ret) + else: + raise tox.exception.InvocationError("%r" % (invoked, ), ret) + if not out and outpath: + out = outpath.read() + 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) + newargs.append(str(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.getcommandpath())] + 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), + universal_newlines=True, + stdout=stdout, stderr=stderr, env=env) + + +class Reporter(object): + actionchar = "-" + + def __init__(self, session): + self.tw = py.io.TerminalWriter() + self.session = session + self._reportedlines = [] + # self.cumulated_time = 0.0 + + 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,)) + else: + 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) + + def startsummary(self): + self.tw.sep("_", "summary") + + def info(self, msg): + if self.session.config.option.verbosity >= 2: + self.logline(msg) + + def using(self, msg): + if self.session.config.option.verbosity >= 1: + self.logline("using %s" % (msg,), bold=True) + + def keyboard_interrupt(self): + self.error("KEYBOARDINTERRUPT") + +# 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 += " " + self.tw.write(name, bold=True) + self.tw.write(value) + self.tw.line() + + 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) + self.tw.line("%s" % msg, **opts) + + def verbosity0(self, msg, **opts): + if self.session.config.option.verbosity >= 0: + self.logline("%s" % msg, **opts) + + def verbosity1(self, msg, **opts): + if self.session.config.option.verbosity >= 1: + self.logline("%s" % msg, **opts) + + def verbosity2(self, msg, **opts): + if self.session.config.option.verbosity >= 2: + self.logline("%s" % msg, **opts) + + # def log(self, msg): + # py.builtin.print_(msg, file=sys.stderr) + + +class Session: + + def __init__(self, config, popen=subprocess.Popen, Report=Reporter): + self.config = config + self.popen = popen + self.resultlog = ResultLog() + self.report = Report(self) + self.make_emptydir(config.logdir) + config.logdir.ensure(dir=1) + # self.report.using("logdir %s" %(self.config.logdir,)) + self.report.using("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) + self._actions = [] + + def _makevenv(self, name): + envconfig = self.config.envconfigs.get(name, None) + if envconfig is None: + self.report.error("unknown environment %r" % name) + raise LookupError(name) + venv = VirtualEnv(envconfig=envconfig, session=self) + self._name2venv[name] = venv + return venv + + 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): + self.report.using("tox-%s from %s" % (tox.__version__, tox.__file__)) + if self.config.minversion: + minversion = NormalizedVersion(self.config.minversion) + toxversion = NormalizedVersion(tox.__version__) + if toxversion < minversion: + self.report.error( + "tox version is %s, required is at least %s" % ( + toxversion, minversion)) + raise SystemExit(1) + 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(): + self.report.error("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("setup.py") + 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(): + self.report.error( + 'setup.py is empty' + ) + raise SystemExit(1) + self.report.error( + 'No dist directory found. Please check setup.py, e.g with:\n' + ' python setup.py sdist' + ) + raise SystemExit(1) + + def make_emptydir(self, path): + if path.check(): + self.report.info(" removing %s" % path) + py.std.shutil.rmtree(str(path), ignore_errors=True) + path.ensure(dir=1) + + def setupenv(self, venv): + action = self.newaction(venv, "getenv", venv.envconfig.envdir) + with action: + venv.status = 0 + envlog = self.resultlog.get_envlog(venv.name) + 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 + self.report.error(str(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 + + 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 + + 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) + self.report.info("using package %r, skipping 'sdist' activity " % + str(path)) + else: + try: + path = self._makesdist() + except tox.exception.InvocationError: + v = sys.exc_info()[1] + self.report.error("FAIL could not package project - v = %r" % + v) + return + sdistfile = self.config.distshare.join(path.basename) + if sdistfile != path: + self.report.info("copying new sdistfile to %r" % + str(sdistfile)) + try: + sdistfile.dirpath().ensure(dir=1) + except py.error.Error: + self.report.warning("could not copy distfile to %s" % + sdistfile.dirpath()) + else: + path.copy(sdistfile) + return path + + def subcommand_test(self): + if self.config.skipsdist: + self.report.info("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 not venv.matching_platform(): + venv.status = "platform mismatch" + continue # we simply omit non-matching platforms + 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: + pip = venv.getcommandpath("pip") + # we can't really call internal helpers here easily :/ + # output = venv._pcall([str(pip), "freeze"], + # cwd=self.config.toxinidir, + # action=action) + output = py.process.cmdexec("%s freeze" % (pip)) + packages = output.strip().split("\n") + action.setactivity("installed", ",".join(packages)) + envlog = self.resultlog.get_envlog(venv.name) + 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 + venv.test(redirect=redirect) + else: + venv.status = "skipped tests" + + def _summary(self): + self.report.startsummary() + 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: + self.report.skip(msg) + else: + retcode = 1 + self.report.error(msg) + elif status == "platform mismatch": + msg = " %s: %s" % (venv.envconfig.envname, str(status)) + self.report.verbosity1(msg) + elif status and status != "skipped tests": + msg = " %s: %s" % (venv.envconfig.envname, str(status)) + self.report.error(msg) + retcode = 1 + else: + if not status: + status = "commands succeeded" + self.report.good(" %s: %s" % (venv.envconfig.envname, status)) + if not retcode: + self.report.good(" congratulations :)") + + path = self.config.option.resultjson + if path: + path = py.path.local(path) + path.write(self.resultlog.dumps_json()) + self.report.line("wrote json report at: %s" % path) + return retcode + + def showconfig(self): + self.info_versions() + self.report.keyvalue("config-file:", self.config.option.configfile) + self.report.keyvalue("toxinipath: ", self.config.toxinipath) + self.report.keyvalue("toxinidir: ", self.config.toxinidir) + self.report.keyvalue("toxworkdir: ", self.config.toxworkdir) + self.report.keyvalue("setupdir: ", self.config.setupdir) + self.report.keyvalue("distshare: ", self.config.distshare) + self.report.keyvalue("skipsdist: ", self.config.skipsdist) + self.report.tw.line() + for envconfig in self.config.envconfigs.values(): + self.report.line("[testenv:%s]" % envconfig.envname, bold=True) + self.report.line(" basepython=%s" % envconfig.basepython) + self.report.line(" pythoninfo=%s" % (envconfig.python_info,)) + self.report.line(" envpython=%s" % envconfig.envpython) + self.report.line(" envtmpdir=%s" % envconfig.envtmpdir) + self.report.line(" envbindir=%s" % envconfig.envbindir) + self.report.line(" envlogdir=%s" % envconfig.envlogdir) + self.report.line(" changedir=%s" % envconfig.changedir) + self.report.line(" args_are_path=%s" % envconfig.args_are_paths) + self.report.line(" install_command=%s" % + envconfig.install_command) + self.report.line(" commands=") + for command in envconfig.commands: + self.report.line(" %s" % command) + self.report.line(" deps=%s" % envconfig.deps) + self.report.line(" envdir= %s" % envconfig.envdir) + self.report.line(" downloadcache=%s" % envconfig.downloadcache) + self.report.line(" usedevelop=%s" % envconfig.usedevelop) + self.report.line(" setenv=%s" % envconfig.setenv) + self.report.line(" passenv=%s" % envconfig.passenv) + + def showenvs(self): + for env in self.config.envlist: + self.report.line("%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()) + self.report.keyvalue("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()) + self.report.info("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: + self.report.warning("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 = py.std.re.compile("[\w_\-\+\.]+-(.*)(\.zip|\.tar.gz)") + + +def getversion(basename): + m = _rex_getversion.match(basename) + if m is None: + return None + version = m.group(1) + try: + return NormalizedVersion(version) + except IrrationalVersionError: + return None diff --git a/tox/venv.py b/tox/venv.py new file mode 100644 index 0000000..eb2fadc --- /dev/null +++ b/tox/venv.py @@ -0,0 +1,422 @@ +from __future__ import with_statement +import os +import sys +import re +import codecs +import py +import tox +from .config import DepConfig + + +class CreationConfig: + def __init__(self, md5, python, version, sitepackages, + usedevelop, deps): + self.md5 = md5 + self.python = python + self.version = version + self.sitepackages = sitepackages + self.usedevelop = usedevelop + self.deps = deps + + def writeconfig(self, path): + lines = ["%s %s" % (self.md5, self.python)] + lines.append("%s %d %d" % (self.version, self.sitepackages, self.usedevelop)) + for dep in self.deps: + lines.append("%s %s" % dep) + path.ensure() + path.write("\n".join(lines)) + + @classmethod + def readconfig(cls, path): + try: + lines = path.readlines(cr=0) + value = lines.pop(0).split(None, 1) + md5, python = value + version, sitepackages, usedevelop = lines.pop(0).split(None, 3) + sitepackages = bool(int(sitepackages)) + usedevelop = bool(int(usedevelop)) + deps = [] + for line in lines: + md5, depstring = line.split(None, 1) + deps.append((md5, depstring)) + return CreationConfig(md5, python, version, sitepackages, usedevelop, deps) + except Exception: + return None + + def matches(self, other): + return (other 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.deps == other.deps) + + +class VirtualEnv(object): + def __init__(self, envconfig=None, session=None): + self.envconfig = envconfig + self.session = session + self.path = envconfig.envdir + self.path_config = self.path.join(".tox-config1") + + @property + def name(self): + return self.envconfig.envname + + def __repr__(self): + return "" % (self.path) + + def getcommandpath(self, name=None, venv=True, cwd=None): + if name is None: + return self.envconfig.envpython + name = str(name) + if os.path.isabs(name): + return name + if os.path.split(name)[0] == ".": + p = cwd.join(name) + if p.check(): + return str(p) + p = None + if venv: + p = py.path.local.sysfind(name, paths=[self.envconfig.envbindir]) + if p is not None: + return p + p = py.path.local.sysfind(name) + if p is None: + raise tox.exception.InvocationError( + "could not find executable %r" % (name,)) + # p is not found in virtualenv script/bin dir + if venv: + if not self.is_allowed_external(p): + self.session.report.warning( + "test command found but not installed in testenv\n" + " cmd: %s\n" + " env: %s\n" + "Maybe you forgot to specify a dependency? " + "See also the whitelist_externals envconfig setting." % ( + p, self.envconfig.envdir)) + return str(p) # will not be rewritten for reporting + + def is_allowed_external(self, p): + tryadd = [""] + if sys.platform == "win32": + 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 + + def _ispython3(self): + return "python3" in str(self.envconfig.basepython) + + def update(self, action=None): + """ return status string for updating actual venv to match configuration. + if status string is empty, all is ok. + """ + if action is None: + action = self.session.newaction(self, "update") + rconfig = CreationConfig.readconfig(self.path_config) + if not self.envconfig.recreate and rconfig and \ + rconfig.matches(self._getliveconfig()): + action.info("reusing", self.envconfig.envdir) + return + if rconfig is None: + action.setactivity("create", self.envconfig.envdir) + else: + action.setactivity("recreate", self.envconfig.envdir) + try: + self.create(action) + except tox.exception.UnsupportedInterpreter: + return sys.exc_info()[1] + except tox.exception.InterpreterNotFound: + return sys.exc_info()[1] + try: + self.install_deps(action) + except tox.exception.InvocationError: + v = sys.exc_info()[1] + return "could not install deps %s; v = %r" % ( + 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 + deps = [] + for dep in self._getresolvedeps(): + raw_dep = dep.name + md5 = getdigest(raw_dep) + deps.append((md5, raw_dep)) + return CreationConfig(md5, python, version, + sitepackages, develop, deps) + + def _getresolvedeps(self): + l = [] + for dep in self.envconfig.deps: + if dep.indexserver is None: + res = self.session._resolve_pkg(dep.name) + if res != dep.name: + dep = dep.__class__(res) + l.append(dep) + return l + + def getsupportedinterpreter(self): + return self.envconfig.getsupportedinterpreter() + + def matching_platform(self): + return re.match(self.envconfig.platform, sys.platform) + + def create(self, action=None): + # if self.getcommandpath("activate").dirpath().check(): + # return + if action is None: + action = self.session.newaction(self, "create") + + config_interpreter = self.getsupportedinterpreter() + args = [sys.executable, '-m', 'virtualenv'] + if self.envconfig.sitepackages: + args.append('--system-site-packages') + # add interpreter explicitly, to prevent using + # default (virtualenv.ini) + args.extend(['--python', str(config_interpreter)]) + # if sys.platform == "win32": + # f, path, _ = py.std.imp.find_module("virtualenv") + # f.close() + # args[:1] = [str(config_interpreter), str(path)] + # else: + self.session.make_emptydir(self.path) + basepath = self.path.dirpath() + basepath.ensure(dir=1) + args.append(self.path.basename) + self._pcall(args, venv=False, action=action, cwd=basepath) + self.just_created = True + + def finish(self): + self._getliveconfig().writeconfig(self.path_config) + + def _needs_reinstall(self, setupdir, action): + setup_py = setupdir.join('setup.py') + setup_cfg = setupdir.join('setup.cfg') + args = [self.envconfig.envpython, str(setup_py), '--name'] + output = action.popen(args, cwd=setupdir, redirect=False, + returnout=True) + name = output.strip() + egg_info = setupdir.join('.'.join((name, 'egg-info'))) + for conf_file in (setup_py, setup_cfg): + if (not egg_info.check() + or (conf_file.check() and conf_file.mtime() > egg_info.mtime())): + return True + return False + + 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'] + 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'] + self._install([sdistpath], extraopts=extraopts, action=action) + + def install_deps(self, action=None): + if action is None: + action = self.session.newaction(self, "install_deps") + deps = self._getresolvedeps() + if deps: + depinfo = ", ".join(map(str, deps)) + action.setactivity("installdeps", "%s" % depinfo) + self._install(deps, action=action) + + def _installopts(self, indexserver): + l = [] + if indexserver: + l += ["-i", indexserver] + if self.envconfig.downloadcache: + self.envconfig.downloadcache.ensure(dir=1) + l.append("--download-cache=%s" % self.envconfig.downloadcache) + if self.envconfig.pip_pre: + l.append("--pre") + return l + + def run_install_command(self, packages, options=(), + indexserver=None, action=None, + extraenv=None): + argv = self.envconfig.install_command[:] + # use pip-script on win32 to avoid the executable locking + 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__'): + try: + del os.environ[x] + except KeyError: + pass + old_stdout = sys.stdout + sys.stdout = codecs.getwriter('utf8')(sys.stdout) + if extraenv is None: + extraenv = {} + self._pcall(argv, cwd=self.envconfig.config.toxinidir, + extraenv=extraenv, action=action) + sys.stdout = old_stdout + + def _install(self, deps, extraopts=None, action=None): + if not deps: + return + d = {} + l = [] + 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(dep.name) + if ixserver not in l: + l.append(ixserver) + assert ixserver.url is None or isinstance(ixserver.url, str) + + for ixserver in l: + if self.envconfig.config.option.sethome: + extraenv = hack_home_env( + homedir=self.envconfig.envtmpdir.join("pseudo-home"), + index_url=ixserver.url) + else: + extraenv = {} + + packages = d[ixserver] + options = self._installopts(ixserver.url) + if extraopts: + options.extend(extraopts) + self.run_install_command(packages=packages, options=options, + action=action, extraenv=extraenv) + + def _getenv(self, extraenv={}): + env = {} + for envname in self.envconfig.passenv: + if envname in os.environ: + env[envname] = os.environ[envname] + + env.update(self.envconfig.setenv) + + env['VIRTUAL_ENV'] = str(self.path) + + env.update(extraenv) + return env + + def test(self, redirect=False): + action = self.session.newaction(self, "runtests") + with action: + self.status = 0 + self.session.make_emptydir(self.envconfig.envtmpdir) + cwd = self.envconfig.changedir + env = self._getenv() + # Display PYTHONHASHSEED to assist with reproducibility. + action.setactivity("runtests", "PYTHONHASHSEED=%r" % 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[%s] | %s" % (i, ' '.join( + [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) + except tox.exception.InvocationError as err: + self.session.report.error(str(err)) + self.status = "commands failed" + if not self.envconfig.ignore_errors: + break # Don't process remaining commands + except KeyboardInterrupt: + self.status = "keyboardinterrupt" + self.session.report.error(self.status) + raise + + def _pcall(self, args, venv=True, cwd=None, extraenv={}, + action=None, redirect=True, ignore_ret=False): + for name in ("VIRTUALENV_PYTHON", "PYTHONDONTWRITEBYTECODE"): + try: + del os.environ[name] + except KeyError: + pass + assert cwd + cwd.ensure(dir=1) + old = self.patchPATH() + try: + args[0] = self.getcommandpath(args[0], venv, cwd) + env = self._getenv(extraenv) + return action.popen(args, cwd=cwd, env=env, redirect=redirect, ignore_ret=ignore_ret) + finally: + os.environ['PATH'] = old + + def patchPATH(self): + oldPATH = os.environ['PATH'] + bindir = str(self.envconfig.envbindir) + os.environ['PATH'] = os.pathsep.join([bindir, oldPATH]) + self.session.report.verbosity2("setting PATH=%s" % os.environ["PATH"]) + return oldPATH + + +def getdigest(path): + path = py.path.local(path) + if not path.check(file=1): + return "0" * 32 + return path.computehash() + + +def hack_home_env(homedir, index_url=None): + # XXX HACK (this could also live with tox itself, consider) + # if tox uses pip on a package that requires setup_requires + # the index url set with pip is usually not recognized + # because it is setuptools executing very early. + # We therefore run the tox command in an artifical home + # directory and set .pydistutils.cfg and pip.conf files + # accordingly. + if not homedir.check(): + homedir.ensure(dir=1) + d = dict(HOME=str(homedir)) + if not index_url: + index_url = os.environ.get("TOX_INDEX_URL") + if index_url: + homedir.join(".pydistutils.cfg").write( + "[easy_install]\n" + "index_url = %s\n" % index_url) + d["PIP_INDEX_URL"] = index_url + d["TOX_INDEX_URL"] = index_url + return d -- cgit v1.2.1