Scrapy 单元测试

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

更新:

我一直在谈论 Scrapy-用户,我想我应该“在测试代码中构建响应,然后用响应调用方法,并断言[我]在输出中获得预期的项/请求”。不过我好像没法让这个起作用。

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

  • 创建响应对象
  • 尝试使用 response 对象调用爬行器的解析方法

但是它最终产生了 这个回溯。知道为什么吗?

28796 次浏览

您可以按照 Scrapy 站点的 这个片段从脚本运行它。然后您可以对返回的项目进行任何类型的断言。

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

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

我目前的工作流程是,每当有一个错误,我会发送电子邮件到管理,与网址。然后,对于这个特定的错误,我创建了一个带有导致错误的内容的 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/test/response/osdir/sample.html 中

那么测试用例可以看起来如下: 测试用例位置是 scrapyproject/test/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)

这基本上就是我测试解析方法的方式,但这不仅仅是为了解析方法。如果它变得更复杂,我建议看看 莫克斯

新增加的 蜘蛛合同值得一试。它为您提供了一种添加测试的简单方法,而不需要大量代码。

我第一次使用 录像带在实际站点上运行测试,并在本地保持 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()

顺便说一句,多亏了 伊恩 · 科达斯科的演讲我在2015年的 Pycon 上发现了 Betamax。

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

checkparse Scrapy 命令中窃取了一些想法,我最终使用以下基类 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)

我正在使用 scrapy 1.3.0和函数: false _ 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)

非常感谢。

简单一点,从选择的答案中去掉 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()

这是一个非常晚的答案,但是我已经厌烦了 Scrapy 测试,所以我编写了 刮擦试验框架来测试 Scrapy 爬行器对定义的规范。

它通过定义测试规范而不是静态输出来工作。 例如,如果我们正在爬行这种类型的项目:

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

我们可以定义抓挠测试 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')

还有一些与 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

我一直在运行缓存运行的开发更改和检测网站更改的日常 cronjob。

Https://github.com/thomasaitken/scrapy-testmaster

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

类似于 哈德良的回答,但是对于 pytest: (英文)

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