python - Prueba de Unidad de Scrapy

unit-testing nose

Question

Me gustaría implementar algunas pruebas de unidad en un Scrapy (raspador de pantalla / rastreador web). Como un proyecto se ejecuta a través del comando "scrapy crawl", puedo ejecutarlo a través de algo como nose. Dado que el scrapy está construido sobre la parte superior del twisted, ¿puedo usar el marco de pruebas de la unidad Trial? ¿Si es así, cómo? De lo contrario, me gustaría ponerme a trabajar.

Actualizar:

He estado hablando de Scrapy-Users y creo que se supone que debo "construir la Respuesta en el código de prueba, y luego llamar al método con la respuesta y afirmar que [I] obtengo los elementos / solicitudes esperados en la salida". Parece que no puedo hacer que esto funcione.

Puedo construir una clase de prueba de unidad de prueba y en una prueba:

• crear un objeto de respuesta
• tratar de llamar al método parse de mi araña con el objeto de respuesta

Sin embargo, termina generando this rastreo. ¿Alguna idea de por qué?

Answer 1

Estoy usando la trial de Twisted para ejecutar pruebas, similar a las pruebas de Scrapy. Ya inicia un reactor, por lo que utilizo CrawlerRunner sin preocuparme por iniciar y detener uno en las pruebas.

Robando algunas ideas del check y parse comandos de Scrapy, terminé con la siguiente clase base de TestCase para ejecutar las afirmaciones contra los sitios en vivo:

from twisted.trial import unittest

from scrapy.crawler import CrawlerRunner
from scrapy.http import Request
from scrapy.item import BaseItem
from scrapy.utils.spider import iterate_spider_output

class SpiderTestCase(unittest.TestCase):
    def setUp(self):
        self.runner = CrawlerRunner()

    def make_test_class(self, cls, url):
        """
        Make a class that proxies to the original class,
        sets up a URL to be called, and gathers the items
        and requests returned by the parse function.
        """
        class TestSpider(cls):
            # This is a once used class, so writing into
            # the class variables is fine. The framework
            # will instantiate it, not us.
            items = []
            requests = []

            def start_requests(self):
                req = super(TestSpider, self).make_requests_from_url(url)
                req.meta["_callback"] = req.callback or self.parse
                req.callback = self.collect_output
                yield req

            def collect_output(self, response):
                try:
                    cb = response.request.meta["_callback"]
                    for x in iterate_spider_output(cb(response)):
                        if isinstance(x, (BaseItem, dict)):
                            self.items.append(x)
                        elif isinstance(x, Request):
                            self.requests.append(x)
                except Exception as ex:
                    print("ERROR", "Could not execute callback: ",     ex)
                    raise ex

                # Returning any requests here would make the     crawler follow them.
                return None

        return TestSpider

Ejemplo:

@defer.inlineCallbacks
def test_foo(self):
    tester = self.make_test_class(FooSpider, ''https://foo.com'')
    yield self.runner.crawl(tester)
    self.assertEqual(len(tester.items), 1)
    self.assertEqual(len(tester.requests), 2)

o realice una solicitud en la configuración y ejecute pruebas múltiples contra los resultados:

@defer.inlineCallbacks
def setUp(self):
    super(FooTestCase, self).setUp()
    if FooTestCase.tester is None:
        FooTestCase.tester = self.make_test_class(FooSpider, ''https://foo.com'')
        yield self.runner.crawl(self.tester)

def test_foo(self):
    self.assertEqual(len(self.tester.items), 1)

Answer 2

Estoy usando scrapy 1.3.0 y la función: fake_response_from_file, provoca un error:

response = Response(url=url, request=request, body=file_content)

Yo obtengo:

raise AttributeError("Response content isn''t text")

La solución es usar TextResponse en su lugar, y funciona bien, como ejemplo:

response = TextResponse(url=url, request=request, body=file_content)     

Muchas gracias.

Answer 3

La forma en que lo hice es crear respuestas falsas, de esta manera puede probar la función de análisis sin conexión. Pero obtienes la situación real usando HTML real.

Un problema con este enfoque es que su archivo HTML local puede no reflejar el último estado en línea. Entonces, si el HTML cambia en línea, es posible que tengas un gran error, pero tus casos de prueba aún pasarán. Entonces, puede que no sea la mejor manera de probarlo de esta manera.

Mi flujo de trabajo actual es que, siempre que haya un error, enviaré un correo electrónico a admin, con la url. Luego, para ese error específico, creo un archivo html con el contenido que está causando el error. Luego creo una prueba unitaria para ello.

Este es el código que uso para crear ejemplos de respuesta http de Scrapy para las pruebas de un archivo html local:

# scrapyproject/tests/responses/__init__.py

import os

from scrapy.http import Response, Request

def fake_response_from_file(file_name, url=None):
    """
    Create a Scrapy fake HTTP response from a HTML file
    @param file_name: The relative filename from the responses directory,
                      but absolute paths are also accepted.
    @param url: The URL of the response.
    returns: A scrapy HTTP response which can be used for unittesting.
    """
    if not url:
        url = ''http://www.example.com''

    request = Request(url=url)
    if not file_name[0] == ''/'':
        responses_dir = os.path.dirname(os.path.realpath(__file__))
        file_path = os.path.join(responses_dir, file_name)
    else:
        file_path = file_name

    file_content = open(file_path, ''r'').read()

    response = Response(url=url,
        request=request,
        body=file_content)
    response.encoding = ''utf-8''
    return response

El archivo html de muestra se encuentra en scrapyproject / tests / responses / osdir / sample.html

Entonces, el caso de prueba podría verse de la siguiente manera: La ubicación del caso de prueba es scrapyproject / tests / test_osdir.py

import unittest
from scrapyproject.spiders import osdir_spider
from responses import fake_response_from_file

class OsdirSpiderTest(unittest.TestCase):

    def setUp(self):
        self.spider = osdir_spider.DirectorySpider()

    def _test_item_results(self, results, expected_length):
        count = 0
        permalinks = set()
        for item in results:
            self.assertIsNotNone(item[''content''])
            self.assertIsNotNone(item[''title''])
        self.assertEqual(count, expected_length)

    def test_parse(self):
        results = self.spider.parse(fake_response_from_file(''osdir/sample.html''))
        self._test_item_results(results, 10)

Así es básicamente como pruebo mis métodos de análisis, pero no solo para los métodos de análisis. Si se vuelve más complejo, sugiero mirar Mox

Answer 4

Ligeramente más simple, eliminando la def fake_response_from_file de la respuesta elegida:

import unittest
from spiders.my_spider import MySpider
from scrapy.selector import Selector


class TestParsers(unittest.TestCase):


    def setUp(self):
        self.spider = MySpider(limit=1)
        self.html = Selector(text=open("some.htm", ''r'').read())


    def test_some_parse(self):
        expected = "some-text"
        result = self.spider.some_parse(self.html)
        self.assertEqual(result, expected)


if __name__ == ''__main__'':
    unittest.main()

Answer 5

Merece la pena intentar los nuevos Contratos Spider . Le brinda una manera simple de agregar pruebas sin requerir mucho código.

Answer 6

Puede seguir this fragmento del sitio de scrapy para ejecutarlo desde un script. Luego puede hacer cualquier tipo de afirmaciones que desee sobre los artículos devueltos.

Answer 7

Uso Betamax para ejecutar la prueba en el sitio real la primera vez y mantener las respuestas http localmente para que las próximas pruebas se ejecuten súper rápido después de:

Betamax intercepta cada solicitud que realiza e intenta encontrar una solicitud de coincidencia que ya ha sido interceptada y registrada.

Cuando necesite obtener la última versión del sitio, solo elimine lo que betamax ha registrado y vuelva a ejecutar la prueba.

Ejemplo:

from scrapy import Spider, Request
from scrapy.http import HtmlResponse


class Example(Spider):
    name = ''example''

    url = ''http://doc.scrapy.org/en/latest/_static/selectors-sample1.html''

    def start_requests(self):
        yield Request(self.url, self.parse)

    def parse(self, response):
        for href in response.xpath(''//a/@href'').extract():
            yield {''image_href'': href}


# Test part
from betamax import Betamax
from betamax.fixtures.unittest import BetamaxTestCase


with Betamax.configure() as config:
    # where betamax will store cassettes (http responses):
    config.cassette_library_dir = ''cassettes''
    config.preserve_exact_body_bytes = True


class TestExample(BetamaxTestCase):  # superclass provides self.session

    def test_parse(self):
        example = Example()

        # http response is recorded in a betamax cassette:
        response = self.session.get(example.url)

        # forge a scrapy response to test
        scrapy_response = HtmlResponse(body=response.content, url=example.url)

        result = example.parse(scrapy_response)

        self.assertEqual({''image_href'': u''image1.html''}, result.next())
        self.assertEqual({''image_href'': u''image2.html''}, result.next())
        self.assertEqual({''image_href'': u''image3.html''}, result.next())
        self.assertEqual({''image_href'': u''image4.html''}, result.next())
        self.assertEqual({''image_href'': u''image5.html''}, result.next())

        with self.assertRaises(StopIteration):
            result.next()

FYI, descubro betamax en pycon 2015 gracias a la charla de Ian Cordasco .