proper unittest-like output for scrapy contracts

This commit is contained in:
Alex Cepoi 2012-10-27 23:26:40 +02:00 committed by Daniel Graña
parent c9b690deb2
commit 86635e4afe
4 changed files with 77 additions and 42 deletions

View File

@ -1,11 +1,14 @@
from collections import defaultdict
from functools import wraps
from unittest.runner import TextTestRunner
from scrapy import signals
from scrapy.command import ScrapyCommand
from scrapy.contracts import ContractsManager
from scrapy.utils.misc import load_object
from scrapy.utils.spider import iterate_spider_output
from scrapy.utils.conf import build_component_list
from scrapy.xlib.pydispatch import dispatcher
def _generate(cb):
@ -39,6 +42,7 @@ class Command(ScrapyCommand):
self.settings['SPIDER_CONTRACTS'],
)
self.conman = ContractsManager([load_object(c) for c in contracts])
self.results = TextTestRunner()._makeResult()
# contract requests
contract_reqs = defaultdict(list)
@ -61,6 +65,8 @@ class Command(ScrapyCommand):
for method in sorted(methods):
print ' * %s' % method
else:
dispatcher.connect(self.results.printErrors,
signals.engine_stopped)
self.crawler.start()
def get_requests(self, spider):
@ -69,7 +75,7 @@ class Command(ScrapyCommand):
for key, value in vars(type(spider)).items():
if callable(value) and value.__doc__:
bound_method = value.__get__(spider, type(spider))
request = self.conman.from_method(bound_method)
request = self.conman.from_method(bound_method, self.results)
if request:
request.callback = _generate(request.callback)

View File

@ -1,10 +1,22 @@
import sys
import re
from functools import wraps
from unittest.case import TestCase
from scrapy.http import Request
from scrapy.utils.spider import iterate_spider_output
from scrapy.utils.python import get_spec
from scrapy.exceptions import ContractFail
def create_testcase(method):
name = '%s.%s' % (method.__self__.__class__.__name__, method.__name__)
class ContractTestCase(TestCase):
def __str__(self):
return "%s (%s)" % (name, method.__self__.name)
setattr(ContractTestCase, name, lambda x: x)
return ContractTestCase(name)
class ContractsManager(object):
@ -27,7 +39,7 @@ class ContractsManager(object):
return contracts
def from_method(self, method, fail=False):
def from_method(self, method, results):
contracts = self.extract_contracts(method)
if contracts:
# calculate request args
@ -43,9 +55,9 @@ class ContractsManager(object):
# execute pre and post hooks in order
for contract in reversed(contracts):
request = contract.add_pre_hook(request, fail)
request = contract.add_pre_hook(request, results)
for contract in contracts:
request = contract.add_post_hook(request, fail)
request = contract.add_post_hook(request, results)
return request
@ -54,40 +66,44 @@ class Contract(object):
""" Abstract class for contracts """
def __init__(self, method, *args):
self.method = method
self.testcase = create_testcase(method)
self.args = args
def add_pre_hook(self, request, fail=False):
def add_pre_hook(self, request, results):
cb = request.callback
@wraps(cb)
def wrapper(response):
try:
self.pre_process(response)
except ContractFail as e:
if fail:
raise
else:
print e.format(self.method)
return list(iterate_spider_output(cb(response)))
except AssertionError:
results.addFailure(self.testcase, sys.exc_info())
except Exception:
results.addError(self.testcase, sys.exc_info())
else:
results.addSuccess(self.testcase)
finally:
return list(iterate_spider_output(cb(response)))
request.callback = wrapper
return request
def add_post_hook(self, request, fail=False):
def add_post_hook(self, request, results):
cb = request.callback
@wraps(cb)
def wrapper(response):
output = list(iterate_spider_output(cb(response)))
try:
output = list(iterate_spider_output(cb(response)))
self.post_process(output)
except ContractFail as e:
if fail:
raise
else:
print e.format(self.method)
return output
except AssertionError:
results.addFailure(self.testcase, sys.exc_info())
except Exception:
results.addError(self.testcase, sys.exc_info())
else:
results.addSuccess(self.testcase)
finally:
return output
request.callback = wrapper
return request

View File

@ -52,7 +52,4 @@ class ScrapyDeprecationWarning(Warning):
class ContractFail(AssertionError):
"""Error raised in case of a failing contract"""
def format(self, method):
return '[FAILED] %s:%s\n>>> %s\n' % \
(method.im_class.name, method.__name__, self)
pass

View File

@ -1,9 +1,10 @@
from unittest.runner import TextTestRunner
from twisted.trial import unittest
from scrapy.spider import BaseSpider
from scrapy.http import Request
from scrapy.item import Item, Field
from scrapy.exceptions import ContractFail
from scrapy.contracts import ContractsManager
from scrapy.contracts.default import (
UrlContract,
@ -71,54 +72,69 @@ class TestSpider(BaseSpider):
class ContractsManagerTest(unittest.TestCase):
contracts = [UrlContract, ReturnsContract, ScrapesContract]
def test_contracts(self):
conman = ContractsManager(self.contracts)
def setUp(self):
self.conman = ContractsManager(self.contracts)
self.results = TextTestRunner()._makeResult()
self.results.stream = None
def should_succeed(self):
self.assertFalse(self.results.failures)
self.assertFalse(self.results.errors)
def should_fail(self):
self.assertTrue(self.results.failures)
self.assertFalse(self.results.errors)
def test_contracts(self):
# extract contracts correctly
contracts = conman.extract_contracts(TestSpider.returns_request)
contracts = self.conman.extract_contracts(TestSpider.returns_request)
self.assertEqual(len(contracts), 2)
self.assertEqual(frozenset(map(type, contracts)),
frozenset([UrlContract, ReturnsContract]))
# returns request for valid method
request = conman.from_method(TestSpider.returns_request)
request = self.conman.from_method(TestSpider.returns_request,
self.results)
self.assertNotEqual(request, None)
# no request for missing url
request = conman.from_method(TestSpider.parse_no_url)
request = self.conman.from_method(TestSpider.parse_no_url,
self.results)
self.assertEqual(request, None)
def test_returns(self):
conman = ContractsManager(self.contracts)
spider = TestSpider()
response = ResponseMock()
# returns_item
request = conman.from_method(spider.returns_item, fail=True)
request = self.conman.from_method(spider.returns_item, self.results)
output = request.callback(response)
self.assertEqual(map(type, output), [TestItem])
self.should_succeed()
# returns_request
request = conman.from_method(spider.returns_request, fail=True)
request = self.conman.from_method(spider.returns_request, self.results)
output = request.callback(response)
self.assertEqual(map(type, output), [Request])
self.should_succeed()
# returns_fail
request = conman.from_method(spider.returns_fail, fail=True)
self.assertRaises(ContractFail, request.callback, response)
request = self.conman.from_method(spider.returns_fail, self.results)
request.callback(response)
self.should_fail()
def test_scrapes(self):
conman = ContractsManager(self.contracts)
spider = TestSpider()
response = ResponseMock()
# scrapes_item_ok
request = conman.from_method(spider.scrapes_item_ok, fail=True)
request = self.conman.from_method(spider.scrapes_item_ok, self.results)
output = request.callback(response)
self.assertEqual(map(type, output), [TestItem])
self.should_succeed()
# scrapes_item_fail
request = conman.from_method(spider.scrapes_item_fail, fail=True)
self.assertRaises(ContractFail, request.callback, response)
request = self.conman.from_method(spider.scrapes_item_fail,
self.results)
request.callback(response)
self.should_fail()