70

我想在 Scrapy(屏幕抓取器/网络爬虫)中实现一些单元测试。由于一个项目是通过“scrapy crawl”命令运行的,所以我可以通过鼻子之类的东西来运行它。由于scrapy是建立在twisted之上的,我可以使用它的单元测试框架Trial吗?如果是这样,怎么做?否则我想让鼻子工作。

更新:

我一直在谈论Scrapy-Users,我想我应该“在测试代码中构建响应,然后使用响应调用方法并断言 [I] 在输出中获得预期的项目/请求”。我似乎无法让它工作。

我可以构建一个单元测试测试类并在测试中:

  • 创建响应对象
  • 尝试使用响应对象调用我的蜘蛛的解析方法

但是,它最终会生成回溯。关于为什么的任何见解?

4

10 回答 10

77

我这样做的方式是创建虚假响应,这样您就可以离线测试解析功能。但是你可以通过使用真正的 HTML 来获得真实的情况。

这种方法的一个问题是您的本地 HTML 文件可能无法反映在线的最新状态。因此,如果 HTML 在线更改,您可能会遇到很大的错误,但您的测试用例仍然会通过。因此,以这种方式进行测试可能不是最好的方法。

我目前的工作流程是,每当出现错误时,我都会向管理员发送一封带有 url 的电子邮件。然后对于那个特定的错误,我创建一个包含导致错误的内容的 html 文件。然后我为它创建一个单元测试。

这是我用来创建示例 Scrapy http 响应以从本地 html 文件进行测试的代码:

# 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

示例 html 文件位于 scrapyproject/tests/responses/osdir/sample.html

那么测试用例可能如下所示: 测试用例位置是 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)

这基本上就是我测试我的解析方法的方式,但它不仅适用于解析方法。如果它变得更复杂,我建议看看Mox

于 2012-10-05T06:51:32.237 回答
26

我第一次使用Betamax在真实站点上运行测试并将 http 响应保留在本地,以便下一个测试在以下情况下运行得非常快:

Betamax 拦截您发出的每个请求,并尝试找到已被拦截和记录的匹配请求。

当您需要获取最新版本的站点时,只需删除 betamax 记录的内容并重新运行测试即可。

例子:

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

仅供参考,感谢Ian Cordasco 的演讲,我在 pycon 2015 上发现了 betamax 。

于 2016-07-05T23:00:22.137 回答
23

新加入的蜘蛛契约值得一试。它为您提供了一种无需大量代码即可添加测试的简单方法。

于 2012-10-05T18:01:30.087 回答
9

这是一个很晚的答案,但我一直对scrapy测试感到恼火,所以我编写了scrapy-test一个框架,用于根据定义的规范测试scrapy爬虫。

它通过定义测试规范而不是静态输出来工作。例如,如果我们正在抓取此类项目:

{
    "name": "Alex",
    "age": 21,
    "gender": "Female",
}

我们可以定义 scrapy-test ItemSpec

from scrapytest.tests import Match, MoreThan, LessThan
from scrapytest.spec import ItemSpec

class MySpec(ItemSpec):
    name_test = Match('{3,}')  # name should be at least 3 characters long
    age_test = Type(int), MoreThan(18), LessThan(99)
    gender_test = Match('Female|Male')

对于scrapy stats也有同样的想法测试StatsSpec

from scrapytest.spec import StatsSpec
from scrapytest.tests import Morethan

class MyStatsSpec(StatsSpec):
    validate = {
        "item_scraped_count": MoreThan(0),
    }

之后,它可以针对实时或缓存结果运行:

$ scrapy-test 
# or
$ scrapy-test --cache

我一直在为开发更改运行缓存运行,并为检测网站更改运行每日 cronjobs。

于 2019-02-25T16:53:08.567 回答
4

我正在使用 Twistedtrial运行测试,类似于 Scrapy 自己的测试。它已经启动了一个反应器,所以我使用了CrawlerRunner而不用担心在测试中启动和停止一个反应器。

check我从 Scrapy 命令和Scrapy 命令中窃取了一些想法,parse最终得到了以下基TestCase类来针对实时站点运行断言:

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

例子:

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

或在设置中执行一个请求并针对结果运行多个测试:

@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)
于 2016-10-13T12:06:48.430 回答
4

稍微简单一点,通过def fake_response_from_file从选择的答案中删除:

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()
于 2018-02-08T23:57:54.167 回答
3

我正在使用scrapy 1.3.0和函数:fake_response_from_file,引发错误:

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

我得到:

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

解决方案是改用 TextResponse,它可以正常工作,例如:

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

非常感谢。

于 2017-01-02T15:54:51.900 回答
2

您可以按照scrapy 站点中的代码段从脚本中运行它。然后,您可以对退回的物品进行任何类型的断言。

于 2011-06-27T14:16:12.667 回答
1

https://github.com/ThomasAitken/Scrapy-Testmaster

这是我编写的一个包,它显着扩展了 Scrapy Autounit 库的功能并将其带向不同的方向(允许轻松地动态更新测试用例并合并调试/测试用例生成的过程)。它还包括 Scrapyparse命令的修改版本(https://docs.scrapy.org/en/latest/topics/commands.html#std-command-parse

于 2020-05-22T10:39:49.003 回答
1

类似于Hadrien 的答案,但对于 pytest: pytest-vcr

import requests
import pytest
from scrapy.http import HtmlResponse

@pytest.mark.vcr()
def test_parse(url, target):
    response = requests.get(url)
    scrapy_response = HtmlResponse(url, body=response.content)
    assert Spider().parse(scrapy_response) == target

于 2020-11-05T20:27:23.233 回答