-import collections.abc
import typing
from ..concurrency import AsyncioBackend
CONNECTIONS_DICT = typing.Dict[Origin, typing.List[HTTPConnection]]
-class ConnectionStore(collections.abc.Sequence):
+class ConnectionStore:
"""
We need to maintain collections of connections in a way that allows us to:
def __iter__(self) -> typing.Iterator[HTTPConnection]:
return iter(self.all.keys())
- def __getitem__(self, key: typing.Any) -> typing.Any:
- if key in self.all:
- return key
- return None
-
def __len__(self) -> int:
return len(self.all)
AuthTypes = typing.Union[
typing.Tuple[typing.Union[str, bytes], typing.Union[str, bytes]],
- typing.Callable[["Request"], "Request"]
+ typing.Callable[["Request"], "Request"],
]
RequestData = typing.Union[dict, bytes, typing.AsyncIterator[bytes]]
@property
def username(self) -> str:
userinfo = self.components.userinfo or ""
- return userinfo.partition(':')[0]
+ return userinfo.partition(":")[0]
@property
def password(self) -> str:
userinfo = self.components.userinfo or ""
- return userinfo.partition(':')[2]
+ return userinfo.partition(":")[2]
@property
def host(self) -> str:
import json
-from urllib.parse import parse_qs
import pytest
ssl: SSLConfig = None,
timeout: TimeoutConfig = None,
) -> Response:
- body = json.dumps({"auth": request.headers.get('Authorization')}).encode()
+ body = json.dumps({"auth": request.headers.get("Authorization")}).encode()
return Response(200, content=body, request=request)
def test_basic_auth():
url = "https://example.org/"
- auth = ('tomchristie', 'password123')
+ auth = ("tomchristie", "password123")
with Client(dispatch=MockDispatch()) as client:
response = client.get(url, auth=auth)
assert response.status_code == 200
- assert json.loads(response.text) == {'auth': 'Basic dG9tY2hyaXN0aWU6cGFzc3dvcmQxMjM='}
+ assert json.loads(response.text) == {
+ "auth": "Basic dG9tY2hyaXN0aWU6cGFzc3dvcmQxMjM="
+ }
def test_basic_auth_in_url():
response = client.get(url)
assert response.status_code == 200
- assert json.loads(response.text) == {'auth': 'Basic dG9tY2hyaXN0aWU6cGFzc3dvcmQxMjM='}
+ assert json.loads(response.text) == {
+ "auth": "Basic dG9tY2hyaXN0aWU6cGFzc3dvcmQxMjM="
+ }
def test_basic_auth_on_session():
url = "https://example.org/"
- auth = ('tomchristie', 'password123')
+ auth = ("tomchristie", "password123")
with Client(dispatch=MockDispatch(), auth=auth) as client:
response = client.get(url)
assert response.status_code == 200
- assert json.loads(response.text) == {'auth': 'Basic dG9tY2hyaXN0aWU6cGFzc3dvcmQxMjM='}
+ assert json.loads(response.text) == {
+ "auth": "Basic dG9tY2hyaXN0aWU6cGFzc3dvcmQxMjM="
+ }
def test_custom_auth():
url = "https://example.org/"
def auth(request):
- request.headers['Authorization'] = 'Token 123'
+ request.headers["Authorization"] = "Token 123"
return request
with Client(dispatch=MockDispatch()) as client:
response = client.get(url, auth=auth)
assert response.status_code == 200
- assert json.loads(response.text) == {'auth': 'Token 123'}
+ assert json.loads(response.text) == {"auth": "Token 123"}
async def echo_body(scope, receive, send):
- body = b''
+ body = b""
more_body = True
while more_body:
message = await receive()
- body += message.get('body', b'')
- more_body = message.get('more_body', False)
+ body += message.get("body", b"")
+ more_body = message.get("more_body", False)
await send(
{