スクレイピーパイプラインの統合テストを作成できますか?私はこれを行う方法を理解することはできません。特に、私はFilesPipelineのテストを書こうとしていますが、Amazon S3に私の嘲笑されたレスポンスを永続させたいと思っています。ここでスケジューリングパイプラインの統合テストが遅延返す
は私のテストで:
def _mocked_download_func(request, info):
return Response(url=response.url, status=200, body="test", request=request)
class FilesPipelineTests(unittest.TestCase):
def setUp(self):
self.settings = get_project_settings()
crawler = Crawler(self.settings)
crawler.configure()
self.pipeline = FilesPipeline.from_crawler(crawler)
self.pipeline.open_spider(None)
self.pipeline.download_func = _mocked_download_func
@defer.inlineCallbacks
def test_file_should_be_directly_available_from_s3_when_processed(self):
item = CrawlResult()
item['id'] = "test"
item['file_urls'] = ['http://localhost/test']
result = yield self.pipeline.process_item(item, None)
self.assertEquals(result['files'][0]['path'], "full/002338a87aab86c6b37ffa22100504ad1262f21b")
私はいつも次のようなエラーに遭遇:
DirtyReactorAggregateError: Reactor was unclean.
は、どのように私はツイストとscrapyを使用して、適切なテストを作成するのですか?