Scrapy单元测试

Question

我想在Scrapy（屏幕抓取/网络爬虫）中实现一些单元测试。由于一个项目是通过“scrapy crawl”命令运行的，我可以通过鼻子之类的东西运行它。由于scrapy建立在扭曲的顶端，我可以使用它的单元测试框架Trial？如果是这样，怎么样？否则，我想获得鼻子工作。

更新：

我一直在谈论关于Scrapy-Users，我想我应该“建立测试码的响应，然后调用与响应的方法和断言[I]得到输出中的预期项目/请求“。我似乎无法得到这个工作，虽然。

我可以建立一个单元测试测试类和测试：

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

但它最终会产生this回溯。任何洞察力为什么？

Answer 1

我这样做的方式是创建假响应，这种方式可以离线测试解析函数。但是你通过使用真实的HTML来获得真实的情况。

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

我目前的工作流程是，无论何时出现错误，我都会发送一封电子邮件给管理员，并附有网址。然后，对于那个特定的错误，我创建了一个html文件，其中包含导致错误的内容。然后我为它创建一个单元测试。

这是我使用从一个本地HTML文件创建用于测试样品Scrapy HTTP响应代码：

# 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 /测试/响应/ osdir/sample.html

然后测试用例可以看看如下： 测试用例位置是scrapyproject /测试/ 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) 

这是BASICA lly我如何测试我的解析方法，但它不仅用于解析方法。如果它变得更复杂，我建议看看Mox

Answer 2

您可以按照scrapy网站上的this代码片段从脚本中运行它。然后，您可以对返回的项目进行任何形式的声明。

Answer 3

新增加的Spider Contracts值得一试。它给你一个简单的方法来添加测试，而不需要太多的代码。

Answer 4

我用Betamax在真正的现场运行测试的第一时间，保持本地的HTTP响应以便下次测试运行超快后：

Betamax intercepts every request you make and attempts to find a matching request that has already been intercepted and recorded.

当你需要获得最新版本的网站，只是删除了什么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() 

仅供参考，我发现在Betamax的2015年PYCON由于Ian Cordasco's talk。

Answer 5

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

从check和parse 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) 

Answer 6

我使用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)  

非常感谢。

Answer 7

稍微简单，由所选择的答案取出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()