1"""
2Automatically package and test a Python project against configurable
3Python2 and Python3 based virtual environments. Environments are
4setup by using virtualenv. Configuration is generally done through an
5INI-style "tox.ini" file.
6"""
7from __future__ import absolute_import, unicode_literals
8
9import json
10import os
11import re
12import subprocess
13import sys
14from collections import OrderedDict
15from contextlib import contextmanager
16
17import py
18
19import tox
20from tox import reporter
21from tox.action import Action
22from tox.config import parseconfig
23from tox.config.parallel import ENV_VAR_KEY as PARALLEL_ENV_VAR_KEY
24from tox.config.parallel import OFF_VALUE as PARALLEL_OFF
25from tox.logs.result import ResultLog
26from tox.reporter import update_default_reporter
27from tox.util import set_os_env_var
28from tox.util.graph import stable_topological_sort
29from tox.util.stdlib import suppress_output
30from tox.venv import VirtualEnv
31
32from .commands.help import show_help
33from .commands.help_ini import show_help_ini
34from .commands.provision import provision_tox
35from .commands.run.parallel import run_parallel
36from .commands.run.sequential import run_sequential
37from .commands.show_config import show_config
38from .commands.show_env import show_envs
39
40
41def cmdline(args=None):
42    if args is None:
43        args = sys.argv[1:]
44    main(args)
45
46
47def setup_reporter(args):
48    from argparse import ArgumentParser
49    from tox.config.reporter import add_verbosity_commands
50
51    parser = ArgumentParser(add_help=False)
52    add_verbosity_commands(parser)
53    with suppress_output():
54        try:
55            options, _ = parser.parse_known_args(args)
56            update_default_reporter(options.quiet_level, options.verbose_level)
57        except SystemExit:
58            pass
59
60
61def main(args):
62    setup_reporter(args)
63    try:
64        config = load_config(args)
65        config.logdir.ensure(dir=1)
66        with set_os_env_var(str("TOX_WORK_DIR"), config.toxworkdir):
67            session = build_session(config)
68            exit_code = session.runcommand()
69        if exit_code is None:
70            exit_code = 0
71        raise SystemExit(exit_code)
72    except tox.exception.BadRequirement:
73        raise SystemExit(1)
74    except KeyboardInterrupt:
75        raise SystemExit(2)
76
77
78def load_config(args):
79    try:
80        config = parseconfig(args)
81        if config.option.help:
82            show_help(config)
83            raise SystemExit(0)
84        elif config.option.helpini:
85            show_help_ini(config)
86            raise SystemExit(0)
87    except tox.exception.MissingRequirement as exception:
88        config = exception.config
89    return config
90
91
92def build_session(config):
93    return Session(config)
94
95
96class Session(object):
97    """The session object that ties together configuration, reporting, venv creation, testing."""
98
99    def __init__(self, config, popen=subprocess.Popen):
100        self._reset(config, popen)
101
102    def _reset(self, config, popen=subprocess.Popen):
103        self.config = config
104        self.popen = popen
105        self.resultlog = ResultLog()
106        self.existing_venvs = OrderedDict()
107        self.venv_dict = {} if self.config.run_provision else self._build_venvs()
108
109    def _build_venvs(self):
110        try:
111            need_to_run = OrderedDict((v, self.getvenv(v)) for v in self._evaluated_env_list)
112            try:
113                venv_order = stable_topological_sort(
114                    OrderedDict((name, v.envconfig.depends) for name, v in need_to_run.items())
115                )
116
117                venvs = OrderedDict((v, need_to_run[v]) for v in venv_order)
118                return venvs
119            except ValueError as exception:
120                reporter.error("circular dependency detected: {}".format(exception))
121        except LookupError:
122            pass
123        except tox.exception.ConfigError as exception:
124            reporter.error(str(exception))
125        raise SystemExit(1)
126
127    def getvenv(self, name):
128        if name in self.existing_venvs:
129            return self.existing_venvs[name]
130        env_config = self.config.envconfigs.get(name, None)
131        if env_config is None:
132            reporter.error("unknown environment {!r}".format(name))
133            raise LookupError(name)
134        elif env_config.envdir == self.config.toxinidir:
135            reporter.error("venv {!r} in {} would delete project".format(name, env_config.envdir))
136            raise tox.exception.ConfigError("envdir must not equal toxinidir")
137        env_log = self.resultlog.get_envlog(name)
138        venv = VirtualEnv(envconfig=env_config, popen=self.popen, env_log=env_log)
139        self.existing_venvs[name] = venv
140        return venv
141
142    @property
143    def _evaluated_env_list(self):
144        tox_env_filter = os.environ.get("TOX_SKIP_ENV")
145        tox_env_filter_re = re.compile(tox_env_filter) if tox_env_filter is not None else None
146        visited = set()
147        for name in self.config.envlist:
148            if name in visited:
149                continue
150            visited.add(name)
151            if tox_env_filter_re is not None and tox_env_filter_re.match(name):
152                msg = "skip environment {}, matches filter {!r}".format(
153                    name, tox_env_filter_re.pattern
154                )
155                reporter.verbosity1(msg)
156                continue
157            yield name
158
159    @property
160    def hook(self):
161        return self.config.pluginmanager.hook
162
163    def newaction(self, name, msg, *args):
164        return Action(
165            name,
166            msg,
167            args,
168            self.config.logdir,
169            self.config.option.resultjson,
170            self.resultlog.command_log,
171            self.popen,
172            sys.executable,
173        )
174
175    def runcommand(self):
176        reporter.using(
177            "tox-{} from {} (pid {})".format(tox.__version__, tox.__file__, os.getpid())
178        )
179        show_description = reporter.has_level(reporter.Verbosity.DEFAULT)
180        if self.config.run_provision:
181            provision_tox_venv = self.getvenv(self.config.provision_tox_env)
182            return provision_tox(provision_tox_venv, self.config.args)
183        else:
184            if self.config.option.showconfig:
185                self.showconfig()
186            elif self.config.option.listenvs:
187                self.showenvs(all_envs=False, description=show_description)
188            elif self.config.option.listenvs_all:
189                self.showenvs(all_envs=True, description=show_description)
190            else:
191                with self.cleanup():
192                    return self.subcommand_test()
193
194    @contextmanager
195    def cleanup(self):
196        self.config.temp_dir.ensure(dir=True)
197        try:
198            yield
199        finally:
200            self.hook.tox_cleanup(session=self)
201
202    def subcommand_test(self):
203        if self.config.skipsdist:
204            reporter.info("skipping sdist step")
205        else:
206            for venv in self.venv_dict.values():
207                if not venv.envconfig.skip_install:
208                    venv.package = self.hook.tox_package(session=self, venv=venv)
209                    if not venv.package:
210                        return 2
211                    venv.envconfig.setenv[str("TOX_PACKAGE")] = str(venv.package)
212        if self.config.option.sdistonly:
213            return
214
215        within_parallel = PARALLEL_ENV_VAR_KEY in os.environ
216        try:
217            if not within_parallel and self.config.option.parallel != PARALLEL_OFF:
218                run_parallel(self.config, self.venv_dict)
219            else:
220                run_sequential(self.config, self.venv_dict)
221        finally:
222            retcode = self._summary()
223        return retcode
224
225    def _add_parallel_summaries(self):
226        if self.config.option.parallel != PARALLEL_OFF and "testenvs" in self.resultlog.dict:
227            result_log = self.resultlog.dict["testenvs"]
228            for tox_env in self.venv_dict.values():
229                data = self._load_parallel_env_report(tox_env)
230                if data and "testenvs" in data and tox_env.name in data["testenvs"]:
231                    result_log[tox_env.name] = data["testenvs"][tox_env.name]
232
233    @staticmethod
234    def _load_parallel_env_report(tox_env):
235        """Load report data into memory, remove disk file"""
236        result_json_path = tox_env.get_result_json_path()
237        if result_json_path and result_json_path.exists():
238            with result_json_path.open("r") as file_handler:
239                data = json.load(file_handler)
240            result_json_path.remove()
241            return data
242
243    def _summary(self):
244        is_parallel_child = PARALLEL_ENV_VAR_KEY in os.environ
245        if not is_parallel_child:
246            reporter.separator("_", "summary", reporter.Verbosity.QUIET)
247        exit_code = 0
248        for venv in self.venv_dict.values():
249            report = reporter.good
250            status = getattr(venv, "status", "undefined")
251            if isinstance(status, tox.exception.InterpreterNotFound):
252                msg = " {}: {}".format(venv.envconfig.envname, str(status))
253                if self.config.option.skip_missing_interpreters == "true":
254                    report = reporter.skip
255                else:
256                    exit_code = 1
257                    report = reporter.error
258            elif status == "platform mismatch":
259                msg = " {}: {} ({!r} does not match {!r})".format(
260                    venv.envconfig.envname, str(status), sys.platform, venv.envconfig.platform
261                )
262                report = reporter.skip
263            elif status and status == "ignored failed command":
264                msg = "  {}: {}".format(venv.envconfig.envname, str(status))
265            elif status and status != "skipped tests":
266                msg = "  {}: {}".format(venv.envconfig.envname, str(status))
267                report = reporter.error
268                exit_code = 1
269            else:
270                if not status:
271                    status = "commands succeeded"
272                msg = "  {}: {}".format(venv.envconfig.envname, status)
273            if not is_parallel_child:
274                report(msg)
275        if not exit_code and not is_parallel_child:
276            reporter.good("  congratulations :)")
277        path = self.config.option.resultjson
278        if path:
279            if not is_parallel_child:
280                self._add_parallel_summaries()
281            path = py.path.local(path)
282            data = self.resultlog.dumps_json()
283            reporter.line("write json report at: {}".format(path))
284            path.write(data)
285        return exit_code
286
287    def showconfig(self):
288        show_config(self.config)
289
290    def showenvs(self, all_envs=False, description=False):
291        show_envs(self.config, all_envs=all_envs, description=description)
292