Repository navigation
Expand file tree
/
Copy pathtest_request_attribute_binding.py
More file actions
216 lines (180 loc) · 7.46 KB
/
Copy pathtest_request_attribute_binding.py
File metadata and controls
216 lines (180 loc) · 7.46 KB
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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
from testfixtures import LogCapture
from twisted.internet import defer
from twisted.trial.unittest import TestCase
from scrapy import Request, signals
from scrapy.http.response import Response
from scrapy.utils.test import get_crawler
from tests.mockserver import MockServer
from tests.spiders import SingleRequestSpider
OVERRIDDEN_URL = "https://example.org"
class ProcessResponseMiddleware:
def process_response(self, request, response, spider):
return response.replace(request=Request(OVERRIDDEN_URL))
class RaiseExceptionRequestMiddleware:
def process_request(self, request, spider):
1 / 0
return request
class CatchExceptionOverrideRequestMiddleware:
def process_exception(self, request, exception, spider):
return Response(
url="http://localhost/",
body=b"Caught " + exception.__class__.__name__.encode("utf-8"),
request=Request(OVERRIDDEN_URL),
)
class CatchExceptionDoNotOverrideRequestMiddleware:
def process_exception(self, request, exception, spider):
return Response(
url="http://localhost/",
body=b"Caught " + exception.__class__.__name__.encode("utf-8"),
)
class AlternativeCallbacksSpider(SingleRequestSpider):
name = "alternative_callbacks_spider"
def alt_callback(self, response, foo=None):
self.logger.info("alt_callback was invoked with foo=%s", foo)
class AlternativeCallbacksMiddleware:
def process_response(self, request, response, spider):
new_request = request.replace(
url=OVERRIDDEN_URL,
callback=spider.alt_callback,
cb_kwargs={"foo": "bar"},
)
return response.replace(request=new_request)
class CrawlTestCase(TestCase):
def setUp(self):
self.mockserver = MockServer()
self.mockserver.__enter__()
def tearDown(self):
self.mockserver.__exit__(None, None, None)
@defer.inlineCallbacks
def test_response_200(self):
url = self.mockserver.url("/status?n=200")
crawler = get_crawler(SingleRequestSpider)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.request.url, url)
@defer.inlineCallbacks
def test_response_error(self):
for status in ("404", "500"):
url = self.mockserver.url(f"/status?n={status}")
crawler = get_crawler(SingleRequestSpider)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
failure = crawler.spider.meta["failure"]
response = failure.value.response
self.assertEqual(failure.request.url, url)
self.assertEqual(response.request.url, url)
@defer.inlineCallbacks
def test_downloader_middleware_raise_exception(self):
url = self.mockserver.url("/status?n=200")
crawler = get_crawler(
SingleRequestSpider,
{
"DOWNLOADER_MIDDLEWARES": {
RaiseExceptionRequestMiddleware: 590,
},
},
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
failure = crawler.spider.meta["failure"]
self.assertEqual(failure.request.url, url)
self.assertIsInstance(failure.value, ZeroDivisionError)
@defer.inlineCallbacks
def test_downloader_middleware_override_request_in_process_response(self):
"""
Downloader middleware which returns a response with an specific 'request' attribute.
* The spider callback should receive the overridden response.request
* Handlers listening to the response_received signal should receive the overridden response.request
* The "crawled" log message should show the overridden response.request
"""
signal_params = {}
def signal_handler(response, request, spider):
signal_params["response"] = response
signal_params["request"] = request
url = self.mockserver.url("/status?n=200")
crawler = get_crawler(
SingleRequestSpider,
{
"DOWNLOADER_MIDDLEWARES": {
ProcessResponseMiddleware: 595,
}
},
)
crawler.signals.connect(signal_handler, signal=signals.response_received)
with LogCapture() as log:
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.request.url, OVERRIDDEN_URL)
self.assertEqual(signal_params["response"].url, url)
self.assertEqual(signal_params["request"].url, OVERRIDDEN_URL)
log.check_present(
(
"scrapy.core.engine",
"DEBUG",
f"Crawled (200) <GET {OVERRIDDEN_URL}> (referer: None)",
),
)
@defer.inlineCallbacks
def test_downloader_middleware_override_in_process_exception(self):
"""
An exception is raised but caught by the next middleware, which
returns a Response with a specific 'request' attribute.
The spider callback should receive the overridden response.request
"""
url = self.mockserver.url("/status?n=200")
crawler = get_crawler(
SingleRequestSpider,
{
"DOWNLOADER_MIDDLEWARES": {
RaiseExceptionRequestMiddleware: 590,
CatchExceptionOverrideRequestMiddleware: 595,
},
},
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.body, b"Caught ZeroDivisionError")
self.assertEqual(response.request.url, OVERRIDDEN_URL)
@defer.inlineCallbacks
def test_downloader_middleware_do_not_override_in_process_exception(self):
"""
An exception is raised but caught by the next middleware, which
returns a Response without a specific 'request' attribute.
The spider callback should receive the original response.request
"""
url = self.mockserver.url("/status?n=200")
crawler = get_crawler(
SingleRequestSpider,
{
"DOWNLOADER_MIDDLEWARES": {
RaiseExceptionRequestMiddleware: 590,
CatchExceptionDoNotOverrideRequestMiddleware: 595,
},
},
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.body, b"Caught ZeroDivisionError")
self.assertEqual(response.request.url, url)
@defer.inlineCallbacks
def test_downloader_middleware_alternative_callback(self):
"""
Downloader middleware which returns a response with a
specific 'request' attribute, with an alternative callback
"""
crawler = get_crawler(
AlternativeCallbacksSpider,
{
"DOWNLOADER_MIDDLEWARES": {
AlternativeCallbacksMiddleware: 595,
}
},
)
with LogCapture() as log:
url = self.mockserver.url("/status?n=200")
yield crawler.crawl(seed=url, mockserver=self.mockserver)
log.check_present(
(
"alternative_callbacks_spider",
"INFO",
"alt_callback was invoked with foo=bar",
),
)