Integrationtest di pipeline scrapy restituita differita

Question

È possibile creare un test di integrazione di una pipeline di scrapy? Non riesco a capire come farlo. In particolare sto cercando di scrivere un test per FilesPipeline e voglio anche che mantenga la mia risposta derisoria ad Amazon S3.

Ecco la mia prova:

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") 

ho sempre incontrato il seguente errore:

DirtyReactorAggregateError: Reactor was unclean. 

Come posso creare un test adeguato utilizzando Scrapy contorto e?

Answer 1

Su ora ho fatto i miei test di pipeline senza la chiamata a from_crawler, quindi non sono l'ideale, perché non testano la funzionalità di from_crawler, ma funzionano.

li fanno utilizzando un vuoto Spider esempio:

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 

(in realtà, ho dimenticato di chiamare open_spider())

Si può anche avere uno sguardo a come Scrapy si fa il test delle condotte , 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 

si può anche avere uno sguardo attraverso i loro altri unit test. Per me, queste sono sempre buone ispirazioni su come testare i componenti di scrapy.

Se si desidera testare anche la funzione from_crawler, è possibile dare un'occhiata ai test Middleware. In questi test, spesso usano from_crawler per creare i middleware, 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) 

Suppongo che il componente chiave qui è quello di chiamare get_crawler da scrapy.utils.test. Sembra che abbiano preso in considerazione alcune chiamate che devi fare per avere un ambiente di test.