test_parse_with_creds_in_netrc() - Code Metrics - Inspection of "Merge pull request #73 from drebs/use-netrc-file" - ionelmc/pytest-benchmark - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b996cf...bbb940 )

by Ionel Cristian

created 2017-04-04 18:50 UTC

test_parse_with_creds_in_netrc() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

from __future__ import absolute_import

import json
import logging
import os
from io import BytesIO
from io import StringIO

import elasticsearch
import py
import pytest
from freezegun import freeze_time

from pytest_benchmark import plugin
from pytest_benchmark.plugin import BenchmarkSession
from pytest_benchmark.plugin import pytest_benchmark_compare_machine_info
from pytest_benchmark.plugin import pytest_benchmark_generate_json
from pytest_benchmark.plugin import pytest_benchmark_group_stats
from pytest_benchmark.storage.elasticsearch import ElasticsearchStorage
from pytest_benchmark.storage.elasticsearch import _mask_hosts
from pytest_benchmark.utils import parse_elasticsearch_storage

try:
    import unittest.mock as mock
except ImportError:
    import mock

logger = logging.getLogger(__name__)

THIS = py.path.local(__file__)
BENCHFILE = THIS.dirpath('test_storage/0030_5b78858eb718649a31fb93d8dc96ca2cee41a4cd_20150815_030419_uncommitted-changes.json')
SAVE_DATA = json.load(BENCHFILE.open('rU'))
SAVE_DATA["machine_info"] = {'foo': 'bar'}
SAVE_DATA["commit_info"] = {'foo': 'bar'}

tmp = SAVE_DATA.copy()

ES_DATA = tmp.pop("benchmarks")[0]
ES_DATA.update(tmp)
ES_DATA["benchmark_id"] = "FoobarOS_commitId"


class Namespace(object):
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

    def __getitem__(self, item):
        return self.__dict__[item]


class LooseFileLike(BytesIO):
    def close(self):
        value = self.getvalue()
        super(LooseFileLike, self).close()
        self.getvalue = lambda: value


class MockStorage(ElasticsearchStorage):
    def __init__(self):
        self._es = mock.Mock(spec=elasticsearch.Elasticsearch)
        self._es_hosts = self._es_index = self._es_doctype = 'mocked'
        self.logger = logger
        self.default_machine_id = "FoobarOS"


class MockSession(BenchmarkSession):
    def __init__(self):
        self.verbose = False
        self.histogram = True
        self.benchmarks = []
        self.performance_regressions = []
        self.sort = u"min"
        self.compare = '0001'
        self.logger = logging.getLogger(__name__)
        self.machine_id = "FoobarOS"
        self.machine_info = {'foo': 'bar'}
        self.save = self.autosave = self.json = False
        self.options = {
            'min_rounds': 123,
            'min_time': 234,
            'max_time': 345,
        }
        self.compare_fail = []
        self.config = Namespace(hook=Namespace(
            pytest_benchmark_group_stats=pytest_benchmark_group_stats,
            pytest_benchmark_generate_machine_info=lambda **kwargs: {'foo': 'bar'},
            pytest_benchmark_update_machine_info=lambda **kwargs: None,
            pytest_benchmark_compare_machine_info=pytest_benchmark_compare_machine_info,
            pytest_benchmark_generate_json=pytest_benchmark_generate_json,
            pytest_benchmark_update_json=lambda **kwargs: None,
            pytest_benchmark_generate_commit_info=lambda **kwargs: {'foo': 'bar'},
            pytest_benchmark_update_commit_info=lambda **kwargs: None,
        ))
        self.elasticsearch_host = "localhost:9200"
        self.elasticsearch_index = "benchmark"
        self.elasticsearch_doctype = "benchmark"
        self.storage = MockStorage()
        self.group_by = 'group'
        self.columns = ['min', 'max', 'mean', 'stddev', 'median', 'iqr',
                        'outliers', 'rounds', 'iterations']
        self.benchmarks = []
        with BENCHFILE.open('rU') as fh:
            data = json.load(fh)
        self.benchmarks.extend(
            Namespace(
                as_dict=lambda include_data=False, stats=True, flat=False, _bench=bench:
                    dict(_bench, **_bench["stats"]) if flat else dict(_bench),
                name=bench['name'],
                fullname=bench['fullname'],
                group=bench['group'],
                options=bench['options'],
                has_error=False,
                params=None,
                **bench['stats']
            )
            for bench in data['benchmarks']
        )


try:
    text_type = unicode
except NameError:
    text_type = str


def force_text(text):
    if isinstance(text, text_type):
        return text
    else:
        return text.decode('utf-8')


def force_bytes(text):
    if isinstance(text, text_type):
        return text.encode('utf-8')
    else:
        return text



def make_logger(sess):
    output = StringIO()
    sess.logger = Namespace(
        warn=lambda code, text, **opts: output.write(u"%s: %s %s\n" % (code, force_text(text), opts)),
        info=lambda text, **opts: output.write(force_text(text) + u'\n'),
        error=lambda text: output.write(force_text(text) + u'\n'),
    )
    sess.storage.logger = Namespace(
        warn=lambda code, text, **opts: output.write(u"%s: %s %s\n" % (code, force_text(text), opts)),
        info=lambda text, **opts: output.write(force_text(text) + u'\n'),
        error=lambda text: output.write(force_text(text) + u'\n'),
    )
    return output


@pytest.fixture
def sess():
    return MockSession()


@pytest.fixture
def logger_output(sess):
    return make_logger(sess)


@freeze_time("2015-08-15T00:04:18.687119")
def test_handle_saving(sess, logger_output, monkeypatch):
    monkeypatch.setattr(plugin, '__version__', '2.5.0')
    sess.save = "commitId"
    sess.autosave = True
    sess.json = None
    sess.save_data = False
    sess.handle_saving()
    sess.storage._es.index.assert_called_with(
        index='mocked',
        doc_type='mocked',
        body=ES_DATA,
        id='FoobarOS_commitId_tests/test_normal.py::test_xfast_parametrized[0]',
    )


def test_parse_with_no_creds():
    string = 'https://example.org,another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string)
    assert len(hosts) == 2
    assert 'https://example.org' in hosts
    assert 'https://another.org' in hosts


def test_parse_with_creds_in_first_host_of_url():
    string = 'https://user:[email protected],another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string)
    assert len(hosts) == 2
    assert 'https://user:[email protected]' in hosts
    assert 'https://another.org' in hosts


def test_parse_with_creds_in_second_host_of_url():
    string = 'https://example.org,user:[email protected]'
    hosts, _, _, _ = parse_elasticsearch_storage(string)
    assert len(hosts) == 2
    assert 'https://example.org' in hosts
    assert 'https://user:[email protected]' in hosts


def test_parse_with_creds_in_netrc(tmpdir):
    netrc_file = os.path.join(tmpdir.strpath, 'netrc')
    with open(netrc_file, 'w') as f:
        f.write('machine example.org login user1 password pass1\n')
        f.write('machine another.org login user2 password pass2\n')
    string = 'https://example.org,another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)
    assert len(hosts) == 2
    assert 'https://user1:[email protected]' in hosts
    assert 'https://user2:[email protected]' in hosts


def test_parse_url_creds_supersedes_netrc_creds(tmpdir):
    netrc_file = os.path.join(tmpdir.strpath, 'netrc')
    with open(netrc_file, 'w') as f:
        f.write('machine example.org login user1 password pass1\n')
        f.write('machine another.org login user2 password pass2\n')
    string = 'https://user3:[email protected],another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)
    assert len(hosts) == 2
    assert 'https://user3:[email protected]' in hosts  # superseded by creds in url
    assert 'https://user2:[email protected]' in hosts  # got creds from netrc file


def test__mask_hosts():
    hosts = ['https://user1:[email protected]', 'https://user2:[email protected]']
    masked_hosts = _mask_hosts(hosts)
    assert len(masked_hosts) == len(hosts)
    assert 'https://***:***@example.org' in masked_hosts
    assert 'https://***:***@another.org' in masked_hosts


1		from __future__ import absolute_import
2
3		import json
4		import logging
5		import os
6		from io import BytesIO
7		from io import StringIO
8
9		import elasticsearch
10		import py
11		import pytest
12		from freezegun import freeze_time
13
14		from pytest_benchmark import plugin
15		from pytest_benchmark.plugin import BenchmarkSession
16		from pytest_benchmark.plugin import pytest_benchmark_compare_machine_info
17		from pytest_benchmark.plugin import pytest_benchmark_generate_json
18		from pytest_benchmark.plugin import pytest_benchmark_group_stats
19		from pytest_benchmark.storage.elasticsearch import ElasticsearchStorage
20		from pytest_benchmark.storage.elasticsearch import _mask_hosts
21		from pytest_benchmark.utils import parse_elasticsearch_storage
22
23		try:
24		import unittest.mock as mock
25		except ImportError:
26		import mock
27
28		logger = logging.getLogger(__name__)
29
30		THIS = py.path.local(__file__)
31		BENCHFILE = THIS.dirpath('test_storage/0030_5b78858eb718649a31fb93d8dc96ca2cee41a4cd_20150815_030419_uncommitted-changes.json')
32		SAVE_DATA = json.load(BENCHFILE.open('rU'))
33		SAVE_DATA["machine_info"] = {'foo': 'bar'}
34		SAVE_DATA["commit_info"] = {'foo': 'bar'}
35
36		tmp = SAVE_DATA.copy()
37
38		ES_DATA = tmp.pop("benchmarks")[0]
39		ES_DATA.update(tmp)
40		ES_DATA["benchmark_id"] = "FoobarOS_commitId"
41
42
43		class Namespace(object):
44		def __init__(self, **kwargs):
45		self.__dict__.update(kwargs)
46
47		def __getitem__(self, item):
48		return self.__dict__[item]
49
50
51		class LooseFileLike(BytesIO):
52		def close(self):
53		value = self.getvalue()
54		super(LooseFileLike, self).close()
55		self.getvalue = lambda: value
56
57
58		class MockStorage(ElasticsearchStorage):
59		def __init__(self):
60		self._es = mock.Mock(spec=elasticsearch.Elasticsearch)
61		self._es_hosts = self._es_index = self._es_doctype = 'mocked'
62		self.logger = logger
63		self.default_machine_id = "FoobarOS"
64
65
66		class MockSession(BenchmarkSession):
67		def __init__(self):
68		self.verbose = False
69		self.histogram = True
70		self.benchmarks = []
71		self.performance_regressions = []
72		self.sort = u"min"
73		self.compare = '0001'
74		self.logger = logging.getLogger(__name__)
75		self.machine_id = "FoobarOS"
76		self.machine_info = {'foo': 'bar'}
77		self.save = self.autosave = self.json = False
78		self.options = {
79		'min_rounds': 123,
80		'min_time': 234,
81		'max_time': 345,
82		}
83		self.compare_fail = []
84		self.config = Namespace(hook=Namespace(
85		pytest_benchmark_group_stats=pytest_benchmark_group_stats,
86		pytest_benchmark_generate_machine_info=lambda **kwargs: {'foo': 'bar'},
87		pytest_benchmark_update_machine_info=lambda **kwargs: None,
88		pytest_benchmark_compare_machine_info=pytest_benchmark_compare_machine_info,
89		pytest_benchmark_generate_json=pytest_benchmark_generate_json,
90		pytest_benchmark_update_json=lambda **kwargs: None,
91		pytest_benchmark_generate_commit_info=lambda **kwargs: {'foo': 'bar'},
92		pytest_benchmark_update_commit_info=lambda **kwargs: None,
93		))
94		self.elasticsearch_host = "localhost:9200"
95		self.elasticsearch_index = "benchmark"
96		self.elasticsearch_doctype = "benchmark"
97		self.storage = MockStorage()
98		self.group_by = 'group'
99		self.columns = ['min', 'max', 'mean', 'stddev', 'median', 'iqr',
100		'outliers', 'rounds', 'iterations']
101		self.benchmarks = []
102		with BENCHFILE.open('rU') as fh:
103		data = json.load(fh)
104		self.benchmarks.extend(
105		Namespace(
106		as_dict=lambda include_data=False, stats=True, flat=False, _bench=bench:
107		dict(_bench, **_bench["stats"]) if flat else dict(_bench),
108		name=bench['name'],
109		fullname=bench['fullname'],
110		group=bench['group'],
111		options=bench['options'],
112		has_error=False,
113		params=None,
114		**bench['stats']
115		)
116		for bench in data['benchmarks']
117		)
118
119
120		try:
121		text_type = unicode
122		except NameError:
123		text_type = str
124
125
126		def force_text(text):
127		if isinstance(text, text_type):
128		return text
129		else:
130		return text.decode('utf-8')
131
132
133		def force_bytes(text):
134		if isinstance(text, text_type):
135		return text.encode('utf-8')
136		else:
137	View Code Duplication	return text
		0 ignored issues – show Duplication introduced 2016-09-09 08:37 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
138
139
140		def make_logger(sess):
141		output = StringIO()
142		sess.logger = Namespace(
143		warn=lambda code, text, **opts: output.write(u"%s: %s %s\n" % (code, force_text(text), opts)),
144		info=lambda text, **opts: output.write(force_text(text) + u'\n'),
145		error=lambda text: output.write(force_text(text) + u'\n'),
146		)
147		sess.storage.logger = Namespace(
148		warn=lambda code, text, **opts: output.write(u"%s: %s %s\n" % (code, force_text(text), opts)),
149		info=lambda text, **opts: output.write(force_text(text) + u'\n'),
150		error=lambda text: output.write(force_text(text) + u'\n'),
151		)
152		return output
153
154
155		@pytest.fixture
156		def sess():
157		return MockSession()
158
159
160		@pytest.fixture
161		def logger_output(sess):
162		return make_logger(sess)
163
164
165		@freeze_time("2015-08-15T00:04:18.687119")
166		def test_handle_saving(sess, logger_output, monkeypatch):
167		monkeypatch.setattr(plugin, '__version__', '2.5.0')
168		sess.save = "commitId"
169		sess.autosave = True
170		sess.json = None
171		sess.save_data = False
172		sess.handle_saving()
173		sess.storage._es.index.assert_called_with(
174		index='mocked',
175		doc_type='mocked',
176		body=ES_DATA,
177		id='FoobarOS_commitId_tests/test_normal.py::test_xfast_parametrized[0]',
178		)
179
180
181		def test_parse_with_no_creds():
182		string = 'https://example.org,another.org'
183		hosts, _, _, _ = parse_elasticsearch_storage(string)
184		assert len(hosts) == 2
185		assert 'https://example.org' in hosts
186		assert 'https://another.org' in hosts
187
188
189		def test_parse_with_creds_in_first_host_of_url():
190		string = 'https://user:[email protected],another.org'
191		hosts, _, _, _ = parse_elasticsearch_storage(string)
192		assert len(hosts) == 2
193		assert 'https://user:[email protected]' in hosts
194		assert 'https://another.org' in hosts
195
196
197		def test_parse_with_creds_in_second_host_of_url():
198		string = 'https://example.org,user:[email protected]'
199		hosts, _, _, _ = parse_elasticsearch_storage(string)
200		assert len(hosts) == 2
201		assert 'https://example.org' in hosts
202		assert 'https://user:[email protected]' in hosts
203
204
205		def test_parse_with_creds_in_netrc(tmpdir):
206		netrc_file = os.path.join(tmpdir.strpath, 'netrc')
207		with open(netrc_file, 'w') as f:
208		f.write('machine example.org login user1 password pass1\n')
209		f.write('machine another.org login user2 password pass2\n')
210		string = 'https://example.org,another.org'
211		hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)
212		assert len(hosts) == 2
213		assert 'https://user1:[email protected]' in hosts
214		assert 'https://user2:[email protected]' in hosts
215
216
217		def test_parse_url_creds_supersedes_netrc_creds(tmpdir):
218		netrc_file = os.path.join(tmpdir.strpath, 'netrc')
219		with open(netrc_file, 'w') as f:
220		f.write('machine example.org login user1 password pass1\n')
221		f.write('machine another.org login user2 password pass2\n')
222		string = 'https://user3:[email protected],another.org'
223		hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)
224		assert len(hosts) == 2
225		assert 'https://user3:[email protected]' in hosts # superseded by creds in url
226		assert 'https://user2:[email protected]' in hosts # got creds from netrc file
227
228
229		def test__mask_hosts():
230		hosts = ['https://user1:[email protected]', 'https://user2:[email protected]']
231		masked_hosts = _mask_hosts(hosts)
232		assert len(masked_hosts) == len(hosts)
233		assert 'https://*:*@example.org' in masked_hosts
234		assert 'https://*:*@another.org' in masked_hosts
235

ionelmc / pytest-benchmark

Push — master ( b996cf...bbb940 )

test_parse_with_creds_in_netrc() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like