feat(proxy): proxy for aiohttp,base sessions (#284)

* feat(proxy): proxy for aiohttp,base sessions

Add support for proxies in aiohttp session with aiohttp_socks library,
edit BaseSession class to support proxies for other sessions in future.

* fix(annotation): missing underscore before "private" typevar

* chore: remove redundant of proxy_url schema for socks version

* test: add missing test

Add missing test, remove BaseSession.cfg and switch to
implementing class' "private" traits, add aiohttp_socks in dependency list as
optional and extra.

* feat(session):

Implement asyncio session for requests [wip]

* feat(proxy chain): Chained proxy support in aiohttp session

Add ChainProxyConnector support, !pin pydantic to "1.4", add
documentation on aiohttp connector.

* style(mypy): apply linter changes

* tests(mock): remove await for magic mock

* fix dangling dependency

* refactor(generic):

get rid of generic behaviour for base session
This commit is contained in:
Martin Winks 2020-05-02 17:12:53 +04:00 committed by GitHub
parent 2553f5f19e
commit 15bcc0ba9f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 245 additions and 8 deletions

View file

@ -1,6 +1,7 @@
from typing import AsyncContextManager, AsyncGenerator
import aiohttp
import aiohttp_socks
import pytest
from aresponses import ResponsesMockServer
@ -29,6 +30,59 @@ class TestAiohttpSession:
assert session._session is not None
assert isinstance(aiohttp_session, aiohttp.ClientSession)
@pytest.mark.asyncio
async def test_create_proxy_session(self):
session = AiohttpSession(
proxy=("socks5://proxy.url/", aiohttp.BasicAuth("login", "password", "encoding"))
)
assert session._connector_type == aiohttp_socks.ProxyConnector
assert isinstance(session._connector_init, dict)
assert session._connector_init["proxy_type"] is aiohttp_socks.ProxyType.SOCKS5
aiohttp_session = await session.create_session()
assert isinstance(aiohttp_session.connector, aiohttp_socks.ProxyConnector)
@pytest.mark.asyncio
async def test_create_proxy_session_proxy_url(self):
session = AiohttpSession(proxy="socks4://proxy.url/")
assert isinstance(session.proxy, str)
assert isinstance(session._connector_init, dict)
assert session._connector_init["proxy_type"] is aiohttp_socks.ProxyType.SOCKS4
aiohttp_session = await session.create_session()
assert isinstance(aiohttp_session.connector, aiohttp_socks.ProxyConnector)
@pytest.mark.asyncio
async def test_create_proxy_session_chained_proxies(self):
session = AiohttpSession(
proxy=[
"socks4://proxy.url/",
"socks5://proxy.url/",
"http://user:password@127.0.0.1:3128",
]
)
assert isinstance(session.proxy, list)
assert isinstance(session._connector_init, dict)
assert isinstance(session._connector_init["proxy_infos"], list)
assert isinstance(session._connector_init["proxy_infos"][0], aiohttp_socks.ProxyInfo)
assert (
session._connector_init["proxy_infos"][0].proxy_type is aiohttp_socks.ProxyType.SOCKS4
)
assert (
session._connector_init["proxy_infos"][1].proxy_type is aiohttp_socks.ProxyType.SOCKS5
)
assert session._connector_init["proxy_infos"][2].proxy_type is aiohttp_socks.ProxyType.HTTP
aiohttp_session = await session.create_session()
assert isinstance(aiohttp_session.connector, aiohttp_socks.ChainProxyConnector)
@pytest.mark.asyncio
async def test_close_session(self):
session = AiohttpSession()