スケジューリングパイプラインの統合テストが遅延返す

Question

スクレイピーパイプラインの統合テストを作成できますか？私はこれを行う方法を理解することはできません。特に、私は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を使用して、適切なテストを作成するのですか？

Answer 1

今ではfrom_crawlerを呼び出さずにパイプラインテストを行ったので、それらはfrom_crawlerの機能をテストしないため理想的ではありませんが、機能します。

私は空Spiderインスタンスを使用してそれらを実行します。

from scrapy.spiders import Spider 
# some other imports for my own stuff and standard libs 

@pytest.fixture 
def mqtt_client(): 
    client = mock.Mock() 

    return client 

def test_mqtt_pipeline_does_return_item_after_process(mqtt_client): 
    spider = Spider(name='spider') 
    pipeline = MqttOutputPipeline(mqtt_client, 'dummy-namespace') 

    item = BasicItem() 
    item['url'] = 'http://example.com/' 
    item['source'] = 'dummy source' 

    ret = pipeline.process_item(item, spider) 

    assert ret is not None 

（実際には、私はopen_spider()に電話をするのを忘れた）

ます。またscrapy自体は、パイプラインのテストを行う方法を見て持つことができます、e.g. for MediaPipeline：

class BaseMediaPipelineTestCase(unittest.TestCase): 

    pipeline_class = MediaPipeline 
    settings = None 

    def setUp(self): 
     self.spider = Spider('media.com') 
     self.pipe = self.pipeline_class(download_func=_mocked_download_func, 
            settings=Settings(self.settings)) 
     self.pipe.open_spider(self.spider) 
     self.info = self.pipe.spiderinfo 

    def test_default_media_to_download(self): 
     request = Request('http://url') 
     assert self.pipe.media_to_download(request, self.info) is None 

また、彼らの他のユニットテストを通して見ることができます。私にとっては、これらは常に、テストスクラピーコンポーネントをどのようにユニット化するかについての良いインスピレーションです。

from_crawlerの機能もテストしたい場合は、Middlewareテストをご覧ください。これらのテストでは、しばしばfrom_crawlerを使用してミドルウェアを作成します。e.g. for OffsiteMiddleware。

from scrapy.spiders import Spider 
from scrapy.utils.test import get_crawler 

class TestOffsiteMiddleware(TestCase): 

    def setUp(self): 
     crawler = get_crawler(Spider) 
     self.spider = crawler._create_spider(**self._get_spiderargs()) 
     self.mw = OffsiteMiddleware.from_crawler(crawler) 
     self.mw.spider_opened(self.spider) 

私はここで重要な要素は、scrapy.utils.testからget_crawlerを呼び出すことであると仮定します。彼らはテスト環境を持つためにあなたがする必要があるいくつかの呼び出しを分解したようだ。