* Added tests for shell/fetch/version commands (closes #255)

* Fixed bug causing Scrapy shell to fail if started without any argument (closes #294)
This commit is contained in:
Pablo Hoffman 2010-11-28 18:14:45 -02:00
parent 6f82ea19de
commit 1d726063d6
6 changed files with 163 additions and 0 deletions

View File

@ -47,6 +47,8 @@ class Shell(object):
elif response:
request = response.request
self.populate_vars(request.url, response, request, spider)
else:
self.populate_vars()
if self.code:
print eval(self.code, globals(), self.vars)
else:

View File

@ -0,0 +1,22 @@
from twisted.trial import unittest
from twisted.internet import defer
from scrapy.utils.testsite import SiteTest
from scrapy.utils.testproc import ProcessTest
class FetchTest(ProcessTest, SiteTest, unittest.TestCase):
command = 'fetch'
@defer.inlineCallbacks
def test_output(self):
_, out, _ = yield self.execute([self.url('/text')])
self.assertEqual(out.strip(), 'Works')
@defer.inlineCallbacks
def test_headers(self):
_, out, _ = yield self.execute([self.url('/text'), '--headers'])
headers = eval(out)
assert 'TwistedWeb' in headers['Server'][0]
self.assertEqual(headers['Content-Type'], ['text/plain'])

View File

@ -0,0 +1,47 @@
from twisted.trial import unittest
from twisted.internet import defer
from scrapy.utils.testsite import SiteTest
from scrapy.utils.testproc import ProcessTest
class ShellTest(ProcessTest, SiteTest, unittest.TestCase):
command = 'shell'
@defer.inlineCallbacks
def test_empty(self):
_, out, _ = yield self.execute(['-c', 'item'])
assert 'Item' in out
@defer.inlineCallbacks
def test_response_body(self):
_, out, _ = yield self.execute([self.url('/text'), '-c', 'response.body'])
assert 'Works' in out
@defer.inlineCallbacks
def test_response_type_text(self):
_, out, _ = yield self.execute([self.url('/text'), '-c', 'type(response)'])
assert 'TextResponse' in out
@defer.inlineCallbacks
def test_response_type_html(self):
_, out, _ = yield self.execute([self.url('/html'), '-c', 'type(response)'])
assert 'HtmlResponse' in out
@defer.inlineCallbacks
def test_response_selector_html(self):
xpath = 'hxs.select("//p[@class=\'one\']/text()").extract()[0]'
_, out, _ = yield self.execute([self.url('/html'), '-c', xpath])
self.assertEqual(out.strip(), 'Works')
@defer.inlineCallbacks
def test_response_encoding_gb18030(self):
_, out, _ = yield self.execute([self.url('/enc-gb18030'), '-c', 'response.encoding'])
self.assertEqual(out.strip(), 'gb18030')
@defer.inlineCallbacks
def test_redirect(self):
_, out, _ = yield self.execute([self.url('/redirect'), '-c', 'response.url'])
assert out.strip().endswith('/redirected')

View File

@ -0,0 +1,15 @@
from twisted.trial import unittest
from twisted.internet import defer
import scrapy
from scrapy.utils.testproc import ProcessTest
class VersionTest(ProcessTest, unittest.TestCase):
command = 'version'
@defer.inlineCallbacks
def test_output(self):
_, out, _ = yield self.execute([])
self.assertEqual(out.strip(), "Scrapy %s" % scrapy.__version__)

46
scrapy/utils/testproc.py Normal file
View File

@ -0,0 +1,46 @@
import sys
import os
from twisted.internet import reactor, defer, protocol
class ProcessTest(object):
command = None
prefix = [sys.executable, '-m', 'scrapy.cmdline']
def execute(self, args, check_code=True, settings='missing'):
env = os.environ.copy()
env['SCRAPY_SETTINGS_MODULE'] = settings
cmd = self.prefix + [self.command] + list(args)
pp = TestProcessProtocol()
pp.deferred.addBoth(self._process_finished, cmd, check_code)
reactor.spawnProcess(pp, cmd[0], cmd, env=env)
return pp.deferred
def _process_finished(self, pp, cmd, check_code):
if pp.exitcode and check_code:
msg = "process %s exit with code %d" % (cmd, pp.exitcode)
msg += "\n>>> stdout <<<\n%s" % pp.out
msg += "\n"
msg += "\n>>> stderr <<<\n%s" % pp.err
raise RuntimeError(msg)
return pp.exitcode, pp.out, pp.err
class TestProcessProtocol(protocol.ProcessProtocol):
def __init__(self):
self.deferred = defer.Deferred()
self.out = ''
self.err = ''
self.exitcode = None
def outReceived(self, data):
self.out += data
def errReceived(self, data):
self.err += data
def processEnded(self, status):
self.exitcode = status.value.exitCode
self.deferred.callback(self)

31
scrapy/utils/testsite.py Normal file
View File

@ -0,0 +1,31 @@
import urlparse
from twisted.internet import reactor
from twisted.web import server, resource, static, util
class SiteTest(object):
def setUp(self):
self.site = reactor.listenTCP(0, test_site(), interface="127.0.0.1")
self.baseurl = "http://localhost:%d/" % self.site.getHost().port
def tearDown(self):
self.site.stopListening()
def url(self, path):
return urlparse.urljoin(self.baseurl, path)
def test_site():
r = resource.Resource()
r.putChild("text", static.Data("Works", "text/plain"))
r.putChild("html", static.Data("<body><p class='one'>Works</p><p class='two'>World</p></body>", "text/html"))
r.putChild("enc-gb18030", static.Data("<p>gb18030 encoding</p>", "text/html; charset=gb18030"))
r.putChild("redirect", util.Redirect("/redirected"))
r.putChild("redirected", static.Data("Redirected here", "text/plain"))
return server.Site(r)
if __name__ == '__main__':
port = reactor.listenTCP(0, test_site(), interface="127.0.0.1")
print "http://localhost:%d/" % port.getHost().port
reactor.run()