Scrapyunit testing

我想在Scrapy中执行一些unit testing(screen scraper / web crawler)。 由于一个项目是通过“scrapy crawl”命令运行的,我可以通过像鼻子这样的东西运行它。 由于scrapy是在扭曲的基础上构build的,我可以使用它的unit testing框架Trial? 如果是这样,怎么样? 否则,我想鼻子工作。

更新:

我一直在讨论Scrapy-Users ,我想我应该“在testing代码中构build响应,然后调用带有响应的方法,并断言[我]在输出中获得预期的项目/请求。 我似乎无法得到这个工作。

我可以build立一个unit testingtesting课程,并在testing中:

  • 创build一个响应对象
  • 尝试用响应对象调用我的蜘蛛的parsing方法

但是它最终会产生这个回溯。 任何洞察力为什么?

我这样做的方式是创build假响应,这样你可以离线testingparsing函数。 但是,通过使用真实的HTML来获得真实的情况。

这种方法的一个问题是您的本地HTML文件可能不会反映最新的状态联机。 所以,如果HTML在线改变,你可能会有一个大的错误,但是你的testing用例仍然会通过。 所以这可能不是testing这种方式的最好方法。

我目前的工作stream程是,每当有错误,我会发送一封电子邮件给pipe理员,与url。 然后,对于那个特定的错误,我创build一个HTML文件,导致错误的内容。 然后我为它创build一个unit testing。

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

# 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中

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

这基本上是我如何testing我的parsing方法,但不仅仅是parsing方法。 如果变得更复杂,我build议看看Mox

新增加的蜘蛛合同是值得尝试的。 它给你一个简单的方法来添加testing,而不需要太多的代码。

我使用Betamax首次在真实网站上运行testing,并在本地保留http响应,以便下次testing运行超快后:

Betamax拦截你所做的每一个请求,并试图find一个已被拦截和logging的匹配请求。

当您需要获取最新版本的网站时,只需删除betamaxlogging的内容并重新运行testing。

例:

 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 。

您可以从scrapy网站上查看这个片段,从脚本中运行它。 然后,您可以对返回的项目进行任何forms的声明。

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

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

我得到:

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

解决scheme是使用TextResponse,而且它工作正常,例如:

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

非常感谢。

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

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) 

或者在设置中执行一个请求,并对结果运行多个testing:

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