contrib: fix a header bug in httpcompression middleware and add tests

--HG--
extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%40969
This commit is contained in:
Daniel Grana 2009-03-04 20:11:12 +00:00
parent 8ad4211e26
commit e3c9f5951a
5 changed files with 105 additions and 5 deletions

View File

@ -14,13 +14,14 @@ class HttpCompressionMiddleware(object):
def process_response(self, request, response, spider):
if isinstance(response, Response):
content_encoding = response.headers.get('Content-Encoding')
content_encoding = response.headers.getlist('Content-Encoding')
if content_encoding:
encoding = content_encoding[0].lower()
raw_body = response.body
decoded_body = self._decode(raw_body, encoding)
encoding = content_encoding.pop()
decoded_body = self._decode(response.body, encoding.lower())
response = response.replace(body=decoded_body)
response.headers['Content-Encoding'] = content_encoding[1:]
if not content_encoding:
del response.headers['Content-Encoding']
return response
def _decode(self, body, encoding):

View File

@ -0,0 +1,99 @@
from __future__ import with_statement
from unittest import TestCase
from os.path import join, abspath, dirname
from scrapy.spider import spiders
from scrapy.http import Response, Request
from scrapy.contrib.downloadermiddleware.httpcompression import HttpCompressionMiddleware
SAMPLEDIR = join(abspath(dirname(__file__)), 'sample_data/compressed')
FORMAT = {
'gzip': ('html-gzip.bin', 'gzip'),
'rawdeflate': ('html-rawdeflate.bin', 'deflate'),
'zlibdeflate': ('html-zlibdeflate.bin', 'deflate'),
}
class HttpCompressionTest(TestCase):
def setUp(self):
spiders.spider_modules = ['scrapy.tests.test_spiders']
spiders.reload()
self.spider = spiders.fromdomain('scrapytest.org')
self.mw = HttpCompressionMiddleware()
def _getresponse(self, coding):
if coding not in FORMAT:
raise ValueError()
samplefile, contentencoding = FORMAT[coding]
with open(join(SAMPLEDIR, samplefile), 'rb') as sample:
body = sample.read()
headers = {
'Server': 'Yaws/1.49 Yet Another Web Server',
'Date': 'Sun, 08 Mar 2009 00:41:03 GMT',
'Content-Length': len(body),
'Content-Type': 'text/html',
'Content-Encoding': contentencoding,
}
response = Response('http://scrapytest.org/', body=body, headers=headers)
response.request = Request('http://scrapytest.org', headers={'Accept-Encoding': 'gzip,deflate'})
return response
def test_process_request(self):
request = Request('http://scrapytest.org')
assert 'Accept-Encoding' not in request.headers
self.mw.process_request(request, self.spider)
self.assertEqual(request.headers.get('Accept-Encoding'), 'gzip,deflate')
def test_process_response_gzip(self):
response = self._getresponse('gzip')
request = response.request
self.assertEqual(response.headers['Content-Encoding'], 'gzip')
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith('<!DOCTYPE')
assert 'Content-Encoding' not in newresponse.headers
def test_process_response_rawdeflate(self):
response = self._getresponse('rawdeflate')
request = response.request
self.assertEqual(response.headers['Content-Encoding'], 'deflate')
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith('<!DOCTYPE')
assert 'Content-Encoding' not in newresponse.headers
def test_process_response_zlibdelate(self):
response = self._getresponse('zlibdeflate')
request = response.request
self.assertEqual(response.headers['Content-Encoding'], 'deflate')
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith('<!DOCTYPE')
assert 'Content-Encoding' not in newresponse.headers
def test_process_response_plain(self):
response = Response('http://scrapytest.org', body='<!DOCTYPE...')
request = Request('http://scrapytest.org')
assert not response.headers.get('Content-Encoding')
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is response
assert newresponse.body.startswith('<!DOCTYPE')
def test_multipleencodings(self):
response = self._getresponse('gzip')
response.headers['Content-Encoding'] = ['uuencode', 'gzip']
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
self.assertEqual(newresponse.headers.getlist('Content-Encoding'), ['uuencode'])