BenchmarkSession.__init__() - Code Metrics - Inspection of "Fix machine_info hooks." - ionelmc/pytest-benchmark - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 779b59...163501 )

by Ionel Cristian

created 2018-06-06 09:38 UTC

BenchmarkSession.init() D

↳ Parent: BenchmarkSession

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
c	0
b	0
f	0
dl	0
loc	66
rs	4.5084

How to fix Long Method

from __future__ import division
from __future__ import print_function

import pytest

from .fixture import statistics
from .fixture import statistics_error
from .logger import Logger
from .table import TableResults
from .utils import NAME_FORMATTERS
from .utils import SecondsDecimal
from .utils import first_or_value
from .utils import get_machine_id
from .utils import load_storage
from .utils import load_timer
from .utils import safe_dumps
from .utils import short_filename


class PerformanceRegression(pytest.UsageError):
    pass


class BenchmarkSession(object):
    compared_mapping = None
    groups = None

    def __init__(self, config):
        self.verbose = config.getoption("benchmark_verbose")
        self.logger = Logger(self.verbose, config)
        self.config = config
        self.performance_regressions = []
        self.benchmarks = []
        self.machine_id = get_machine_id()
        self.storage = load_storage(
            config.getoption("benchmark_storage"),
            logger=self.logger,
            default_machine_id=self.machine_id,
            netrc=config.getoption("benchmark_netrc")
        )
        self.options = dict(
            min_time=SecondsDecimal(config.getoption("benchmark_min_time")),
            min_rounds=config.getoption("benchmark_min_rounds"),
            max_time=SecondsDecimal(config.getoption("benchmark_max_time")),
            timer=load_timer(config.getoption("benchmark_timer")),
            calibration_precision=config.getoption("benchmark_calibration_precision"),
            disable_gc=config.getoption("benchmark_disable_gc"),
            warmup=config.getoption("benchmark_warmup"),
            warmup_iterations=config.getoption("benchmark_warmup_iterations"),
            use_cprofile=bool(config.getoption("benchmark_cprofile")),
        )
        self.skip = config.getoption("benchmark_skip")
        self.disabled = config.getoption("benchmark_disable") and not config.getoption("benchmark_enable")
        self.cprofile_sort_by = config.getoption("benchmark_cprofile")

        if config.getoption("dist", "no") != "no" and not self.skip:
            self.logger.warn(
                "BENCHMARK-U2",
                "Benchmarks are automatically disabled because xdist plugin is active."
                "Benchmarks cannot be performed reliably in a parallelized environment.",
                fslocation="::"
            )
            self.disabled = True
        if hasattr(config, "slaveinput"):
            self.disabled = True
        if not statistics:
            self.logger.warn(
                "BENCHMARK-U3",
                "Benchmarks are automatically disabled because we could not import `statistics`\n\n%s" %
                statistics_error,
                fslocation="::"
            )
            self.disabled = True

        self.only = config.getoption("benchmark_only")
        self.sort = config.getoption("benchmark_sort")
        self.columns = config.getoption("benchmark_columns")
        if self.skip and self.only:
            raise pytest.UsageError("Can't have both --benchmark-only and --benchmark-skip options.")
        if self.disabled and self.only:
            raise pytest.UsageError(
                "Can't have both --benchmark-only and --benchmark-disable options. Note that --benchmark-disable is "
                "automatically activated if xdist is on or you're missing the statistics dependency.")
        self.group_by = config.getoption("benchmark_group_by")
        self.save = config.getoption("benchmark_save")
        self.autosave = config.getoption("benchmark_autosave")
        self.save_data = config.getoption("benchmark_save_data")
        self.json = config.getoption("benchmark_json")
        self.compare = config.getoption("benchmark_compare")
        self.compare_fail = config.getoption("benchmark_compare_fail")
        self.name_format = NAME_FORMATTERS[config.getoption("benchmark_name")]

        self.histogram = first_or_value(config.getoption("benchmark_histogram"), False)

    def get_machine_info(self):
        obj = self.config.hook.pytest_benchmark_generate_machine_info(config=self.config)
        self.config.hook.pytest_benchmark_update_machine_info(
            config=self.config,
            machine_info=obj
        )
        return obj

    def prepare_benchmarks(self):
        for bench in self.benchmarks:
            if bench:
                compared = False
                for path, compared_mapping in self.compared_mapping.items():
                    if bench.fullname in compared_mapping:
                        compared = compared_mapping[bench.fullname]
                        source = short_filename(path, self.machine_id)
                        flat_bench = bench.as_dict(include_data=False, stats=False, cprofile=self.cprofile_sort_by)
                        flat_bench.update(compared["stats"])
                        flat_bench["path"] = str(path)
                        flat_bench["source"] = source
                        if self.compare_fail:
                            for check in self.compare_fail:
                                fail = check.fails(bench, flat_bench)
                                if fail:
                                    self.performance_regressions.append((self.name_format(flat_bench), fail))
                        yield flat_bench
                flat_bench = bench.as_dict(include_data=False, flat=True, cprofile=self.cprofile_sort_by)
                flat_bench["path"] = None
                flat_bench["source"] = compared and "NOW"
                yield flat_bench

    def save_json(self, output_json):
        with self.json as fh:
            fh.write(safe_dumps(output_json, ensure_ascii=True, indent=4).encode())
        self.logger.info("Wrote benchmark data in: %s" % self.json, purple=True)

    def handle_saving(self):
        save = self.save or self.autosave
        if save or self.json:
            if not self.benchmarks:
                self.logger.warn("BENCHMARK-U2", "Not saving anything, no benchmarks have been run!")
                return
            machine_info = self.get_machine_info()
            commit_info = self.config.hook.pytest_benchmark_generate_commit_info(config=self.config)
            self.config.hook.pytest_benchmark_update_commit_info(config=self.config, commit_info=commit_info)

        if self.json:
            output_json = self.config.hook.pytest_benchmark_generate_json(
                config=self.config,
                benchmarks=self.benchmarks,
                include_data=True,
                machine_info=machine_info,
                commit_info=commit_info,
            )
            self.config.hook.pytest_benchmark_update_json(
                config=self.config,
                benchmarks=self.benchmarks,
                output_json=output_json,
            )
            self.save_json(output_json)

        if save:
            output_json = self.config.hook.pytest_benchmark_generate_json(
                config=self.config,
                benchmarks=self.benchmarks,
                include_data=self.save_data,
                machine_info=machine_info,
                commit_info=commit_info,
            )
            self.config.hook.pytest_benchmark_update_json(
                config=self.config,
                benchmarks=self.benchmarks,
                output_json=output_json,
            )
            self.storage.save(output_json, save)

    def handle_loading(self):
        compared_mapping = {}
        if self.compare:
            if self.compare is True:
                compared_benchmarks = list(self.storage.load())[-1:]
            else:
                compared_benchmarks = list(self.storage.load(self.compare))

            if not compared_benchmarks:
                msg = "Can't compare. No benchmark files in %r" % str(self.storage)
                if self.compare is True:
                    msg += ". Can't load the previous benchmark."
                    code = "BENCHMARK-C2"
                else:
                    msg += " match %r." % self.compare
                    code = "BENCHMARK-C1"
                self.logger.warn(code, msg, fslocation=self.storage.location)

            machine_info = self.get_machine_info()
            for path, compared_benchmark in compared_benchmarks:
                self.config.hook.pytest_benchmark_compare_machine_info(
                    config=self.config,
                    benchmarksession=self,
                    machine_info=machine_info,
                    compared_benchmark=compared_benchmark,
                )
                compared_mapping[path] = dict(
                    (bench['fullname'], bench) for bench in compared_benchmark['benchmarks']
                )
                self.logger.info("Comparing against benchmarks from: %s" % path, newline=False)
        self.compared_mapping = compared_mapping

    def finish(self):
        self.handle_saving()
        prepared_benchmarks = list(self.prepare_benchmarks())
        if prepared_benchmarks:
            self.groups = self.config.hook.pytest_benchmark_group_stats(
                config=self.config,
                benchmarks=prepared_benchmarks,
                group_by=self.group_by
            )

    def display(self, tr):
        if not self.groups:
            return

        tr.ensure_newline()
        results_table = TableResults(
            columns=self.columns,
            sort=self.sort,
            histogram=self.histogram,
            name_format=self.name_format,
            logger=self.logger
        )
        results_table.display(tr, self.groups)
        self.check_regressions()
        self.display_cprofile(tr)

    def check_regressions(self):
        if self.compare_fail and not self.compared_mapping:
            raise pytest.UsageError("--benchmark-compare-fail requires valid --benchmark-compare.")

        if self.performance_regressions:
            self.logger.error("Performance has regressed:\n%s" % "\n".join(
                "\t%s - %s" % line for line in self.performance_regressions
            ))
            raise PerformanceRegression("Performance has regressed.")

    def display_cprofile(self, tr):
        if self.options["use_cprofile"]:
            tr.section("cProfile information")
            tr.write_line("Time in s")
            for group in self.groups:
                group_name, benchmarks = group
                for benchmark in benchmarks:
                    tr.write(benchmark["fullname"], yellow=True)
                    if benchmark["source"]:
                        tr.write_line(" ({})".format((benchmark["source"])))
                    else:
                        tr.write("\n")
                    tr.write_line("ncalls\ttottime\tpercall\tcumtime\tpercall\tfilename:lineno(function)")
                    for function_info in benchmark["cprofile"]:
                        line = ("{ncalls_recursion}\t{tottime:.{prec}f}\t{tottime_per:.{prec}f}\t{cumtime:.{prec}f}"
                                "\t{cumtime_per:.{prec}f}\t{function_name}").format(
                            prec=4, **function_info)
                        tr.write_line(line)
                    tr.write("\n")


1			from __future__ import division
2			from __future__ import print_function
3
4			import pytest
5
6			from .fixture import statistics
7			from .fixture import statistics_error
8			from .logger import Logger
9			from .table import TableResults
10			from .utils import NAME_FORMATTERS
11			from .utils import SecondsDecimal
12			from .utils import first_or_value
13			from .utils import get_machine_id
14			from .utils import load_storage
15			from .utils import load_timer
16			from .utils import safe_dumps
17			from .utils import short_filename
18
19
20			class PerformanceRegression(pytest.UsageError):
21			pass
22
23
24			class BenchmarkSession(object):
25			compared_mapping = None
26			groups = None
27
28			def __init__(self, config):
29			self.verbose = config.getoption("benchmark_verbose")
30			self.logger = Logger(self.verbose, config)
31			self.config = config
32			self.performance_regressions = []
33			self.benchmarks = []
34			self.machine_id = get_machine_id()
35			self.storage = load_storage(
36			config.getoption("benchmark_storage"),
37			logger=self.logger,
38			default_machine_id=self.machine_id,
39			netrc=config.getoption("benchmark_netrc")
40			)
41			self.options = dict(
42			min_time=SecondsDecimal(config.getoption("benchmark_min_time")),
43			min_rounds=config.getoption("benchmark_min_rounds"),
44			max_time=SecondsDecimal(config.getoption("benchmark_max_time")),
45			timer=load_timer(config.getoption("benchmark_timer")),
46			calibration_precision=config.getoption("benchmark_calibration_precision"),
47			disable_gc=config.getoption("benchmark_disable_gc"),
48			warmup=config.getoption("benchmark_warmup"),
49			warmup_iterations=config.getoption("benchmark_warmup_iterations"),
50			use_cprofile=bool(config.getoption("benchmark_cprofile")),
51			)
52			self.skip = config.getoption("benchmark_skip")
53			self.disabled = config.getoption("benchmark_disable") and not config.getoption("benchmark_enable")
54			self.cprofile_sort_by = config.getoption("benchmark_cprofile")
55
56			if config.getoption("dist", "no") != "no" and not self.skip:
57			self.logger.warn(
58			"BENCHMARK-U2",
59			"Benchmarks are automatically disabled because xdist plugin is active."
60			"Benchmarks cannot be performed reliably in a parallelized environment.",
61			fslocation="::"
62			)
63			self.disabled = True
64			if hasattr(config, "slaveinput"):
65			self.disabled = True
66			if not statistics:
67			self.logger.warn(
68			"BENCHMARK-U3",
69			"Benchmarks are automatically disabled because we could not import `statistics`\n\n%s" %
70			statistics_error,
71			fslocation="::"
72			)
73			self.disabled = True
74
75			self.only = config.getoption("benchmark_only")
76			self.sort = config.getoption("benchmark_sort")
77			self.columns = config.getoption("benchmark_columns")
78			if self.skip and self.only:
79			raise pytest.UsageError("Can't have both --benchmark-only and --benchmark-skip options.")
80			if self.disabled and self.only:
81			raise pytest.UsageError(
82			"Can't have both --benchmark-only and --benchmark-disable options. Note that --benchmark-disable is "
83			"automatically activated if xdist is on or you're missing the statistics dependency.")
84			self.group_by = config.getoption("benchmark_group_by")
85			self.save = config.getoption("benchmark_save")
86			self.autosave = config.getoption("benchmark_autosave")
87			self.save_data = config.getoption("benchmark_save_data")
88			self.json = config.getoption("benchmark_json")
89			self.compare = config.getoption("benchmark_compare")
90			self.compare_fail = config.getoption("benchmark_compare_fail")
91			self.name_format = NAME_FORMATTERS[config.getoption("benchmark_name")]
92
93			self.histogram = first_or_value(config.getoption("benchmark_histogram"), False)
94
95			def get_machine_info(self):
96			obj = self.config.hook.pytest_benchmark_generate_machine_info(config=self.config)
97			self.config.hook.pytest_benchmark_update_machine_info(
98			config=self.config,
99			machine_info=obj
100			)
101			return obj
102
103			def prepare_benchmarks(self):
104			for bench in self.benchmarks:
105			if bench:
106			compared = False
107			for path, compared_mapping in self.compared_mapping.items():
108			if bench.fullname in compared_mapping:
109			compared = compared_mapping[bench.fullname]
110			source = short_filename(path, self.machine_id)
111			flat_bench = bench.as_dict(include_data=False, stats=False, cprofile=self.cprofile_sort_by)
112			flat_bench.update(compared["stats"])
113			flat_bench["path"] = str(path)
114			flat_bench["source"] = source
115			if self.compare_fail:
116			for check in self.compare_fail:
117			fail = check.fails(bench, flat_bench)
118			if fail:
119			self.performance_regressions.append((self.name_format(flat_bench), fail))
120			yield flat_bench
121			flat_bench = bench.as_dict(include_data=False, flat=True, cprofile=self.cprofile_sort_by)
122			flat_bench["path"] = None
123			flat_bench["source"] = compared and "NOW"
124			yield flat_bench
125
126			def save_json(self, output_json):
127			with self.json as fh:
128			fh.write(safe_dumps(output_json, ensure_ascii=True, indent=4).encode())
129			self.logger.info("Wrote benchmark data in: %s" % self.json, purple=True)
130
131			def handle_saving(self):
132			save = self.save or self.autosave
133			if save or self.json:
134			if not self.benchmarks:
135			self.logger.warn("BENCHMARK-U2", "Not saving anything, no benchmarks have been run!")
136			return
137			machine_info = self.get_machine_info()
138			commit_info = self.config.hook.pytest_benchmark_generate_commit_info(config=self.config)
139			self.config.hook.pytest_benchmark_update_commit_info(config=self.config, commit_info=commit_info)
140
141			if self.json:
142			output_json = self.config.hook.pytest_benchmark_generate_json(
143			config=self.config,
144			benchmarks=self.benchmarks,
145			include_data=True,
146			machine_info=machine_info,
147			commit_info=commit_info,
148			)
149			self.config.hook.pytest_benchmark_update_json(
150			config=self.config,
151			benchmarks=self.benchmarks,
152			output_json=output_json,
153			)
154			self.save_json(output_json)
155
156			if save:
157			output_json = self.config.hook.pytest_benchmark_generate_json(
158			config=self.config,
159			benchmarks=self.benchmarks,
160			include_data=self.save_data,
161			machine_info=machine_info,
162			commit_info=commit_info,
163			)
164			self.config.hook.pytest_benchmark_update_json(
165			config=self.config,
166			benchmarks=self.benchmarks,
167			output_json=output_json,
168			)
169			self.storage.save(output_json, save)
170
171			def handle_loading(self):
172			compared_mapping = {}
173			if self.compare:
174			if self.compare is True:
175			compared_benchmarks = list(self.storage.load())[-1:]
176			else:
177			compared_benchmarks = list(self.storage.load(self.compare))
178
179			if not compared_benchmarks:
180			msg = "Can't compare. No benchmark files in %r" % str(self.storage)
181			if self.compare is True:
182			msg += ". Can't load the previous benchmark."
183			code = "BENCHMARK-C2"
184			else:
185			msg += " match %r." % self.compare
186			code = "BENCHMARK-C1"
187			self.logger.warn(code, msg, fslocation=self.storage.location)
188
189			machine_info = self.get_machine_info()
190			for path, compared_benchmark in compared_benchmarks:
191			self.config.hook.pytest_benchmark_compare_machine_info(
192			config=self.config,
193			benchmarksession=self,
194			machine_info=machine_info,
195			compared_benchmark=compared_benchmark,
196			)
197			compared_mapping[path] = dict(
198			(bench['fullname'], bench) for bench in compared_benchmark['benchmarks']
199			)
200			self.logger.info("Comparing against benchmarks from: %s" % path, newline=False)
201			self.compared_mapping = compared_mapping
202
203			def finish(self):
204			self.handle_saving()
205			prepared_benchmarks = list(self.prepare_benchmarks())
206			if prepared_benchmarks:
207			self.groups = self.config.hook.pytest_benchmark_group_stats(
208			config=self.config,
209			benchmarks=prepared_benchmarks,
210			group_by=self.group_by
211			)
212
213			def display(self, tr):
214			if not self.groups:
215			return
216
217			tr.ensure_newline()
218			results_table = TableResults(
219			columns=self.columns,
220			sort=self.sort,
221			histogram=self.histogram,
222			name_format=self.name_format,
223			logger=self.logger
224			)
225			results_table.display(tr, self.groups)
226			self.check_regressions()
227			self.display_cprofile(tr)
228
229			def check_regressions(self):
230			if self.compare_fail and not self.compared_mapping:
231			raise pytest.UsageError("--benchmark-compare-fail requires valid --benchmark-compare.")
232
233			if self.performance_regressions:
234			self.logger.error("Performance has regressed:\n%s" % "\n".join(
235			"\t%s - %s" % line for line in self.performance_regressions
236			))
237			raise PerformanceRegression("Performance has regressed.")
238
239			def display_cprofile(self, tr):
240			if self.options["use_cprofile"]:
241			tr.section("cProfile information")
242			tr.write_line("Time in s")
243			for group in self.groups:
244			group_name, benchmarks = group
245			for benchmark in benchmarks:
246			tr.write(benchmark["fullname"], yellow=True)
247			if benchmark["source"]:
248			tr.write_line(" ({})".format((benchmark["source"])))
249			else:
250			tr.write("\n")
251			tr.write_line("ncalls\ttottime\tpercall\tcumtime\tpercall\tfilename:lineno(function)")
252			for function_info in benchmark["cprofile"]:
253			line = ("{ncalls_recursion}\t{tottime:.{prec}f}\t{tottime_per:.{prec}f}\t{cumtime:.{prec}f}"
254			"\t{cumtime_per:.{prec}f}\t{function_name}").format(
255			prec=4, **function_info)
256			tr.write_line(line)
257			tr.write("\n")
258

ionelmc / pytest-benchmark

Push — master ( 779b59...163501 )

BenchmarkSession.__init__() D

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

BenchmarkSession.init() D