mirror of https://github.com/scrapy/scrapy.git
Add tests to MediaPipeline. closes #269
--HG-- extra : rebase_source : ccf726e147b5c97f7cba60d20ce2fca58c687a3e
This commit is contained in:
Daniel Grana
parent
d17edd4a59
commit
ad43917322
|
|
@ -1,3 +1,4 @@
|
|||
from collections import defaultdict
|
||||
from twisted.internet.defer import Deferred, DeferredList
|
||||
|
||||
from scrapy.utils.defer import mustbe_deferred, defer_result
|
||||
|
|
@ -8,20 +9,22 @@ from scrapy.utils.misc import arg_to_iter
|
|||
|
||||
class MediaPipeline(object):
|
||||
|
||||
DOWNLOAD_PRIORITY = 1000
|
||||
LOG_FAILED_RESULTS = True
|
||||
|
||||
class SpiderInfo(object):
|
||||
def __init__(self, spider):
|
||||
self.spider = spider
|
||||
self.downloading = {}
|
||||
self.downloading = set()
|
||||
self.downloaded = {}
|
||||
self.waiting = {}
|
||||
self.waiting = defaultdict(list)
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self, download_func=None):
|
||||
self.spiderinfo = {}
|
||||
self._download_func = download_func or self._default_download_func()
|
||||
|
||||
def _default_download_func(self):
|
||||
from scrapy.project import crawler
|
||||
self.crawler = crawler
|
||||
return crawler.engine.download
|
||||
|
||||
def open_spider(self, spider):
|
||||
self.spiderinfo[spider] = self.SpiderInfo(spider)
|
||||
|
|
@ -32,63 +35,53 @@ class MediaPipeline(object):
|
|||
def process_item(self, item, spider):
|
||||
info = self.spiderinfo[spider]
|
||||
requests = arg_to_iter(self.get_media_requests(item, info))
|
||||
dlist = [self._enqueue(r, info) for r in requests]
|
||||
dlist = [self._process_request(r, info) for r in requests]
|
||||
dfd = DeferredList(dlist, consumeErrors=1)
|
||||
return dfd.addCallback(self.item_completed, item, info)
|
||||
|
||||
def _enqueue(self, request, info):
|
||||
def _process_request(self, request, info):
|
||||
fp = request_fingerprint(request)
|
||||
cb = request.callback or (lambda _: _)
|
||||
eb = request.errback
|
||||
|
||||
# if already downloaded, return cached result.
|
||||
# Return cached result if request was already seen
|
||||
if fp in info.downloaded:
|
||||
return defer_result(info.downloaded[fp]).addCallbacks(cb, eb)
|
||||
|
||||
# Otherwise, wait for result
|
||||
wad = Deferred().addCallbacks(cb, eb)
|
||||
# add to pending list for this request, and wait for result like the others.
|
||||
info.waiting.setdefault(fp, []).append(wad)
|
||||
info.waiting[fp].append(wad)
|
||||
|
||||
# if request is not downloading, download it.
|
||||
if fp not in info.downloading:
|
||||
self._download(request, info, fp)
|
||||
# Check if request is downloading right now to avoid doing it twice
|
||||
if fp in info.downloading:
|
||||
return wad
|
||||
|
||||
return wad
|
||||
|
||||
def _download(self, request, info, fp):
|
||||
def _downloaded(result):
|
||||
info.downloading.pop(fp)
|
||||
info.downloaded[fp] = result
|
||||
for wad in info.waiting.pop(fp): # pass result to each waiting client
|
||||
defer_result(result).chainDeferred(wad)
|
||||
|
||||
def _post_media_to_download(result):
|
||||
if result is None: # continue with download
|
||||
dwld = mustbe_deferred(self.download, request, info)
|
||||
dwld.addCallbacks(
|
||||
callback=self.media_downloaded,
|
||||
callbackArgs=(request, info),
|
||||
errback=self.media_failed,
|
||||
errbackArgs=(request, info))
|
||||
else: # or use media_to_download return value as result
|
||||
dwld = defer_result(result)
|
||||
|
||||
info.downloading[fp] = (request, dwld) # fill downloading state data
|
||||
dwld.addBoth(_downloaded) # append post-download hook
|
||||
dwld.addErrback(log.err, spider=info.spider)
|
||||
|
||||
# declare request in downloading state (None is used as place holder)
|
||||
info.downloading[fp] = None
|
||||
|
||||
# defer pre-download request processing
|
||||
# Download request checking media_to_download hook output first
|
||||
info.downloading.add(fp)
|
||||
dfd = mustbe_deferred(self.media_to_download, request, info)
|
||||
dfd.addCallback(_post_media_to_download)
|
||||
dfd.addCallback(self._check_media_to_download, request, info)
|
||||
dfd.addBoth(self._cache_result_and_execute_waiters, fp, info)
|
||||
dfd.addErrback(log.err, spider=info.spider)
|
||||
return dfd.addBoth(lambda _: wad) # it must return wad at last
|
||||
|
||||
def _check_media_to_download(self, result, request, info):
|
||||
if result is not None:
|
||||
return result
|
||||
# Download request and process its response
|
||||
return mustbe_deferred(self.download, request, info).addCallbacks(
|
||||
callback=self.media_downloaded, callbackArgs=(request, info),
|
||||
errback=self.media_failed, errbackArgs=(request, info))
|
||||
|
||||
def _cache_result_and_execute_waiters(self, result, fp, info):
|
||||
info.downloading.remove(fp)
|
||||
info.downloaded[fp] = result # cache result
|
||||
for wad in info.waiting.pop(fp):
|
||||
defer_result(result).chainDeferred(wad)
|
||||
|
||||
### Overradiable Interface
|
||||
def download(self, request, info):
|
||||
"""Defines how to download the media request"""
|
||||
request.priority = self.DOWNLOAD_PRIORITY
|
||||
return self.crawler.engine.download(request, info.spider)
|
||||
return self._download_func(request, info.spider)
|
||||
|
||||
def media_to_download(self, request, info):
|
||||
"""Check request before starting download"""
|
||||
|
|
@ -109,8 +102,8 @@ class MediaPipeline(object):
|
|||
def item_completed(self, results, item, info):
|
||||
"""Called per item when all media requests has been processed"""
|
||||
if self.LOG_FAILED_RESULTS:
|
||||
for success, result in results:
|
||||
if not success:
|
||||
log.err(result, '%s found errors proessing %s' % (self.__class__.__name__, item))
|
||||
msg = '%s found errors proessing %s' % (self.__class__.__name__, item)
|
||||
for ok, value in results:
|
||||
if not ok:
|
||||
log.err(value, msg, spider=info.spider)
|
||||
return item
|
||||
|
||||
|
|
|
|||
|
|
@ -1,85 +1,252 @@
|
|||
from twisted.trial import unittest
|
||||
from twisted.python import failure
|
||||
from twisted.internet import defer, reactor
|
||||
from twisted.python.failure import Failure
|
||||
from twisted.internet import reactor
|
||||
from twisted.internet.defer import Deferred, inlineCallbacks
|
||||
from twisted.python import log as txlog
|
||||
|
||||
from scrapy.settings import Settings
|
||||
from scrapy.crawler import Crawler
|
||||
from scrapy.http import Request, Response
|
||||
from scrapy.spider import BaseSpider
|
||||
from scrapy.utils.request import request_fingerprint
|
||||
from scrapy.contrib.pipeline.media import MediaPipeline
|
||||
from scrapy.utils.test import get_crawler
|
||||
from scrapy import log
|
||||
|
||||
|
||||
class _MockedMediaPipeline(MediaPipeline):
|
||||
|
||||
def download(self, request, info):
|
||||
delay = request.meta.get('delay')
|
||||
response = request.meta.get('response')
|
||||
if delay is None:
|
||||
return response
|
||||
else:
|
||||
dfd = defer.Deferred()
|
||||
reactor.callLater(delay, dfd.callback, None)
|
||||
return dfd.addCallback(lambda _: response)
|
||||
|
||||
def get_media_requests(self, item, info):
|
||||
return item.get('requests')
|
||||
def _mocked_download_func(request, info):
|
||||
response = request.meta.get('response')
|
||||
return response() if callable(response) else response
|
||||
|
||||
|
||||
class MediaPipelineTestCase(unittest.TestCase):
|
||||
class BaseMediaPipelineTestCase(unittest.TestCase):
|
||||
|
||||
pipeline_class = _MockedMediaPipeline
|
||||
pipeline_class = MediaPipeline
|
||||
|
||||
def setUp(self):
|
||||
self.crawler = Crawler(Settings())
|
||||
self.crawler.install()
|
||||
self.spider = BaseSpider('media.com')
|
||||
self.pipe = self.pipeline_class()
|
||||
self.pipe = self.pipeline_class(download_func=_mocked_download_func)
|
||||
self.pipe.open_spider(self.spider)
|
||||
self.info = self.pipe.spiderinfo[self.spider]
|
||||
|
||||
def tearDown(self):
|
||||
self.pipe.close_spider(self.spider)
|
||||
self.crawler.uninstall()
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def test_return_item_by_default(self):
|
||||
item = dict(name='sofa')
|
||||
def test_default_media_to_download(self):
|
||||
request = Request('url')
|
||||
assert self.pipe.media_to_download(request, self.info) is None
|
||||
|
||||
def test_default_get_media_requests(self):
|
||||
item = dict(name='name')
|
||||
assert self.pipe.get_media_requests(item, self.info) is None
|
||||
|
||||
def test_default_media_downloaded(self):
|
||||
request = Request('url')
|
||||
response = Response('url', body='')
|
||||
assert self.pipe.media_downloaded(response, request, self.info) is response
|
||||
|
||||
def test_default_media_failed(self):
|
||||
request = Request('url')
|
||||
fail = Failure(Exception())
|
||||
assert self.pipe.media_failed(fail, request, self.info) is fail
|
||||
|
||||
def test_default_item_completed(self):
|
||||
item = dict(name='name')
|
||||
assert self.pipe.item_completed([], item, self.info) is item
|
||||
|
||||
# Check that failures are logged by default
|
||||
fail = Failure(Exception())
|
||||
results = [(True, 1), (False, fail)]
|
||||
|
||||
events = []
|
||||
txlog.addObserver(events.append)
|
||||
new_item = self.pipe.item_completed(results, item, self.info)
|
||||
txlog.removeObserver(events.append)
|
||||
self.flushLoggedErrors()
|
||||
|
||||
assert new_item is item
|
||||
assert len(events) == 1
|
||||
assert events[0]['logLevel'] == log.ERROR
|
||||
assert events[0]['failure'] is fail
|
||||
|
||||
# disable failure logging and check again
|
||||
self.pipe.LOG_FAILED_RESULTS = False
|
||||
events = []
|
||||
txlog.addObserver(events.append)
|
||||
new_item = self.pipe.item_completed(results, item, self.info)
|
||||
txlog.removeObserver(events.append)
|
||||
self.flushLoggedErrors()
|
||||
assert new_item is item
|
||||
assert len(events) == 0
|
||||
|
||||
@inlineCallbacks
|
||||
def test_default_process_item(self):
|
||||
item = dict(name='name')
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
assert new_item is item
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def test_default_download_func(self):
|
||||
crawler = get_crawler()
|
||||
crawler.install()
|
||||
crawler.configure()
|
||||
try:
|
||||
pipe = MediaPipeline()
|
||||
assert pipe._download_func == crawler.engine.download
|
||||
finally:
|
||||
crawler.uninstall()
|
||||
|
||||
|
||||
class MockedMediaPipeline(MediaPipeline):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(MockedMediaPipeline, self).__init__(*args, **kwargs)
|
||||
self._mockcalled = []
|
||||
|
||||
def download(self, request, info):
|
||||
self._mockcalled.append('download')
|
||||
return super(MockedMediaPipeline, self).download(request, info)
|
||||
|
||||
def media_to_download(self, request, info):
|
||||
self._mockcalled.append('media_to_download')
|
||||
return super(MockedMediaPipeline, self).media_to_download(request, info)
|
||||
|
||||
def get_media_requests(self, item, info):
|
||||
self._mockcalled.append('get_media_requests')
|
||||
return item.get('requests')
|
||||
|
||||
def media_downloaded(self, response, request, info):
|
||||
self._mockcalled.append('media_downloaded')
|
||||
return super(MockedMediaPipeline, self).media_downloaded(response, request, info)
|
||||
|
||||
def media_failed(self, failure, request, info):
|
||||
self._mockcalled.append('media_failed')
|
||||
return super(MockedMediaPipeline, self).media_failed(failure, request, info)
|
||||
|
||||
def item_completed(self, results, item, info):
|
||||
self._mockcalled.append('item_completed')
|
||||
item = super(MockedMediaPipeline, self).item_completed(results, item, info)
|
||||
item['results'] = results
|
||||
return item
|
||||
|
||||
|
||||
class MediaPipelineTestCase(BaseMediaPipelineTestCase):
|
||||
|
||||
pipeline_class = MockedMediaPipeline
|
||||
|
||||
@inlineCallbacks
|
||||
def test_result_succeed(self):
|
||||
cb = lambda _: self.pipe._mockcalled.append('request_callback') or _
|
||||
eb = lambda _: self.pipe._mockcalled.append('request_errback') or _
|
||||
rsp = Response('url1')
|
||||
req = Request('url1', meta=dict(response=rsp), callback=cb, errback=eb)
|
||||
item = dict(requests=req)
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertEqual(new_item['results'], [(True, rsp)])
|
||||
self.assertEqual(self.pipe._mockcalled,
|
||||
['get_media_requests', 'media_to_download', 'download',
|
||||
'media_downloaded', 'request_callback', 'item_completed'])
|
||||
|
||||
@inlineCallbacks
|
||||
def test_result_failure(self):
|
||||
self.pipe.LOG_FAILED_RESULTS = False
|
||||
cb = lambda _: self.pipe._mockcalled.append('request_callback') or _
|
||||
eb = lambda _: self.pipe._mockcalled.append('request_errback') or _
|
||||
fail = Failure(Exception())
|
||||
req = Request('url1', meta=dict(response=fail), callback=cb, errback=eb)
|
||||
item = dict(requests=req)
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertEqual(new_item['results'], [(False, fail)])
|
||||
self.assertEqual(self.pipe._mockcalled,
|
||||
['get_media_requests', 'media_to_download', 'download',
|
||||
'media_failed', 'request_errback', 'item_completed'])
|
||||
|
||||
@inlineCallbacks
|
||||
def test_mix_of_success_and_failure(self):
|
||||
self.pipe.LOG_FAILED_RESULTS = False
|
||||
rsp1 = Response('url1')
|
||||
req1 = Request('url1', meta=dict(response=rsp1))
|
||||
fail = Failure(Exception())
|
||||
req2 = Request('url2', meta=dict(response=fail))
|
||||
item = dict(requests=[req1, req2])
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertEqual(new_item['results'], [(True, rsp1), (False, fail)])
|
||||
m = self.pipe._mockcalled
|
||||
# only once
|
||||
self.assertEqual(m[0], 'get_media_requests') # first hook called
|
||||
self.assertEqual(m.count('get_media_requests'), 1)
|
||||
self.assertEqual(m.count('item_completed'), 1)
|
||||
self.assertEqual(m[-1], 'item_completed') # last hook called
|
||||
# twice, one per request
|
||||
self.assertEqual(m.count('media_to_download'), 2)
|
||||
self.assertEqual(m.count('download'), 2)
|
||||
# one to handle success and other for failure
|
||||
self.assertEqual(m.count('media_downloaded'), 1)
|
||||
self.assertEqual(m.count('media_failed'), 1)
|
||||
|
||||
@inlineCallbacks
|
||||
def test_get_media_requests(self):
|
||||
# returns single Request (without callback)
|
||||
info = self.pipe.spiderinfo[self.spider]
|
||||
req = Request('http://media.com/2.gif')
|
||||
req = Request('url')
|
||||
item = dict(requests=req) # pass a single item
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
assert new_item is item
|
||||
assert request_fingerprint(req) in info.downloaded
|
||||
assert request_fingerprint(req) in self.info.downloaded
|
||||
|
||||
# returns iterable of Requests
|
||||
req1 = Request('http://media.com/1.gif')
|
||||
req2 = Request('http://media.com/1.jpg')
|
||||
req1 = Request('url1')
|
||||
req2 = Request('url2')
|
||||
item = dict(requests=iter([req1, req2]))
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
assert new_item is item
|
||||
assert info.downloaded.get(request_fingerprint(req1)) is None
|
||||
assert info.downloaded.get(request_fingerprint(req2)) is None
|
||||
assert request_fingerprint(req1) in self.info.downloaded
|
||||
assert request_fingerprint(req2) in self.info.downloaded
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def test_requets_callback_is_called(self):
|
||||
collected = []
|
||||
response = Response('http://media.com/2.gif')
|
||||
request = Request('http://media.com/2.gif', meta=dict(response=response), callback=collected.append)
|
||||
item = dict(requests=request) # pass a single item
|
||||
yield self.pipe.process_item(item, self.spider)
|
||||
assert collected == [response]
|
||||
@inlineCallbacks
|
||||
def test_results_are_cached_across_multiple_items(self):
|
||||
rsp1 = Response('url1')
|
||||
req1 = Request('url1', meta=dict(response=rsp1))
|
||||
item = dict(requests=req1)
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertTrue(new_item is item)
|
||||
self.assertEqual(new_item['results'], [(True, rsp1)])
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def test_requets_errback_is_called(self):
|
||||
collected = []
|
||||
fail = failure.Failure(Exception())
|
||||
req = Request('http://media.com/2.gif', meta=dict(response=fail), callback=lambda _:_, errback=collected.append)
|
||||
item = dict(requests=req) # pass a single item
|
||||
yield self.pipe.process_item(item, self.spider)
|
||||
assert collected == [fail]
|
||||
# rsp2 is ignored, rsp1 must be in results because request fingerprints are the same
|
||||
req2 = Request(req1.url, meta=dict(response=Response('http://donot.download.me')))
|
||||
item = dict(requests=req2)
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertTrue(new_item is item)
|
||||
self.assertEqual(request_fingerprint(req1), request_fingerprint(req2))
|
||||
self.assertEqual(new_item['results'], [(True, rsp1)])
|
||||
|
||||
@inlineCallbacks
|
||||
def test_results_are_cached_for_requests_of_single_item(self):
|
||||
rsp1 = Response('url1')
|
||||
req1 = Request('url1', meta=dict(response=rsp1))
|
||||
req2 = Request(req1.url, meta=dict(response=Response('http://donot.download.me')))
|
||||
item = dict(requests=[req1, req2])
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertTrue(new_item is item)
|
||||
self.assertEqual(new_item['results'], [(True, rsp1), (True, rsp1)])
|
||||
|
||||
@inlineCallbacks
|
||||
def test_wait_if_request_is_downloading(self):
|
||||
def _check_downloading(response):
|
||||
fp = request_fingerprint(req1)
|
||||
assert fp in self.info.downloading, self.info.downloading
|
||||
assert fp in self.info.waiting, self.info.waiting
|
||||
assert len(self.info.waiting[fp]) == 2, self.info.waiting
|
||||
assert fp not in self.info.downloaded, self.info.downloaded
|
||||
return response
|
||||
|
||||
rsp1 = Response('url')
|
||||
def rsp1_func():
|
||||
dfd = Deferred().addCallback(_check_downloading)
|
||||
reactor.callLater(.1, dfd.callback, rsp1)
|
||||
return dfd
|
||||
|
||||
def rsp2_func():
|
||||
assert False, 'This can not be called'
|
||||
|
||||
req1 = Request('url', meta=dict(response=rsp1_func))
|
||||
req2 = Request(req1.url, meta=dict(response=rsp2_func))
|
||||
item = dict(requests=[req1, req2])
|
||||
new_item = yield self.pipe.process_item(item, self.spider)
|
||||
self.assertEqual(new_item['results'], [(True, rsp1), (True, rsp1)])
|
||||
|
|
|
|||
Loading…
Reference in New Issue