Best Python code snippet using pytest-benchmark
test_pytest_benchmark_elasticsearch.py
Source:test_pytest_benchmark_elasticsearch.py  
1from __future__ import absolute_import2import json3import logging4import os5from io import BytesIO6from io import StringIO7import elasticsearch8import py9import pytest10from freezegun import freeze_time11from pytest_benchmark import plugin12from pytest_benchmark.plugin import BenchmarkSession13from pytest_benchmark.plugin import pytest_benchmark_compare_machine_info14from pytest_benchmark.plugin import pytest_benchmark_generate_json15from pytest_benchmark.plugin import pytest_benchmark_group_stats16from pytest_benchmark.storage.elasticsearch import ElasticsearchStorage17from pytest_benchmark.storage.elasticsearch import _mask_hosts18from pytest_benchmark.utils import parse_elasticsearch_storage19try:20    import unittest.mock as mock21except ImportError:22    import mock23logger = logging.getLogger(__name__)24THIS = py.path.local(__file__)25BENCHFILE = THIS.dirpath('test_storage/0030_5b78858eb718649a31fb93d8dc96ca2cee41a4cd_20150815_030419_uncommitted-changes.json')26SAVE_DATA = json.loads(BENCHFILE.read_text(encoding='utf8'))27SAVE_DATA["machine_info"] = {'foo': 'bar'}28SAVE_DATA["commit_info"] = {'foo': 'bar'}29tmp = SAVE_DATA.copy()30ES_DATA = tmp.pop("benchmarks")[0]31ES_DATA.update(tmp)32ES_DATA["benchmark_id"] = "FoobarOS_commitId"33class Namespace(object):34    def __init__(self, **kwargs):35        self.__dict__.update(kwargs)36    def __getitem__(self, item):37        return self.__dict__[item]38class LooseFileLike(BytesIO):39    def close(self):40        value = self.getvalue()41        super(LooseFileLike, self).close()42        self.getvalue = lambda: value43class MockStorage(ElasticsearchStorage):44    def __init__(self):45        self._es = mock.Mock(spec=elasticsearch.Elasticsearch)46        self._es_hosts = self._es_index = self._es_doctype = 'mocked'47        self.logger = logger48        self.default_machine_id = "FoobarOS"49class MockSession(BenchmarkSession):50    def __init__(self):51        self.verbose = False52        self.histogram = True53        self.benchmarks = []54        self.performance_regressions = []55        self.sort = u"min"56        self.compare = '0001'57        self.logger = logging.getLogger(__name__)58        self.machine_id = "FoobarOS"59        self.machine_info = {'foo': 'bar'}60        self.save = self.autosave = self.json = False61        self.options = {62            'min_rounds': 123,63            'min_time': 234,64            'max_time': 345,65        }66        self.compare_fail = []67        self.config = Namespace(hook=Namespace(68            pytest_benchmark_group_stats=pytest_benchmark_group_stats,69            pytest_benchmark_generate_machine_info=lambda **kwargs: {'foo': 'bar'},70            pytest_benchmark_update_machine_info=lambda **kwargs: None,71            pytest_benchmark_compare_machine_info=pytest_benchmark_compare_machine_info,72            pytest_benchmark_generate_json=pytest_benchmark_generate_json,73            pytest_benchmark_update_json=lambda **kwargs: None,74            pytest_benchmark_generate_commit_info=lambda **kwargs: {'foo': 'bar'},75            pytest_benchmark_update_commit_info=lambda **kwargs: None,76        ))77        self.elasticsearch_host = "localhost:9200"78        self.elasticsearch_index = "benchmark"79        self.elasticsearch_doctype = "benchmark"80        self.storage = MockStorage()81        self.group_by = 'group'82        self.columns = ['min', 'max', 'mean', 'stddev', 'median', 'iqr',83                        'outliers', 'rounds', 'iterations']84        self.benchmarks = []85        data = json.loads(BENCHFILE.read_text(encoding='utf8'))86        self.benchmarks.extend(87            Namespace(88                as_dict=lambda include_data=False, stats=True, flat=False, _bench=bench:89                    dict(_bench, **_bench["stats"]) if flat else dict(_bench),90                name=bench['name'],91                fullname=bench['fullname'],92                group=bench['group'],93                options=bench['options'],94                has_error=False,95                params=None,96                **bench['stats']97            )98            for bench in data['benchmarks']99        )100try:101    text_type = unicode102except NameError:103    text_type = str104def force_text(text):105    if isinstance(text, text_type):106        return text107    else:108        return text.decode('utf-8')109def force_bytes(text):110    if isinstance(text, text_type):111        return text.encode('utf-8')112    else:113        return text114def make_logger(sess):115    output = StringIO()116    sess.logger = Namespace(117        info=lambda text, **opts: output.write(force_text(text) + u'\n'),118        error=lambda text: output.write(force_text(text) + u'\n'),119    )120    sess.storage.logger = Namespace(121        info=lambda text, **opts: output.write(force_text(text) + u'\n'),122        error=lambda text: output.write(force_text(text) + u'\n'),123    )124    return output125@pytest.fixture126def sess():127    return MockSession()128@pytest.fixture129def logger_output(sess):130    return make_logger(sess)131@freeze_time("2015-08-15T00:04:18.687119")132def test_handle_saving(sess, logger_output, monkeypatch):133    monkeypatch.setattr(plugin, '__version__', '2.5.0')134    sess.save = "commitId"135    sess.autosave = True136    sess.json = None137    sess.save_data = False138    sess.handle_saving()139    sess.storage._es.index.assert_called_with(140        index='mocked',141        doc_type='mocked',142        body=ES_DATA,143        id='FoobarOS_commitId_tests/test_normal.py::test_xfast_parametrized[0]',144    )145def test_parse_with_no_creds():146    string = 'https://example.org,another.org'147    hosts, _, _, _ = parse_elasticsearch_storage(string)148    assert len(hosts) == 2149    assert 'https://example.org' in hosts150    assert 'https://another.org' in hosts151def test_parse_with_creds_in_first_host_of_url():152    string = 'https://user:pass@example.org,another.org'153    hosts, _, _, _ = parse_elasticsearch_storage(string)154    assert len(hosts) == 2155    assert 'https://user:pass@example.org' in hosts156    assert 'https://another.org' in hosts157def test_parse_with_creds_in_second_host_of_url():158    string = 'https://example.org,user:pass@another.org'159    hosts, _, _, _ = parse_elasticsearch_storage(string)160    assert len(hosts) == 2161    assert 'https://example.org' in hosts162    assert 'https://user:pass@another.org' in hosts163def test_parse_with_creds_in_netrc(tmpdir):164    netrc_file = os.path.join(tmpdir.strpath, 'netrc')165    with open(netrc_file, 'w') as f:166        f.write('machine example.org login user1 password pass1\n')167        f.write('machine another.org login user2 password pass2\n')168    string = 'https://example.org,another.org'169    hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)170    assert len(hosts) == 2171    assert 'https://user1:pass1@example.org' in hosts172    assert 'https://user2:pass2@another.org' in hosts173def test_parse_url_creds_supersedes_netrc_creds(tmpdir):174    netrc_file = os.path.join(tmpdir.strpath, 'netrc')175    with open(netrc_file, 'w') as f:176        f.write('machine example.org login user1 password pass1\n')177        f.write('machine another.org login user2 password pass2\n')178    string = 'https://user3:pass3@example.org,another.org'179    hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)180    assert len(hosts) == 2181    assert 'https://user3:pass3@example.org' in hosts  # superseded by creds in url182    assert 'https://user2:pass2@another.org' in hosts  # got creds from netrc file183def test__mask_hosts():184    hosts = ['https://user1:pass1@example.org', 'https://user2:pass2@another.org']185    masked_hosts = _mask_hosts(hosts)186    assert len(masked_hosts) == len(hosts)187    assert 'https://***:***@example.org' in masked_hosts...hookspec.py
Source:hookspec.py  
...46        def pytest_benchmark_update_commit_info(config, commit_info):47            commit_info['message'] = subprocess.check_output(['git', 'log', '-1', '--pretty=%B']).strip()48    """49    pass50def pytest_benchmark_group_stats(config, benchmarks, group_by):51    """52    You may perform grouping customization here, in case the builtin grouping doesn't suit you.53    Example:54    .. sourcecode:: python55        @pytest.mark.hookwrapper56        def pytest_benchmark_group_stats(config, benchmarks, group_by):57            outcome = yield58            if group_by == "special":  # when you use --benchmark-group-by=special59                result = defaultdict(list)60                for bench in benchmarks:61                    # `bench.special` doesn't exist, replace with whatever you need62                    result[bench.special].append(bench)63                outcome.force_result(result.items())64    """65    pass66def pytest_benchmark_generate_json(config, benchmarks, include_data, machine_info, commit_info):67    """68    You should read pytest-benchmark's code if you really need to wholly customize the json.69    .. warning::70        Improperly customizing this may cause breakage if ``--benchmark-compare`` or ``--benchmark-histogram`` are used....Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
