Tool.make_config_dictionary() - Code Metrics - Inspection of "Use (c) instead of unicode character on copyright" - ContinuumIO/ciocheck - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 074cc7...987304 )

by Gonzalo

created 2016-10-22 16:03 UTC

Tool.make_config_dictionary() F

↳ Parent: Tool

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
dl	0
loc	30
rs	3.1764
c	0
b	0
f	0

How to fix Complexity

# -*- 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

        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			# try:
132			# original_init = coverage.summary.SummaryReporter.__init__
133			#
134			# def modified_init(self, coverage, config):
135			# config.skip_covered = True
136			# original_init(self, coverage, config)
137			#
138			# coverage.summary.SummaryReporter.__init__ = modified_init
139			#
140			# print("\nCoverage monkeypatched to skip_covered")
141			# except Exception as e:
142			# print("\nFailed to monkeypatch coverage: {0}".format(str(e)),
143			# file=sys.stderr)
144
145			def run(self, paths):
146			"""Run the tool."""
147			return []
148
149			@classmethod
150			def remove_config(cls, path):
151			"""Remove config file."""
152			pass
153
154
155			class PytestTool(Tool):
156			"""Pytest tool runner."""
157
158			name = 'pytest'
159			language = 'python'
160			extensions = ('py', )
161
162			config_file = 'pytest.ini'
163			config_sections = [('pytest', 'pytest')]
164
165			REPORT_FILE = '.pytestreport.json'
166
167			def __init__(self, cmd_root):
168			"""Pytest tool runner."""
169			super(PytestTool, self).__init__(cmd_root)
170			self.pytest_args = None
171			self.output = None
172			self.coverage_fail = False
173
174			def setup_pytest_coverage_args(self, paths):
175			"""Setup pytest-cov arguments and config file path."""
176			if isinstance(paths, (dict, OrderedDict)):
177			paths = list(sorted(paths.keys()))
178
179			for path in paths:
180			if os.path.isdir(path):
181			cov = '--cov={0}'.format(path)
182			coverage_args = [cov]
183			break
184			else:
185			coverage_args = []
186
187			coverage_config_file = os.path.join(self.cmd_root,
188			COVERAGE_CONFIGURATION_FILE)
189			if os.path.isfile(coverage_config_file):
190			cov_config = ['--cov-config', coverage_config_file]
191			coverage_args = cov_config + coverage_args
192
193			if PY2:
194			# xdist appears to lock up the test suite with python2, maybe due
195			# to an interaction with coverage
196			enable_xdist = []
197			else:
198			enable_xdist = ['-n', str(cpu_count())]
199
200			self.pytest_args = ['--json={0}'.format(self.REPORT_FILE)]
201			self.pytest_args = self.pytest_args + enable_xdist
202			self.pytest_args = self.pytest_args + coverage_args
203
204			def run(self, paths):
205			"""Run pytest test suite."""
206			cmd = paths + self.pytest_args
207
208			try:
209			with ShortOutput(self.cmd_root) as so:
210			errno = pytest.main(cmd)
211			output_lines = ''.join(so.output).lower()
212
213			if 'FAIL Required test coverage'.lower() in output_lines:
214			self.coverage_fail = True
215
216			if errno != 0:
217			print("pytest failed, code {errno}".format(errno=errno))
218			except CoverageError as e:
219			print("Test coverage failure: " + str(e))
220			self.coverage_fail = True
221
222			covered_lines = self.parse_coverage()
223			pytest_report = self.parse_pytest_report()
224
225			results = {'coverage': covered_lines}
226			if pytest_report is not None:
227			results['pytest'] = pytest_report
228			return results
229
230			def parse_pytest_report(self):
231			"""Parse pytest json resport generated by pytest-json."""
232			data = None
233			pytest_report_path = os.path.join(self.cmd_root, self.REPORT_FILE)
234			if os.path.isfile(pytest_report_path):
235			with open(pytest_report_path, 'r') as file_obj:
236			data = json.load(file_obj)
237			return data
238
239			def parse_coverage(self):
240			"""Parse .coverage json report generated by coverage."""
241			coverage_string = ("!coverage.py: This is a private format, don't "
242			"read it directly!")
243			coverage_path = os.path.join(self.cmd_root, '.coverage')
244
245			covered_lines = {}
246			if os.path.isfile(coverage_path):
247			with open(coverage_path, 'r') as file_obj:
248			data = file_obj.read()
249			data = data.replace(coverage_string, '')
250
251			cov = json.loads(data)
252			covered_lines = OrderedDict()
253			lines = cov['lines']
254			for path in sorted(lines):
255			covered_lines[path] = lines[path]
256			return covered_lines
257
258			@classmethod
259			def remove_config(cls, path):
260			"""Remove config file."""
261			super(PytestTool, cls).remove_config(path)
262			remove_file = os.path.join(path, cls.REPORT_FILE)
263			if os.path.isfile(remove_file):
264			os.remove(remove_file)
265
266
267			TOOLS = [
268			CoverageTool,
269			PytestTool,
270			]
271
272
273			def test():
274			"""Main local test."""
275			pass
276
277
278			if __name__ == '__main__':
279			test()
280

ContinuumIO / ciocheck

GitHub Access Token became invalid

Push — master ( 074cc7...987304 )

Tool.make_config_dictionary() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like