RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

reconfigured scrapy Response in order to disallow direct use of ResponseBody class that should be private 

--HG--
extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%40643
This commit is contained in:
samus_ 2009-01-04 18:23:12 +00:00
parent 58e4cf74b6
commit 2ea631398e
7 changed files with 40 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
scrapy/trunk/scrapy/contrib/downloadermiddleware/compression.py
View File

@ -2,7 +2,7 @@ import zlib
from gzip import GzipFile
from cStringIO import StringIO
from scrapy.http import Response, ResponseBody
from scrapy.http import Response
class CompressionMiddleware(object):
@ -18,9 +18,8 @@ class CompressionMiddleware(object):
if content_encoding:
encoding = content_encoding[0].lower()
raw_body = response.body.get_content()
declared_encoding = response.body.declared_encoding
decoded_body = self._decode(raw_body, encoding)
response.body = ResponseBody(decoded_body, declared_encoding)
response = response.replace(body=decoded_body)
response.headers['Content-Encoding'] = content_encoding[1:]
return response

8
scrapy/trunk/scrapy/contrib/downloadermiddleware/decompression.py
View File

@ -8,7 +8,7 @@ except:
from StringIO import StringIO
from scrapy import log
from scrapy.http import Response, ResponseBody
from scrapy.http import Response
class DecompressionMiddleware(object):
""" This middleware tries to recognise and extract the possibly compressed
@ -28,7 +28,7 @@ class DecompressionMiddleware(object):
except tarfile.ReadError:
return False
if tar_file.members:
return response.replace(body=ResponseBody(tar_file.extractfile(tar_file.members[0]).read()))
return response.replace(body=tar_file.extractfile(tar_file.members[0]).read())
else:
raise self.ArchiveIsEmpty
@ -39,7 +39,7 @@ class DecompressionMiddleware(object):
return False
namelist = zip_file.namelist()
if namelist:
return response.replace(body=ResponseBody(zip_file.read(namelist[0])))
return response.replace(body=zip_file.read(namelist[0]))
else:
raise self.ArchiveIsEmpty
@ -56,7 +56,7 @@ class DecompressionMiddleware(object):
decompressed_body = bz2.decompress(self.body)
except IOError:
return False
return response.replace(body=ResponseBody(decompressed_body))
return response.replace(body=decompressed_body)
def extract(self, response):
""" This method tries to decompress the given response, if possible,

8
scrapy/trunk/scrapy/http/response.py
View File

@ -29,11 +29,9 @@ class Response(object) :
self.original_url = Url(original_url) if original_url else url # different if redirected or escaped
self.headers = Headers(headers or {})
self.status = status
# FIXME this should be replaced with assert(isinstance(body, basestring)) since ResponseBody is not meant to be used directly
if isinstance(body, ResponseBody):
self.body = body
else:
self.body = ResponseBody(body, self.headers_encoding())
# ResponseBody is not meant to be used directly (use .replace instead)
assert(isinstance(body, basestring) or body is None)
self.body = ResponseBody(body, self.headers_encoding())
self.cached = False
self.request = None # request which originated this response

28
scrapy/trunk/scrapy/tests/test_adaptors.py
View File

@ -4,7 +4,7 @@ import unittest
import re
from scrapy.contrib import adaptors
from scrapy.http import Response, ResponseBody, Headers
from scrapy.http import Response, Headers
from scrapy.xpath.selector import HtmlXPathSelector, XmlXPathSelector
class AdaptorsTestCase(unittest.TestCase):
@ -89,7 +89,7 @@ class AdaptorsTestCase(unittest.TestCase):
<a onclick="javascript: opensomething('dummy/my_html2.html');">something2</a>
</div>
</body></html>"""
sample_response = Response('foobar.com', 'http://foobar.com/dummy', body=ResponseBody(test_data))
sample_response = Response('foobar.com', 'http://foobar.com/dummy', body=test_data)
sample_xsel = XmlXPathSelector(sample_response)
sample_adaptor = adaptors.ExtractImages(response=sample_response)
@ -105,56 +105,56 @@ class AdaptorsTestCase(unittest.TestCase):
self.assertEqual(sample_adaptor(sample_xsel.x('//a[@onclick]').re(r'opensomething\(\'(.*?)\'\)')),
[u'http://foobar.com/my_html1.html', u'http://foobar.com/dummy/my_html2.html'])
def test_to_unicode(self):
self.assertEqual(adaptors.to_unicode(['lala', 'lele', 'lulu\xc3\xb1', 1, '\xc3\xa1\xc3\xa9']),
[u'lala', u'lele', u'lulu\xf1', u'1', u'\xe1\xe9'])
def test_regex(self):
adaptor = adaptors.Regex(regex=r'href="(.*?)"')
self.assertEqual(adaptor(['<a href="lala.com">dsa</a><a href="pepe.co.uk"></a>',
'<a href="das.biz">href="lelelel.net"</a>']),
['lala.com', 'pepe.co.uk', 'das.biz', 'lelelel.net'])
def test_unquote_all(self):
self.assertEqual(adaptors.Unquote()([u'hello&copy;&amp;welcome', u'&lt;br /&gt;&amp;']), [u'hello\xa9&welcome', u'<br />&'])
def test_unquote(self):
self.assertEqual(adaptors.Unquote(keep=['amp', 'lt'])([u'hello&copy;&amp;welcome', u'&lt;br /&gt;&amp;']), [u'hello\xa9&amp;welcome', u'&lt;br />&amp;'])
def test_remove_tags(self):
test_data = ['<a href="lala">adsaas<br /></a>', '<div id="1"><table>dsadasf</table></div>']
self.assertEqual(adaptors.remove_tags(test_data), ['adsaas', 'dsadasf'])
def test_remove_root(self):
self.assertEqual(adaptors.remove_root(['<div>lallaa<a href="coso">dsfsdfds</a>pepepep<br /></div>']),
['lallaa<a href="coso">dsfsdfds</a>pepepep<br />'])
def test_remove_multispaces(self):
self.assertEqual(adaptors.clean_spaces([' hello, whats up?', 'testing testingtesting testing']),
[' hello, whats up?', 'testing testingtesting testing'])
def test_strip(self):
self.assertEqual(adaptors.strip([' hi there, sweety ;D ', ' I CAN HAZ TEST?? ']),
['hi there, sweety ;D', 'I CAN HAZ TEST??'])
self.assertEqual(adaptors.strip(' hello there, this is my test '),
'hello there, this is my test')
def test_drop_empty_elements(self):
self.assertEqual(adaptors.drop_empty([1, 2, None, 5, 0, 6, False, 'hi']),
[1, 2, 5, 6, 'hi'])
def test_delist(self):
self.assertEqual(adaptors.Delist()(['hi', 'there', 'fellas.', 'this', 'is', 'my', 'test.']),
'hi there fellas. this is my test.')

6
scrapy/trunk/scrapy/tests/test_libxml2.py
View File

@ -1,6 +1,6 @@
from unittest import TestCase, main
import libxml2
from scrapy.http import ResponseBody, Response
from scrapy.http import Response
class Libxml2Test(TestCase):
def setUp(self):
@ -35,8 +35,8 @@ class ResponseLibxml2DocTest(TestCase):
scrapymanager.configure()
self.body_content = 'test problematic \x00 body'
self.problematic_body = ResponseBody(self.body_content, 'utf-8')
response = Response('example.com', 'http://example.com/catalog/product/blabla-123', body=self.problematic_body)
response = Response('example.com', 'http://example.com/catalog/product/blabla-123',
headers={'Content-Type': 'text/plain; charset=utf-8'}, body=self.body_content)
response.getlibxml2doc()
if __name__ == "__main__":

12
scrapy/trunk/scrapy/tests/test_link.py
View File

@ -1,7 +1,7 @@
import os
import unittest
from scrapy.http.response import Response, ResponseBody
from scrapy.http.response import Response
from scrapy.link import LinkExtractor, Link
from scrapy.link.extractors import RegexLinkExtractor
from scrapy.contrib.link_extractors import HTMLImageLinkExtractor
@ -37,10 +37,10 @@ class LinkExtractorTestCase(unittest.TestCase):
def test_extraction_encoding(self):
base_path = os.path.join(os.path.dirname(__file__), 'sample_data', 'link_extractor')
body = open(os.path.join(base_path, 'linkextractor_noenc.html'), 'r').read()
response_utf8 = Response(url='http://example.com/utf8', domain='example.com', body=ResponseBody(body), headers={'Content-Type': ['text/html; charset=utf-8']})
response_noenc = Response(url='http://example.com/noenc', domain='example.com', body=ResponseBody(body))
response_utf8 = Response(url='http://example.com/utf8', domain='example.com', body=body, headers={'Content-Type': ['text/html; charset=utf-8']})
response_noenc = Response(url='http://example.com/noenc', domain='example.com', body=body)
body = open(os.path.join(base_path, 'linkextractor_latin1.html'), 'r').read()
response_latin1 = Response(url='http://example.com/latin1', domain='example.com', body=ResponseBody(body))
response_latin1 = Response(url='http://example.com/latin1', domain='example.com', body=body)
lx = LinkExtractor()
self.assertEqual(lx.extract_links(response_utf8),
@ -67,7 +67,7 @@ class RegexLinkExtractorTestCase(unittest.TestCase):
def setUp(self):
base_path = os.path.join(os.path.dirname(__file__), 'sample_data', 'link_extractor')
body = open(os.path.join(base_path, 'regex_linkextractor.html'), 'r').read()
self.response = Response(url='http://example.com/index', domain='example.com', body=ResponseBody(body))
self.response = Response(url='http://example.com/index', domain='example.com', body=body)
def test_urls_type(self):
'''Test that the resulting urls are regular strings and not a unicode objects'''
@ -141,7 +141,7 @@ class RegexLinkExtractorTestCase(unittest.TestCase):
# def setUp(self):
# base_path = os.path.join(os.path.dirname(__file__), 'sample_data', 'link_extractor')
# body = open(os.path.join(base_path 'image_linkextractor.html'), 'r').read()
# self.response = Response(url='http://example.com/index', domain='example.com', body=ResponseBody(body))
# self.response = Response(url='http://example.com/index', domain='example.com', body=body)
# def test_urls_type(self):
# '''Test that the resulting urls are regular strings and not a unicode objects'''

16
scrapy/trunk/scrapy/tests/test_middleware_decompression.py
View File

@ -1,34 +1,34 @@
import os
from unittest import TestCase, main
from scrapy.http import Response, ResponseBody
from scrapy.http import Response
from scrapy.contrib.downloadermiddleware.decompression import DecompressionMiddleware
class ScrapyDecompressionTest(TestCase):
uncompressed_body = ''
test_responses = {}
middleware = DecompressionMiddleware()
def setUp(self):
self.datadir = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'sample_data', 'compressed')
formats = ['tar', 'xml.bz2', 'xml.gz', 'zip']
uncompressed_fd = open(os.path.join(self.datadir, 'feed-sample1.xml'), 'r')
self.uncompressed_body = uncompressed_fd.read()
uncompressed_fd.close()
for format in formats:
fd = open(os.path.join(self.datadir, 'feed-sample1.' + format), 'r')
body = ResponseBody(fd.read())
body = fd.read()
fd.close()
self.test_responses[format] = Response('foo.com', 'http://foo.com/bar', body=body)
def test_tar(self):
response, format = self.middleware.extract(self.test_responses['tar'])
self.assertEqual(response.body.to_string(), self.uncompressed_body)
def test_zip(self):
response, format = self.middleware.extract(self.test_responses['zip'])
self.assertEqual(response.body.to_string(), self.uncompressed_body)
def test_gz(self):
response, format = self.middleware.extract(self.test_responses['xml.gz'])
self.assertEqual(response.body.to_string(), self.uncompressed_body)
@ -36,6 +36,6 @@ class ScrapyDecompressionTest(TestCase):
def test_bz2(self):
response, format = self.middleware.extract(self.test_responses['xml.bz2'])
self.assertEqual(response.body.to_string(), self.uncompressed_body)
if __name__ == '__main__':
main()