main() - Code Metrics - ContinuumIO/ciocheck - Measure and Improve Code Quality continuously with Scrutinizer

main() D
last analyzed 2017-01-21 23:18 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	8
dl	0
loc	94
rs	4.1815
c	2
b	0
f	0

How to fix Long Method

# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) 2016 Continuum Analytics, Inc.
#
# Licensed under the terms of the MIT License
# (see LICENSE.txt for details)
# -----------------------------------------------------------------------------
"""CLI Parser for `ciocheck`."""

# Standard library imports
from collections import OrderedDict
import argparse
import os
import shutil
import sys

# Local imports
from ciocheck.config import ALL_FILES, load_config
from ciocheck.files import FileManager
from ciocheck.formatters import FORMATTERS, MULTI_FORMATTERS, MultiFormatter
from ciocheck.linters import LINTERS
from ciocheck.tools import TOOLS


class Runner(object):
    """Main tool runner."""

    def __init__(self, cmd_root, cli_args, folders=None, files=None):
        """Main tool runner."""
        # Run options
        self.cmd_root = cmd_root  # Folder on which the command was executed
        self.config = load_config(cmd_root, cli_args)
        self.file_manager = FileManager(folders=folders, files=files)
        self.folders = folders
        self.files = files
        self.all_results = OrderedDict()
        self.all_tools = {}
        self.test_results = None
        self.failed_checks = set()

        self.check = self.config.get_value('check')
        self.enforce = self.config.get_value('enforce')
        self.diff_mode = self.config.get_value('diff_mode')
        self.file_mode = self.config.get_value('file_mode')
        self.branch = self.config.get_value('branch')
        self.disable_formatters = cli_args.disable_formatters
        self.disable_linters = cli_args.disable_linters
        self.disable_tests = cli_args.disable_tests

    def run(self):
        """Run tools."""
        msg = 'Running ciocheck'
        print('')
        print('=' * len(msg))
        print(msg)
        print('=' * len(msg))
        print('')
        self.clean()

        check_linters = [l for l in LINTERS if l.name in self.check]
        check_formatters = [f for f in FORMATTERS if f.name in self.check]
        check_testers = [t for t in TOOLS if t.name in self.check]
        run_multi = any(f for f in MULTI_FORMATTERS if f.name in self.check)

        # Format before lint, linters may complain about bad formatting

        # Formatters
        if not self.disable_formatters:
            for formatter in check_formatters:
                print('Running "{}" ...'.format(formatter.name))
                tool = formatter(self.cmd_root)
                files = self.file_manager.get_files(
                    branch=self.branch,
                    diff_mode=self.diff_mode,
                    file_mode=self.file_mode,
                    extensions=tool.extensions)
                tool.create_config(self.config)
                self.all_tools[tool.name] = tool
                results = tool.run(files)
                # Pyformat might include files in results that are not in files
                # like when an init is created
                if results:
                    self.all_results[tool.name] = {
                        'files': files,
                        'results': results,
                    }

            # The result of the the multi formatter is special!
            if run_multi:
                print('Running "Multi formatter"')
                tool = MultiFormatter(self.cmd_root, self.check)
                files = self.file_manager.get_files(
                    branch=self.branch,
                    diff_mode=self.diff_mode,
                    file_mode=self.file_mode,
                    extensions=tool.extensions)
                multi_results = tool.run(files)
                for key, values in multi_results.items():
                    self.all_results[key] = {
                        'files': files,
                        'results': values,
                    }

        # Linters
        if not self.disable_linters:
            for linter in check_linters:
                print('Running "{}" ...'.format(linter.name))
                tool = linter(self.cmd_root)
                files = self.file_manager.get_files(
                    branch=self.branch,
                    diff_mode=self.diff_mode,
                    file_mode=self.file_mode,
                    extensions=tool.extensions)
                self.all_tools[tool.name] = tool
                tool.create_config(self.config)
                self.all_results[tool.name] = {
                    'files': files,
                    'results': tool.run(files),
                }

        # Tests
        if not self.disable_tests:
            for tester in check_testers:
                print('Running "{}" ...'.format(tester.name))
                tool = tester(self.cmd_root)
                tool.create_config(self.config)
                self.all_tools[tool.name] = tool

                if tool.name == 'pytest':
                    tool.setup_pytest_coverage_args(self.folders)

                files = self.file_manager.get_files(
                    branch=self.branch,
                    diff_mode=self.diff_mode,
                    file_mode=ALL_FILES,
                    extensions=tool.extensions)
                results = tool.run(files)
                if results:
                    results['files'] = files
                    self.test_results = results

        for tool in LINTERS + FORMATTERS + TOOLS:
            tool.remove_config(self.cmd_root)
        self.clean()

        self.process_results(self.all_results)
        if self.enforce_checks():
            msg = 'Ciocheck successful run'
            print('\n\n' + '=' * len(msg))
            print(msg)
            print('=' * len(msg))
            print('')

    def process_results(self, all_results):
        """Group all results by file path."""
        all_changed_paths = []
        for tool_name, data in all_results.items():
            if data:
                files, results = data['files'], data['results']
                all_changed_paths += [result['path'] for result in results]

        all_changed_paths = list(sorted(set(all_changed_paths)))

        if self.test_results:
            test_files = self.test_results.get('files')
            test_coverage = self.test_results.get('coverage')
        else:
            test_files = []
            test_coverage = []

        for path in all_changed_paths:
            short_path = path.replace(self.cmd_root, '...')
            print('')
            print(short_path)
            print('-' * len(short_path))
            for tool_name, data in all_results.items():
                if data:
                    files, results = data['files'], data['results']
                    lines = [[-1], range(100000)]

                    if isinstance(files, dict):
                        added_lines = files.get(path, lines)[-1]
                    else:
                        added_lines = lines[-1]

                    messages = []
                    for result in results:
                        res_path = result['path']
                        if path == res_path:
                            # LINTERS
                            line = int(result.get('line', -1))
                            created = result.get('created')
                            added_copy = result.get('added-copy')
                            added_header = result.get('added-header')
                            diff = result.get('diff')
                            if line and line in list(added_lines):
                                spaces = (8 - len(str(line))) * ' '
                                args = result.copy()
                                args['spaces'] = spaces
                                msg = ('    {line}:{spaces}'
                                       '{type}: {message}').format(**args)
                                messages.append(msg)

                            # Formatters
                            if created:
                                msg = '    __init__ file created.'
                                messages.append(msg)
                            if added_copy:
                                msg = '    added copyright.'
                                messages.append(msg)
                            if added_header:
                                msg = '    added header.'
                                messages.append(msg)
                            if diff:
                                msg = self.format_diff(diff)
                                messages.append(msg)

                            # TESTERS / COVERAGE

                    test = [r['path'] for r in results if path == r['path']]
                    if test and messages:
                        print('\n  ' + tool_name)
                        print('  ' + '-' * len(tool_name))
                        self.failed_checks.add(tool_name)
                        for message in messages:
                            print(message)

            if isinstance(test_files, dict) and test_files:
                # Asked for lines changed
                if test_coverage:
                    lines_changed_not_covered = []
                    lines = test_files.get(path)
                    lines_added = lines[-1] if lines else []
                    lines_covered = test_coverage.get(path)
                    for line in lines_added:
                        if line not in lines_covered:
                            lines_changed_not_covered.append(str(line))

                    if lines_changed_not_covered:
                        uncov_perc = ((1.0 * len(lines_changed_not_covered)) /
                                      (1.0 * len(lines_added)))
                        cov_perc = (1 - uncov_perc) * 100
                        tool_name = 'coverage'
                        print('\n  ' + tool_name)
                        print('  ' + '-' * len(tool_name))
                        print('    The following lines changed and are not '
                              'covered by tests ({0}%):'.format(cov_perc))
                        print('    ' + ', '.join(lines_changed_not_covered))

        print('')
        pytest_tool = self.all_tools.get('pytest')
        if pytest_tool:
            if pytest_tool.coverage_fail:
                self.failed_checks.add('coverage')

    def enforce_checks(self):
        """Check that enforced checks did not generate reports."""
        if self.test_results:
            if 'pytest' in self.test_results:
                test_summary = self.test_results['pytest']['report']['summary']
                if test_summary.get('failed'):
                    self.failed_checks.add('pytest')
            else:
                self.failed_checks.add('pytest')

        for enforce_tool in self.enforce:
            if enforce_tool in self.failed_checks:
                msg = "Ciocheck failures in: {0}".format(
                    repr(self.failed_checks))
                print('\n\n' + '=' * len(msg))
                print(msg)
                print('=' * len(msg))
                print('')
                sys.exit(1)
                break

        return True

    def format_diff(self, diff, indent='    '):
        """Format diff to include an indentation for console printing."""
        lines = diff.split('\n')
        new_lines = []
        for line in lines:
            new_lines.append(indent + line)
        return '\n'.join(new_lines)

    def clean(self):
        """Remove build directories and temporal config files."""
        # Clean up leftover trash as best we can
        build_tmp = os.path.join(self.cmd_root, 'build', 'tmp')
        if os.path.isdir(build_tmp):
            try:
                shutil.rmtree(build_tmp, ignore_errors=True)
            except Exception:
                pass


def main():
    """CLI `Parser for ciocheck`."""
    description = 'Run Continuum IO test suite.'
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument(
        'folders', help='Folders to analyze. Use from repo root.', nargs='+')
    parser.add_argument(
        '--disable-formatters',
        '-df',
        action='store_true',
        default=False,
        help=('Skip all configured formatters'))
    parser.add_argument(
        '--disable-linters',
        '-dl',
        action='store_true',
        default=False,
        help=('Skip all configured linters'))
    parser.add_argument(
        '--disable-tests',
        '-dt',
        action='store_true',
        default=False,
        help=('Skip running tests'))
    parser.add_argument(
        '--file-mode',
        '-fm',
        dest='file_mode',
        choices=['lines', 'files', 'all'],
        default=None,
        help=('Define if the tool should run on modified '
              'lines of files (default), modified files or '
              'all files'))
    parser.add_argument(
        '--diff-mode',
        '-dm',
        dest='diff_mode',
        choices=['commited', 'staged', 'unstaged'],
        default=None,
        help='Define diff mode. Default mode is commited.')
    parser.add_argument(
        '--branch',
        '-b',
        dest='branch',
        default=None,
        help=('Define branch to compare to. Default branch is '
              '"origin/master"'))
    parser.add_argument(
        '--check',
        '-c',
        dest='check',
        nargs='+',
        choices=[
            'pep8', 'pydocstyle', 'flake8', 'pylint', 'pyformat', 'isort',
            'yapf', 'autopep8', 'coverage', 'pytest'
        ],
        default=None,
        help='Select tools to run. Default is "pep8"')
    parser.add_argument(
        '--enforce',
        '-e',
        dest='enforce',
        choices=[
            'pep8', 'pydocstyle', 'flake8', 'pylint', 'pyformat', 'isort',
            'yapf', 'autopep8', 'coverage', 'pytest'
        ],
        default=None,
        nargs='+',
        help=('Select tools to enforce. Enforced tools will '
              'fail if a result is obtained. Default is '
              'none.'))
    parser.add_argument(
        '--config',
        '-cf',
        dest='config_file',
        default=None,
        help=('Select a config file to use. Default is none.'))

    cli_args = parser.parse_args()
    root = os.getcwd()
    folders = []
    files = []
    for folder_or_file in cli_args.folders:
        folder_or_file = os.path.abspath(folder_or_file)
        if os.path.isfile(folder_or_file):
            files.append(folder_or_file)
        elif os.path.isdir(folder_or_file):
            folders.append(folder_or_file)

    if folders or files:
        test = Runner(root, cli_args, folders=folders, files=files)
        test.run()
    elif not folders and not files:
        print('Invalid folders or files!')


if __name__ == '__main__':
    main()


1			# -- coding: utf-8 --
2			# -----------------------------------------------------------------------------
3			# Copyright (c) 2016 Continuum Analytics, Inc.
4			#
5			# Licensed under the terms of the MIT License
6			# (see LICENSE.txt for details)
7			# -----------------------------------------------------------------------------
8			"""CLI Parser for `ciocheck`."""
9
10			# Standard library imports
11			from collections import OrderedDict
12			import argparse
13			import os
14			import shutil
15			import sys
16
17			# Local imports
18			from ciocheck.config import ALL_FILES, load_config
19			from ciocheck.files import FileManager
20			from ciocheck.formatters import FORMATTERS, MULTI_FORMATTERS, MultiFormatter
21			from ciocheck.linters import LINTERS
22			from ciocheck.tools import TOOLS
23
24
25			class Runner(object):
26			"""Main tool runner."""
27
28			def __init__(self, cmd_root, cli_args, folders=None, files=None):
29			"""Main tool runner."""
30			# Run options
31			self.cmd_root = cmd_root # Folder on which the command was executed
32			self.config = load_config(cmd_root, cli_args)
33			self.file_manager = FileManager(folders=folders, files=files)
34			self.folders = folders
35			self.files = files
36			self.all_results = OrderedDict()
37			self.all_tools = {}
38			self.test_results = None
39			self.failed_checks = set()
40
41			self.check = self.config.get_value('check')
42			self.enforce = self.config.get_value('enforce')
43			self.diff_mode = self.config.get_value('diff_mode')
44			self.file_mode = self.config.get_value('file_mode')
45			self.branch = self.config.get_value('branch')
46			self.disable_formatters = cli_args.disable_formatters
47			self.disable_linters = cli_args.disable_linters
48			self.disable_tests = cli_args.disable_tests
49
50			def run(self):
51			"""Run tools."""
52			msg = 'Running ciocheck'
53			print('')
54			print('=' * len(msg))
55			print(msg)
56			print('=' * len(msg))
57			print('')
58			self.clean()
59
60			check_linters = [l for l in LINTERS if l.name in self.check]
61			check_formatters = [f for f in FORMATTERS if f.name in self.check]
62			check_testers = [t for t in TOOLS if t.name in self.check]
63			run_multi = any(f for f in MULTI_FORMATTERS if f.name in self.check)
64
65			# Format before lint, linters may complain about bad formatting
66
67			# Formatters
68			if not self.disable_formatters:
69			for formatter in check_formatters:
70			print('Running "{}" ...'.format(formatter.name))
71			tool = formatter(self.cmd_root)
72			files = self.file_manager.get_files(
73			branch=self.branch,
74			diff_mode=self.diff_mode,
75			file_mode=self.file_mode,
76			extensions=tool.extensions)
77			tool.create_config(self.config)
78			self.all_tools[tool.name] = tool
79			results = tool.run(files)
80			# Pyformat might include files in results that are not in files
81			# like when an init is created
82			if results:
83			self.all_results[tool.name] = {
84			'files': files,
85			'results': results,
86			}
87
88			# The result of the the multi formatter is special!
89			if run_multi:
90			print('Running "Multi formatter"')
91			tool = MultiFormatter(self.cmd_root, self.check)
92			files = self.file_manager.get_files(
93			branch=self.branch,
94			diff_mode=self.diff_mode,
95			file_mode=self.file_mode,
96			extensions=tool.extensions)
97			multi_results = tool.run(files)
98			for key, values in multi_results.items():
99			self.all_results[key] = {
100			'files': files,
101			'results': values,
102			}
103
104			# Linters
105			if not self.disable_linters:
106			for linter in check_linters:
107			print('Running "{}" ...'.format(linter.name))
108			tool = linter(self.cmd_root)
109			files = self.file_manager.get_files(
110			branch=self.branch,
111			diff_mode=self.diff_mode,
112			file_mode=self.file_mode,
113			extensions=tool.extensions)
114			self.all_tools[tool.name] = tool
115			tool.create_config(self.config)
116			self.all_results[tool.name] = {
117			'files': files,
118			'results': tool.run(files),
119			}
120
121			# Tests
122			if not self.disable_tests:
123			for tester in check_testers:
124			print('Running "{}" ...'.format(tester.name))
125			tool = tester(self.cmd_root)
126			tool.create_config(self.config)
127			self.all_tools[tool.name] = tool
128
129			if tool.name == 'pytest':
130			tool.setup_pytest_coverage_args(self.folders)
131
132			files = self.file_manager.get_files(
133			branch=self.branch,
134			diff_mode=self.diff_mode,
135			file_mode=ALL_FILES,
136			extensions=tool.extensions)
137			results = tool.run(files)
138			if results:
139			results['files'] = files
140			self.test_results = results
141
142			for tool in LINTERS + FORMATTERS + TOOLS:
143			tool.remove_config(self.cmd_root)
144			self.clean()
145
146			self.process_results(self.all_results)
147			if self.enforce_checks():
148			msg = 'Ciocheck successful run'
149			print('\n\n' + '=' * len(msg))
150			print(msg)
151			print('=' * len(msg))
152			print('')
153
154			def process_results(self, all_results):
155			"""Group all results by file path."""
156			all_changed_paths = []
157			for tool_name, data in all_results.items():
158			if data:
159			files, results = data['files'], data['results']
160			all_changed_paths += [result['path'] for result in results]
161
162			all_changed_paths = list(sorted(set(all_changed_paths)))
163
164			if self.test_results:
165			test_files = self.test_results.get('files')
166			test_coverage = self.test_results.get('coverage')
167			else:
168			test_files = []
169			test_coverage = []
170
171			for path in all_changed_paths:
172			short_path = path.replace(self.cmd_root, '...')
173			print('')
174			print(short_path)
175			print('-' * len(short_path))
176			for tool_name, data in all_results.items():
177			if data:
178			files, results = data['files'], data['results']
179			lines = [[-1], range(100000)]
180
181			if isinstance(files, dict):
182			added_lines = files.get(path, lines)[-1]
183			else:
184			added_lines = lines[-1]
185
186			messages = []
187			for result in results:
188			res_path = result['path']
189			if path == res_path:
190			# LINTERS
191			line = int(result.get('line', -1))
192			created = result.get('created')
193			added_copy = result.get('added-copy')
194			added_header = result.get('added-header')
195			diff = result.get('diff')
196			if line and line in list(added_lines):
197			spaces = (8 - len(str(line))) * ' '
198			args = result.copy()
199			args['spaces'] = spaces
200			msg = (' {line}:{spaces}'
201			'{type}: {message}').format(**args)
202			messages.append(msg)
203
204			# Formatters
205			if created:
206			msg = ' __init__ file created.'
207			messages.append(msg)
208			if added_copy:
209			msg = ' added copyright.'
210			messages.append(msg)
211			if added_header:
212			msg = ' added header.'
213			messages.append(msg)
214			if diff:
215			msg = self.format_diff(diff)
216			messages.append(msg)
217
218			# TESTERS / COVERAGE
219
220			test = [r['path'] for r in results if path == r['path']]
221			if test and messages:
222			print('\n ' + tool_name)
223			print(' ' + '-' * len(tool_name))
224			self.failed_checks.add(tool_name)
225			for message in messages:
226			print(message)
227
228			if isinstance(test_files, dict) and test_files:
229			# Asked for lines changed
230			if test_coverage:
231			lines_changed_not_covered = []
232			lines = test_files.get(path)
233			lines_added = lines[-1] if lines else []
234			lines_covered = test_coverage.get(path)
235			for line in lines_added:
236			if line not in lines_covered:
237			lines_changed_not_covered.append(str(line))
238
239			if lines_changed_not_covered:
240			uncov_perc = ((1.0 * len(lines_changed_not_covered)) /
241			(1.0 * len(lines_added)))
242			cov_perc = (1 - uncov_perc) * 100
243			tool_name = 'coverage'
244			print('\n ' + tool_name)
245			print(' ' + '-' * len(tool_name))
246			print(' The following lines changed and are not '
247			'covered by tests ({0}%):'.format(cov_perc))
248			print(' ' + ', '.join(lines_changed_not_covered))
249
250			print('')
251			pytest_tool = self.all_tools.get('pytest')
252			if pytest_tool:
253			if pytest_tool.coverage_fail:
254			self.failed_checks.add('coverage')
255
256			def enforce_checks(self):
257			"""Check that enforced checks did not generate reports."""
258			if self.test_results:
259			if 'pytest' in self.test_results:
260			test_summary = self.test_results['pytest']['report']['summary']
261			if test_summary.get('failed'):
262			self.failed_checks.add('pytest')
263			else:
264			self.failed_checks.add('pytest')
265
266			for enforce_tool in self.enforce:
267			if enforce_tool in self.failed_checks:
268			msg = "Ciocheck failures in: {0}".format(
269			repr(self.failed_checks))
270			print('\n\n' + '=' * len(msg))
271			print(msg)
272			print('=' * len(msg))
273			print('')
274			sys.exit(1)
275			break
276
277			return True
278
279			def format_diff(self, diff, indent=' '):
280			"""Format diff to include an indentation for console printing."""
281			lines = diff.split('\n')
282			new_lines = []
283			for line in lines:
284			new_lines.append(indent + line)
285			return '\n'.join(new_lines)
286
287			def clean(self):
288			"""Remove build directories and temporal config files."""
289			# Clean up leftover trash as best we can
290			build_tmp = os.path.join(self.cmd_root, 'build', 'tmp')
291			if os.path.isdir(build_tmp):
292			try:
293			shutil.rmtree(build_tmp, ignore_errors=True)
294			except Exception:
295			pass
296
297
298			def main():
299			"""CLI `Parser for ciocheck`."""
300			description = 'Run Continuum IO test suite.'
301			parser = argparse.ArgumentParser(description=description)
302			parser.add_argument(
303			'folders', help='Folders to analyze. Use from repo root.', nargs='+')
304			parser.add_argument(
305			'--disable-formatters',
306			'-df',
307			action='store_true',
308			default=False,
309			help=('Skip all configured formatters'))
310			parser.add_argument(
311			'--disable-linters',
312			'-dl',
313			action='store_true',
314			default=False,
315			help=('Skip all configured linters'))
316			parser.add_argument(
317			'--disable-tests',
318			'-dt',
319			action='store_true',
320			default=False,
321			help=('Skip running tests'))
322			parser.add_argument(
323			'--file-mode',
324			'-fm',
325			dest='file_mode',
326			choices=['lines', 'files', 'all'],
327			default=None,
328			help=('Define if the tool should run on modified '
329			'lines of files (default), modified files or '
330			'all files'))
331			parser.add_argument(
332			'--diff-mode',
333			'-dm',
334			dest='diff_mode',
335			choices=['commited', 'staged', 'unstaged'],
336			default=None,
337			help='Define diff mode. Default mode is commited.')
338			parser.add_argument(
339			'--branch',
340			'-b',
341			dest='branch',
342			default=None,
343			help=('Define branch to compare to. Default branch is '
344			'"origin/master"'))
345			parser.add_argument(
346			'--check',
347			'-c',
348			dest='check',
349			nargs='+',
350			choices=[
351			'pep8', 'pydocstyle', 'flake8', 'pylint', 'pyformat', 'isort',
352			'yapf', 'autopep8', 'coverage', 'pytest'
353			],
354			default=None,
355			help='Select tools to run. Default is "pep8"')
356			parser.add_argument(
357			'--enforce',
358			'-e',
359			dest='enforce',
360			choices=[
361			'pep8', 'pydocstyle', 'flake8', 'pylint', 'pyformat', 'isort',
362			'yapf', 'autopep8', 'coverage', 'pytest'
363			],
364			default=None,
365			nargs='+',
366			help=('Select tools to enforce. Enforced tools will '
367			'fail if a result is obtained. Default is '
368			'none.'))
369			parser.add_argument(
370			'--config',
371			'-cf',
372			dest='config_file',
373			default=None,
374			help=('Select a config file to use. Default is none.'))
375
376			cli_args = parser.parse_args()
377			root = os.getcwd()
378			folders = []
379			files = []
380			for folder_or_file in cli_args.folders:
381			folder_or_file = os.path.abspath(folder_or_file)
382			if os.path.isfile(folder_or_file):
383			files.append(folder_or_file)
384			elif os.path.isdir(folder_or_file):
385			folders.append(folder_or_file)
386
387			if folders or files:
388			test = Runner(root, cli_args, folders=folders, files=files)
389			test.run()
390			elif not folders and not files:
391			print('Invalid folders or files!')
392
393
394			if __name__ == '__main__':
395			main()
396

ContinuumIO / ciocheck

GitHub Access Token became invalid

main() D last analyzed 2017-01-21 23:18 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

main() D
last analyzed 2017-01-21 23:18 UTC