Merge pull request #1396 from scrapy/py3-utils-response

PY3 port scrapy.utils.response
This commit is contained in:
Daniel Graña 2015-07-31 17:53:53 -03:00
commit 71c6bd920e
4 changed files with 50 additions and 31 deletions

View File

@ -71,7 +71,7 @@ def request_authenticate(request, username, password):
def request_httprepr(request):
"""Return the raw HTTP representation (as string) of the given request.
"""Return the raw HTTP representation (as bytes) of the given request.
This is provided only for reference since it's not the actual stream of
bytes that will be send when performing the request (that's controlled
by Twisted).

View File

@ -2,7 +2,6 @@
This module provides some useful functions for working with
scrapy.http.Response objects
"""
import os
import re
import weakref
@ -11,6 +10,7 @@ import tempfile
from twisted.web import http
from twisted.web.http import RESPONSES
from scrapy.utils.python import to_bytes
from w3lib import html
from scrapy.utils.decorators import deprecated
@ -27,10 +27,11 @@ def get_base_url(response):
"""Return the base url of the given response, joined with the response url"""
if response not in _baseurl_cache:
text = response.body_as_unicode()[0:4096]
_baseurl_cache[response] = html.get_base_url(text, response.url, \
_baseurl_cache[response] = html.get_base_url(text, response.url,
response.encoding)
return _baseurl_cache[response]
_noscript_re = re.compile(u'<noscript>.*?</noscript>', re.IGNORECASE | re.DOTALL)
_script_re = re.compile(u'<script.*?>.*?</script>', re.IGNORECASE | re.DOTALL)
_metaref_cache = weakref.WeakKeyDictionary()
@ -40,10 +41,11 @@ def get_meta_refresh(response):
text = response.body_as_unicode()[0:4096]
text = _noscript_re.sub(u'', text)
text = _script_re.sub(u'', text)
_metaref_cache[response] = html.get_meta_refresh(text, response.url, \
_metaref_cache[response] = html.get_meta_refresh(text, response.url,
response.encoding)
return _metaref_cache[response]
def response_status_message(status):
"""Return status code plus status text descriptive message
@ -55,19 +57,21 @@ def response_status_message(status):
"""
return '%s %s' % (status, http.responses.get(int(status)))
def response_httprepr(response):
"""Return raw HTTP representation (as string) of the given response. This
"""Return raw HTTP representation (as bytes) of the given response. This
is provided only for reference, since it's not the exact stream of bytes
that was received (that's not exposed by Twisted).
"""
s = "HTTP/1.1 %d %s\r\n" % (response.status, RESPONSES.get(response.status, ''))
s = b"HTTP/1.1 " + to_bytes(str(response.status)) + b" " + \
to_bytes(RESPONSES.get(response.status, b'')) + b"\r\n"
if response.headers:
s += response.headers.to_string() + "\r\n"
s += "\r\n"
s += response.headers.to_string() + b"\r\n"
s += b"\r\n"
s += response.body
return s
def open_in_browser(response, _openfunc=webbrowser.open):
"""Open the given response in a local web browser, populating the <base>
tag for external links to work
@ -76,14 +80,15 @@ def open_in_browser(response, _openfunc=webbrowser.open):
# XXX: this implementation is a bit dirty and could be improved
body = response.body
if isinstance(response, HtmlResponse):
if '<base' not in body:
body = body.replace('<head>', '<head><base href="%s">' % response.url)
if b'<base' not in body:
repl = '<head><base href="%s">' % response.url
body = body.replace(b'<head>', to_bytes(repl))
ext = '.html'
elif isinstance(response, TextResponse):
ext = '.txt'
else:
raise TypeError("Unsupported response type: %s" % \
response.__class__.__name__)
raise TypeError("Unsupported response type: %s" %
response.__class__.__name__)
fd, fname = tempfile.mkstemp(ext)
os.write(fd, body)
os.close(fd)

View File

@ -42,7 +42,6 @@ tests/test_stats.py
tests/test_utils_iterators.py
tests/test_utils_log.py
tests/test_utils_reqser.py
tests/test_utils_response.py
tests/test_utils_template.py
tests/test_webclient.py

View File

@ -3,53 +3,59 @@ import unittest
from six.moves.urllib.parse import urlparse
from scrapy.http import Response, TextResponse, HtmlResponse
from scrapy.utils.response import response_httprepr, open_in_browser, get_meta_refresh
from scrapy.utils.python import to_bytes
from scrapy.utils.response import (response_httprepr, open_in_browser,
get_meta_refresh, get_base_url)
__doctests__ = ['scrapy.utils.response']
class ResponseUtilsTest(unittest.TestCase):
dummy_response = TextResponse(url='http://example.org/', body='dummy_response')
dummy_response = TextResponse(url='http://example.org/', body=b'dummy_response')
def test_response_httprepr(self):
r1 = Response("http://www.example.com")
self.assertEqual(response_httprepr(r1), 'HTTP/1.1 200 OK\r\n\r\n')
self.assertEqual(response_httprepr(r1), b'HTTP/1.1 200 OK\r\n\r\n')
r1 = Response("http://www.example.com", status=404, headers={"Content-type": "text/html"}, body="Some body")
self.assertEqual(response_httprepr(r1), 'HTTP/1.1 404 Not Found\r\nContent-Type: text/html\r\n\r\nSome body')
r1 = Response("http://www.example.com", status=404, headers={"Content-type": "text/html"}, body=b"Some body")
self.assertEqual(response_httprepr(r1), b'HTTP/1.1 404 Not Found\r\nContent-Type: text/html\r\n\r\nSome body')
r1 = Response("http://www.example.com", status=6666, headers={"Content-type": "text/html"}, body="Some body")
self.assertEqual(response_httprepr(r1), 'HTTP/1.1 6666 \r\nContent-Type: text/html\r\n\r\nSome body')
r1 = Response("http://www.example.com", status=6666, headers={"Content-type": "text/html"}, body=b"Some body")
self.assertEqual(response_httprepr(r1), b'HTTP/1.1 6666 \r\nContent-Type: text/html\r\n\r\nSome body')
def test_open_in_browser(self):
url = "http:///www.example.com/some/page.html"
body = "<html> <head> <title>test page</title> </head> <body>test body</body> </html>"
body = b"<html> <head> <title>test page</title> </head> <body>test body</body> </html>"
def browser_open(burl):
path = urlparse(burl).path
if not os.path.exists(path):
path = burl.replace('file://', '')
bbody = open(path).read()
assert '<base href="%s">' % url in bbody, "<base> tag not added"
with open(path, "rb") as f:
bbody = f.read()
self.assertIn(b'<base href="' + to_bytes(url) + b'">', bbody)
return True
response = HtmlResponse(url, body=body)
assert open_in_browser(response, _openfunc=browser_open), \
"Browser not called"
self.assertRaises(TypeError, open_in_browser, Response(url, body=body), \
debug=True)
resp = Response(url, body=body)
self.assertRaises(TypeError, open_in_browser, resp, debug=True)
def test_get_meta_refresh(self):
r1 = HtmlResponse("http://www.example.com", body="""
r1 = HtmlResponse("http://www.example.com", body=b"""
<html>
<head><title>Dummy</title><meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head>
<body>blahablsdfsal&amp;</body>
</html>""")
r2 = HtmlResponse("http://www.example.com", body="""
r2 = HtmlResponse("http://www.example.com", body=b"""
<html>
<head><title>Dummy</title><noScript>
<meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head>
</noSCRIPT>
<body>blahablsdfsal&amp;</body>
</html>""")
r3 = HtmlResponse("http://www.example.com", body="""
r3 = HtmlResponse("http://www.example.com", body=b"""
<noscript><meta http-equiv="REFRESH" content="0;url=http://www.example.com/newpage</noscript>
<script type="text/javascript">
if(!checkCookies()){
@ -61,5 +67,14 @@ class ResponseUtilsTest(unittest.TestCase):
self.assertEqual(get_meta_refresh(r2), (None, None))
self.assertEqual(get_meta_refresh(r3), (None, None))
if __name__ == "__main__":
unittest.main()
def test_get_base_url(self):
resp = HtmlResponse("http://www.example.com", body=b"""
<html>
<head><base href="http://www.example.com/img/" target="_blank"></head>
<body>blahablsdfsal&amp;</body>
</html>""")
self.assertEqual(get_base_url(resp), "http://www.example.com/img/")
resp2 = HtmlResponse("http://www.example.com", body=b"""
<html><body>blahablsdfsal&amp;</body></html>""")
self.assertEqual(get_base_url(resp2), "http://www.example.com")