diff options
| author | maxim-yurchuk <maxim-yurchuk@yandex-team.com> | 2024-10-09 12:29:46 +0300 |
|---|---|---|
| committer | maxim-yurchuk <maxim-yurchuk@yandex-team.com> | 2024-10-09 13:14:22 +0300 |
| commit | 9731d8a4bb7ee2cc8554eaf133bb85498a4c7d80 (patch) | |
| tree | a8fb3181d5947c0d78cf402aa56e686130179049 /contrib/python/requests-mock/py3 | |
| parent | a44b779cd359f06c3ebbef4ec98c6b38609d9d85 (diff) | |
| download | ydb-9731d8a4bb7ee2cc8554eaf133bb85498a4c7d80.tar.gz | |
publishFullContrib: true for ydb
<HIDDEN_URL>
commit_hash:c82a80ac4594723cebf2c7387dec9c60217f603e
Diffstat (limited to 'contrib/python/requests-mock/py3')
15 files changed, 2279 insertions, 0 deletions
diff --git a/contrib/python/requests-mock/py3/patches/01-fix-tests.patch b/contrib/python/requests-mock/py3/patches/01-fix-tests.patch new file mode 100644 index 0000000000..dc5164a523 --- /dev/null +++ b/contrib/python/requests-mock/py3/patches/01-fix-tests.patch @@ -0,0 +1,13 @@ +--- contrib/python/requests-mock/py3/tests/base.py (index) ++++ contrib/python/requests-mock/py3/tests/base.py (working tree) +@@ -10,8 +10,8 @@ + # License for the specific language governing permissions and limitations + # under the License. + +-import testtools ++import unittest + + +-class TestCase(testtools.TestCase): ++class TestCase(unittest.TestCase): + pass diff --git a/contrib/python/requests-mock/py3/patches/02-support-python-3.12.patch b/contrib/python/requests-mock/py3/patches/02-support-python-3.12.patch new file mode 100644 index 0000000000..c876bc8243 --- /dev/null +++ b/contrib/python/requests-mock/py3/patches/02-support-python-3.12.patch @@ -0,0 +1,5 @@ +--- contrib/python/requests-mock/py3/tests/test_mocker.py (index) ++++ contrib/python/requests-mock/py3/tests/test_mocker.py (working tree) +@@ -222 +222 @@ class MockerTests(base.TestCase): +- self.assertEquals( ++ self.assertEqual( diff --git a/contrib/python/requests-mock/py3/tests/__init__.py b/contrib/python/requests-mock/py3/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/__init__.py diff --git a/contrib/python/requests-mock/py3/tests/base.py b/contrib/python/requests-mock/py3/tests/base.py new file mode 100644 index 0000000000..ecc872ecb7 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/base.py @@ -0,0 +1,17 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import unittest + + +class TestCase(unittest.TestCase): + pass diff --git a/contrib/python/requests-mock/py3/tests/pytest/__init__.py b/contrib/python/requests-mock/py3/tests/pytest/__init__.py new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/pytest/__init__.py diff --git a/contrib/python/requests-mock/py3/tests/pytest/pytest.ini b/contrib/python/requests-mock/py3/tests/pytest/pytest.ini new file mode 100644 index 0000000000..58c87f21b7 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/pytest/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +requests_mock_case_sensitive=false diff --git a/contrib/python/requests-mock/py3/tests/pytest/test_with_pytest.py b/contrib/python/requests-mock/py3/tests/pytest/test_with_pytest.py new file mode 100644 index 0000000000..d74908bc38 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/pytest/test_with_pytest.py @@ -0,0 +1,119 @@ +try: + from http import HTTPStatus + HTTP_STATUS_FOUND = HTTPStatus.FOUND +except ImportError: + from httplib import FOUND as HTTP_STATUS_FOUND + +import pytest +import requests +import requests_mock + + +def test_simple(requests_mock): + requests_mock.get('https://httpbin.org/get', text='data') + assert 'data' == requests.get('https://httpbin.org/get').text + + +def test_redirect_and_nesting(): + url_inner = "inner-mock://example.test/" + url_middle = "middle-mock://example.test/" + url_outer = "outer-mock://example.test/" + url_base = "https://www.example.com/" + + text_middle = 'middle' + url_middle + text_outer = 'outer' + url_outer + text_base = 'outer' + url_base + + with requests_mock.Mocker() as outer_mock: + outer_mock.get(url_base, text=text_base) + outer_mock.get(url_outer, text=text_outer) + + with requests_mock.Mocker(real_http=True) as middle_mock: + middle_mock.get(url_middle, text=text_middle) + + with requests_mock.Mocker() as inner_mock: + inner_mock.post(url_inner, + status_code=HTTP_STATUS_FOUND, + headers={'location': url_base}) + inner_mock.get(url_base, real_http=True) + + assert text_base == requests.post(url_inner).text # nosec + + with pytest.raises(requests_mock.NoMockAddress): + requests.get(url_middle) + + with pytest.raises(requests_mock.NoMockAddress): + requests.get(url_outer) + + # back to middle mock + with pytest.raises(requests_mock.NoMockAddress): + requests.post(url_inner) + + assert text_middle == requests.get(url_middle).text # nosec + assert text_outer == requests.get(url_outer).text # nosec + + # back to outter mock + with pytest.raises(requests_mock.NoMockAddress): + requests.post(url_inner) + + with pytest.raises(requests_mock.NoMockAddress): + requests.get(url_middle) + + assert text_outer == requests.get(url_outer).text # nosec + + +def test_mixed_mocks(): + url = 'mock://example.test/' + with requests_mock.Mocker() as global_mock: + global_mock.get(url, text='global') + session = requests.Session() + text = session.get(url).text + assert text == 'global' # nosec + with requests_mock.Mocker(session=session) as session_mock: + session_mock.get(url, real_http=True) + text = session.get(url).text + assert text == 'global' # nosec + + +def test_threaded_sessions(): + """ + When using requests_futures.FuturesSession() with a ThreadPoolExecutor + there is a race condition where one threaded request removes the + monkeypatched get_adapter() method from the Session before another threaded + request is finished using it. + """ + from requests_futures.sessions import FuturesSession + + url1 = 'http://www.example.com/requests-mock-fake-url1' + url2 = 'http://www.example.com/requests-mock-fake-url2' + + with requests_mock.Mocker() as m: + # respond with 204 so we know its us + m.get(url1, status_code=204) + m.get(url2, status_code=204) + + # NOTE(phodge): just firing off two .get() requests right after each + # other was a pretty reliable way to reproduce the race condition on my + # intel Macbook Pro but YMMV. Guaranteeing the race condition to + # reappear might require replacing the Session.send() with a wrapper + # that delays kicking off the request for url1 until the request for + # url2 has restored the original session.get_adapter(), but replacing + # Session.send() could be difficult because the requests_mock.Mocker() + # context manager has *already* monkeypatched this method. + session = FuturesSession() + future1 = session.get(url1) + future2 = session.get(url2) + + # verify both requests were handled by the mock dispatcher + assert future1.result().status_code == 204 + assert future2.result().status_code == 204 + + +class TestClass(object): + + def configure(self, requests_mock): + requests_mock.get('https://httpbin.org/get', text='data') + + def test_one(self, requests_mock): + self.configure(requests_mock) + assert 'data' == requests.get('https://httpbin.org/get').text diff --git a/contrib/python/requests-mock/py3/tests/test_adapter.py b/contrib/python/requests-mock/py3/tests/test_adapter.py new file mode 100644 index 0000000000..4d6a0b4df2 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_adapter.py @@ -0,0 +1,724 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import http.client +import io +import json +import re +import urllib.parse + +import purl +import requests +from urllib3 import HTTPResponse + +import requests_mock +from . import base + + +class MyExc(Exception): + pass + + +class SessionAdapterTests(base.TestCase): + + PREFIX = "mock" + + def setUp(self): + super(SessionAdapterTests, self).setUp() + + self.adapter = requests_mock.Adapter() + self.session = requests.Session() + self.session.mount(self.PREFIX, self.adapter) + + self.url = '%s://example.com/test' % self.PREFIX + self.headers = {'header_a': 'A', 'header_b': 'B'} + + def assertHeaders(self, resp): + for k, v in self.headers.items(): + self.assertEqual(v, resp.headers[k]) + + def assertLastRequest(self, method='GET', body=None): + self.assertEqual(self.url, self.adapter.last_request.url) + self.assertEqual(method, self.adapter.last_request.method) + self.assertEqual(body, self.adapter.last_request.body) + + url_parts = urllib.parse.urlparse(self.url) + qs = urllib.parse.parse_qs(url_parts.query) + self.assertEqual(url_parts.scheme, self.adapter.last_request.scheme) + self.assertEqual(url_parts.netloc, self.adapter.last_request.netloc) + self.assertEqual(url_parts.path, self.adapter.last_request.path) + self.assertEqual(url_parts.query, self.adapter.last_request.query) + self.assertEqual(url_parts.query, self.adapter.last_request.query) + self.assertEqual(qs, self.adapter.last_request.qs) + + def test_content(self): + data = b'testdata' + + self.adapter.register_uri('GET', + self.url, + content=data, + headers=self.headers) + resp = self.session.get(self.url) + self.assertEqual(data, resp.content) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_content_callback(self): + status_code = 401 + data = b'testdata' + + def _content_cb(request, context): + context.status_code = status_code + context.headers.update(self.headers) + return data + + self.adapter.register_uri('GET', + self.url, + content=_content_cb) + resp = self.session.get(self.url) + self.assertEqual(status_code, resp.status_code) + self.assertEqual(data, resp.content) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_text(self): + data = u'testdata' + + self.adapter.register_uri('GET', + self.url, + text=data, + headers=self.headers) + resp = self.session.get(self.url) + self.assertEqual(data.encode('utf-8'), resp.content) + self.assertEqual(data, resp.text) + self.assertEqual('utf-8', resp.encoding) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_text_callback(self): + status_code = 401 + data = u'testdata' + + def _text_cb(request, context): + context.status_code = status_code + context.headers.update(self.headers) + return data + + self.adapter.register_uri('GET', self.url, text=_text_cb) + resp = self.session.get(self.url) + self.assertEqual(status_code, resp.status_code) + self.assertEqual(data, resp.text) + self.assertEqual(data.encode('utf-8'), resp.content) + self.assertEqual('utf-8', resp.encoding) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_raw_callback(self): + status_code = 401 + data = 'testdata' + + def _raw_cb(request, context): + return HTTPResponse( + status=status_code, + headers=self.headers, + body=io.BytesIO(data.encode('utf-8')), + preload_content=False, + reason=http.client.responses.get(status_code), + ) + + self.adapter.register_uri('GET', self.url, raw=_raw_cb) + resp = self.session.get(self.url) + self.assertEqual(status_code, resp.status_code) + self.assertEqual(data, resp.text) + self.assertEqual(data.encode('utf-8'), resp.content) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_json(self): + json_data = {'hello': 'world'} + self.adapter.register_uri('GET', + self.url, + json=json_data, + headers=self.headers) + resp = self.session.get(self.url) + self.assertEqual(b'{"hello": "world"}', resp.content) + self.assertEqual(u'{"hello": "world"}', resp.text) + self.assertEqual(json_data, resp.json()) + self.assertEqual('utf-8', resp.encoding) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_json_callback(self): + status_code = 401 + json_data = {'hello': 'world'} + data = u'{"hello": "world"}' + + def _json_cb(request, context): + context.status_code = status_code + context.headers.update(self.headers) + return json_data + + self.adapter.register_uri('GET', self.url, json=_json_cb) + resp = self.session.get(self.url) + self.assertEqual(status_code, resp.status_code) + self.assertEqual(json_data, resp.json()) + self.assertEqual(data, resp.text) + self.assertEqual(data.encode('utf-8'), resp.content) + self.assertEqual('utf-8', resp.encoding) + self.assertHeaders(resp) + self.assertLastRequest() + + def test_no_body(self): + self.adapter.register_uri('GET', self.url) + resp = self.session.get(self.url) + self.assertEqual(b'', resp.content) + self.assertEqual(200, resp.status_code) + + def test_multiple_body_elements(self): + self.assertRaises(RuntimeError, + self.adapter.register_uri, + self.url, + 'GET', + content=b'b', + text=u'u') + + def test_multiple_responses(self): + inp = [{'status_code': 400, 'text': 'abcd'}, + {'status_code': 300, 'text': 'defg'}, + {'status_code': 200, 'text': 'hijk'}] + + self.adapter.register_uri('GET', self.url, inp) + out = [self.session.get(self.url) for i in range(0, len(inp))] + + for i, o in zip(inp, out): + for k, v in i.items(): + self.assertEqual(v, getattr(o, k)) + + last = self.session.get(self.url) + for k, v in inp[-1].items(): + self.assertEqual(v, getattr(last, k)) + + def test_callback_optional_status(self): + headers = {'a': 'b'} + + def _test_cb(request, context): + context.headers.update(headers) + return '' + + self.adapter.register_uri('GET', + self.url, + text=_test_cb, + status_code=300) + resp = self.session.get(self.url) + self.assertEqual(300, resp.status_code) + + for k, v in headers.items(): + self.assertEqual(v, resp.headers[k]) + + def test_callback_optional_headers(self): + headers = {'a': 'b'} + + def _test_cb(request, context): + context.status_code = 300 + return '' + + self.adapter.register_uri('GET', + self.url, + text=_test_cb, + headers=headers) + + resp = self.session.get(self.url) + self.assertEqual(300, resp.status_code) + + for k, v in headers.items(): + self.assertEqual(v, resp.headers[k]) + + def test_latest_register_overrides(self): + self.adapter.register_uri('GET', self.url, text='abc') + self.adapter.register_uri('GET', self.url, text='def') + + resp = self.session.get(self.url) + self.assertEqual('def', resp.text) + + def test_no_last_request(self): + self.assertIsNone(self.adapter.last_request) + self.assertEqual(0, len(self.adapter.request_history)) + + def test_dont_pass_list_and_kwargs(self): + self.assertRaises(RuntimeError, + self.adapter.register_uri, + 'GET', + self.url, + [{'text': 'a'}], + headers={'a': 'b'}) + + def test_empty_string_return(self): + # '' evaluates as False, so make sure an empty string is not ignored. + self.adapter.register_uri('GET', self.url, text='') + resp = self.session.get(self.url) + self.assertEqual('', resp.text) + + def test_dont_pass_multiple_bodies(self): + self.assertRaises(RuntimeError, + self.adapter.register_uri, + 'GET', + self.url, + json={'abc': 'def'}, + text='ghi') + + def test_dont_pass_unexpected_kwargs(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + unknown='argument') + + def test_dont_pass_unicode_as_content(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + content=u'unicode') + + def test_dont_pass_empty_string_as_content(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + content=u'') + + def test_dont_pass_bytes_as_text(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + text=b'bytes') + + def test_dont_pass_empty_string_as_text(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + text=b'') + + def test_dont_pass_non_str_as_content(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + content=5) + + def test_dont_pass_non_str_as_text(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + text=5) + + def test_with_any_method(self): + self.adapter.register_uri(requests_mock.ANY, self.url, text='resp') + + for m in ('GET', 'HEAD', 'POST', 'UNKNOWN'): + resp = self.session.request(m, self.url) + self.assertEqual('resp', resp.text) + + def test_with_any_url(self): + self.adapter.register_uri('GET', requests_mock.ANY, text='resp') + + for u in ('mock://a', 'mock://b', 'mock://c'): + resp = self.session.get(u) + self.assertEqual('resp', resp.text) + + def test_with_regexp(self): + self.adapter.register_uri('GET', re.compile('tester.com'), text='resp') + + for u in ('mock://www.tester.com/a', 'mock://abc.tester.com'): + resp = self.session.get(u) + self.assertEqual('resp', resp.text) + + def test_with_purl(self): + self.adapter.register_uri('GET', + purl.URL('mock://www.tester.com/a'), + text='resp') + + resp = self.session.get('mock://www.tester.com/a') + self.assertEqual('resp', resp.text) + + def test_requests_in_history_on_no_match(self): + self.assertRaises(requests_mock.NoMockAddress, + self.session.get, + self.url) + + self.assertEqual(self.url, self.adapter.last_request.url) + + def test_requests_in_history_on_exception(self): + + def _test_cb(request, ctx): + raise MyExc() + + self.adapter.register_uri('GET', self.url, text=_test_cb) + + self.assertRaises(MyExc, + self.session.get, + self.url) + + self.assertEqual(self.url, self.adapter.last_request.url) + + def test_not_called_and_called_count(self): + m = self.adapter.register_uri('GET', self.url, text='resp') + self.assertEqual(0, m.call_count) + self.assertFalse(m.called) + self.assertFalse(m.called_once) + + self.assertEqual(0, self.adapter.call_count) + self.assertFalse(self.adapter.called) + self.assertFalse(m.called_once) + + def test_called_and_called_count(self): + m = self.adapter.register_uri('GET', self.url, text='resp') + + resps = [self.session.get(self.url) for i in range(0, 3)] + + for r in resps: + self.assertEqual('resp', r.text) + self.assertEqual(200, r.status_code) + + self.assertEqual(len(resps), m.call_count) + self.assertTrue(m.called) + self.assertFalse(m.called_once) + + self.assertEqual(len(resps), self.adapter.call_count) + self.assertTrue(self.adapter.called) + self.assertFalse(m.called_once) + + def test_reset_reverts_call_count(self): + # Create matchers and add calls to history + call_count = 3 + matcher_count = 3 + for i in range(matcher_count): + url = self.url + str(i) + self.adapter.register_uri('GET', url, text='resp') + for _ in range(call_count): + self.session.get(url) + + # Verify call counts on adapter and matchers + self.assertEqual(self.adapter.call_count, matcher_count * call_count) + for matcher in self.adapter._matchers: + self.assertEqual(matcher.call_count, call_count) + + self.adapter.reset() + + # Verify call counts are 0 after reset + self.assertEqual(self.adapter.call_count, 0) + for matcher in self.adapter._matchers: + self.assertEqual(matcher.call_count, 0) + + def test_adapter_picks_correct_adapter(self): + good = '%s://test3.url/' % self.PREFIX + self.adapter.register_uri('GET', + '%s://test1.url' % self.PREFIX, + text='bad') + self.adapter.register_uri('GET', + '%s://test2.url' % self.PREFIX, + text='bad') + self.adapter.register_uri('GET', good, text='good') + self.adapter.register_uri('GET', + '%s://test4.url' % self.PREFIX, + text='bad') + + resp = self.session.get(good) + + self.assertEqual('good', resp.text) + + def test_adapter_is_connection(self): + url = '%s://test.url' % self.PREFIX + text = 'text' + self.adapter.register_uri('GET', url, text=text) + resp = self.session.get(url) + + self.assertEqual(text, resp.text) + self.assertIs(self.adapter, resp.connection) + + def test_send_to_connection(self): + url1 = '%s://test1.url/' % self.PREFIX + url2 = '%s://test2.url/' % self.PREFIX + + text1 = 'text1' + text2 = 'text2' + + self.adapter.register_uri('GET', url1, text=text1) + self.adapter.register_uri('GET', url2, text=text2) + + req = requests.Request(method='GET', url=url2).prepare() + + resp1 = self.session.get(url1) + self.assertEqual(text1, resp1.text) + + resp2 = resp1.connection.send(req) + self.assertEqual(text2, resp2.text) + + def test_request_json_with_str_data(self): + dict_req = {'hello': 'world'} + dict_resp = {'goodbye': 'world'} + + m = self.adapter.register_uri('POST', self.url, json=dict_resp) + + data = json.dumps(dict_req) + resp = self.session.post(self.url, data=data) + + self.assertIs(data, m.last_request.body) + self.assertEqual(dict_resp, resp.json()) + self.assertEqual(dict_req, m.last_request.json()) + + def test_request_json_with_bytes_data(self): + dict_req = {'hello': 'world'} + dict_resp = {'goodbye': 'world'} + + m = self.adapter.register_uri('POST', self.url, json=dict_resp) + + data = json.dumps(dict_req).encode('utf-8') + resp = self.session.post(self.url, data=data) + + self.assertIs(data, m.last_request.body) + self.assertEqual(dict_resp, resp.json()) + self.assertEqual(dict_req, m.last_request.json()) + + def test_request_json_with_cb(self): + dict_req = {'hello': 'world'} + dict_resp = {'goodbye': 'world'} + data = json.dumps(dict_req) + + def _cb(req, context): + self.assertEqual(dict_req, req.json()) + return dict_resp + + m = self.adapter.register_uri('POST', self.url, json=_cb) + resp = self.session.post(self.url, data=data) + + self.assertEqual(1, m.call_count) + self.assertTrue(m.called_once) + self.assertEqual(dict_resp, resp.json()) + + def test_raises_exception(self): + self.adapter.register_uri('GET', self.url, exc=MyExc) + + self.assertRaises(MyExc, + self.session.get, + self.url) + + self.assertTrue(self.adapter.called_once) + self.assertEqual(self.url, self.adapter.last_request.url) + + def test_raises_exception_with_body_args_fails(self): + self.assertRaises(TypeError, + self.adapter.register_uri, + 'GET', + self.url, + exc=MyExc, + text='fail') + + def test_sets_request_matcher_in_history(self): + url1 = '%s://test1.url/' % self.PREFIX + url2 = '%s://test2.url/' % self.PREFIX + + text1 = 'text1' + text2 = 'text2' + + m1 = self.adapter.register_uri('GET', url1, text=text1) + m2 = self.adapter.register_uri('GET', url2, text=text2) + + resp1 = self.session.get(url1) + resp2 = self.session.get(url2) + + self.assertEqual(text1, resp1.text) + self.assertEqual(text2, resp2.text) + + self.assertEqual(2, self.adapter.call_count) + self.assertFalse(self.adapter.called_once) + + self.assertEqual(url1, self.adapter.request_history[0].url) + self.assertEqual(url2, self.adapter.request_history[1].url) + + self.assertIs(m1, self.adapter.request_history[0].matcher) + self.assertIs(m2, self.adapter.request_history[1].matcher) + + def test_sets_request_matcher_on_exception(self): + m = self.adapter.register_uri('GET', self.url, exc=MyExc) + + self.assertRaises(MyExc, + self.session.get, + self.url) + + self.assertEqual(self.url, self.adapter.last_request.url) + self.assertIs(m, self.adapter.last_request.matcher) + + def test_cookies_from_header(self): + headers = {'Set-Cookie': 'fig=newton; Path=/test; domain=.example.com'} + self.adapter.register_uri('GET', + self.url, + text='text', + headers=headers) + + resp = self.session.get(self.url) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual(['/test'], resp.cookies.list_paths()) + self.assertEqual(['.example.com'], resp.cookies.list_domains()) + + def test_cookies_from_dict(self): + # This is a syntax we get from requests. I'm not sure i like it. + self.adapter.register_uri('GET', + self.url, + text='text', + cookies={'fig': 'newton', 'sugar': 'apple'}) + + resp = self.session.get(self.url) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual('apple', resp.cookies['sugar']) + + def test_cookies_with_jar(self): + jar = requests_mock.CookieJar() + jar.set('fig', 'newton', path='/foo', domain='.example.com') + jar.set('sugar', 'apple', path='/bar', domain='.example.com') + + self.adapter.register_uri('GET', self.url, text='text', cookies=jar) + resp = self.session.get(self.url) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual('apple', resp.cookies['sugar']) + self.assertEqual({'/foo', '/bar'}, set(resp.cookies.list_paths())) + self.assertEqual(['.example.com'], resp.cookies.list_domains()) + + def test_cookies_header_with_cb(self): + + def _cb(request, context): + val = 'fig=newton; Path=/test; domain=.example.com' + context.headers['Set-Cookie'] = val + return 'text' + + self.adapter.register_uri('GET', self.url, text=_cb) + resp = self.session.get(self.url) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual(['/test'], resp.cookies.list_paths()) + self.assertEqual(['.example.com'], resp.cookies.list_domains()) + + def test_cookies_from_dict_with_cb(self): + def _cb(request, context): + # converted into a jar by now + context.cookies.set('sugar', 'apple', path='/test') + return 'text' + + self.adapter.register_uri('GET', + self.url, + text=_cb, + cookies={'fig': 'newton'}) + + resp = self.session.get(self.url) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual('apple', resp.cookies['sugar']) + self.assertEqual(['/', '/test'], resp.cookies.list_paths()) + + def test_cookies_with_jar_cb(self): + def _cb(request, context): + context.cookies.set('sugar', + 'apple', + path='/bar', + domain='.example.com') + return 'text' + + jar = requests_mock.CookieJar() + jar.set('fig', 'newton', path='/foo', domain='.example.com') + + self.adapter.register_uri('GET', self.url, text=_cb, cookies=jar) + resp = self.session.get(self.url) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual('apple', resp.cookies['sugar']) + self.assertEqual({'/foo', '/bar'}, set(resp.cookies.list_paths())) + self.assertEqual(['.example.com'], resp.cookies.list_domains()) + + def test_reading_closed_fp(self): + self.adapter.register_uri('GET', self.url, text='abc') + resp = self.session.get(self.url) + + # raw will have been closed during the request reading + self.assertTrue(resp.raw.closed) + + data = resp.raw.read() + + self.assertIsInstance(data, bytes) + self.assertEqual(0, len(data)) + + def test_case_sensitive_headers(self): + data = 'testdata' + headers = {'aBcDe': 'FgHiJ'} + + self.adapter.register_uri('GET', self.url, text=data) + resp = self.session.get(self.url, headers=headers) + + self.assertEqual('GET', self.adapter.last_request.method) + self.assertEqual(200, resp.status_code) + self.assertEqual(data, resp.text) + + for k, v in headers.items(): + self.assertEqual(v, self.adapter.last_request.headers[k]) + + def test_case_sensitive_history(self): + self.adapter._case_sensitive = True + + data = 'testdata' + netloc = 'examPlE.CoM' + path = '/TesTER' + query = 'aBC=deF' + + mock_url = '%s://%s%s' % (self.PREFIX, netloc.lower(), path) + request_url = '%s://%s%s?%s' % (self.PREFIX, netloc, path, query) + + # test that the netloc is ignored when actually making the request + self.adapter.register_uri('GET', mock_url, text=data) + resp = self.session.get(request_url) + + self.assertEqual('GET', self.adapter.last_request.method) + self.assertEqual(200, resp.status_code) + self.assertEqual(data, resp.text) + + # but even still the mixed case parameters come out in history + self.assertEqual(netloc, self.adapter.last_request.netloc) + self.assertEqual(path, self.adapter.last_request.path) + self.assertEqual(query, self.adapter.last_request.query) + + def test_stream_none(self): + text = 'hello world' + + self.adapter.register_uri('GET', + self.url, + text=text, + headers=self.headers) + + resp = self.session.get(self.url, stream=True) + resps = [c for c in resp.iter_content(None, decode_unicode=True)] + self.assertEqual([text], resps) + + def test_stream_size(self): + text = 'hello world' + + self.adapter.register_uri('GET', + self.url, + text=text, + headers=self.headers) + + resp = self.session.get(self.url, stream=True) + resps = [c for c in resp.iter_content(3, decode_unicode=True)] + self.assertEqual(['hel', 'lo ', 'wor', 'ld'], resps) diff --git a/contrib/python/requests-mock/py3/tests/test_custom_matchers.py b/contrib/python/requests-mock/py3/tests/test_custom_matchers.py new file mode 100644 index 0000000000..73e23822f5 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_custom_matchers.py @@ -0,0 +1,74 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import requests + +import requests_mock +from . import base + + +class FailMatcher(object): + + def __init___(self): + self.called = False + + def __call__(self, request): + self.called = True + return None + + +def match_all(request): + return requests_mock.create_response(request, content=b'data') + + +class CustomMatchersTests(base.TestCase): + + def assertMatchAll(self, resp): + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.text, u'data') + + @requests_mock.Mocker() + def test_custom_matcher(self, mocker): + mocker.add_matcher(match_all) + + resp = requests.get('http://any/thing') + self.assertMatchAll(resp) + + @requests_mock.Mocker() + def test_failing_matcher(self, mocker): + failer = FailMatcher() + + mocker.add_matcher(match_all) + mocker.add_matcher(failer) + + resp = requests.get('http://any/thing') + + self.assertMatchAll(resp) + self.assertTrue(failer.called) + + @requests_mock.Mocker() + def test_some_pass(self, mocker): + + def matcher_a(request): + if 'a' in request.url: + return match_all(request) + + return None + + mocker.add_matcher(matcher_a) + + resp = requests.get('http://any/thing') + self.assertMatchAll(resp) + + self.assertRaises(requests_mock.NoMockAddress, + requests.get, + 'http://other/thing') diff --git a/contrib/python/requests-mock/py3/tests/test_fixture.py b/contrib/python/requests-mock/py3/tests/test_fixture.py new file mode 100644 index 0000000000..63cbb99141 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_fixture.py @@ -0,0 +1,39 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import requests +import requests_mock +from requests_mock.contrib import fixture +from . import base + + +class MockingTests(base.TestCase): + + def setUp(self): + super(MockingTests, self).setUp() + self.mocker = self.useFixture(fixture.Fixture()) + + def test_failure(self): + self.assertRaises(requests_mock.NoMockAddress, + requests.get, + 'http://www.google.com') + + def test_basic(self): + test_url = 'http://www.google.com/' + self.mocker.register_uri('GET', test_url, text='response') + + resp = requests.get(test_url) + self.assertEqual('response', resp.text) + self.assertEqual(test_url, self.mocker.last_request.url) + + def test_fixture_has_normal_attr_error(self): + self.assertRaises(AttributeError, lambda: self.mocker.unknown) diff --git a/contrib/python/requests-mock/py3/tests/test_matcher.py b/contrib/python/requests-mock/py3/tests/test_matcher.py new file mode 100644 index 0000000000..a30e195966 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_matcher.py @@ -0,0 +1,324 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import re + +from requests_mock import adapter +from . import base +from requests_mock.response import _MatcherResponse + +ANY = adapter.ANY + + +class TestMatcher(base.TestCase): + + def match(self, + target, + url, + matcher_method='GET', + request_method='GET', + complete_qs=False, + headers=None, + request_data=None, + request_headers={}, + additional_matcher=None, + real_http=False, + case_sensitive=False): + matcher = adapter._Matcher(matcher_method, + target, + [], + complete_qs=complete_qs, + additional_matcher=additional_matcher, + request_headers=request_headers, + real_http=real_http, + case_sensitive=case_sensitive) + request = adapter._RequestObjectProxy._create(request_method, + url, + headers, + data=request_data) + return matcher._match(request) + + def assertMatch(self, + target=ANY, + url='http://example.com/requests-mock', + matcher_method='GET', + request_method='GET', + **kwargs): + self.assertEqual(True, + self.match(target, + url, + matcher_method=matcher_method, + request_method=request_method, + **kwargs), + 'Matcher %s %s failed to match %s %s' % + (matcher_method, target, request_method, url)) + + def assertMatchBoth(self, + target=ANY, + url='http://example.com/requests-mock', + matcher_method='GET', + request_method='GET', + **kwargs): + self.assertMatch(target, + url, + matcher_method=matcher_method, + request_method=request_method, + **kwargs) + self.assertMatch(url, + target, + matcher_method=request_method, + request_method=matcher_method, + **kwargs) + + def assertNoMatch(self, + target=ANY, + url='http://example.com/requests-mock', + matcher_method='GET', + request_method='GET', + **kwargs): + self.assertEqual(False, + self.match(target, + url, + matcher_method=matcher_method, + request_method=request_method, + **kwargs), + 'Matcher %s %s unexpectedly matched %s %s' % + (matcher_method, target, request_method, url)) + + def assertNoMatchBoth(self, + target=ANY, + url='http://example.com/requests-mock', + matcher_method='GET', + request_method='GET', + **kwargs): + self.assertNoMatch(target, + url, + matcher_method=matcher_method, + request_method=request_method, + **kwargs) + self.assertNoMatch(url, + target, + matcher_method=request_method, + request_method=matcher_method, + **kwargs) + + def assertMatchMethodBoth(self, matcher_method, request_method, **kwargs): + url = 'http://www.test.com' + + self.assertMatchBoth(url, + url, + request_method=request_method, + matcher_method=matcher_method, + **kwargs) + + def assertNoMatchMethodBoth(self, + matcher_method, + request_method, + **kwargs): + url = 'http://www.test.com' + + self.assertNoMatchBoth(url, + url, + request_method=request_method, + matcher_method=matcher_method, + **kwargs) + + def test_url_matching(self): + self.assertMatchBoth('http://www.test.com', + 'http://www.test.com') + self.assertMatchBoth('http://www.test.com', + 'http://www.test.com/') + self.assertMatchBoth('http://www.test.com/abc', + 'http://www.test.com/abc') + self.assertMatchBoth('http://www.test.com:5000/abc', + 'http://www.test.com:5000/abc') + self.assertNoMatchBoth('https://www.test.com', + 'http://www.test.com') + self.assertNoMatchBoth('http://www.test.com/abc', + 'http://www.test.com') + self.assertNoMatchBoth('http://test.com', + 'http://www.test.com') + self.assertNoMatchBoth('http://test.com', + 'http://www.test.com') + self.assertNoMatchBoth('http://test.com/abc', + 'http://www.test.com/abc/') + self.assertNoMatchBoth('http://test.com/abc/', + 'http://www.test.com/abc') + self.assertNoMatchBoth('http://test.com:5000/abc/', + 'http://www.test.com/abc') + self.assertNoMatchBoth('http://test.com/abc/', + 'http://www.test.com:5000/abc') + + def test_quotation(self): + self.assertMatchBoth('http://www.test.com/a string%url', + 'http://www.test.com/a string%url') + self.assertMatchBoth('http://www.test.com/ABC 123', + 'http://www.test.com/ABC%20123') + self.assertMatchBoth('http://www.test.com/user@example.com', + 'http://www.test.com/user@example.com') + + def test_subset_match(self): + self.assertMatch('/path', 'http://www.test.com/path') + self.assertMatch('/path', 'http://www.test.com/path') + self.assertMatch('//www.test.com/path', 'http://www.test.com/path') + self.assertMatch('//www.test.com/path', 'https://www.test.com/path') + + def test_query_string(self): + self.assertMatch('/path?a=1&b=2', + 'http://www.test.com/path?a=1&b=2') + self.assertMatch('/path?a=1', + 'http://www.test.com/path?a=1&b=2', + complete_qs=False) + self.assertNoMatch('/path?a=1', + 'http://www.test.com/path?a=1&b=2', + complete_qs=True) + self.assertNoMatch('/path?a=1&b=2', + 'http://www.test.com/path?a=1') + + def test_query_empty_string(self): + self.assertMatch('/path?a', + 'http://www.test.com/path?a') + self.assertMatch('/path?bob&paul', + 'http://www.test.com/path?paul&bob') + self.assertNoMatch('/path?bob', + 'http://www.test.com/path?paul') + self.assertNoMatch('/path?pual&bob', + 'http://www.test.com/path?bob') + + def test_method_match(self): + self.assertNoMatchMethodBoth('GET', 'POST') + self.assertMatchMethodBoth('GET', 'get') + self.assertMatchMethodBoth('GeT', 'geT') + + def test_match_ANY_url(self): + self.assertMatch(ANY, 'http://anything') + self.assertMatch(ANY, 'http://somethingelse') + self.assertNoMatch(ANY, 'http://somethingelse', request_method='POST') + + def test_match_ANY_method(self): + for m in ('GET', 'POST', 'HEAD', 'OPTION'): + self.assertMatch('http://www.test.com', + 'http://www.test.com', + matcher_method=ANY, + request_method=m) + + self.assertNoMatch('http://www.test.com', + 'http://another', + matcher_method=ANY) + + def test_match_with_regex(self): + r1 = re.compile('test.com/a') + r2 = re.compile('/b/c') + + self.assertMatch(r1, 'http://mock.test.com/a/b') + self.assertMatch(r1, 'http://test.com/a/') + self.assertMatch(r1, 'mock://test.com/a/b') + self.assertNoMatch(r1, 'mock://test.com/') + + self.assertMatch(r2, 'http://anything/a/b/c/d') + self.assertMatch(r2, 'mock://anything/a/b/c/d') + + def test_match_with_headers(self): + self.assertMatch('/path', + 'http://www.test.com/path', + headers={'A': 'abc', 'b': 'def'}, + request_headers={'a': 'abc'}) + + self.assertMatch('/path', + 'http://www.test.com/path', + headers={'A': 'abc', 'b': 'def'}) + + self.assertNoMatch('/path', + 'http://www.test.com/path', + headers={'A': 'abc', 'b': 'def'}, + request_headers={'b': 'abc'}) + + self.assertNoMatch('/path', + 'http://www.test.com/path', + headers={'A': 'abc', 'b': 'def'}, + request_headers={'c': 'ghi'}) + + # headers should be key insensitive and value sensitive, we have no + # choice here because they go into an insensitive dict. + self.assertMatch('/path', + 'http://www.test.com/path', + headers={'aBc': 'abc', 'DEF': 'def'}, + request_headers={'abC': 'abc'}) + + self.assertNoMatch('/path', + 'http://www.test.com/path', + headers={'abc': 'aBC', 'DEF': 'def'}, + request_headers={'abc': 'Abc'}) + + def test_case_sensitive_ignored_for_netloc_and_protocol(self): + for case_sensitive in (True, False): + self.assertMatch('http://AbC.CoM', + 'http://aBc.CoM', + case_sensitive=case_sensitive) + + self.assertMatch('htTP://abc.com', + 'hTTp://abc.com', + case_sensitive=case_sensitive) + + self.assertMatch('htTP://aBC.cOm', + 'hTTp://AbC.Com', + case_sensitive=case_sensitive) + + def assertSensitiveMatch(self, target, url, **kwargs): + self.assertMatch(target, url, case_sensitive=False, **kwargs) + self.assertNoMatch(target, url, case_sensitive=True, **kwargs) + + def test_case_sensitive_paths(self): + self.assertSensitiveMatch('http://abc.com/pAtH', 'http://abc.com/path') + self.assertSensitiveMatch('/pAtH', 'http://abc.com/path') + + def test_case_sensitive_query(self): + self.assertSensitiveMatch('http://abc.com/path?abCD=efGH', + 'http://abc.com/path?abCd=eFGH') + + self.assertSensitiveMatch('http://abc.com/path?abcd=efGH', + 'http://abc.com/path?abcd=eFGH') + + def test_additional_matcher(self): + + def test_match_body(request): + return 'hello' in request.text + + self.assertMatch(request_method='POST', + matcher_method='POST', + request_data='hello world', + additional_matcher=test_match_body) + + self.assertNoMatch(request_method='POST', + matcher_method='POST', + request_data='goodbye world', + additional_matcher=test_match_body) + + def test_reset_reverts_count(self): + url = 'mock://test/site/' + matcher = adapter._Matcher('GET', + url, + [_MatcherResponse()], + complete_qs=False, + additional_matcher=None, + request_headers={}, + real_http=False, + case_sensitive=False) + request = adapter._RequestObjectProxy._create('GET', url) + + call_count = 3 + for _ in range(call_count): + matcher(request) + + self.assertEqual(matcher.call_count, call_count) + matcher.reset() + self.assertEqual(matcher.call_count, 0) diff --git a/contrib/python/requests-mock/py3/tests/test_mocker.py b/contrib/python/requests-mock/py3/tests/test_mocker.py new file mode 100644 index 0000000000..c2e88813ed --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_mocker.py @@ -0,0 +1,646 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import json +import pickle + +try: + from unittest import mock +except ImportError: + import mock +import requests + +import requests_mock +from requests_mock import adapter +from requests_mock import exceptions +from requests_mock import response +from . import base + +original_send = requests.Session.send + + +class MockerTests(base.TestCase): + + def assertMockStarted(self): + self.assertNotEqual(original_send, requests.Session.send) + + def assertMockStopped(self): + self.assertEqual(original_send, requests.Session.send) + + def _do_test(self, m): + self.assertMockStarted() + matcher = m.register_uri('GET', 'http://www.test.com', text='resp') + resp = requests.get('http://www.test.com') + self.assertEqual('resp', resp.text) + return matcher + + def test_multiple_starts(self): + mocker = requests_mock.Mocker() + self.assertMockStopped() + mocker.start() + self.assertMockStarted() + self.assertRaises(RuntimeError, mocker.start) + mocker.stop() + self.assertMockStopped() + mocker.stop() + + def test_with_session(self): + url = 'http://test.url/path' + url_inner = 'http://test.url/inner' + url_outer = 'http://test.url/outer' + with requests_mock.Mocker() as global_mock: + global_mock.get(url_outer, text='global') + + session_a = requests.Session() + session_b = requests.Session() + + session_a_original_send = session_a.send + session_b_original_send = session_b.send + self.assertNotEqual(session_a_original_send, + session_b_original_send) + + mocker_a = requests_mock.Mocker(session=session_a) + mocker_b = requests_mock.Mocker(session=session_b) + + mocker_a.start() + mocker_b.start() + + mocker_a.register_uri('GET', url, text='resp_a') + mocker_a.register_uri('GET', url_outer, real_http=True) + mocker_b.register_uri('GET', url, text='resp_b') + + with requests_mock.Mocker(session=session_b) as mocker_b_inner: + mocker_b_inner.register_uri('GET', url, real_http=True) + mocker_b_inner.register_uri('GET', + url_inner, + text='resp_b_inner') + + self.assertEqual('resp_a', session_a.get(url).text) + self.assertEqual('resp_b', session_b.get(url).text) + self.assertRaises(exceptions.NoMockAddress, + session_a.get, + url_inner) + self.assertEqual('resp_b_inner', session_b.get(url_inner).text) + + self.assertEqual('resp_a', session_a.get(url).text) + self.assertEqual('resp_b', session_b.get(url).text) + self.assertRaises(exceptions.NoMockAddress, + session_a.get, + url_inner) + self.assertRaises(exceptions.NoMockAddress, + session_b.get, + url_inner) + self.assertEqual('global', session_a.get(url_outer).text) + self.assertRaises(exceptions.NoMockAddress, + session_b.get, + url_outer) + + self.assertNotEqual(session_a.send, session_a_original_send) + self.assertNotEqual(session_b.send, session_b_original_send) + self.assertNotEqual(session_a.send, session_b.send) + + mocker_a.stop() + mocker_b.stop() + + self.assertEqual(session_a.send, session_a_original_send) + self.assertEqual(session_b.send, session_b_original_send) + + self.assertEqual(requests.Session.send, original_send) + + def test_with_context_manager(self): + self.assertMockStopped() + with requests_mock.Mocker() as m: + self._do_test(m) + self.assertMockStopped() + + @mock.patch('requests.adapters.HTTPAdapter.send') + @requests_mock.mock(real_http=True) + def test_real_http(self, real_send, mocker): + url = 'http://www.google.com/' + test_text = 'real http data' + test_bytes = test_text.encode('utf-8') + + # using create_response is a bit bootstrappy here but so long as it's + # coming from HTTPAdapter.send it's ok + req = requests.Request(method='GET', url=url) + real_send.return_value = response.create_response(req.prepare(), + status_code=200, + content=test_bytes) + resp = requests.get(url) + + self.assertEqual(1, real_send.call_count) + self.assertEqual(url, real_send.call_args[0][0].url) + + self.assertEqual(test_text, resp.text) + self.assertEqual(test_bytes, resp.content) + + @mock.patch('requests.adapters.HTTPAdapter.send') + def test_real_http_changes(self, real_send): + url = 'http://www.google.com/' + test_text = 'real http data' + test_bytes = test_text.encode('utf-8') + + req = requests.Request(method='GET', url=url) + real_send.return_value = response.create_response(req.prepare(), + status_code=200, + content=test_bytes) + + with requests_mock.Mocker() as m: + # real_http defaults to false so should raise NoMockAddress + + self.assertRaises(exceptions.NoMockAddress, + requests.get, + url) + + self.assertEqual(1, m.call_count) + self.assertEqual(0, real_send.call_count) + + # change the value of real_http mid test + m.real_http = True + + # fetch the url again and it should go through to the real url that + # we've mocked out at a lower level. + resp = requests.get(url) + + self.assertEqual(1, real_send.call_count) + self.assertEqual(url, real_send.call_args[0][0].url) + + self.assertEqual(test_text, resp.text) + self.assertEqual(test_bytes, resp.content) + + @mock.patch('requests.adapters.HTTPAdapter.send') + def test_real_http_and_session(self, real_send): + url = 'http://www.google.com/' + test_text = 'real http data' + test_bytes = test_text.encode('utf-8') + + req = requests.Request(method='GET', url=url) + real_send.return_value = response.create_response(req.prepare(), + status_code=200, + content=test_bytes) + + session = requests.Session() + with requests_mock.Mocker(session=session, real_http=True): + resp = session.get(url) + + self.assertEqual(test_text, resp.text) + self.assertEqual(test_bytes, resp.content) + + @requests_mock.mock() + def test_with_test_decorator(self, m): + self._do_test(m) + + @requests_mock.mock(kw='mock') + def test_with_mocker_kwargs(self, **kwargs): + self._do_test(kwargs['mock']) + + def test_with_decorator(self): + + @requests_mock.mock() + def inner(m): + self.assertMockStarted() + self._do_test(m) + + self.assertMockStopped() + inner() + self.assertMockStopped() + + def test_with_decorator_called_multiple_times(self): + + @requests_mock.Mocker() + def inner(arg1, m): + self._do_test(m) + self.assertEqual( + len(m.request_history), 1, + "Failed to provide clean mock on subsequent calls" + ) + + inner('a') + # if we call the same decorated method again should get + # a new request mock + inner('b') + + def test_with_class_decorator(self): + outer = self + + @requests_mock.mock() + class Decorated(object): + + def test_will_be_decorated(self, m): + outer.assertMockStarted() + outer._do_test(m) + + def will_not_be_decorated(self): + outer.assertMockStopped() + + decorated_class = Decorated() + + self.assertMockStopped() + decorated_class.test_will_be_decorated() + self.assertMockStopped() + decorated_class.will_not_be_decorated() + self.assertMockStopped() + + def test_with_class_decorator_and_custom_kw(self): + outer = self + + @requests_mock.mock(kw='custom_m') + class Decorated(object): + + def test_will_be_decorated(self, **kwargs): + outer.assertMockStarted() + outer._do_test(kwargs['custom_m']) + + def will_not_be_decorated(self): + outer.assertMockStopped() + + decorated_class = Decorated() + + self.assertMockStopped() + decorated_class.test_will_be_decorated() + self.assertMockStopped() + decorated_class.will_not_be_decorated() + self.assertMockStopped() + + @mock.patch.object(requests_mock.mock, 'TEST_PREFIX', 'foo') + def test_with_class_decorator_and_custom_test_prefix(self): + outer = self + + @requests_mock.mock() + class Decorated(object): + + def foo_will_be_decorated(self, m): + outer.assertMockStarted() + outer._do_test(m) + + def will_not_be_decorated(self): + outer.assertMockStopped() + + decorated_class = Decorated() + + self.assertMockStopped() + decorated_class.foo_will_be_decorated() + self.assertMockStopped() + decorated_class.will_not_be_decorated() + self.assertMockStopped() + + @requests_mock.mock() + def test_query_string(self, m): + url = 'http://test.url/path' + qs = 'a=1&b=2' + m.register_uri('GET', url, text='resp') + resp = requests.get("%s?%s" % (url, qs)) + + self.assertEqual('resp', resp.text) + + self.assertEqual(qs, m.last_request.query) + self.assertEqual(['1'], m.last_request.qs['a']) + self.assertEqual(['2'], m.last_request.qs['b']) + + @requests_mock.mock() + def test_mock_matcher_attributes(self, m): + matcher = self._do_test(m) + + self.assertEqual(1, matcher.call_count) + self.assertEqual(1, m.call_count) + + self.assertTrue(matcher.called) + self.assertTrue(matcher.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + self.assertEqual(m.request_history, matcher.request_history) + self.assertIs(m.last_request, matcher.last_request) + + def test_copy(self): + mocker = requests_mock.mock(kw='foo', real_http=True) + copy_of_mocker = mocker.copy() + self.assertIsNot(copy_of_mocker, mocker) + self.assertEqual(copy_of_mocker._kw, mocker._kw) + self.assertEqual(copy_of_mocker.real_http, mocker.real_http) + + @requests_mock.mock() + def test_reset_mock_reverts_call_count(self, request_mock): + url = 'http://test.url/path' + request_mock.get(url, text='resp') + requests.get(url) + + self.assertEqual(request_mock.call_count, 1) + + # reset count and verify it is 0 + request_mock.reset_mock() + self.assertEqual(request_mock.call_count, 0) + + +class MockerHttpMethodsTests(base.TestCase): + + URL = 'http://test.com/path' + TEXT = 'resp' + + def assertResponse(self, resp): + self.assertEqual(self.TEXT, resp.text) + + @requests_mock.Mocker() + def test_mocker_request(self, m): + method = 'XXX' + mock_obj = m.request(method, self.URL, text=self.TEXT) + resp = requests.request(method, self.URL) + self.assertResponse(resp) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_get(self, m): + mock_obj = m.get(self.URL, text=self.TEXT) + self.assertResponse(requests.get(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_options(self, m): + mock_obj = m.options(self.URL, text=self.TEXT) + self.assertResponse(requests.options(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_head(self, m): + mock_obj = m.head(self.URL, text=self.TEXT) + self.assertResponse(requests.head(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_post(self, m): + mock_obj = m.post(self.URL, text=self.TEXT) + self.assertResponse(requests.post(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_put(self, m): + mock_obj = m.put(self.URL, text=self.TEXT) + self.assertResponse(requests.put(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_patch(self, m): + mock_obj = m.patch(self.URL, text=self.TEXT) + self.assertResponse(requests.patch(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_delete(self, m): + mock_obj = m.delete(self.URL, text=self.TEXT) + self.assertResponse(requests.delete(self.URL)) + self.assertTrue(mock_obj.called) + self.assertTrue(mock_obj.called_once) + self.assertTrue(m.called) + self.assertTrue(m.called_once) + + @requests_mock.Mocker() + def test_mocker_real_http_and_responses(self, m): + self.assertRaises(RuntimeError, + m.get, + self.URL, + text='abcd', + real_http=True) + + @requests_mock.Mocker() + def test_mocker_real_http(self, m): + data = 'testdata' + + uri1 = 'fake://example.com/foo' + uri2 = 'fake://example.com/bar' + uri3 = 'fake://example.com/baz' + + m.get(uri1, text=data) + m.get(uri2, real_http=True) + + self.assertEqual(data, requests.get(uri1).text) + + # This should fail because requests can't get an adapter for mock:// + # but it shows that it has tried and would have made a request. + self.assertRaises(requests.exceptions.InvalidSchema, + requests.get, + uri2) + + # This fails because real_http is not set on the mocker + self.assertRaises(exceptions.NoMockAddress, + requests.get, + uri3) + + # do it again to make sure the mock is still in place + self.assertEqual(data, requests.get(uri1).text) + + @requests_mock.Mocker(case_sensitive=True) + def test_case_sensitive_query(self, m): + data = 'testdata' + query = {'aBcDe': 'FgHiJ'} + + m.get(self.URL, text=data) + resp = requests.get(self.URL, params=query) + + self.assertEqual('GET', m.last_request.method) + self.assertEqual(200, resp.status_code) + self.assertEqual(data, resp.text) + + for k, v in query.items(): + self.assertEqual([v], m.last_request.qs[k]) + + @mock.patch.object(requests_mock.Mocker, 'case_sensitive', True) + def test_global_case_sensitive(self): + with requests_mock.mock() as m: + data = 'testdata' + query = {'aBcDe': 'FgHiJ'} + + m.get(self.URL, text=data) + resp = requests.get(self.URL, params=query) + + self.assertEqual('GET', m.last_request.method) + self.assertEqual(200, resp.status_code) + self.assertEqual(data, resp.text) + + for k, v in query.items(): + self.assertEqual([v], m.last_request.qs[k]) + + def test_nested_mocking(self): + url1 = 'http://url1.com/path1' + url2 = 'http://url2.com/path2' + url3 = 'http://url3.com/path3' + + data1 = 'data1' + data2 = 'data2' + data3 = 'data3' + + with requests_mock.mock() as m1: + + r1 = m1.get(url1, text=data1) + + resp1a = requests.get(url1) + self.assertRaises(exceptions.NoMockAddress, requests.get, url2) + self.assertRaises(exceptions.NoMockAddress, requests.get, url3) + + self.assertEqual(data1, resp1a.text) + + # call count = 3 because there are 3 calls above, url 1-3 + self.assertEqual(3, m1.call_count) + self.assertEqual(1, r1.call_count) + + with requests_mock.mock() as m2: + + r2 = m2.get(url2, text=data2) + + self.assertRaises(exceptions.NoMockAddress, requests.get, url1) + resp2a = requests.get(url2) + self.assertRaises(exceptions.NoMockAddress, requests.get, url3) + + self.assertEqual(data2, resp2a.text) + + with requests_mock.mock() as m3: + + r3 = m3.get(url3, text=data3) + + self.assertRaises(exceptions.NoMockAddress, + requests.get, + url1) + self.assertRaises(exceptions.NoMockAddress, + requests.get, + url2) + resp3 = requests.get(url3) + + self.assertEqual(data3, resp3.text) + + self.assertEqual(3, m3.call_count) + self.assertEqual(1, r3.call_count) + + resp2b = requests.get(url2) + self.assertRaises(exceptions.NoMockAddress, requests.get, url1) + self.assertEqual(data2, resp2b.text) + self.assertRaises(exceptions.NoMockAddress, requests.get, url3) + + self.assertEqual(3, m1.call_count) + self.assertEqual(1, r1.call_count) + self.assertEqual(6, m2.call_count) + self.assertEqual(2, r2.call_count) + self.assertEqual(3, m3.call_count) + self.assertEqual(1, r3.call_count) + + resp1b = requests.get(url1) + self.assertEqual(data1, resp1b.text) + self.assertRaises(exceptions.NoMockAddress, requests.get, url2) + self.assertRaises(exceptions.NoMockAddress, requests.get, url3) + + self.assertEqual(6, m1.call_count) + self.assertEqual(2, r1.call_count) + + @requests_mock.mock() + def test_mocker_additional(self, m): + url = 'http://www.example.com' + good_text = 'success' + + def additional_cb(req): + return 'hello' in req.text + + m.post(url, additional_matcher=additional_cb, text=good_text) + + self.assertEqual(good_text, + requests.post(url, data='hello world').text) + self.assertRaises(exceptions.NoMockAddress, + requests.post, + url, + data='goodbye world') + + @requests_mock.mock() + def test_mocker_pickle(self, m): + url = 'http://www.example.com' + text = 'hello world' + m.get(url, text=text) + + orig_resp = requests.get(url) + self.assertEqual(text, orig_resp.text) + + d = pickle.dumps(orig_resp) + new_resp = pickle.loads(d) + + self.assertEqual(text, new_resp.text) + self.assertIsInstance(orig_resp.request.matcher, adapter._Matcher) + self.assertIsNone(new_resp.request.matcher) + + @requests_mock.mock() + def test_stream_zero_bytes(self, m): + content = b'blah' + + m.get("http://test", content=content) + res = requests.get("http://test", stream=True) + zero_val = res.raw.read(0) + self.assertEqual(b'', zero_val) + self.assertFalse(res.raw.closed) + + full_val = res.raw.read() + self.assertEqual(content, full_val) + + def test_with_json_encoder_on_mocker(self): + test_val = 'hello world' + + class MyJsonEncoder(json.JSONEncoder): + def encode(s, o): + return test_val + + with requests_mock.Mocker(json_encoder=MyJsonEncoder) as m: + m.get("http://test", json={"a": "b"}) + res = requests.get("http://test") + self.assertEqual(test_val, res.text) + + @requests_mock.mock() + def test_with_json_encoder_on_endpoint(self, m): + test_val = 'hello world' + + class MyJsonEncoder(json.JSONEncoder): + def encode(s, o): + return test_val + + m.get("http://test", json={"a": "b"}, json_encoder=MyJsonEncoder) + res = requests.get("http://test") + self.assertEqual(test_val, res.text) + + @requests_mock.mock() + def test_mismatch_content_length_streaming(self, m): + url = "https://test/package.tar.gz" + + def f(request, context): + context.headers["Content-Length"] = "300810" + return None + + m.head( + url=url, + status_code=200, + text=f, + ) + + requests.head(url) diff --git a/contrib/python/requests-mock/py3/tests/test_request.py b/contrib/python/requests-mock/py3/tests/test_request.py new file mode 100644 index 0000000000..69a71556e2 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_request.py @@ -0,0 +1,139 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import uuid + +import requests +import requests_mock +from . import base + + +class RequestTests(base.TestCase): + + def setUp(self): + super(RequestTests, self).setUp() + + self.mocker = requests_mock.Mocker() + self.addCleanup(self.mocker.stop) + self.mocker.start() + + def do_request(self, **kwargs): + method = kwargs.pop('method', 'GET') + url = kwargs.pop('url', 'http://test.example.com/path') + status_code = kwargs.pop('status_code', 200) + data = uuid.uuid4().hex + + m = self.mocker.register_uri(method, + url, + text=data, + status_code=status_code) + + resp = requests.request(method, url, **kwargs) + + self.assertEqual(status_code, resp.status_code) + self.assertEqual(data, resp.text) + + self.assertTrue(m.called_once) + return m.last_request + + def test_base_params(self): + req = self.do_request(method='GET', status_code=200) + + self.assertIs(None, req.allow_redirects) + self.assertIs(None, req.timeout) + self.assertIs(True, req.verify) + self.assertIs(None, req.cert) + self.assertIs(False, req.stream) + + # actually it's an OrderedDict, but equality works fine + # Skipping this check - it's problematic based on people's environments + # and in CI systems where there are proxies set up at the environment + # level. gh #127 + # self.assertEqual({}, req.proxies) + + def test_allow_redirects(self): + req = self.do_request(allow_redirects=False, status_code=300) + self.assertFalse(req.allow_redirects) + + def test_timeout(self): + timeout = 300 + req = self.do_request(timeout=timeout) + self.assertEqual(timeout, req.timeout) + + def test_verify_false(self): + verify = False + req = self.do_request(verify=verify) + self.assertIs(verify, req.verify) + + def test_verify_path(self): + verify = '/path/to/cacerts.pem' + req = self.do_request(verify=verify) + self.assertEqual(verify, req.verify) + + def test_stream(self): + req = self.do_request() + self.assertIs(False, req.stream) + req = self.do_request(stream=False) + self.assertIs(False, req.stream) + req = self.do_request(stream=True) + self.assertIs(True, req.stream) + + def test_certs(self): + cert = ('/path/to/cert.pem', 'path/to/key.pem') + req = self.do_request(cert=cert) + self.assertEqual(cert, req.cert) + self.assertTrue(req.verify) + + def test_proxies(self): + proxies = {'http': 'foo.bar:3128', + 'http://host.name': 'foo.bar:4012'} + + req = self.do_request(proxies=proxies) + + self.assertEqual(proxies, req.proxies) + self.assertIsNot(proxies, req.proxies) + + def test_hostname_port_http(self): + req = self.do_request(url='http://host.example.com:81/path') + + self.assertEqual('host.example.com:81', req.netloc) + self.assertEqual('host.example.com', req.hostname) + self.assertEqual(81, req.port) + + def test_hostname_port_https(self): + req = self.do_request(url='https://host.example.com:8080/path') + + self.assertEqual('host.example.com:8080', req.netloc) + self.assertEqual('host.example.com', req.hostname) + self.assertEqual(8080, req.port) + + def test_hostname_default_port_http(self): + req = self.do_request(url='http://host.example.com/path') + + self.assertEqual('host.example.com', req.netloc) + self.assertEqual('host.example.com', req.hostname) + self.assertEqual(80, req.port) + + def test_hostname_default_port_https(self): + req = self.do_request(url='https://host.example.com/path') + + self.assertEqual('host.example.com', req.netloc) + self.assertEqual('host.example.com', req.hostname) + self.assertEqual(443, req.port) + + def test_to_string(self): + req = self.do_request(url='https://host.example.com/path') + self.assertEqual('GET https://host.example.com/path', str(req)) + + def test_empty_query_string(self): + req = self.do_request(url='https://host.example.com/path?key') + self.assertEqual([''], req.qs['key']) diff --git a/contrib/python/requests-mock/py3/tests/test_response.py b/contrib/python/requests-mock/py3/tests/test_response.py new file mode 100644 index 0000000000..b560eea8d7 --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/test_response.py @@ -0,0 +1,151 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import io +import pickle + +from requests_mock import exceptions +from requests_mock import request +from requests_mock import response +from . import base + + +class ResponseTests(base.TestCase): + + def setUp(self): + super(ResponseTests, self).setUp() + self.method = 'GET' + self.url = 'http://test.url/path' + self.request = request._RequestObjectProxy._create(self.method, + self.url, + {}) + + def create_response(self, **kwargs): + return response.create_response(self.request, **kwargs) + + def test_create_response_body_args(self): + self.assertRaises(RuntimeError, + self.create_response, + raw='abc', + body='abc') + + self.assertRaises(RuntimeError, + self.create_response, + text='abc', + json={'a': 1}) + + def test_content_type(self): + self.assertRaises(TypeError, self.create_response, text=55) + self.assertRaises(TypeError, self.create_response, text={'a': 1}) + self.assertRaises(TypeError, self.create_response, text=b'') + + def test_text_type(self): + self.assertRaises(TypeError, self.create_response, content=u't') + self.assertRaises(TypeError, self.create_response, content={'a': 1}) + self.assertRaises(TypeError, self.create_response, content=u'') + + def test_json_body(self): + data = {'a': 1} + resp = self.create_response(json=data) + + self.assertEqual('{"a": 1}', resp.text) + self.assertIsInstance(resp.text, str) + self.assertIsInstance(resp.content, bytes) + self.assertEqual(data, resp.json()) + + def test_body_body(self): + value = b'data' + body = io.BytesIO(value) + resp = self.create_response(body=body) + + self.assertEqual(value.decode(), resp.text) + self.assertIsInstance(resp.text, str) + self.assertIsInstance(resp.content, bytes) + + def test_setting_connection(self): + conn = object() + resp = self.create_response(connection=conn) + self.assertIs(conn, resp.connection) + + def test_send_from_no_connection(self): + resp = self.create_response() + self.assertRaises(exceptions.InvalidRequest, + resp.connection.send, self.request) + + def test_cookies_from_header(self): + # domain must be same as request url to pass policy check + headers = {'Set-Cookie': 'fig=newton; Path=/test; domain=.test.url'} + resp = self.create_response(headers=headers) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual(['/test'], resp.cookies.list_paths()) + self.assertEqual(['.test.url'], resp.cookies.list_domains()) + + def test_cookies_from_dict(self): + # This is a syntax we get from requests. I'm not sure i like it. + resp = self.create_response(cookies={'fig': 'newton', + 'sugar': 'apple'}) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual('apple', resp.cookies['sugar']) + + def test_cookies_with_jar(self): + jar = response.CookieJar() + jar.set('fig', 'newton', path='/foo', domain='.test.url') + jar.set('sugar', 'apple', path='/bar', domain='.test.url') + resp = self.create_response(cookies=jar) + + self.assertEqual('newton', resp.cookies['fig']) + self.assertEqual('apple', resp.cookies['sugar']) + self.assertEqual({'/foo', '/bar'}, set(resp.cookies.list_paths())) + self.assertEqual(['.test.url'], resp.cookies.list_domains()) + + def test_response_pickle(self): + text = 'hello world' + jar = response.CookieJar() + jar.set('fig', 'newton', path='/foo', domain='.test.url') + orig_resp = self.create_response(cookies=jar, text=text) + + d = pickle.dumps(orig_resp) + new_resp = pickle.loads(d) + + self.assertEqual(text, new_resp.text) + self.assertEqual('newton', new_resp.cookies['fig']) + self.assertIsNone(new_resp.request.matcher) + + def test_response_encoding(self): + headers = {"content-type": "text/html; charset=ISO-8859-1"} + resp = self.create_response(headers=headers, + text="<html><body></body></html") + self.assertEqual('ISO-8859-1', resp.encoding) + + def test_default_reason(self): + resp = self.create_response() + self.assertEqual('OK', resp.reason) + + def test_custom_reason(self): + reason = 'Live long and prosper' + resp = self.create_response(status_code=201, reason=reason) + + self.assertEqual(201, resp.status_code) + self.assertEqual(reason, resp.reason) + + def test_some_other_response_reasons(self): + reasons = { + 301: 'Moved Permanently', + 410: 'Gone', + 503: 'Service Unavailable', + } + + for code, reason in reasons.items(): + self.assertEqual(reason, + self.create_response(status_code=code).reason) diff --git a/contrib/python/requests-mock/py3/tests/ya.make b/contrib/python/requests-mock/py3/tests/ya.make new file mode 100644 index 0000000000..f52ec22dba --- /dev/null +++ b/contrib/python/requests-mock/py3/tests/ya.make @@ -0,0 +1,26 @@ +PY3TEST() + +SUBSCRIBER(g:python-contrib) + +PEERDIR( + contrib/python/mock + contrib/python/requests-futures + contrib/python/requests-mock +) + +TEST_SRCS( + base.py + pytest/__init__.py + pytest/test_with_pytest.py + #test_adapter.py - need purl + test_custom_matchers.py + #test_fixture.py - need fixtures + test_matcher.py + test_mocker.py + test_request.py + test_response.py +) + +NO_LINT() + +END() |