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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
--- contrib/python/aioresponses/tests/test_aioresponses.py (index)
+++ contrib/python/aioresponses/tests/test_aioresponses.py (working tree)
@@ -8,9 +8,10 @@ from unittest.mock import patch
from aiohttp import hdrs
from aiohttp import http
+from aiohttp import web
from aiohttp.client import ClientSession
from aiohttp.client_reqrep import ClientResponse
-from ddt import ddt, data
+from aiohttp.test_utils import TestServer
from packaging.version import Version
try:
@@ -31,10 +32,11 @@ from aioresponses import CallbackResult, aioresponses
-@ddt
class AIOResponsesTestCase(AsyncTestCase):
async def setup(self):
self.url = 'http://example.com/api?foo=bar#fragment'
self.session = ClientSession()
+ self.external_server = self.make_external_server()
+ await self.external_server.start_server()
async def teardown(self):
@@ -46,24 +48,43 @@ class AIOResponsesTestCase(AsyncTestCase):
def run_async(self, coroutine: Union[Coroutine, Generator]):
return self.loop.run_until_complete(coroutine)
+ def make_external_server(self):
+ """
+ В оригинальном коде для тестирования passthrough исполняются настоящие
+ запросы до сайта http://httpbin.org. В sandbox нет интернета, потому я заменил
+ httpbin на локальный сервер. Осторожнее при обновлении!
+ """
+ async def testserver_status_201(request):
+ return web.Response(status=201)
+
+ async def testserver_get(request):
+ return web.Response()
+
+ app = web.Application()
+ app.router.add_get('/status/201', testserver_status_201)
+ app.router.add_get('/get', testserver_get)
+
+ return TestServer(app)
+
async def request(self, url: str):
return await self.session.get(url)
- @data(
- hdrs.METH_HEAD,
- hdrs.METH_GET,
- hdrs.METH_POST,
- hdrs.METH_PUT,
- hdrs.METH_PATCH,
- hdrs.METH_DELETE,
- hdrs.METH_OPTIONS,
- )
- @patch('aioresponses.aioresponses.add')
@fail_on(unused_loop=False)
- def test_shortcut_method(self, http_method, mocked):
- with aioresponses() as m:
- getattr(m, http_method.lower())(self.url)
- mocked.assert_called_once_with(self.url, method=http_method)
+ def test_shortcut_method(self):
+ for http_method in (
+ hdrs.METH_HEAD,
+ hdrs.METH_GET,
+ hdrs.METH_POST,
+ hdrs.METH_PUT,
+ hdrs.METH_PATCH,
+ hdrs.METH_DELETE,
+ hdrs.METH_OPTIONS,
+ ):
+ with self.subTest(), \
+ patch('aioresponses.aioresponses.add') as mocked, \
+ aioresponses() as m:
+ getattr(m, http_method.lower())(self.url)
+ mocked.assert_called_once_with(self.url, method=http_method)
@aioresponses()
def test_returned_instance(self, m):
@@ -369,7 +390,7 @@ class AIOResponsesTestCase(AsyncTestCase):
assert str(exception_info.exception) == "Session is closed"
async def test_address_as_instance_of_url_combined_with_pass_through(self):
- external_api = 'http://httpbin.org/status/201'
+ external_api = str(self.external_server.make_url('/status/201'))
async def doit():
api_resp = await self.session.get(self.url)
@@ -386,7 +407,7 @@ class AIOResponsesTestCase(AsyncTestCase):
self.assertEqual(ext.status, 201)
async def test_pass_through_with_origin_params(self):
- external_api = 'http://httpbin.org/get'
+ external_api = str(self.external_server.make_url('/get'))
async def doit(params):
# we have to hit actual url,
@@ -400,7 +421,7 @@ class AIOResponsesTestCase(AsyncTestCase):
params = {'foo': 'bar'}
ext = await doit(params=params)
self.assertEqual(ext.status, 200)
- self.assertEqual(str(ext.url), 'http://httpbin.org/get?foo=bar')
+ self.assertEqual(str(ext.url), external_api + '?foo=bar')
@aioresponses()
async def test_custom_response_class(self, m):
--- contrib/python/aioresponses/tests/test_compat.py (index)
+++ contrib/python/aioresponses/tests/test_compat.py (working tree)
@@ -2,7 +2,6 @@
from typing import Union
from unittest import TestCase
-from ddt import ddt, data
from yarl import URL
from aioresponses.compat import merge_params
@@ -12,7 +11,6 @@ def get_url(url: str, as_str: bool) -> Union[URL, str]:
return url if as_str else URL(url)
-@ddt
class CompatTestCase(TestCase):
use_default_loop = False
@@ -20,29 +18,33 @@ class CompatTestCase(TestCase):
self.url_with_parameters = 'http://example.com/api?foo=bar#fragment'
self.url_without_parameters = 'http://example.com/api?#fragment'
- @data(True, False)
- def test_no_params_returns_same_url__as_str(self, as_str):
- url = get_url(self.url_with_parameters, as_str)
- self.assertEqual(
- merge_params(url, None), URL(self.url_with_parameters)
- )
-
- @data(True, False)
- def test_empty_params_returns_same_url__as_str(self, as_str):
- url = get_url(self.url_with_parameters, as_str)
- self.assertEqual(merge_params(url, {}), URL(self.url_with_parameters))
-
- @data(True, False)
- def test_both_with_params_returns_corrected_url__as_str(self, as_str):
- url = get_url(self.url_with_parameters, as_str)
- self.assertEqual(
- merge_params(url, {'x': 42}),
- URL('http://example.com/api?foo=bar&x=42#fragment'),
- )
-
- @data(True, False)
- def test_base_without_params_returns_corrected_url__as_str(self, as_str):
- expected_url = URL('http://example.com/api?x=42#fragment')
- url = get_url(self.url_without_parameters, as_str)
-
- self.assertEqual(merge_params(url, {'x': 42}), expected_url)
+ def test_no_params_returns_same_url__as_str(self):
+ for as_str in (True, False):
+ with self.subTest():
+ url = get_url(self.url_with_parameters, as_str)
+ self.assertEqual(
+ merge_params(url, None), URL(self.url_with_parameters)
+ )
+
+ def test_empty_params_returns_same_url__as_str(self):
+ for as_str in (True, False):
+ with self.subTest():
+ url = get_url(self.url_with_parameters, as_str)
+ self.assertEqual(merge_params(url, {}), URL(self.url_with_parameters))
+
+ def test_both_with_params_returns_corrected_url__as_str(self):
+ for as_str in (True, False):
+ with self.subTest():
+ url = get_url(self.url_with_parameters, as_str)
+ self.assertEqual(
+ merge_params(url, {'x': 42}),
+ URL('http://example.com/api?foo=bar&x=42#fragment'),
+ )
+
+ def test_base_without_params_returns_corrected_url__as_str(self):
+ for as_str in (True, False):
+ with self.subTest():
+ expected_url = URL('http://example.com/api?x=42#fragment')
+ url = get_url(self.url_without_parameters, as_str)
+
+ self.assertEqual(merge_params(url, {'x': 42}), expected_url)
|
