From 1d726063d6b241c96444094551583698ae156e0e Mon Sep 17 00:00:00 2001 From: Pablo Hoffman Date: Sun, 28 Nov 2010 18:14:45 -0200 Subject: [PATCH] * Added tests for shell/fetch/version commands (closes #255) * Fixed bug causing Scrapy shell to fail if started without any argument (closes #294) --- scrapy/shell.py | 2 ++ scrapy/tests/test_command_fetch.py | 22 +++++++++++++ scrapy/tests/test_command_shell.py | 47 ++++++++++++++++++++++++++++ scrapy/tests/test_command_version.py | 15 +++++++++ scrapy/utils/testproc.py | 46 +++++++++++++++++++++++++++ scrapy/utils/testsite.py | 31 ++++++++++++++++++ 6 files changed, 163 insertions(+) create mode 100644 scrapy/tests/test_command_fetch.py create mode 100644 scrapy/tests/test_command_shell.py create mode 100644 scrapy/tests/test_command_version.py create mode 100644 scrapy/utils/testproc.py create mode 100644 scrapy/utils/testsite.py diff --git a/scrapy/shell.py b/scrapy/shell.py index 81d24d604..3cd0dda05 100644 --- a/scrapy/shell.py +++ b/scrapy/shell.py @@ -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: diff --git a/scrapy/tests/test_command_fetch.py b/scrapy/tests/test_command_fetch.py new file mode 100644 index 000000000..c1b429c2a --- /dev/null +++ b/scrapy/tests/test_command_fetch.py @@ -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']) diff --git a/scrapy/tests/test_command_shell.py b/scrapy/tests/test_command_shell.py new file mode 100644 index 000000000..ffba30415 --- /dev/null +++ b/scrapy/tests/test_command_shell.py @@ -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') + diff --git a/scrapy/tests/test_command_version.py b/scrapy/tests/test_command_version.py new file mode 100644 index 000000000..6f0380d77 --- /dev/null +++ b/scrapy/tests/test_command_version.py @@ -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__) diff --git a/scrapy/utils/testproc.py b/scrapy/utils/testproc.py new file mode 100644 index 000000000..b00f8ae47 --- /dev/null +++ b/scrapy/utils/testproc.py @@ -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) diff --git a/scrapy/utils/testsite.py b/scrapy/utils/testsite.py new file mode 100644 index 000000000..d673950e6 --- /dev/null +++ b/scrapy/utils/testsite.py @@ -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("

Works

World

", "text/html")) + r.putChild("enc-gb18030", static.Data("

gb18030 encoding

", "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()