partial port of Request and Response

This commit is contained in:
Mikhail Korobov 2015-07-25 13:08:44 +02:00
parent f576b3ffee
commit e853d9e910
8 changed files with 161 additions and 169 deletions

View File

@ -8,6 +8,7 @@ import six
from w3lib.url import safe_url_string
from scrapy.http.headers import Headers
from scrapy.utils.python import to_native_str, to_bytes
from scrapy.utils.trackref import object_ref
from scrapy.utils.url import escape_ajax
from scrapy.http.common import obsolete_setter
@ -46,15 +47,12 @@ class Request(object_ref):
return self._url
def _set_url(self, url):
if isinstance(url, str):
self._url = escape_ajax(safe_url_string(url))
elif isinstance(url, six.text_type):
if self.encoding is None:
raise TypeError('Cannot convert unicode url - %s has no encoding' %
type(self).__name__)
self._set_url(url.encode(self.encoding))
else:
if not isinstance(url, six.string_types):
raise TypeError('Request url must be str or unicode, got %s:' % type(url).__name__)
url = to_native_str(url, self.encoding)
self._url = escape_ajax(safe_url_string(url))
if ':' not in self._url:
raise ValueError('Missing scheme in request url: %s' % self._url)
@ -64,17 +62,10 @@ class Request(object_ref):
return self._body
def _set_body(self, body):
if isinstance(body, str):
self._body = body
elif isinstance(body, six.text_type):
if self.encoding is None:
raise TypeError('Cannot convert unicode body - %s has no encoding' %
type(self).__name__)
self._body = body.encode(self.encoding)
elif body is None:
self._body = ''
if body is None:
self._body = b''
else:
raise TypeError("Request body must either str or unicode. Got: '%s'" % type(body).__name__)
self._body = to_bytes(body, self.encoding)
body = property(_get_body, obsolete_setter(_set_body, 'body'))

View File

@ -9,7 +9,7 @@ from six.moves.urllib.parse import urljoin, urlencode
import lxml.html
import six
from scrapy.http.request import Request
from scrapy.utils.python import to_bytes
from scrapy.utils.python import to_bytes, is_listlike
class FormRequest(Request):
@ -25,7 +25,7 @@ class FormRequest(Request):
items = formdata.items() if isinstance(formdata, dict) else formdata
querystr = _urlencode(items, self.encoding)
if self.method == 'POST':
self.headers.setdefault('Content-Type', 'application/x-www-form-urlencoded')
self.headers.setdefault(b'Content-Type', b'application/x-www-form-urlencoded')
self._set_body(querystr)
else:
self._set_url(self.url + ('&' if '?' in self.url else '?') + querystr)
@ -50,7 +50,7 @@ def _get_form_url(form, url):
def _urlencode(seq, enc):
values = [(to_bytes(k, enc), to_bytes(v, enc))
for k, vs in seq
for v in (vs if hasattr(vs, '__iter__') else [vs])]
for v in (vs if is_listlike(vs) else [vs])]
return urlencode(values, doseq=1)

View File

@ -4,9 +4,6 @@ responses in Scrapy.
See documentation in docs/topics/request-response.rst
"""
import copy
from six.moves.urllib.parse import urljoin
from scrapy.http.headers import Headers
@ -15,7 +12,7 @@ from scrapy.http.common import obsolete_setter
class Response(object_ref):
def __init__(self, url, status=200, headers=None, body='', flags=None, request=None):
def __init__(self, url, status=200, headers=None, body=b'', flags=None, request=None):
self.headers = Headers(headers or {})
self.status = int(status)
self._set_body(body)
@ -28,8 +25,10 @@ class Response(object_ref):
try:
return self.request.meta
except AttributeError:
raise AttributeError("Response.meta not available, this response " \
"is not tied to any request")
raise AttributeError(
"Response.meta not available, this response "
"is not tied to any request"
)
def _get_url(self):
return self._url
@ -38,7 +37,7 @@ class Response(object_ref):
if isinstance(url, str):
self._url = url
else:
raise TypeError('%s url must be str, got %s:' % (type(self).__name__, \
raise TypeError('%s url must be str, got %s:' % (type(self).__name__,
type(url).__name__))
url = property(_get_url, obsolete_setter(_set_url, 'url'))
@ -47,16 +46,15 @@ class Response(object_ref):
return self._body
def _set_body(self, body):
if isinstance(body, str):
self._body = body
elif isinstance(body, unicode):
raise TypeError("Cannot assign a unicode body to a raw Response. " \
"Use TextResponse, HtmlResponse, etc")
elif body is None:
self._body = ''
if body is None:
self._body = b''
elif not isinstance(body, bytes):
raise TypeError(
"Response body must be bytes. "
"If you want to pass unicode body use TextResponse "
"or HtmlResponse.")
else:
raise TypeError("Response body must either be str or unicode. Got: '%s'" \
% type(body).__name__)
self._body = body
body = property(_get_body, obsolete_setter(_set_body, 'body'))

View File

@ -5,13 +5,14 @@ discovering (through HTTP headers) to base Response class.
See documentation in docs/topics/request-response.rst
"""
import six
from six.moves.urllib.parse import urljoin
from w3lib.encoding import html_to_unicode, resolve_encoding, \
html_body_declared_encoding, http_content_type_encoding
from scrapy.http.response import Response
from scrapy.utils.response import get_base_url
from scrapy.utils.python import memoizemethod_noargs
from scrapy.utils.python import memoizemethod_noargs, to_native_str
class TextResponse(Response):
@ -26,18 +27,18 @@ class TextResponse(Response):
super(TextResponse, self).__init__(*args, **kwargs)
def _set_url(self, url):
if isinstance(url, unicode):
if self.encoding is None:
raise TypeError('Cannot convert unicode url - %s has no encoding' %
type(self).__name__)
self._url = url.encode(self.encoding)
if isinstance(url, six.text_type):
if six.PY2 and self.encoding is None:
raise TypeError("Cannot convert unicode url - %s "
"has no encoding" % type(self).__name__)
self._url = to_native_str(url, self.encoding)
else:
super(TextResponse, self)._set_url(url)
def _set_body(self, body):
self._body = ''
if isinstance(body, unicode):
if self.encoding is None:
self._body = b'' # used by encoding detection
if isinstance(body, six.text_type):
if self._encoding is None:
raise TypeError('Cannot convert unicode body - %s has no encoding' %
type(self).__name__)
self._body = body.encode(self._encoding)
@ -73,14 +74,14 @@ class TextResponse(Response):
@memoizemethod_noargs
def _headers_encoding(self):
content_type = self.headers.get('Content-Type')
return http_content_type_encoding(content_type)
content_type = self.headers.get(b'Content-Type', b'')
return http_content_type_encoding(to_native_str(content_type))
def _body_inferred_encoding(self):
if self._cached_benc is None:
content_type = self.headers.get('Content-Type')
benc, ubody = html_to_unicode(content_type, self.body, \
auto_detect_fun=self._auto_detect_fun, \
content_type = to_native_str(self.headers.get(b'Content-Type', b''))
benc, ubody = html_to_unicode(content_type, self.body,
auto_detect_fun=self._auto_detect_fun,
default_encoding=self._DEFAULT_ENCODING)
self._cached_benc = benc
self._cached_ubody = ubody

View File

@ -121,7 +121,7 @@ class Selector(object_ref):
try:
return etree.tostring(self._root,
method=self._tostring_method,
encoding=unicode,
encoding="unicode",
with_tail=False)
except (AttributeError, TypeError):
if self._root is True:
@ -129,7 +129,7 @@ class Selector(object_ref):
elif self._root is False:
return u'0'
else:
return unicode(self._root)
return six.text_type(self._root)
def register_namespace(self, prefix, uri):
if self.namespaces is None:

View File

@ -7,7 +7,7 @@ from pkgutil import iter_modules
import six
from w3lib.html import replace_entities
from scrapy.utils.python import flatten
from scrapy.utils.python import flatten, to_unicode
from scrapy.item import BaseItem
@ -81,7 +81,7 @@ def extract_regex(regex, text, encoding='utf-8'):
* if the regex doesn't contain any group the entire regex matching is returned
"""
if isinstance(regex, basestring):
if isinstance(regex, six.string_types):
regex = re.compile(regex, re.UNICODE)
try:
@ -90,10 +90,11 @@ def extract_regex(regex, text, encoding='utf-8'):
strings = regex.findall(text) # full regex or numbered groups
strings = flatten(strings)
if isinstance(text, unicode):
if isinstance(text, six.text_type):
return [replace_entities(s, keep=['lt', 'amp']) for s in strings]
else:
return [replace_entities(unicode(s, encoding), keep=['lt', 'amp']) for s in strings]
return [replace_entities(to_unicode(s, encoding), keep=['lt', 'amp'])
for s in strings]
def md5sum(file):

View File

@ -1,9 +1,12 @@
import cgi
import unittest
import six
from six.moves import xmlrpc_client as xmlrpclib
from six.moves.urllib.parse import urlparse
from scrapy.http import Request, FormRequest, XmlRpcRequest, Headers, HtmlResponse
from scrapy.utils.python import to_bytes, to_native_str
class RequestTest(unittest.TestCase):
@ -31,13 +34,13 @@ class RequestTest(unittest.TestCase):
self.assertEqual(r.meta, self.default_meta)
meta = {"lala": "lolo"}
headers = {"caca": "coco"}
headers = {b"caca": b"coco"}
r = self.request_class("http://www.example.com", meta=meta, headers=headers, body="a body")
assert r.meta is not meta
self.assertEqual(r.meta, meta)
assert r.headers is not headers
self.assertEqual(r.headers["caca"], "coco")
self.assertEqual(r.headers[b"caca"], b"coco")
def test_url_no_scheme(self):
self.assertRaises(ValueError, self.request_class, 'foo')
@ -45,7 +48,7 @@ class RequestTest(unittest.TestCase):
def test_headers(self):
# Different ways of setting headers attribute
url = 'http://www.scrapy.org'
headers = {'Accept':'gzip', 'Custom-Header':'nothing to tell you'}
headers = {b'Accept':'gzip', b'Custom-Header':'nothing to tell you'}
r = self.request_class(url=url, headers=headers)
p = self.request_class(url=url, headers=r.headers)
@ -57,9 +60,9 @@ class RequestTest(unittest.TestCase):
h = Headers({'key1': u'val1', u'key2': 'val2'})
h[u'newkey'] = u'newval'
for k, v in h.iteritems():
self.assert_(isinstance(k, str))
self.assert_(isinstance(k, bytes))
for s in v:
self.assert_(isinstance(s, str))
self.assert_(isinstance(s, bytes))
def test_eq(self):
url = 'http://www.scrapy.org'
@ -73,17 +76,17 @@ class RequestTest(unittest.TestCase):
self.assertEqual(len(set_), 2)
def test_url(self):
"""Request url tests"""
r = self.request_class(url="http://www.scrapy.org/path")
self.assertEqual(r.url, "http://www.scrapy.org/path")
# url quoting on creation
def test_url_quoting(self):
r = self.request_class(url="http://www.scrapy.org/blank%20space")
self.assertEqual(r.url, "http://www.scrapy.org/blank%20space")
r = self.request_class(url="http://www.scrapy.org/blank space")
self.assertEqual(r.url, "http://www.scrapy.org/blank%20space")
# url encoding
@unittest.skipUnless(six.PY2, "TODO")
def test_url_encoding(self):
r1 = self.request_class(url=u"http://www.scrapy.org/price/\xa3", encoding="utf-8")
r2 = self.request_class(url=u"http://www.scrapy.org/price/\xa3", encoding="latin1")
self.assertEqual(r1.url, "http://www.scrapy.org/price/%C2%A3")
@ -91,19 +94,19 @@ class RequestTest(unittest.TestCase):
def test_body(self):
r1 = self.request_class(url="http://www.example.com/")
assert r1.body == ''
assert r1.body == b''
r2 = self.request_class(url="http://www.example.com/", body="")
assert isinstance(r2.body, str)
r2 = self.request_class(url="http://www.example.com/", body=b"")
assert isinstance(r2.body, bytes)
self.assertEqual(r2.encoding, 'utf-8') # default encoding
r3 = self.request_class(url="http://www.example.com/", body=u"Price: \xa3100", encoding='utf-8')
assert isinstance(r3.body, str)
self.assertEqual(r3.body, "Price: \xc2\xa3100")
assert isinstance(r3.body, bytes)
self.assertEqual(r3.body, b"Price: \xc2\xa3100")
r4 = self.request_class(url="http://www.example.com/", body=u"Price: \xa3100", encoding='latin1')
assert isinstance(r4.body, str)
self.assertEqual(r4.body, "Price: \xa3100")
assert isinstance(r4.body, bytes)
self.assertEqual(r4.body, b"Price: \xa3100")
def test_ajax_url(self):
# ascii url
@ -155,18 +158,19 @@ class RequestTest(unittest.TestCase):
def test_replace(self):
"""Test Request.replace() method"""
r1 = self.request_class("http://www.example.com", method='GET')
hdrs = Headers(dict(r1.headers, key='value'))
hdrs = Headers(r1.headers)
hdrs[b'key'] = b'value'
r2 = r1.replace(method="POST", body="New body", headers=hdrs)
self.assertEqual(r1.url, r2.url)
self.assertEqual((r1.method, r2.method), ("GET", "POST"))
self.assertEqual((r1.body, r2.body), ('', "New body"))
self.assertEqual((r1.body, r2.body), (b'', b"New body"))
self.assertEqual((r1.headers, r2.headers), (self.default_headers, hdrs))
# Empty attributes (which may fail if not compared properly)
r3 = self.request_class("http://www.example.com", meta={'a': 1}, dont_filter=True)
r4 = r3.replace(url="http://www.example.com/2", body='', meta={}, dont_filter=False)
r4 = r3.replace(url="http://www.example.com/2", body=b'', meta={}, dont_filter=False)
self.assertEqual(r4.url, "http://www.example.com/2")
self.assertEqual(r4.body, '')
self.assertEqual(r4.body, b'')
self.assertEqual(r4.meta, {})
assert r4.dont_filter is False
@ -184,39 +188,41 @@ class FormRequestTest(RequestTest):
request_class = FormRequest
def assertSortedEqual(self, first, second, msg=None):
def assertQueryEqual(self, first, second, msg=None):
first = to_native_str(first).split("&")
second = to_native_str(second).split("&")
return self.assertEqual(sorted(first), sorted(second), msg)
def test_empty_formdata(self):
r1 = self.request_class("http://www.example.com", formdata={})
self.assertEqual(r1.body, '')
self.assertEqual(r1.body, b'')
@unittest.skipUnless(six.PY2, "TODO")
def test_default_encoding(self):
# using default encoding (utf-8)
data = {'one': 'two', 'price': '\xc2\xa3 100'}
r2 = self.request_class("http://www.example.com", formdata=data)
self.assertEqual(r2.method, 'POST')
self.assertEqual(r2.encoding, 'utf-8')
self.assertSortedEqual(r2.body.split('&'),
'price=%C2%A3+100&one=two'.split('&'))
self.assertEqual(r2.headers['Content-Type'], 'application/x-www-form-urlencoded')
self.assertQueryEqual(r2.body, b'price=%C2%A3+100&one=two')
self.assertEqual(r2.headers[b'Content-Type'], b'application/x-www-form-urlencoded')
def test_custom_encoding(self):
data = {'price': u'\xa3 100'}
r3 = self.request_class("http://www.example.com", formdata=data, encoding='latin1')
self.assertEqual(r3.encoding, 'latin1')
self.assertEqual(r3.body, 'price=%A3+100')
self.assertEqual(r3.body, b'price=%A3+100')
def test_multi_key_values(self):
# using multiples values for a single key
data = {'price': u'\xa3 100', 'colours': ['red', 'blue', 'green']}
r3 = self.request_class("http://www.example.com", formdata=data)
self.assertSortedEqual(r3.body.split('&'),
'colours=red&colours=blue&colours=green&price=%C2%A3+100'.split('&'))
self.assertQueryEqual(r3.body,
b'colours=red&colours=blue&colours=green&price=%C2%A3+100')
def test_from_response_post(self):
response = _buildresponse(
"""<form action="post.php" method="POST">
b"""<form action="post.php" method="POST">
<input type="hidden" name="test" value="val1">
<input type="hidden" name="test" value="val2">
<input type="hidden" name="test2" value="xxx">
@ -225,13 +231,13 @@ class FormRequestTest(RequestTest):
req = self.request_class.from_response(response,
formdata={'one': ['two', 'three'], 'six': 'seven'})
self.assertEqual(req.method, 'POST')
self.assertEqual(req.headers['Content-type'], 'application/x-www-form-urlencoded')
self.assertEqual(req.headers[b'Content-type'], b'application/x-www-form-urlencoded')
self.assertEqual(req.url, "http://www.example.com/this/post.php")
fs = _qs(req)
self.assertEqual(set(fs["test"]), set(["val1", "val2"]))
self.assertEqual(set(fs["one"]), set(["two", "three"]))
self.assertEqual(fs['test2'], ['xxx'])
self.assertEqual(fs['six'], ['seven'])
self.assertEqual(set(fs[b"test"]), {b"val1", b"val2"})
self.assertEqual(set(fs[b"one"]), {b"two", b"three"})
self.assertEqual(fs[b'test2'], [b'xxx'])
self.assertEqual(fs[b'six'], [b'seven'])
def test_from_response_extra_headers(self):
response = _buildresponse(
@ -244,8 +250,8 @@ class FormRequestTest(RequestTest):
formdata={'one': ['two', 'three'], 'six': 'seven'},
headers={"Accept-Encoding": "gzip,deflate"})
self.assertEqual(req.method, 'POST')
self.assertEqual(req.headers['Content-type'], 'application/x-www-form-urlencoded')
self.assertEqual(req.headers['Accept-Encoding'], 'gzip,deflate')
self.assertEqual(req.headers['Content-type'], b'application/x-www-form-urlencoded')
self.assertEqual(req.headers['Accept-Encoding'], b'gzip,deflate')
def test_from_response_get(self):
response = _buildresponse(
@ -274,8 +280,8 @@ class FormRequestTest(RequestTest):
</form>""")
req = self.request_class.from_response(response, formdata={'two': '2'})
fs = _qs(req)
self.assertEqual(fs['one'], ['1'])
self.assertEqual(fs['two'], ['2'])
self.assertEqual(fs[b'one'], [b'1'])
self.assertEqual(fs[b'two'], [b'2'])
def test_from_response_override_method(self):
response = _buildresponse(
@ -379,7 +385,7 @@ class FormRequestTest(RequestTest):
req = self.request_class.from_response(response, \
clickdata={'name': u'price in \u00a3'})
fs = _qs(req)
self.assertTrue(fs[u'price in \u00a3'.encode('utf-8')])
self.assertTrue(fs[to_native_str(u'price in \u00a3')])
def test_from_response_multiple_forms_clickdata(self):
response = _buildresponse(
@ -489,9 +495,9 @@ class FormRequestTest(RequestTest):
</form>""")
r1 = self.request_class.from_response(response, formdata={'two':'3'})
self.assertEqual(r1.method, 'POST')
self.assertEqual(r1.headers['Content-type'], 'application/x-www-form-urlencoded')
self.assertEqual(r1.headers['Content-type'], b'application/x-www-form-urlencoded')
fs = _qs(r1)
self.assertEqual(fs, {'one': ['1'], 'two': ['3']})
self.assertEqual(fs, {b'one': [b'1'], b'two': [b'3']})
def test_from_response_formname_exists(self):
response = _buildresponse(
@ -506,7 +512,7 @@ class FormRequestTest(RequestTest):
r1 = self.request_class.from_response(response, formname="form2")
self.assertEqual(r1.method, 'POST')
fs = _qs(r1)
self.assertEqual(fs, {'four': ['4'], 'three': ['3']})
self.assertEqual(fs, {b'four': [b'4'], b'three': [b'3']})
def test_from_response_formname_notexist(self):
response = _buildresponse(
@ -519,7 +525,7 @@ class FormRequestTest(RequestTest):
r1 = self.request_class.from_response(response, formname="form3")
self.assertEqual(r1.method, 'POST')
fs = _qs(r1)
self.assertEqual(fs, {'one': ['1']})
self.assertEqual(fs, {b'one': [b'1']})
def test_from_response_formname_errors_formnumber(self):
response = _buildresponse(
@ -664,11 +670,11 @@ class FormRequestTest(RequestTest):
</form>""")
r1 = self.request_class.from_response(response, formxpath="//form[@action='post.php']")
fs = _qs(r1)
self.assertEqual(fs['one'], ['1'])
self.assertEqual(fs[b'one'], [b'1'])
r1 = self.request_class.from_response(response, formxpath="//form/input[@name='four']")
fs = _qs(r1)
self.assertEqual(fs['three'], ['3'])
self.assertEqual(fs[b'three'], [b'3'])
self.assertRaises(ValueError, self.request_class.from_response,
response, formxpath="//form/input[@name='abc']")
@ -691,12 +697,12 @@ class XmlRpcRequestTest(RequestTest):
request_class = XmlRpcRequest
default_method = 'POST'
default_headers = {'Content-Type': ['text/xml']}
default_headers = {b'Content-Type': [b'text/xml']}
def _test_request(self, **kwargs):
r = self.request_class('http://scrapytest.org/rpc2', **kwargs)
self.assertEqual(r.headers['Content-Type'], 'text/xml')
self.assertEqual(r.body, xmlrpclib.dumps(**kwargs))
self.assertEqual(r.headers[b'Content-Type'], b'text/xml')
self.assertEqual(r.body, to_bytes(xmlrpclib.dumps(**kwargs)))
self.assertEqual(r.method, 'POST')
self.assertEqual(r.encoding, kwargs.get('encoding', 'utf-8'))
self.assertTrue(r.dont_filter, True)
@ -706,11 +712,14 @@ class XmlRpcRequestTest(RequestTest):
self._test_request(params=('username', 'password'), methodname='login')
self._test_request(params=('response', ), methodresponse='login')
self._test_request(params=(u'pas\xa3',), encoding='utf-8')
self._test_request(params=(u'pas\xa3',), encoding='latin')
self._test_request(params=(None,), allow_none=1)
self.assertRaises(TypeError, self._test_request)
self.assertRaises(TypeError, self._test_request, params=(None,))
@unittest.skipUnless(six.PY2, "TODO")
def test_latin1(self):
self._test_request(params=(u'pas\xa3',), encoding='latin')
if __name__ == "__main__":
unittest.main()

View File

@ -1,8 +1,12 @@
import unittest
import six
from w3lib.encoding import resolve_encoding
from scrapy.http import Request, Response, TextResponse, HtmlResponse, XmlResponse, Headers
from scrapy.http import (Request, Response, TextResponse, HtmlResponse,
XmlResponse, Headers)
from scrapy.selector import Selector
from scrapy.utils.python import to_native_str
class BaseResponseTest(unittest.TestCase):
@ -14,10 +18,10 @@ class BaseResponseTest(unittest.TestCase):
self.assertRaises(Exception, self.response_class)
self.assertTrue(isinstance(self.response_class('http://example.com/'), self.response_class))
# body can be str or None
self.assertTrue(isinstance(self.response_class('http://example.com/', body=''), self.response_class))
self.assertTrue(isinstance(self.response_class('http://example.com/', body='body'), self.response_class))
self.assertTrue(isinstance(self.response_class('http://example.com/', body=b''), self.response_class))
self.assertTrue(isinstance(self.response_class('http://example.com/', body=b'body'), self.response_class))
# test presence of all optional parameters
self.assertTrue(isinstance(self.response_class('http://example.com/', headers={}, status=200, body=''), self.response_class))
self.assertTrue(isinstance(self.response_class('http://example.com/', body=b'', headers={}, status=200), self.response_class))
r = self.response_class("http://www.example.com")
assert isinstance(r.url, str)
@ -27,12 +31,12 @@ class BaseResponseTest(unittest.TestCase):
assert isinstance(r.headers, Headers)
self.assertEqual(r.headers, {})
headers = {"caca": "coco"}
body = "a body"
headers = {"foo": "bar"}
body = b"a body"
r = self.response_class("http://www.example.com", headers=headers, body=body)
assert r.headers is not headers
self.assertEqual(r.headers["caca"], "coco")
self.assertEqual(r.headers[b"foo"], b"bar")
r = self.response_class("http://www.example.com", status=301)
self.assertEqual(r.status, 301)
@ -43,7 +47,7 @@ class BaseResponseTest(unittest.TestCase):
def test_copy(self):
"""Test Response copy"""
r1 = self.response_class("http://www.example.com", body="Some body")
r1 = self.response_class("http://www.example.com", body=b"Some body")
r1.flags.append('cached')
r2 = r1.copy()
@ -61,7 +65,7 @@ class BaseResponseTest(unittest.TestCase):
def test_copy_meta(self):
req = Request("http://www.example.com")
req.meta['foo'] = 'bar'
r1 = self.response_class("http://www.example.com", body="Some body", request=req)
r1 = self.response_class("http://www.example.com", body=b"Some body", request=req)
assert r1.meta is req.meta
def test_copy_inherited_classes(self):
@ -79,30 +83,30 @@ class BaseResponseTest(unittest.TestCase):
"""Test Response.replace() method"""
hdrs = Headers({"key": "value"})
r1 = self.response_class("http://www.example.com")
r2 = r1.replace(status=301, body="New body", headers=hdrs)
assert r1.body == ''
r2 = r1.replace(status=301, body=b"New body", headers=hdrs)
assert r1.body == b''
self.assertEqual(r1.url, r2.url)
self.assertEqual((r1.status, r2.status), (200, 301))
self.assertEqual((r1.body, r2.body), ('', "New body"))
self.assertEqual((r1.body, r2.body), (b'', b"New body"))
self.assertEqual((r1.headers, r2.headers), ({}, hdrs))
# Empty attributes (which may fail if not compared properly)
r3 = self.response_class("http://www.example.com", flags=['cached'])
r4 = r3.replace(body='', flags=[])
self.assertEqual(r4.body, '')
r4 = r3.replace(body=b'', flags=[])
self.assertEqual(r4.body, b'')
self.assertEqual(r4.flags, [])
def _assert_response_values(self, response, encoding, body):
if isinstance(body, unicode):
if isinstance(body, six.text_type):
body_unicode = body
body_str = body.encode(encoding)
body_bytes = body.encode(encoding)
else:
body_unicode = body.decode(encoding)
body_str = body
body_bytes = body
assert isinstance(response.body, str)
assert isinstance(response.body, bytes)
self._assert_response_encoding(response, encoding)
self.assertEqual(response.body, body_str)
self.assertEqual(response.body, body_bytes)
self.assertEqual(response.body_as_unicode(), body_unicode)
def _assert_response_encoding(self, response, encoding):
@ -120,12 +124,6 @@ class BaseResponseTest(unittest.TestCase):
self.assertEqual(joined, absolute)
class ResponseText(BaseResponseTest):
def test_no_unicode_url(self):
self.assertRaises(TypeError, self.response_class, u'http://www.example.com')
class TextResponseTest(BaseResponseTest):
response_class = TextResponse
@ -152,11 +150,11 @@ class TextResponseTest(BaseResponseTest):
assert isinstance(resp.url, str)
resp = self.response_class(url=u"http://www.example.com/price/\xa3", encoding='utf-8')
self.assertEqual(resp.url, 'http://www.example.com/price/\xc2\xa3')
self.assertEqual(resp.url, to_native_str(b'http://www.example.com/price/\xc2\xa3'))
resp = self.response_class(url=u"http://www.example.com/price/\xa3", encoding='latin-1')
self.assertEqual(resp.url, 'http://www.example.com/price/\xa3')
resp = self.response_class(u"http://www.example.com/price/\xa3", headers={"Content-type": ["text/html; charset=utf-8"]})
self.assertEqual(resp.url, 'http://www.example.com/price/\xc2\xa3')
self.assertEqual(resp.url, to_native_str(b'http://www.example.com/price/\xc2\xa3'))
resp = self.response_class(u"http://www.example.com/price/\xa3", headers={"Content-type": ["text/html; charset=iso-8859-1"]})
self.assertEqual(resp.url, 'http://www.example.com/price/\xa3')
@ -168,17 +166,17 @@ class TextResponseTest(BaseResponseTest):
r1 = self.response_class('http://www.example.com', body=original_string, encoding='cp1251')
# check body_as_unicode
self.assertTrue(isinstance(r1.body_as_unicode(), unicode))
self.assertTrue(isinstance(r1.body_as_unicode(), six.text_type))
self.assertEqual(r1.body_as_unicode(), unicode_string)
def test_encoding(self):
r1 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=utf-8"]}, body="\xc2\xa3")
r1 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=utf-8"]}, body=b"\xc2\xa3")
r2 = self.response_class("http://www.example.com", encoding='utf-8', body=u"\xa3")
r3 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=iso-8859-1"]}, body="\xa3")
r4 = self.response_class("http://www.example.com", body="\xa2\xa3")
r5 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=None"]}, body="\xc2\xa3")
r6 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=gb2312"]}, body="\xa8D")
r7 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=gbk"]}, body="\xa8D")
r3 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=iso-8859-1"]}, body=b"\xa3")
r4 = self.response_class("http://www.example.com", body=b"\xa2\xa3")
r5 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=None"]}, body=b"\xc2\xa3")
r6 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=gb2312"]}, body=b"\xa8D")
r7 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=gbk"]}, body=b"\xa8D")
self.assertEqual(r1._headers_encoding(), "utf-8")
self.assertEqual(r2._headers_encoding(), None)
@ -203,21 +201,21 @@ class TextResponseTest(BaseResponseTest):
"""Check that unknown declared encodings are ignored"""
r = self.response_class("http://www.example.com",
headers={"Content-type": ["text/html; charset=UKNOWN"]},
body="\xc2\xa3")
body=b"\xc2\xa3")
self.assertEqual(r._declared_encoding(), None)
self._assert_response_values(r, 'utf-8', u"\xa3")
def test_utf16(self):
"""Test utf-16 because UnicodeDammit is known to have problems with"""
r = self.response_class("http://www.example.com",
body='\xff\xfeh\x00i\x00',
body=b'\xff\xfeh\x00i\x00',
encoding='utf-16')
self._assert_response_values(r, 'utf-16', u"hi")
def test_invalid_utf8_encoded_body_with_valid_utf8_BOM(self):
r6 = self.response_class("http://www.example.com",
headers={"Content-type": ["text/html; charset=utf-8"]},
body="\xef\xbb\xbfWORD\xe3\xab")
body=b"\xef\xbb\xbfWORD\xe3\xab")
self.assertEqual(r6.encoding, 'utf-8')
self.assertEqual(r6.body_as_unicode(), u'WORD\ufffd\ufffd')
@ -227,7 +225,7 @@ class TextResponseTest(BaseResponseTest):
# response.body_as_unicode() in indistint order doesn't affect final
# values for encoding and decoded body.
url = 'http://example.com'
body = "\xef\xbb\xbfWORD"
body = b"\xef\xbb\xbfWORD"
headers = {"Content-type": ["text/html; charset=utf-8"]}
# Test response without content-type and BOM encoding
@ -250,7 +248,7 @@ class TextResponseTest(BaseResponseTest):
def test_replace_wrong_encoding(self):
"""Test invalid chars are replaced properly"""
r = self.response_class("http://www.example.com", encoding='utf-8', body='PREFIX\xe3\xabSUFFIX')
r = self.response_class("http://www.example.com", encoding='utf-8', body=b'PREFIX\xe3\xabSUFFIX')
# XXX: Policy for replacing invalid chars may suffer minor variations
# but it should always contain the unicode replacement char (u'\ufffd')
assert u'\ufffd' in r.body_as_unicode(), repr(r.body_as_unicode())
@ -259,7 +257,7 @@ class TextResponseTest(BaseResponseTest):
# Do not destroy html tags due to encoding bugs
r = self.response_class("http://example.com", encoding='utf-8', \
body='\xf0<span>value</span>')
body=b'\xf0<span>value</span>')
assert u'<span>value</span>' in r.body_as_unicode(), repr(r.body_as_unicode())
# FIXME: This test should pass once we stop using BeautifulSoup's UnicodeDammit in TextResponse
@ -267,7 +265,7 @@ class TextResponseTest(BaseResponseTest):
#assert u'\ufffd' in r.body_as_unicode(), repr(r.body_as_unicode())
def test_selector(self):
body = "<html><head><title>Some page</title><body></body></html>"
body = b"<html><head><title>Some page</title><body></body></html>"
response = self.response_class("http://www.example.com", body=body)
self.assertIsInstance(response.selector, Selector)
@ -289,7 +287,7 @@ class TextResponseTest(BaseResponseTest):
)
def test_selector_shortcuts(self):
body = "<html><head><title>Some page</title><body></body></html>"
body = b"<html><head><title>Some page</title><body></body></html>"
response = self.response_class("http://www.example.com", body=body)
self.assertEqual(
@ -303,17 +301,17 @@ class TextResponseTest(BaseResponseTest):
def test_urljoin_with_base_url(self):
"""Test urljoin shortcut which also evaluates base-url through get_base_url()."""
body = '<html><body><base href="https://example.net"></body></html>'
body = b'<html><body><base href="https://example.net"></body></html>'
joined = self.response_class('http://www.example.com', body=body).urljoin('/test')
absolute = 'https://example.net/test'
self.assertEqual(joined, absolute)
body = '<html><body><base href="/elsewhere"></body></html>'
body = b'<html><body><base href="/elsewhere"></body></html>'
joined = self.response_class('http://www.example.com', body=body).urljoin('test')
absolute = 'http://www.example.com/test'
self.assertEqual(joined, absolute)
body = '<html><body><base href="/elsewhere/"></body></html>'
body = b'<html><body><base href="/elsewhere/"></body></html>'
joined = self.response_class('http://www.example.com', body=body).urljoin('test')
absolute = 'http://www.example.com/elsewhere/test'
self.assertEqual(joined, absolute)
@ -325,13 +323,13 @@ class HtmlResponseTest(TextResponseTest):
def test_html_encoding(self):
body = """<html><head><title>Some page</title><meta http-equiv="Content-Type" content="text/html; charset=iso-8859-1">
body = b"""<html><head><title>Some page</title><meta http-equiv="Content-Type" content="text/html; charset=iso-8859-1">
</head><body>Price: \xa3100</body></html>'
"""
r1 = self.response_class("http://www.example.com", body=body)
self._assert_response_values(r1, 'iso-8859-1', body)
body = """<?xml version="1.0" encoding="iso-8859-1"?>
body = b"""<?xml version="1.0" encoding="iso-8859-1"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN" "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
Price: \xa3100
"""
@ -339,19 +337,19 @@ class HtmlResponseTest(TextResponseTest):
self._assert_response_values(r2, 'iso-8859-1', body)
# for conflicting declarations headers must take precedence
body = """<html><head><title>Some page</title><meta http-equiv="Content-Type" content="text/html; charset=utf-8">
body = b"""<html><head><title>Some page</title><meta http-equiv="Content-Type" content="text/html; charset=utf-8">
</head><body>Price: \xa3100</body></html>'
"""
r3 = self.response_class("http://www.example.com", headers={"Content-type": ["text/html; charset=iso-8859-1"]}, body=body)
self._assert_response_values(r3, 'iso-8859-1', body)
# make sure replace() preserves the encoding of the original response
body = "New body \xa3"
body = b"New body \xa3"
r4 = r3.replace(body=body)
self._assert_response_values(r4, 'iso-8859-1', body)
def test_html5_meta_charset(self):
body = """<html><head><meta charset="gb2312" /><title>Some page</title><body>bla bla</body>"""
body = b"""<html><head><meta charset="gb2312" /><title>Some page</title><body>bla bla</body>"""
r1 = self.response_class("http://www.example.com", body=body)
self._assert_response_values(r1, 'gb2312', body)
@ -361,26 +359,25 @@ class XmlResponseTest(TextResponseTest):
response_class = XmlResponse
def test_xml_encoding(self):
body = "<xml></xml>"
body = b"<xml></xml>"
r1 = self.response_class("http://www.example.com", body=body)
self._assert_response_values(r1, self.response_class._DEFAULT_ENCODING, body)
body = """<?xml version="1.0" encoding="iso-8859-1"?><xml></xml>"""
body = b"""<?xml version="1.0" encoding="iso-8859-1"?><xml></xml>"""
r2 = self.response_class("http://www.example.com", body=body)
self._assert_response_values(r2, 'iso-8859-1', body)
# make sure replace() preserves the explicit encoding passed in the constructor
body = """<?xml version="1.0" encoding="iso-8859-1"?><xml></xml>"""
body = b"""<?xml version="1.0" encoding="iso-8859-1"?><xml></xml>"""
r3 = self.response_class("http://www.example.com", body=body, encoding='utf-8')
body2 = "New body"
body2 = b"New body"
r4 = r3.replace(body=body2)
self._assert_response_values(r4, 'utf-8', body2)
def test_replace_encoding(self):
# make sure replace() keeps the previous encoding unless overridden explicitly
body = """<?xml version="1.0" encoding="iso-8859-1"?><xml></xml>"""
body2 = """<?xml version="1.0" encoding="utf-8"?><xml></xml>"""
body = b"""<?xml version="1.0" encoding="iso-8859-1"?><xml></xml>"""
body2 = b"""<?xml version="1.0" encoding="utf-8"?><xml></xml>"""
r5 = self.response_class("http://www.example.com", body=body)
r6 = r5.replace(body=body2)
r7 = r5.replace(body=body2, encoding='utf-8')
@ -389,7 +386,7 @@ class XmlResponseTest(TextResponseTest):
self._assert_response_values(r7, 'utf-8', body2)
def test_selector(self):
body = '<?xml version="1.0" encoding="utf-8"?><xml><elem>value</elem></xml>'
body = b'<?xml version="1.0" encoding="utf-8"?><xml><elem>value</elem></xml>'
response = self.response_class("http://www.example.com", body=body)
self.assertIsInstance(response.selector, Selector)
@ -403,15 +400,10 @@ class XmlResponseTest(TextResponseTest):
)
def test_selector_shortcuts(self):
body = '<?xml version="1.0" encoding="utf-8"?><xml><elem>value</elem></xml>'
body = b'<?xml version="1.0" encoding="utf-8"?><xml><elem>value</elem></xml>'
response = self.response_class("http://www.example.com", body=body)
self.assertEqual(
response.xpath("//elem/text()").extract(),
response.selector.xpath("//elem/text()").extract(),
)
if __name__ == "__main__":
unittest.main()