BenchmarkSession

Complexity

Total Complexity

27

Size/Duplication

Total Lines	127
Duplicated Lines	0 %

Importance

Changes	15
Bugs	0	Features	0

6 Methods

Rating	Name	Duplication	Size	Complexity
B	check_regressions()	0	9	5
A	finish()	0	9	2
A	display()	0	14	2
C	__init__()	0	55	9
C	prepare_benchmarks()	0	22	8
A	machine_info()	0	8	1

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 cached_property
from .utils import first_or_value
from .utils import get_machine_id
from .utils import load_timer
from .utils import short_filename


class PerformanceRegression(pytest.UsageError):
    pass


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

    def __init__(self, config, report_backend):
        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.report_backend = report_backend

        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"),
        )
        self.skip = config.getoption("benchmark_skip")
        self.disabled = config.getoption("benchmark_disable") and not config.getoption("benchmark_enable")

        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.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)

    @cached_property
    def 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)
                        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)
                flat_bench["path"] = None
                flat_bench["source"] = compared and "NOW"
                yield flat_bench

    def finish(self):
        self.report_backend.handle_saving(self.benchmarks, self.machine_info)
        self.compared_mapping = self.report_backend.handle_loading(self.machine_info)
        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()

    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.")