diff options
author | robot-piglet <robot-piglet@yandex-team.com> | 2024-10-06 13:42:43 +0300 |
---|---|---|
committer | robot-piglet <robot-piglet@yandex-team.com> | 2024-10-06 13:52:30 +0300 |
commit | 52aed29f744afda4549ef5d64acd0fa8c2092789 (patch) | |
tree | e40c9abd25653990d13b68936aee518454df424e /contrib/python/aioresponses | |
parent | 813943fcad905eee1235d764be4268dddd07ce64 (diff) | |
download | ydb-52aed29f744afda4549ef5d64acd0fa8c2092789.tar.gz |
Intermediate changes
commit_hash:cc4365f5a0e443b92d87079a9c91e77fea2ddcaf
Diffstat (limited to 'contrib/python/aioresponses')
-rw-r--r-- | contrib/python/aioresponses/.dist-info/METADATA | 333 | ||||
-rw-r--r-- | contrib/python/aioresponses/.dist-info/top_level.txt | 1 | ||||
-rw-r--r-- | contrib/python/aioresponses/AUTHORS | 51 | ||||
-rw-r--r-- | contrib/python/aioresponses/AUTHORS.rst | 13 | ||||
-rw-r--r-- | contrib/python/aioresponses/LICENSE | 21 | ||||
-rw-r--r-- | contrib/python/aioresponses/README.rst | 306 | ||||
-rw-r--r-- | contrib/python/aioresponses/aioresponses/__init__.py | 9 | ||||
-rw-r--r-- | contrib/python/aioresponses/aioresponses/compat.py | 68 | ||||
-rw-r--r-- | contrib/python/aioresponses/aioresponses/core.py | 549 | ||||
-rw-r--r-- | contrib/python/aioresponses/aioresponses/py.typed | 0 | ||||
-rw-r--r-- | contrib/python/aioresponses/ya.make | 33 |
11 files changed, 1384 insertions, 0 deletions
diff --git a/contrib/python/aioresponses/.dist-info/METADATA b/contrib/python/aioresponses/.dist-info/METADATA new file mode 100644 index 0000000000..54b686eb71 --- /dev/null +++ b/contrib/python/aioresponses/.dist-info/METADATA @@ -0,0 +1,333 @@ +Metadata-Version: 2.1 +Name: aioresponses +Version: 0.7.6 +Summary: Mock out requests made by ClientSession from aiohttp package +Home-page: https://github.com/pnuckowski/aioresponses +Author: Pawel Nuckowski +Author-email: p.nuckowski@gmail.com +Classifier: Development Status :: 4 - Beta +Classifier: Intended Audience :: Developers +Classifier: Operating System :: OS Independent +Classifier: Topic :: Internet :: WWW/HTTP +Classifier: Topic :: Software Development :: Testing +Classifier: Topic :: Software Development :: Testing :: Mocking +Classifier: License :: OSI Approved :: MIT License +Classifier: Natural Language :: English +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3.7 +Classifier: Programming Language :: Python :: 3.8 +Classifier: Programming Language :: Python :: 3.9 +Classifier: Programming Language :: Python :: 3.10 +Classifier: Programming Language :: Python :: 3.11 +License-File: LICENSE +License-File: AUTHORS +License-File: AUTHORS.rst +Requires-Dist: aiohttp (<4.0.0,>=3.3.0) + +=============================== +aioresponses +=============================== + +.. image:: https://travis-ci.org/pnuckowski/aioresponses.svg?branch=master + :target: https://travis-ci.org/pnuckowski/aioresponses + +.. image:: https://coveralls.io/repos/github/pnuckowski/aioresponses/badge.svg?branch=master + :target: https://coveralls.io/github/pnuckowski/aioresponses?branch=master + +.. image:: https://landscape.io/github/pnuckowski/aioresponses/master/landscape.svg?style=flat + :target: https://landscape.io/github/pnuckowski/aioresponses/master + :alt: Code Health + +.. image:: https://pyup.io/repos/github/pnuckowski/aioresponses/shield.svg + :target: https://pyup.io/repos/github/pnuckowski/aioresponses/ + :alt: Updates + +.. image:: https://img.shields.io/pypi/v/aioresponses.svg + :target: https://pypi.python.org/pypi/aioresponses + +.. image:: https://readthedocs.org/projects/aioresponses/badge/?version=latest + :target: https://aioresponses.readthedocs.io/en/latest/?badge=latest + :alt: Documentation Status + + +Aioresponses is a helper to mock/fake web requests in python aiohttp package. + +For *requests* module there are a lot of packages that help us with testing (eg. *httpretty*, *responses*, *requests-mock*). + +When it comes to testing asynchronous HTTP requests it is a bit harder (at least at the beginning). +The purpose of this package is to provide an easy way to test asynchronous HTTP requests. + +Installing +---------- + +.. code:: bash + + $ pip install aioresponses + +Supported versions +------------------ +- Python 3.7+ +- aiohttp>=3.3.0,<4.0.0 + +Usage +-------- + +To mock out HTTP request use *aioresponses* as a method decorator or as a context manager. + +Response *status* code, *body*, *payload* (for json response) and *headers* can be mocked. + +Supported HTTP methods: **GET**, **POST**, **PUT**, **PATCH**, **DELETE** and **OPTIONS**. + +.. code:: python + + import aiohttp + import asyncio + from aioresponses import aioresponses + + @aioresponses() + def test_request(mocked): + loop = asyncio.get_event_loop() + mocked.get('http://example.com', status=200, body='test') + session = aiohttp.ClientSession() + resp = loop.run_until_complete(session.get('http://example.com')) + + assert resp.status == 200 + mocked.assert_called_once_with('http://example.com') + + +for convenience use *payload* argument to mock out json response. Example below. + +**as a context manager** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + def test_ctx(): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + with aioresponses() as m: + m.get('http://test.example.com', payload=dict(foo='bar')) + + resp = loop.run_until_complete(session.get('http://test.example.com')) + data = loop.run_until_complete(resp.json()) + + assert dict(foo='bar') == data + m.assert_called_once_with('http://test.example.com') + +**aioresponses allows to mock out any HTTP headers** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_http_headers(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + m.post( + 'http://example.com', + payload=dict(), + headers=dict(connection='keep-alive'), + ) + + resp = loop.run_until_complete(session.post('http://example.com')) + + # note that we pass 'connection' but get 'Connection' (capitalized) + # under the neath `multidict` is used to work with HTTP headers + assert resp.headers['Connection'] == 'keep-alive' + m.assert_called_once_with('http://example.com', method='POST') + +**allows to register different responses for the same url** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_multiple_responses(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + m.get('http://example.com', status=500) + m.get('http://example.com', status=200) + + resp1 = loop.run_until_complete(session.get('http://example.com')) + resp2 = loop.run_until_complete(session.get('http://example.com')) + + assert resp1.status == 500 + assert resp2.status == 200 + + +**Repeat response for the same url** + +E.g. for cases you want to test retrying mechanisms + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_multiple_responses(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + m.get('http://example.com', status=500, repeat=True) + m.get('http://example.com', status=200) # will not take effect + + resp1 = loop.run_until_complete(session.get('http://example.com')) + resp2 = loop.run_until_complete(session.get('http://example.com')) + + assert resp1.status == 500 + assert resp2.status == 500 + + +**match URLs with regular expressions** + +.. code:: python + + import asyncio + import aiohttp + import re + from aioresponses import aioresponses + + @aioresponses() + def test_regexp_example(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + pattern = re.compile(r'^http://example\.com/api\?foo=.*$') + m.get(pattern, status=200) + + resp = loop.run_until_complete(session.get('http://example.com/api?foo=bar')) + + assert resp.status == 200 + +**allows to make redirects responses** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_redirect_example(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + + # absolute urls are supported + m.get( + 'http://example.com/', + headers={'Location': 'http://another.com/'}, + status=307 + ) + + resp = loop.run_until_complete( + session.get('http://example.com/', allow_redirects=True) + ) + assert resp.url == 'http://another.com/' + + # and also relative + m.get( + 'http://example.com/', + headers={'Location': '/test'}, + status=307 + ) + resp = loop.run_until_complete( + session.get('http://example.com/', allow_redirects=True) + ) + assert resp.url == 'http://example.com/test' + +**allows to passthrough to a specified list of servers** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses(passthrough=['http://backend']) + def test_passthrough(m, test_client): + session = aiohttp.ClientSession() + # this will actually perform a request + resp = loop.run_until_complete(session.get('http://backend/api')) + + +**aioresponses allows to throw an exception** + +.. code:: python + + import asyncio + from aiohttp import ClientSession + from aiohttp.http_exceptions import HttpProcessingError + from aioresponses import aioresponses + + @aioresponses() + def test_how_to_throw_an_exception(m, test_client): + loop = asyncio.get_event_loop() + session = ClientSession() + m.get('http://example.com/api', exception=HttpProcessingError('test')) + + # calling + # loop.run_until_complete(session.get('http://example.com/api')) + # will throw an exception. + + +**aioresponses allows to use callbacks to provide dynamic responses** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import CallbackResult, aioresponses + + def callback(url, **kwargs): + return CallbackResult(status=418) + + @aioresponses() + def test_callback(m, test_client): + loop = asyncio.get_event_loop() + session = ClientSession() + m.get('http://example.com', callback=callback) + + resp = loop.run_until_complete(session.get('http://example.com')) + + assert resp.status == 418 + + +**aioresponses can be used in a pytest fixture** + +.. code:: python + + import pytest + from aioresponses import aioresponses + + @pytest.fixture + def mock_aioresponse(): + with aioresponses() as m: + yield m + + +Features +-------- +* Easy to mock out HTTP requests made by *aiohttp.ClientSession* + + +License +------- +* Free software: MIT license + +Credits +------- + +This package was created with Cookiecutter_ and the `audreyr/cookiecutter-pypackage`_ project template. + +.. _Cookiecutter: https://github.com/audreyr/cookiecutter +.. _`audreyr/cookiecutter-pypackage`: https://github.com/audreyr/cookiecutter-pypackage + diff --git a/contrib/python/aioresponses/.dist-info/top_level.txt b/contrib/python/aioresponses/.dist-info/top_level.txt new file mode 100644 index 0000000000..46cd566df0 --- /dev/null +++ b/contrib/python/aioresponses/.dist-info/top_level.txt @@ -0,0 +1 @@ +aioresponses diff --git a/contrib/python/aioresponses/AUTHORS b/contrib/python/aioresponses/AUTHORS new file mode 100644 index 0000000000..3854a29412 --- /dev/null +++ b/contrib/python/aioresponses/AUTHORS @@ -0,0 +1,51 @@ +Alan Briolat <alan.briolat@gmail.com> +Aleksei Maslakov <lesha.maslakov@gmail.com> +Alexey Nikitenko <wblxyxolb.khv@mail.ru> +Alexey Sveshnikov <a.sveshnikov@rambler-co.ru> +Alexey Sveshnikov <alexey.sveshnikov@gmail.com> +Allisson Azevedo <allisson@gmail.com> +Andrew Grinevich <andrew.grinevich@pandadoc.com> +Anthony Lukach <anthonylukach@gmail.com> +Ben Greiner <code@bnavigator.de> +Brett Wandel <brett.wandel@interferex.com> +Bryce Drennan <github@accounts.brycedrennan.com> +Colin-b <Colin-b@users.noreply.github.com> +Daniel Hahler <git@thequod.de> +Daniel Tan <danieltanjiawang@gmail.com> +David Buxton <david@gasmark6.com> +Fred Thomsen <fred.thomsen@sciencelogic.com> +Georg Sauthoff <mail@gms.tf> +Gordon Rogers <gordonrogers@skyscanner.net> +Hadrien David <hadrien.david@dialogue.co> +Hadrien David <hadrien@ectobal.com> +Ibrahim <8592115+iamibi@users.noreply.github.com> +Ilaï Deutel <ilai-deutel@users.noreply.github.com> +Jakub Boukal <www.bagr@gmail.com> +Joongi Kim <me@daybreaker.info> +Jordi Soucheiron <jordi@soucheiron.cat> +Jordi Soucheiron <jsoucheiron@users.noreply.github.com> +Joshua Coats <joshu@fearchar.net> +Juan Cruz <juancruzmencia@gmail.com> +Lee Treveil <leetreveil@gmail.com> +Louis Sautier <sautier.louis@gmail.com> +Lukasz Jernas <lukasz.jernas@allegrogroup.com> +Marat Sharafutdinov <decaz89@gmail.com> +Marcin Sulikowski <marcin.k.sulikowski@gmail.com> +Marek Kowalski <kowalski0123@gmail.com> +Pavel Savchenko <asfaltboy@gmail.com> +Pawel Nuckowski <p.nuckowski@gmail.com> +Petr Belskiy <petr.belskiy@gmail.com> +Rémy HUBSCHER <rhubscher@mozilla.com> +Sam Bull <aa6bs0@sambull.org> +TyVik <tyvik8@gmail.com> +Ulrik Johansson <ulrik.johansson@blocket.se> +Ville Skyttä <ville.skytta@iki.fi> +d-ryzhikov <d.ryzhykau@gmail.com> +iamnotaprogrammer <iamnotaprogrammer@yandex.ru> +iamnotaprogrammer <issmirnov@domclick.ru> +konstantin <konstantin.klein@hochfrequenz.de> +oren0e <countx@gmail.com> +pnuckowski <p.nuckowski@gmail.com> +pnuckowski <pnuckowski@users.noreply.github.com> +pyup-bot <github-bot@pyup.io> +vangheem <vangheem@gmail.com> diff --git a/contrib/python/aioresponses/AUTHORS.rst b/contrib/python/aioresponses/AUTHORS.rst new file mode 100644 index 0000000000..3b1fc8e0ec --- /dev/null +++ b/contrib/python/aioresponses/AUTHORS.rst @@ -0,0 +1,13 @@ +======= +Credits +======= + +Development Lead +---------------- + +* Pawel Nuckowski <p.nuckowski@gmail.com> + +Contributors +------------ + +None yet. Why not be the first? diff --git a/contrib/python/aioresponses/LICENSE b/contrib/python/aioresponses/LICENSE new file mode 100644 index 0000000000..fe5490da64 --- /dev/null +++ b/contrib/python/aioresponses/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2016 pnuckowski + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/contrib/python/aioresponses/README.rst b/contrib/python/aioresponses/README.rst new file mode 100644 index 0000000000..ae63650d0a --- /dev/null +++ b/contrib/python/aioresponses/README.rst @@ -0,0 +1,306 @@ +=============================== +aioresponses +=============================== + +.. image:: https://travis-ci.org/pnuckowski/aioresponses.svg?branch=master + :target: https://travis-ci.org/pnuckowski/aioresponses + +.. image:: https://coveralls.io/repos/github/pnuckowski/aioresponses/badge.svg?branch=master + :target: https://coveralls.io/github/pnuckowski/aioresponses?branch=master + +.. image:: https://landscape.io/github/pnuckowski/aioresponses/master/landscape.svg?style=flat + :target: https://landscape.io/github/pnuckowski/aioresponses/master + :alt: Code Health + +.. image:: https://pyup.io/repos/github/pnuckowski/aioresponses/shield.svg + :target: https://pyup.io/repos/github/pnuckowski/aioresponses/ + :alt: Updates + +.. image:: https://img.shields.io/pypi/v/aioresponses.svg + :target: https://pypi.python.org/pypi/aioresponses + +.. image:: https://readthedocs.org/projects/aioresponses/badge/?version=latest + :target: https://aioresponses.readthedocs.io/en/latest/?badge=latest + :alt: Documentation Status + + +Aioresponses is a helper to mock/fake web requests in python aiohttp package. + +For *requests* module there are a lot of packages that help us with testing (eg. *httpretty*, *responses*, *requests-mock*). + +When it comes to testing asynchronous HTTP requests it is a bit harder (at least at the beginning). +The purpose of this package is to provide an easy way to test asynchronous HTTP requests. + +Installing +---------- + +.. code:: bash + + $ pip install aioresponses + +Supported versions +------------------ +- Python 3.7+ +- aiohttp>=3.3.0,<4.0.0 + +Usage +-------- + +To mock out HTTP request use *aioresponses* as a method decorator or as a context manager. + +Response *status* code, *body*, *payload* (for json response) and *headers* can be mocked. + +Supported HTTP methods: **GET**, **POST**, **PUT**, **PATCH**, **DELETE** and **OPTIONS**. + +.. code:: python + + import aiohttp + import asyncio + from aioresponses import aioresponses + + @aioresponses() + def test_request(mocked): + loop = asyncio.get_event_loop() + mocked.get('http://example.com', status=200, body='test') + session = aiohttp.ClientSession() + resp = loop.run_until_complete(session.get('http://example.com')) + + assert resp.status == 200 + mocked.assert_called_once_with('http://example.com') + + +for convenience use *payload* argument to mock out json response. Example below. + +**as a context manager** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + def test_ctx(): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + with aioresponses() as m: + m.get('http://test.example.com', payload=dict(foo='bar')) + + resp = loop.run_until_complete(session.get('http://test.example.com')) + data = loop.run_until_complete(resp.json()) + + assert dict(foo='bar') == data + m.assert_called_once_with('http://test.example.com') + +**aioresponses allows to mock out any HTTP headers** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_http_headers(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + m.post( + 'http://example.com', + payload=dict(), + headers=dict(connection='keep-alive'), + ) + + resp = loop.run_until_complete(session.post('http://example.com')) + + # note that we pass 'connection' but get 'Connection' (capitalized) + # under the neath `multidict` is used to work with HTTP headers + assert resp.headers['Connection'] == 'keep-alive' + m.assert_called_once_with('http://example.com', method='POST') + +**allows to register different responses for the same url** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_multiple_responses(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + m.get('http://example.com', status=500) + m.get('http://example.com', status=200) + + resp1 = loop.run_until_complete(session.get('http://example.com')) + resp2 = loop.run_until_complete(session.get('http://example.com')) + + assert resp1.status == 500 + assert resp2.status == 200 + + +**Repeat response for the same url** + +E.g. for cases you want to test retrying mechanisms + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_multiple_responses(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + m.get('http://example.com', status=500, repeat=True) + m.get('http://example.com', status=200) # will not take effect + + resp1 = loop.run_until_complete(session.get('http://example.com')) + resp2 = loop.run_until_complete(session.get('http://example.com')) + + assert resp1.status == 500 + assert resp2.status == 500 + + +**match URLs with regular expressions** + +.. code:: python + + import asyncio + import aiohttp + import re + from aioresponses import aioresponses + + @aioresponses() + def test_regexp_example(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + pattern = re.compile(r'^http://example\.com/api\?foo=.*$') + m.get(pattern, status=200) + + resp = loop.run_until_complete(session.get('http://example.com/api?foo=bar')) + + assert resp.status == 200 + +**allows to make redirects responses** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses() + def test_redirect_example(m): + loop = asyncio.get_event_loop() + session = aiohttp.ClientSession() + + # absolute urls are supported + m.get( + 'http://example.com/', + headers={'Location': 'http://another.com/'}, + status=307 + ) + + resp = loop.run_until_complete( + session.get('http://example.com/', allow_redirects=True) + ) + assert resp.url == 'http://another.com/' + + # and also relative + m.get( + 'http://example.com/', + headers={'Location': '/test'}, + status=307 + ) + resp = loop.run_until_complete( + session.get('http://example.com/', allow_redirects=True) + ) + assert resp.url == 'http://example.com/test' + +**allows to passthrough to a specified list of servers** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses(passthrough=['http://backend']) + def test_passthrough(m, test_client): + session = aiohttp.ClientSession() + # this will actually perform a request + resp = loop.run_until_complete(session.get('http://backend/api')) + + +**aioresponses allows to throw an exception** + +.. code:: python + + import asyncio + from aiohttp import ClientSession + from aiohttp.http_exceptions import HttpProcessingError + from aioresponses import aioresponses + + @aioresponses() + def test_how_to_throw_an_exception(m, test_client): + loop = asyncio.get_event_loop() + session = ClientSession() + m.get('http://example.com/api', exception=HttpProcessingError('test')) + + # calling + # loop.run_until_complete(session.get('http://example.com/api')) + # will throw an exception. + + +**aioresponses allows to use callbacks to provide dynamic responses** + +.. code:: python + + import asyncio + import aiohttp + from aioresponses import CallbackResult, aioresponses + + def callback(url, **kwargs): + return CallbackResult(status=418) + + @aioresponses() + def test_callback(m, test_client): + loop = asyncio.get_event_loop() + session = ClientSession() + m.get('http://example.com', callback=callback) + + resp = loop.run_until_complete(session.get('http://example.com')) + + assert resp.status == 418 + + +**aioresponses can be used in a pytest fixture** + +.. code:: python + + import pytest + from aioresponses import aioresponses + + @pytest.fixture + def mock_aioresponse(): + with aioresponses() as m: + yield m + + +Features +-------- +* Easy to mock out HTTP requests made by *aiohttp.ClientSession* + + +License +------- +* Free software: MIT license + +Credits +------- + +This package was created with Cookiecutter_ and the `audreyr/cookiecutter-pypackage`_ project template. + +.. _Cookiecutter: https://github.com/audreyr/cookiecutter +.. _`audreyr/cookiecutter-pypackage`: https://github.com/audreyr/cookiecutter-pypackage diff --git a/contrib/python/aioresponses/aioresponses/__init__.py b/contrib/python/aioresponses/aioresponses/__init__.py new file mode 100644 index 0000000000..c61652c9aa --- /dev/null +++ b/contrib/python/aioresponses/aioresponses/__init__.py @@ -0,0 +1,9 @@ +# -*- coding: utf-8 -*- +from .core import CallbackResult, aioresponses + +__version__ = '0.7.3' + +__all__ = [ + 'CallbackResult', + 'aioresponses', +] diff --git a/contrib/python/aioresponses/aioresponses/compat.py b/contrib/python/aioresponses/aioresponses/compat.py new file mode 100644 index 0000000000..aa8771d8d6 --- /dev/null +++ b/contrib/python/aioresponses/aioresponses/compat.py @@ -0,0 +1,68 @@ +# -*- coding: utf-8 -*- +import asyncio # noqa: F401 +import sys +from typing import Dict, Optional, Union # noqa +from urllib.parse import parse_qsl, urlencode + +from aiohttp import __version__ as aiohttp_version, StreamReader +from aiohttp.client_proto import ResponseHandler +from multidict import MultiDict +from packaging.version import Version +from yarl import URL + +if sys.version_info < (3, 7): + from re import _pattern_type as Pattern +else: + from re import Pattern + +AIOHTTP_VERSION = Version(aiohttp_version) + + +def stream_reader_factory( # noqa + loop: 'Optional[asyncio.AbstractEventLoop]' = None +) -> StreamReader: + protocol = ResponseHandler(loop=loop) + return StreamReader(protocol, limit=2 ** 16, loop=loop) + + +def merge_params( + url: 'Union[URL, str]', + params: Optional[Dict] = None +) -> 'URL': + url = URL(url) + if params: + query_params = MultiDict(url.query) + query_params.extend(url.with_query(params).query) + return url.with_query(query_params) + return url + + +def normalize_url(url: 'Union[URL, str]') -> 'URL': + """Normalize url to make comparisons.""" + url = URL(url) + return url.with_query(urlencode(sorted(parse_qsl(url.query_string)))) + + +try: + from aiohttp import RequestInfo +except ImportError: + class RequestInfo(object): + __slots__ = ('url', 'method', 'headers', 'real_url') + + def __init__( + self, url: URL, method: str, headers: Dict, real_url: str + ): + self.url = url + self.method = method + self.headers = headers + self.real_url = real_url + +__all__ = [ + 'URL', + 'Pattern', + 'RequestInfo', + 'AIOHTTP_VERSION', + 'merge_params', + 'stream_reader_factory', + 'normalize_url', +] diff --git a/contrib/python/aioresponses/aioresponses/core.py b/contrib/python/aioresponses/aioresponses/core.py new file mode 100644 index 0000000000..2bb6d57365 --- /dev/null +++ b/contrib/python/aioresponses/aioresponses/core.py @@ -0,0 +1,549 @@ +# -*- coding: utf-8 -*- +import asyncio +import copy +import inspect +import json +from collections import namedtuple +from functools import wraps +from typing import ( + Any, + Callable, + cast, + Dict, + List, + Optional, + Tuple, + Type, + TypeVar, + Union, +) +from unittest.mock import Mock, patch +from uuid import uuid4 + +from aiohttp import ( + ClientConnectionError, + ClientResponse, + ClientSession, + hdrs, + http +) +from aiohttp.helpers import TimerNoop +from multidict import CIMultiDict, CIMultiDictProxy + +from .compat import ( + URL, + Pattern, + stream_reader_factory, + merge_params, + normalize_url, + RequestInfo, +) + + +_FuncT = TypeVar("_FuncT", bound=Callable[..., Any]) + + +class CallbackResult: + + def __init__(self, method: str = hdrs.METH_GET, + status: int = 200, + body: Union[str, bytes] = '', + content_type: str = 'application/json', + payload: Optional[Dict] = None, + headers: Optional[Dict] = None, + response_class: Optional[Type[ClientResponse]] = None, + reason: Optional[str] = None): + self.method = method + self.status = status + self.body = body + self.content_type = content_type + self.payload = payload + self.headers = headers + self.response_class = response_class + self.reason = reason + + +class RequestMatch(object): + url_or_pattern = None # type: Union[URL, Pattern] + + def __init__(self, url: Union[URL, str, Pattern], + method: str = hdrs.METH_GET, + status: int = 200, + body: Union[str, bytes] = '', + payload: Optional[Dict] = None, + exception: Optional[Exception] = None, + headers: Optional[Dict] = None, + content_type: str = 'application/json', + response_class: Optional[Type[ClientResponse]] = None, + timeout: bool = False, + repeat: bool = False, + reason: Optional[str] = None, + callback: Optional[Callable] = None): + if isinstance(url, Pattern): + self.url_or_pattern = url + self.match_func = self.match_regexp + else: + self.url_or_pattern = normalize_url(url) + self.match_func = self.match_str + self.method = method.lower() + self.status = status + self.body = body + self.payload = payload + self.exception = exception + if timeout: + self.exception = asyncio.TimeoutError('Connection timeout test') + self.headers = headers + self.content_type = content_type + self.response_class = response_class + self.repeat = repeat + self.reason = reason + if self.reason is None: + try: + self.reason = http.RESPONSES[self.status][0] + except (IndexError, KeyError): + self.reason = '' + self.callback = callback + + def match_str(self, url: URL) -> bool: + return self.url_or_pattern == url + + def match_regexp(self, url: URL) -> bool: + # This method is used if and only if self.url_or_pattern is a pattern. + return bool( + self.url_or_pattern.match(str(url)) # type:ignore[union-attr] + ) + + def match(self, method: str, url: URL) -> bool: + if self.method != method.lower(): + return False + return self.match_func(url) + + def _build_raw_headers(self, headers: Dict) -> Tuple: + """ + Convert a dict of headers to a tuple of tuples + + Mimics the format of ClientResponse. + """ + raw_headers = [] + for k, v in headers.items(): + raw_headers.append((k.encode('utf8'), v.encode('utf8'))) + return tuple(raw_headers) + + def _build_response(self, url: 'Union[URL, str]', + method: str = hdrs.METH_GET, + request_headers: Optional[Dict] = None, + status: int = 200, + body: Union[str, bytes] = '', + content_type: str = 'application/json', + payload: Optional[Dict] = None, + headers: Optional[Dict] = None, + response_class: Optional[Type[ClientResponse]] = None, + reason: Optional[str] = None) -> ClientResponse: + if response_class is None: + response_class = ClientResponse + if payload is not None: + body = json.dumps(payload) + if not isinstance(body, bytes): + body = str.encode(body) + if request_headers is None: + request_headers = {} + loop = Mock() + loop.get_debug = Mock() + loop.get_debug.return_value = True + kwargs = {} # type: Dict[str, Any] + kwargs['request_info'] = RequestInfo( + url=url, + method=method, + headers=CIMultiDictProxy(CIMultiDict(**request_headers)), + ) + kwargs['writer'] = None + kwargs['continue100'] = None + kwargs['timer'] = TimerNoop() + kwargs['traces'] = [] + kwargs['loop'] = loop + kwargs['session'] = None + + # We need to initialize headers manually + _headers = CIMultiDict({hdrs.CONTENT_TYPE: content_type}) + if headers: + _headers.update(headers) + raw_headers = self._build_raw_headers(_headers) + resp = response_class(method, url, **kwargs) + + for hdr in _headers.getall(hdrs.SET_COOKIE, ()): + resp.cookies.load(hdr) + + # Reified attributes + resp._headers = _headers + resp._raw_headers = raw_headers + + resp.status = status + resp.reason = reason + resp.content = stream_reader_factory(loop) + resp.content.feed_data(body) + resp.content.feed_eof() + return resp + + async def build_response( + self, url: URL, **kwargs: Any + ) -> 'Union[ClientResponse, Exception]': + if callable(self.callback): + if asyncio.iscoroutinefunction(self.callback): + result = await self.callback(url, **kwargs) + else: + result = self.callback(url, **kwargs) + else: + result = None + + if self.exception is not None: + return self.exception + + result = self if result is None else result + resp = self._build_response( + url=url, + method=result.method, + request_headers=kwargs.get("headers"), + status=result.status, + body=result.body, + content_type=result.content_type, + payload=result.payload, + headers=result.headers, + response_class=result.response_class, + reason=result.reason) + return resp + + +RequestCall = namedtuple('RequestCall', ['args', 'kwargs']) + + +class aioresponses(object): + """Mock aiohttp requests made by ClientSession.""" + _matches = None # type: Dict[str, RequestMatch] + _responses = None # type: List[ClientResponse] + requests = None # type: Dict + + def __init__(self, **kwargs: Any): + self._param = kwargs.pop('param', None) + self._passthrough = kwargs.pop('passthrough', []) + self.patcher = patch('aiohttp.client.ClientSession._request', + side_effect=self._request_mock, + autospec=True) + self.requests = {} + + def __enter__(self) -> 'aioresponses': + self.start() + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.stop() + + def __call__(self, f: _FuncT) -> _FuncT: + def _pack_arguments(ctx, *args, **kwargs) -> Tuple[Tuple, Dict]: + if self._param: + kwargs[self._param] = ctx + else: + args += (ctx,) + return args, kwargs + + if asyncio.iscoroutinefunction(f): + @wraps(f) + async def wrapped(*args, **kwargs): + with self as ctx: + args, kwargs = _pack_arguments(ctx, *args, **kwargs) + return await f(*args, **kwargs) + else: + @wraps(f) + def wrapped(*args, **kwargs): + with self as ctx: + args, kwargs = _pack_arguments(ctx, *args, **kwargs) + return f(*args, **kwargs) + return cast(_FuncT, wrapped) + + def clear(self) -> None: + self._responses.clear() + self._matches.clear() + + def start(self) -> None: + self._responses = [] + self._matches = {} + self.patcher.start() + self.patcher.return_value = self._request_mock + + def stop(self) -> None: + for response in self._responses: + response.close() + self.patcher.stop() + self.clear() + + def head(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_HEAD, **kwargs) + + def get(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_GET, **kwargs) + + def post(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_POST, **kwargs) + + def put(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_PUT, **kwargs) + + def patch(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_PATCH, **kwargs) + + def delete(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_DELETE, **kwargs) + + def options(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: + self.add(url, method=hdrs.METH_OPTIONS, **kwargs) + + def add(self, url: 'Union[URL, str, Pattern]', method: str = hdrs.METH_GET, + status: int = 200, + body: Union[str, bytes] = '', + exception: Optional[Exception] = None, + content_type: str = 'application/json', + payload: Optional[Dict] = None, + headers: Optional[Dict] = None, + response_class: Optional[Type[ClientResponse]] = None, + repeat: bool = False, + timeout: bool = False, + reason: Optional[str] = None, + callback: Optional[Callable] = None) -> None: + + self._matches[str(uuid4())] = (RequestMatch( + url, + method=method, + status=status, + content_type=content_type, + body=body, + exception=exception, + payload=payload, + headers=headers, + response_class=response_class, + repeat=repeat, + timeout=timeout, + reason=reason, + callback=callback, + )) + + def _format_call_signature(self, *args, **kwargs) -> str: + message = '%s(%%s)' % self.__class__.__name__ or 'mock' + formatted_args = '' + args_string = ', '.join([repr(arg) for arg in args]) + kwargs_string = ', '.join([ + '%s=%r' % (key, value) for key, value in kwargs.items() + ]) + if args_string: + formatted_args = args_string + if kwargs_string: + if formatted_args: + formatted_args += ', ' + formatted_args += kwargs_string + + return message % formatted_args + + def assert_not_called(self): + """assert that the mock was never called. + """ + if len(self.requests) != 0: + msg = ("Expected '%s' to not have been called. Called %s times." + % (self.__class__.__name__, + len(self._responses))) + raise AssertionError(msg) + + def assert_called(self): + """assert that the mock was called at least once. + """ + if len(self.requests) == 0: + msg = ("Expected '%s' to have been called." + % (self.__class__.__name__,)) + raise AssertionError(msg) + + def assert_called_once(self): + """assert that the mock was called only once. + """ + call_count = len(self.requests) + if call_count == 1: + call_count = len(list(self.requests.values())[0]) + if not call_count == 1: + msg = ("Expected '%s' to have been called once. Called %s times." + % (self.__class__.__name__, + call_count)) + + raise AssertionError(msg) + + def assert_called_with(self, url: 'Union[URL, str, Pattern]', + method: str = hdrs.METH_GET, + *args: Any, + **kwargs: Any): + """assert that the last call was made with the specified arguments. + + Raises an AssertionError if the args and keyword args passed in are + different to the last call to the mock.""" + url = normalize_url(merge_params(url, kwargs.get('params'))) + method = method.upper() + key = (method, url) + try: + expected = self.requests[key][-1] + except KeyError: + expected_string = self._format_call_signature( + url, method=method, *args, **kwargs + ) + raise AssertionError( + '%s call not found' % expected_string + ) + actual = self._build_request_call(method, *args, **kwargs) + if not expected == actual: + expected_string = self._format_call_signature( + expected, + ) + actual_string = self._format_call_signature( + actual + ) + raise AssertionError( + '%s != %s' % (expected_string, actual_string) + ) + + def assert_any_call(self, url: 'Union[URL, str, Pattern]', + method: str = hdrs.METH_GET, + *args: Any, + **kwargs: Any): + """assert the mock has been called with the specified arguments. + The assert passes if the mock has *ever* been called, unlike + `assert_called_with` and `assert_called_once_with` that only pass if + the call is the most recent one.""" + url = normalize_url(merge_params(url, kwargs.get('params'))) + method = method.upper() + key = (method, url) + + try: + self.requests[key] + except KeyError: + expected_string = self._format_call_signature( + url, method=method, *args, **kwargs + ) + raise AssertionError( + '%s call not found' % expected_string + ) + + def assert_called_once_with(self, *args: Any, **kwargs: Any): + """assert that the mock was called once with the specified arguments. + Raises an AssertionError if the args and keyword args passed in are + different to the only call to the mock.""" + self.assert_called_once() + self.assert_called_with(*args, **kwargs) + + @staticmethod + def is_exception(resp_or_exc: Union[ClientResponse, Exception]) -> bool: + if inspect.isclass(resp_or_exc): + parent_classes = set(inspect.getmro(resp_or_exc)) + if {Exception, BaseException} & parent_classes: + return True + else: + if isinstance(resp_or_exc, (Exception, BaseException)): + return True + return False + + async def match( + self, method: str, + url: URL, + allow_redirects: bool = True, + **kwargs: Any + ) -> Optional['ClientResponse']: + history = [] + while True: + for key, matcher in self._matches.items(): + if matcher.match(method, url): + response_or_exc = await matcher.build_response( + url, allow_redirects=allow_redirects, **kwargs + ) + break + else: + return None + + if matcher.repeat is False: + del self._matches[key] + + if self.is_exception(response_or_exc): + raise response_or_exc + # If response_or_exc was an exception, it would have been raised. + # At this point we can be sure it's a ClientResponse + response: ClientResponse + response = response_or_exc # type:ignore[assignment] + is_redirect = response.status in (301, 302, 303, 307, 308) + if is_redirect and allow_redirects: + if hdrs.LOCATION not in response.headers: + break + history.append(response) + redirect_url = URL(response.headers[hdrs.LOCATION]) + if redirect_url.is_absolute(): + url = redirect_url + else: + url = url.join(redirect_url) + method = 'get' + continue + else: + break + + response._history = tuple(history) + return response + + async def _request_mock(self, orig_self: ClientSession, + method: str, url: 'Union[URL, str]', + *args: Tuple, + **kwargs: Any) -> 'ClientResponse': + """Return mocked response object or raise connection error.""" + if orig_self.closed: + raise RuntimeError('Session is closed') + + url_origin = url + url = normalize_url(merge_params(url, kwargs.get('params'))) + url_str = str(url) + for prefix in self._passthrough: + if url_str.startswith(prefix): + return (await self.patcher.temp_original( + orig_self, method, url_origin, *args, **kwargs + )) + + key = (method, url) + self.requests.setdefault(key, []) + request_call = self._build_request_call(method, *args, **kwargs) + self.requests[key].append(request_call) + + response = await self.match(method, url, **kwargs) + + if response is None: + raise ClientConnectionError( + 'Connection refused: {} {}'.format(method, url) + ) + self._responses.append(response) + + # Automatically call response.raise_for_status() on a request if the + # request was initialized with raise_for_status=True. Also call + # response.raise_for_status() if the client session was initialized + # with raise_for_status=True, unless the request was called with + # raise_for_status=False. + raise_for_status = kwargs.get('raise_for_status') + if raise_for_status is None: + raise_for_status = getattr( + orig_self, '_raise_for_status', False + ) + if raise_for_status: + response.raise_for_status() + + return response + + def _build_request_call(self, method: str = hdrs.METH_GET, + *args: Any, + allow_redirects: bool = True, + **kwargs: Any): + """Return request call.""" + kwargs.setdefault('allow_redirects', allow_redirects) + if method == 'POST': + kwargs.setdefault('data', None) + + try: + kwargs_copy = copy.deepcopy(kwargs) + except (TypeError, ValueError): + # Handle the fact that some values cannot be deep copied + kwargs_copy = kwargs + return RequestCall(args, kwargs_copy) diff --git a/contrib/python/aioresponses/aioresponses/py.typed b/contrib/python/aioresponses/aioresponses/py.typed new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/contrib/python/aioresponses/aioresponses/py.typed diff --git a/contrib/python/aioresponses/ya.make b/contrib/python/aioresponses/ya.make new file mode 100644 index 0000000000..574b5f85f1 --- /dev/null +++ b/contrib/python/aioresponses/ya.make @@ -0,0 +1,33 @@ +# Generated by devtools/yamaker (pypi). + +PY3_LIBRARY() + +VERSION(0.7.6) + +LICENSE(MIT) + +PEERDIR( + contrib/python/aiohttp +) + +NO_LINT() + +PY_SRCS( + TOP_LEVEL + aioresponses/__init__.py + aioresponses/compat.py + aioresponses/core.py +) + +RESOURCE_FILES( + PREFIX contrib/python/aioresponses/ + .dist-info/METADATA + .dist-info/top_level.txt + aioresponses/py.typed +) + +END() + +RECURSE_FOR_TESTS( + tests +) |