1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See LICENSE.txt in the project root for
# license information.
# -------------------------------------------------------------------------
import pytest
import types
import io
############################## LISTS USED TO PARAMETERIZE TESTS ##############################
from corehttp.rest import HttpRequest
from corehttp.rest._http_response_impl import HttpResponseImpl as RestHttpResponse
from corehttp.rest._http_response_impl_async import AsyncHttpResponseImpl as RestAsyncHttpResponse
SYNC_TRANSPORTS = []
ASYNC_TRANSPORTS = []
SYNC_TRANSPORT_RESPONSES = []
ASYNC_TRANSPORT_RESPONSES = []
try:
from corehttp.rest._requests_basic import RestRequestsTransportResponse
from corehttp.transport.requests import RequestsTransport
SYNC_TRANSPORTS.append(RequestsTransport)
SYNC_TRANSPORT_RESPONSES.append(RestRequestsTransportResponse)
except (ImportError, SyntaxError):
pass
try:
from corehttp.rest._aiohttp import RestAioHttpTransportResponse
from corehttp.transport.aiohttp import AioHttpTransport
ASYNC_TRANSPORTS.append(AioHttpTransport)
ASYNC_TRANSPORT_RESPONSES.append(RestAioHttpTransportResponse)
except (ImportError, SyntaxError):
pass
try:
from corehttp.rest._httpx import HttpXTransportResponse, AsyncHttpXTransportResponse
from corehttp.transport.httpx import HttpXTransport, AsyncHttpXTransport
SYNC_TRANSPORTS.append(HttpXTransport)
SYNC_TRANSPORT_RESPONSES.append(HttpXTransportResponse)
ASYNC_TRANSPORTS.append(AsyncHttpXTransport)
ASYNC_TRANSPORT_RESPONSES.append(AsyncHttpXTransportResponse)
except (ImportError, SyntaxError):
pass
HTTP_RESPONSES = [RestHttpResponse]
ASYNC_HTTP_RESPONSES = [RestAsyncHttpResponse]
############################## HELPER FUNCTIONS ##############################
def create_transport_response(http_response, *args, **kwargs):
# this creates transport-specific responses,
# like requests responses / aiohttp responses
block_size = args[2] if len(args) > 2 else None
return http_response(request=args[0], internal_response=args[1], block_size=block_size, **kwargs)
def create_http_response(http_response, *args, **kwargs):
# since the actual http_response object is
# an ABC for our new responses, it's a little more
# complicated creating a pure http response.
# here, we use the HttpResponsImpl, but we still have to pass
# more things to the init, so making a separate operation
block_size = args[2] if len(args) > 2 else None
return http_response(
request=args[0],
internal_response=args[1],
block_size=block_size,
status_code=kwargs.pop("status_code", 200),
reason=kwargs.pop("reason", "OK"),
content_type=kwargs.pop("content_type", "application/json"),
headers=kwargs.pop("headers", {}),
stream_download_generator=kwargs.pop("stream_download_generator", None),
**kwargs
)
def readonly_checks(response):
# We want these properties to be completely readonly.
assert isinstance(response.request, HttpRequest)
assert isinstance(response.status_code, int)
assert response.headers
assert response.content_type == "text/html; charset=utf-8"
assert response.is_closed
with pytest.raises(AttributeError):
response.is_closed = False
assert response.is_stream_consumed
with pytest.raises(AttributeError):
response.is_stream_consumed = False
# you can set encoding
assert response.encoding == "utf-8"
response.encoding = "blah"
assert response.encoding == "blah"
assert isinstance(response.url, str)
with pytest.raises(AttributeError):
response.url = "http://fakeurl"
assert response.content is not None
with pytest.raises(AttributeError):
response.content = b"bad"
for attr in dir(response):
if attr[0] == "_":
# don't care about private variables
continue
if type(getattr(response, attr)) == types.MethodType:
# methods aren't "readonly"
continue
if attr == "encoding":
# encoding is the only settable new attr
continue
class NamedIo(io.BytesIO):
def __init__(self, name: str, *args, **kwargs):
super(NamedIo, self).__init__(*args, **kwargs)
self.name = name
|
