Tool.__init__() - Code Metrics - ContinuumIO/ciocheck - Measure and Improve Code Quality continuously with Scrutinizer

Tool.init() A
last analyzed 2017-01-21 23:18 UTC

↳ Parent: Tool

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	5
rs	9.4285
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) 2016 Continuum Analytics, Inc.
#
# Licensed under the terms of the MIT License
# (see LICENSE.txt for details)
# -----------------------------------------------------------------------------
"""Generic tools and custom test runner."""

# Standard library imports
from collections import OrderedDict
import ast
import json
import os

# Third party imports
from pytest_cov.plugin import CoverageError
from six import PY2
from six.moves import configparser
import pytest

# Local imports
from ciocheck.config import COVERAGE_CONFIGURATION_FILE
from ciocheck.utils import ShortOutput, cpu_count


class Tool(object):
    """Generic tool object."""

    name = None
    language = None
    extensions = None

    command = None

    # Config
    config_file = None  # '.validconfigfilename'
    config_sections = None  # (('ciocheck:section', 'section'))

    def __init__(self, cmd_root):
        """A Generic tool object."""
        self.cmd_root = cmd_root
        self.config = None
        self.config_options = None  # dict version of the config

    def create_config(self, config):
        """Create a config file for for a given config fname and sections."""
        self.config = config

        if self.config_file and self.config_sections:
            new_config = configparser.ConfigParser()
            new_config_file = os.path.join(self.cmd_root, self.config_file)

            for (cio_config_section, config_section) in self.config_sections:
                if config.has_section(cio_config_section):
                    items = config.items(cio_config_section)
                    new_config.add_section(config_section)

                    for option, value in items:
                        new_config.set(config_section, option, value)

            with open(new_config_file, 'w') as file_obj:
                new_config.write(file_obj)

    @classmethod
    def make_config_dictionary(cls):
        """Turn config into a dictionary for later usage."""
        config_path = os.path.join(cls.cmd_root, cls.config_file)
        config_options = {}

        if os.path.exists(config_path):
            config = configparser.ConfigParser()

            with open(config_path, 'r') as file_obj:
                config.readfp(file_obj)

            for section in config.sections():
                for key in config[section]:
                    value = config[section][key]
                    if ',' in value:
                        value = [v for v in value.split(',') if v]
                    elif value.lower() == 'false':
                        value = False
                    elif value.lower() == 'true':
                        value = True
                    else:
                        try:
                            value = ast.literal_eval(value)  # Numbers
                        except Exception as err:
                            print(err)

                    config_options[key.replace('-', '_')] = value

        return config_options

    @classmethod
    def remove_config(cls, path):
        """Remove config file."""
        if cls.config_file and cls.config_sections:
            remove_file = os.path.join(path, cls.config_file)
            if os.path.isfile(remove_file):
                os.remove(remove_file)

    def run(self, paths):
        """Run the tool."""
        raise NotImplementedError


class CoverageTool(Tool):
    """Coverage tool runner."""

    name = 'coverage'
    language = 'python'
    extensions = ('py', )

    # Config
    config_file = COVERAGE_CONFIGURATION_FILE
    config_sections = [
        ('coverage:run', 'run'),
        ('coverage:report', 'report'),
        ('coverage:html', 'html'),
        ('coverage:xml', 'xml'),
    ]

    def _monkey_path_coverage(self):
        """Enforce the value of `skip_covered`, ignored by pytest-cov.

        pytest-cov ignores the option even if included in the .coveragerc
        configuration file.
        """

#        try:
#            original_init = coverage.summary.SummaryReporter.__init__
#
#            def modified_init(self, coverage, config):
#                config.skip_covered = True
#                original_init(self, coverage, config)
#
#            coverage.summary.SummaryReporter.__init__ = modified_init
#
#            print("\nCoverage monkeypatched to skip_covered")
#        except Exception as e:
#            print("\nFailed to monkeypatch coverage: {0}".format(str(e)),
#                  file=sys.stderr)

    def run(self, paths):
        """Run the tool."""
        return []

    @classmethod
    def remove_config(cls, path):
        """Remove config file."""
        pass


class PytestTool(Tool):
    """Pytest tool runner."""

    name = 'pytest'
    language = 'python'
    extensions = ('py', )

    config_file = 'pytest.ini'
    config_sections = [('pytest', 'pytest')]

    REPORT_FILE = '.pytestreport.json'

    def __init__(self, cmd_root):
        """Pytest tool runner."""
        super(PytestTool, self).__init__(cmd_root)
        self.pytest_args = None
        self.output = None
        self.coverage_fail = False

    def setup_pytest_coverage_args(self, paths):
        """Setup pytest-cov arguments and config file path."""
        if isinstance(paths, (dict, OrderedDict)):
            paths = list(sorted(paths.keys()))

        for path in paths:
            if os.path.isdir(path):
                cov = '--cov={0}'.format(path)
                coverage_args = [cov]
                break
        else:
            coverage_args = []

        coverage_config_file = os.path.join(self.cmd_root,
                                            COVERAGE_CONFIGURATION_FILE)
        if os.path.isfile(coverage_config_file):
            cov_config = ['--cov-config', coverage_config_file]
            coverage_args = cov_config + coverage_args

        if PY2:
            # xdist appears to lock up the test suite with python2, maybe due
            # to an interaction with coverage
            enable_xdist = []
        else:
            enable_xdist = ['-n', str(cpu_count())]

        self.pytest_args = ['--json={0}'.format(self.REPORT_FILE)]
        self.pytest_args = self.pytest_args + enable_xdist
        self.pytest_args = self.pytest_args + coverage_args

    def run(self, paths):
        """Run pytest test suite."""
        cmd = paths + self.pytest_args
        print(cmd)

        try:
            with ShortOutput(self.cmd_root) as so:
                errno = pytest.main(cmd)
            output_lines = ''.join(so.output).lower()

            if 'FAIL Required test coverage'.lower() in output_lines:
                self.coverage_fail = True

            if errno != 0:
                print("pytest failed, code {errno}".format(errno=errno))
        except CoverageError as e:
            print("Test coverage failure: " + str(e))
            self.coverage_fail = True

        covered_lines = self.parse_coverage()
        pytest_report = self.parse_pytest_report()

        results = {'coverage': covered_lines}
        if pytest_report is not None:
            results['pytest'] = pytest_report
        return results

    def parse_pytest_report(self):
        """Parse pytest json resport generated by pytest-json."""
        data = None
        pytest_report_path = os.path.join(self.cmd_root, self.REPORT_FILE)
        if os.path.isfile(pytest_report_path):
            with open(pytest_report_path, 'r') as file_obj:
                data = json.load(file_obj)
        return data

    def parse_coverage(self):
        """Parse .coverage json report generated by coverage."""
        coverage_string = ("!coverage.py: This is a private format, don't "
                           "read it directly!")
        coverage_path = os.path.join(self.cmd_root, '.coverage')

        covered_lines = {}
        if os.path.isfile(coverage_path):
            with open(coverage_path, 'r') as file_obj:
                data = file_obj.read()
                data = data.replace(coverage_string, '')

            cov = json.loads(data)
            covered_lines = OrderedDict()
            lines = cov['lines']
            for path in sorted(lines):
                covered_lines[path] = lines[path]
        return covered_lines

    @classmethod
    def remove_config(cls, path):
        """Remove config file."""
        super(PytestTool, cls).remove_config(path)
        remove_file = os.path.join(path, cls.REPORT_FILE)
        if os.path.isfile(remove_file):
            os.remove(remove_file)


TOOLS = [
    CoverageTool,
    PytestTool,
]


def test():
    """Main local test."""
    pass


if __name__ == '__main__':
    test()


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			"""Generic tools and custom test runner."""
9
10			# Standard library imports
11			from collections import OrderedDict
12			import ast
13			import json
14			import os
15
16			# Third party imports
17			from pytest_cov.plugin import CoverageError
18			from six import PY2
19			from six.moves import configparser
20			import pytest
21
22			# Local imports
23			from ciocheck.config import COVERAGE_CONFIGURATION_FILE
24			from ciocheck.utils import ShortOutput, cpu_count
25
26
27			class Tool(object):
28			"""Generic tool object."""
29
30			name = None
31			language = None
32			extensions = None
33
34			command = None
35
36			# Config
37			config_file = None # '.validconfigfilename'
38			config_sections = None # (('ciocheck:section', 'section'))
39
40			def __init__(self, cmd_root):
41			"""A Generic tool object."""
42			self.cmd_root = cmd_root
43			self.config = None
44			self.config_options = None # dict version of the config
45
46			def create_config(self, config):
47			"""Create a config file for for a given config fname and sections."""
48			self.config = config
49
50			if self.config_file and self.config_sections:
51			new_config = configparser.ConfigParser()
52			new_config_file = os.path.join(self.cmd_root, self.config_file)
53
54			for (cio_config_section, config_section) in self.config_sections:
55			if config.has_section(cio_config_section):
56			items = config.items(cio_config_section)
57			new_config.add_section(config_section)
58
59			for option, value in items:
60			new_config.set(config_section, option, value)
61
62			with open(new_config_file, 'w') as file_obj:
63			new_config.write(file_obj)
64
65			@classmethod
66			def make_config_dictionary(cls):
67			"""Turn config into a dictionary for later usage."""
68			config_path = os.path.join(cls.cmd_root, cls.config_file)
69			config_options = {}
70
71			if os.path.exists(config_path):
72			config = configparser.ConfigParser()
73
74			with open(config_path, 'r') as file_obj:
75			config.readfp(file_obj)
76
77			for section in config.sections():
78			for key in config[section]:
79			value = config[section][key]
80			if ',' in value:
81			value = [v for v in value.split(',') if v]
82			elif value.lower() == 'false':
83			value = False
84			elif value.lower() == 'true':
85			value = True
86			else:
87			try:
88			value = ast.literal_eval(value) # Numbers
89			except Exception as err:
90			print(err)
91
92			config_options[key.replace('-', '_')] = value
93
94			return config_options
95
96			@classmethod
97			def remove_config(cls, path):
98			"""Remove config file."""
99			if cls.config_file and cls.config_sections:
100			remove_file = os.path.join(path, cls.config_file)
101			if os.path.isfile(remove_file):
102			os.remove(remove_file)
103
104			def run(self, paths):
105			"""Run the tool."""
106			raise NotImplementedError
107
108
109			class CoverageTool(Tool):
110			"""Coverage tool runner."""
111
112			name = 'coverage'
113			language = 'python'
114			extensions = ('py', )
115
116			# Config
117			config_file = COVERAGE_CONFIGURATION_FILE
118			config_sections = [
119			('coverage:run', 'run'),
120			('coverage:report', 'report'),
121			('coverage:html', 'html'),
122			('coverage:xml', 'xml'),
123			]
124
125			def _monkey_path_coverage(self):
126			"""Enforce the value of `skip_covered`, ignored by pytest-cov.
127
128			pytest-cov ignores the option even if included in the .coveragerc
129			configuration file.
130			"""
131
132			# try:
133			# original_init = coverage.summary.SummaryReporter.__init__
134			#
135			# def modified_init(self, coverage, config):
136			# config.skip_covered = True
137			# original_init(self, coverage, config)
138			#
139			# coverage.summary.SummaryReporter.__init__ = modified_init
140			#
141			# print("\nCoverage monkeypatched to skip_covered")
142			# except Exception as e:
143			# print("\nFailed to monkeypatch coverage: {0}".format(str(e)),
144			# file=sys.stderr)
145
146			def run(self, paths):
147			"""Run the tool."""
148			return []
149
150			@classmethod
151			def remove_config(cls, path):
152			"""Remove config file."""
153			pass
154
155
156			class PytestTool(Tool):
157			"""Pytest tool runner."""
158
159			name = 'pytest'
160			language = 'python'
161			extensions = ('py', )
162
163			config_file = 'pytest.ini'
164			config_sections = [('pytest', 'pytest')]
165
166			REPORT_FILE = '.pytestreport.json'
167
168			def __init__(self, cmd_root):
169			"""Pytest tool runner."""
170			super(PytestTool, self).__init__(cmd_root)
171			self.pytest_args = None
172			self.output = None
173			self.coverage_fail = False
174
175			def setup_pytest_coverage_args(self, paths):
176			"""Setup pytest-cov arguments and config file path."""
177			if isinstance(paths, (dict, OrderedDict)):
178			paths = list(sorted(paths.keys()))
179
180			for path in paths:
181			if os.path.isdir(path):
182			cov = '--cov={0}'.format(path)
183			coverage_args = [cov]
184			break
185			else:
186			coverage_args = []
187
188			coverage_config_file = os.path.join(self.cmd_root,
189			COVERAGE_CONFIGURATION_FILE)
190			if os.path.isfile(coverage_config_file):
191			cov_config = ['--cov-config', coverage_config_file]
192			coverage_args = cov_config + coverage_args
193
194			if PY2:
195			# xdist appears to lock up the test suite with python2, maybe due
196			# to an interaction with coverage
197			enable_xdist = []
198			else:
199			enable_xdist = ['-n', str(cpu_count())]
200
201			self.pytest_args = ['--json={0}'.format(self.REPORT_FILE)]
202			self.pytest_args = self.pytest_args + enable_xdist
203			self.pytest_args = self.pytest_args + coverage_args
204
205			def run(self, paths):
206			"""Run pytest test suite."""
207			cmd = paths + self.pytest_args
208			print(cmd)
209
210			try:
211			with ShortOutput(self.cmd_root) as so:
212			errno = pytest.main(cmd)
213			output_lines = ''.join(so.output).lower()
214
215			if 'FAIL Required test coverage'.lower() in output_lines:
216			self.coverage_fail = True
217
218			if errno != 0:
219			print("pytest failed, code {errno}".format(errno=errno))
220			except CoverageError as e:
221			print("Test coverage failure: " + str(e))
222			self.coverage_fail = True
223
224			covered_lines = self.parse_coverage()
225			pytest_report = self.parse_pytest_report()
226
227			results = {'coverage': covered_lines}
228			if pytest_report is not None:
229			results['pytest'] = pytest_report
230			return results
231
232			def parse_pytest_report(self):
233			"""Parse pytest json resport generated by pytest-json."""
234			data = None
235			pytest_report_path = os.path.join(self.cmd_root, self.REPORT_FILE)
236			if os.path.isfile(pytest_report_path):
237			with open(pytest_report_path, 'r') as file_obj:
238			data = json.load(file_obj)
239			return data
240
241			def parse_coverage(self):
242			"""Parse .coverage json report generated by coverage."""
243			coverage_string = ("!coverage.py: This is a private format, don't "
244			"read it directly!")
245			coverage_path = os.path.join(self.cmd_root, '.coverage')
246
247			covered_lines = {}
248			if os.path.isfile(coverage_path):
249			with open(coverage_path, 'r') as file_obj:
250			data = file_obj.read()
251			data = data.replace(coverage_string, '')
252
253			cov = json.loads(data)
254			covered_lines = OrderedDict()
255			lines = cov['lines']
256			for path in sorted(lines):
257			covered_lines[path] = lines[path]
258			return covered_lines
259
260			@classmethod
261			def remove_config(cls, path):
262			"""Remove config file."""
263			super(PytestTool, cls).remove_config(path)
264			remove_file = os.path.join(path, cls.REPORT_FILE)
265			if os.path.isfile(remove_file):
266			os.remove(remove_file)
267
268
269			TOOLS = [
270			CoverageTool,
271			PytestTool,
272			]
273
274
275			def test():
276			"""Main local test."""
277			pass
278
279
280			if __name__ == '__main__':
281			test()
282

ContinuumIO / ciocheck

GitHub Access Token became invalid

Tool.__init__() A last analyzed 2017-01-21 23:18 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Tool.init() A
last analyzed 2017-01-21 23:18 UTC