python-2.7 - library - scrapy ruby

Prueba de integración de tubería de retorno de diferido

Question

¿Es posible crear una prueba de integración de un pipeline de scrapy? No puedo entender cómo hacer esto. En particular, estoy tratando de escribir una prueba para FilesPipeline y también quiero que persista mi respuesta burlada a Amazon S3.

Aquí está mi prueba:

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

Siempre me tropiezo con el siguiente error:

DirtyReactorAggregateError: Reactor was unclean.

¿Cómo creo una prueba adecuada usando twisted y scrapy?

Answer 1

Up do ahora hice mis pruebas de pipeline sin la llamada a from_crawler , por lo que no son ideales, porque no prueban la funcionalidad de from_crawler , pero funcionan.

Los hago usando una instancia vacía de 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

(De hecho, me olvidé de llamar a open_spider() )

También puede ver cómo el propio filtro hace las pruebas de las tuberías, por ejemplo , para 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

También puede echar un vistazo a sus otras pruebas unitarias. Para mí, estos siempre son una buena inspiración sobre cómo probar los componentes de la prueba de Scrapy.

Si también quieres probar la función from_crawler , puedes echar un vistazo a sus pruebas de Middleware . En estas pruebas, a menudo usan from_crawler para crear middlewares, por ejemplo , para 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)

Supongo que el componente clave aquí es llamar a get_crawler desde scrapy.utils.test . Parece que tomaron en cuenta algunas llamadas que debes hacer para tener un entorno de prueba.