Source code for tox.session

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

import json
import os
import re
import subprocess
import sys
from collections import OrderedDict
from contextlib import contextmanager

import py

import tox
from tox import reporter
from tox.action import Action
from tox.config import INTERRUPT_TIMEOUT, SUICIDE_TIMEOUT, TERMINATE_TIMEOUT, parseconfig
from tox.config.parallel import ENV_VAR_KEY_PRIVATE as PARALLEL_ENV_VAR_KEY_PRIVATE
from tox.config.parallel import OFF_VALUE as PARALLEL_OFF
from tox.logs.result import ResultLog
from tox.reporter import update_default_reporter
from tox.util import set_os_env_var
from tox.util.graph import stable_topological_sort
from tox.util.stdlib import suppress_output
from tox.venv import VirtualEnv

from .commands.help import show_help
from .commands.help_ini import show_help_ini
from .commands.provision import provision_tox
from .commands.run.parallel import run_parallel
from .commands.run.sequential import run_sequential
from .commands.show_config import show_config
from .commands.show_env import show_envs


def cmdline(args=None):
    if args is None:
        args = sys.argv[1:]
    main(args)


def setup_reporter(args):
    from argparse import ArgumentParser

    from tox.config.reporter import add_verbosity_commands

    parser = ArgumentParser(add_help=False)
    add_verbosity_commands(parser)
    with suppress_output():
        try:
            options, _ = parser.parse_known_args(args)
            update_default_reporter(options.quiet_level, options.verbose_level)
        except SystemExit:
            pass


def main(args):
    setup_reporter(args)
    try:
        config = load_config(args)
        config.logdir.ensure(dir=1)
        with set_os_env_var(str("TOX_WORK_DIR"), config.toxworkdir):
            session = build_session(config)
            exit_code = session.runcommand()
        if exit_code is None:
            exit_code = 0
        raise SystemExit(exit_code)
    except tox.exception.BadRequirement:
        raise SystemExit(1)
    except KeyboardInterrupt:
        raise SystemExit(2)


def load_config(args):
    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)
    except tox.exception.MissingRequirement as exception:
        config = exception.config
    return config


def build_session(config):
    return Session(config)


[docs]class Session(object): """The session object that ties together configuration, reporting, venv creation, testing.""" def __init__(self, config, popen=subprocess.Popen): self._reset(config, popen) def _reset(self, config, popen=subprocess.Popen): self.config = config self.popen = popen self.resultlog = ResultLog() self.existing_venvs = OrderedDict() self.venv_dict = {} if self.config.run_provision else self._build_venvs() def _build_venvs(self): try: need_to_run = OrderedDict((v, self.getvenv(v)) for v in self._evaluated_env_list) try: venv_order = stable_topological_sort( OrderedDict((name, v.envconfig.depends) for name, v in need_to_run.items()), ) venvs = OrderedDict((v, need_to_run[v]) for v in venv_order) return venvs except ValueError as exception: reporter.error("circular dependency detected: {}".format(exception)) except LookupError: pass except tox.exception.ConfigError as exception: reporter.error(str(exception)) raise SystemExit(1) def getvenv(self, name): if name in self.existing_venvs: return self.existing_venvs[name] env_config = self.config.envconfigs.get(name, None) if env_config is None: reporter.error("unknown environment {!r}".format(name)) raise LookupError(name) elif env_config.envdir == self.config.toxinidir: reporter.error("venv {!r} in {} would delete project".format(name, env_config.envdir)) raise tox.exception.ConfigError("envdir must not equal toxinidir") env_log = self.resultlog.get_envlog(name) venv = VirtualEnv(envconfig=env_config, popen=self.popen, env_log=env_log) self.existing_venvs[name] = venv return venv @property def _evaluated_env_list(self): tox_env_filter = os.environ.get("TOX_SKIP_ENV") tox_env_filter_re = re.compile(tox_env_filter) if tox_env_filter is not None else None visited = set() for name in self.config.envlist: if name in visited: continue visited.add(name) if tox_env_filter_re is not None and tox_env_filter_re.match(name): msg = "skip environment {}, matches filter {!r}".format( name, tox_env_filter_re.pattern, ) reporter.verbosity1(msg) continue yield name @property def hook(self): return self.config.pluginmanager.hook def newaction(self, name, msg, *args): return Action( name, msg, args, self.config.logdir, self.config.option.resultjson, self.resultlog.command_log, self.popen, sys.executable, SUICIDE_TIMEOUT, INTERRUPT_TIMEOUT, TERMINATE_TIMEOUT, ) def runcommand(self): reporter.using( "tox-{} from {} (pid {})".format(tox.__version__, tox.__file__, os.getpid()), ) show_description = reporter.has_level(reporter.Verbosity.DEFAULT) if self.config.run_provision: provision_tox_venv = self.getvenv(self.config.provision_tox_env) return provision_tox(provision_tox_venv, self.config.args) else: if self.config.option.showconfig: self.showconfig() elif self.config.option.listenvs: self.showenvs(all_envs=False, description=show_description) elif self.config.option.listenvs_all: self.showenvs(all_envs=True, description=show_description) else: with self.cleanup(): return self.subcommand_test() @contextmanager def cleanup(self): self.config.temp_dir.ensure(dir=True) try: yield finally: self.hook.tox_cleanup(session=self) def subcommand_test(self): if self.config.skipsdist: reporter.info("skipping sdist step") else: for venv in self.venv_dict.values(): if not venv.envconfig.skip_install: venv.package = self.hook.tox_package(session=self, venv=venv) if not venv.package: return 2 venv.envconfig.setenv[str("TOX_PACKAGE")] = str(venv.package) if self.config.option.sdistonly: return within_parallel = PARALLEL_ENV_VAR_KEY_PRIVATE in os.environ try: if not within_parallel and self.config.option.parallel != PARALLEL_OFF: run_parallel(self.config, self.venv_dict) else: run_sequential(self.config, self.venv_dict) finally: retcode = self._summary() return retcode def _add_parallel_summaries(self): if self.config.option.parallel != PARALLEL_OFF and "testenvs" in self.resultlog.dict: result_log = self.resultlog.dict["testenvs"] for tox_env in self.venv_dict.values(): data = self._load_parallel_env_report(tox_env) if data and "testenvs" in data and tox_env.name in data["testenvs"]: result_log[tox_env.name] = data["testenvs"][tox_env.name] @staticmethod def _load_parallel_env_report(tox_env): """Load report data into memory, remove disk file""" result_json_path = tox_env.get_result_json_path() if result_json_path and result_json_path.exists(): with result_json_path.open("r") as file_handler: data = json.load(file_handler) result_json_path.remove() return data def _summary(self): is_parallel_child = PARALLEL_ENV_VAR_KEY_PRIVATE in os.environ if not is_parallel_child: reporter.separator("_", "summary", reporter.Verbosity.QUIET) exit_code = 0 for venv in self.venv_dict.values(): report = reporter.good status = getattr(venv, "status", "undefined") if isinstance(status, tox.exception.InterpreterNotFound): msg = " {}: {}".format(venv.envconfig.envname, str(status)) if self.config.option.skip_missing_interpreters == "true": report = reporter.skip else: exit_code = 1 report = reporter.error elif status == "platform mismatch": msg = " {}: {} ({!r} does not match {!r})".format( venv.envconfig.envname, str(status), sys.platform, venv.envconfig.platform, ) report = reporter.skip elif status and status == "ignored failed command": msg = " {}: {}".format(venv.envconfig.envname, str(status)) elif status and status != "skipped tests": msg = " {}: {}".format(venv.envconfig.envname, str(status)) report = reporter.error exit_code = 1 else: if not status: status = "commands succeeded" msg = " {}: {}".format(venv.envconfig.envname, status) if not is_parallel_child: report(msg) if not exit_code and not is_parallel_child: reporter.good(" congratulations :)") path = self.config.option.resultjson if path: if not is_parallel_child: self._add_parallel_summaries() path = py.path.local(path) data = self.resultlog.dumps_json() reporter.line("write json report at: {}".format(path)) path.write(data) return exit_code def showconfig(self): show_config(self.config) def showenvs(self, all_envs=False, description=False): show_envs(self.config, all_envs=all_envs, description=description)