mirror of https://github.com/scrapy/scrapy.git
518 lines
20 KiB
Python
518 lines
20 KiB
Python
import warnings
|
|
import xmlrpc.client
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from scrapy.http import Headers, Request, XmlRpcRequest
|
|
from scrapy.http.request import NO_CALLBACK
|
|
from scrapy.utils.python import to_bytes
|
|
|
|
|
|
class TestRequest:
|
|
request_class = Request
|
|
default_method = "GET"
|
|
default_headers: dict[bytes, list[bytes]] = {}
|
|
default_meta: dict[str, Any] = {}
|
|
|
|
def test_init(self):
|
|
# Request requires url in the __init__ method
|
|
with pytest.raises(TypeError):
|
|
self.request_class()
|
|
|
|
# url argument must be basestring
|
|
with pytest.raises(TypeError):
|
|
self.request_class(123)
|
|
|
|
# priority argument must be an integer
|
|
with pytest.raises(TypeError, match="Request priority not an integer"):
|
|
self.request_class("http://www.example.com", priority="1")
|
|
|
|
r = self.request_class("http://www.example.com")
|
|
assert isinstance(r.url, str)
|
|
assert r.url == "http://www.example.com"
|
|
assert r.method == self.default_method
|
|
|
|
assert isinstance(r.headers, Headers)
|
|
assert r.headers == self.default_headers
|
|
assert r.meta == self.default_meta
|
|
|
|
meta = {"lala": "lolo"}
|
|
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
|
|
assert r.meta == meta
|
|
assert r.headers is not headers
|
|
assert r.headers[b"caca"] == b"coco"
|
|
|
|
def test_url_scheme(self):
|
|
# This test passes by not raising any (ValueError) exception
|
|
self.request_class("http://example.org")
|
|
self.request_class("https://example.org")
|
|
self.request_class("s3://example.org")
|
|
self.request_class("ftp://example.org")
|
|
self.request_class("about:config")
|
|
self.request_class("data:,Hello%2C%20World!")
|
|
|
|
def test_url_no_scheme(self):
|
|
msg = "Missing scheme in request url:"
|
|
with pytest.raises(ValueError, match=msg):
|
|
self.request_class("foo")
|
|
with pytest.raises(ValueError, match=msg):
|
|
self.request_class("/foo/")
|
|
with pytest.raises(ValueError, match=msg):
|
|
self.request_class("/foo:bar")
|
|
|
|
def test_headers(self):
|
|
# Different ways of setting headers attribute
|
|
url = "http://www.scrapy.org"
|
|
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)
|
|
|
|
assert r.headers == p.headers
|
|
assert r.headers is not headers
|
|
assert p.headers is not r.headers
|
|
|
|
# headers must not be unicode
|
|
h = Headers({"key1": "val1", "key2": "val2"})
|
|
h["newkey"] = "newval"
|
|
for k, v in h.items():
|
|
assert isinstance(k, bytes)
|
|
for s in v:
|
|
assert isinstance(s, bytes)
|
|
|
|
def test_eq(self):
|
|
url = "http://www.scrapy.org"
|
|
r1 = self.request_class(url=url)
|
|
r2 = self.request_class(url=url)
|
|
assert r1 != r2
|
|
|
|
set_ = set()
|
|
set_.add(r1)
|
|
set_.add(r2)
|
|
assert len(set_) == 2
|
|
|
|
def test_url(self):
|
|
r = self.request_class(url="http://www.scrapy.org/path")
|
|
assert r.url == "http://www.scrapy.org/path"
|
|
|
|
def test_url_quoting(self):
|
|
r = self.request_class(url="http://www.scrapy.org/blank%20space")
|
|
assert r.url == "http://www.scrapy.org/blank%20space"
|
|
r = self.request_class(url="http://www.scrapy.org/blank space")
|
|
assert r.url == "http://www.scrapy.org/blank%20space"
|
|
|
|
def test_url_encoding(self):
|
|
r = self.request_class(url="http://www.scrapy.org/price/£")
|
|
assert r.url == "http://www.scrapy.org/price/%C2%A3"
|
|
|
|
def test_url_encoding_other(self):
|
|
# encoding affects only query part of URI, not path
|
|
# path part should always be UTF-8 encoded before percent-escaping
|
|
r = self.request_class(url="http://www.scrapy.org/price/£", encoding="utf-8")
|
|
assert r.url == "http://www.scrapy.org/price/%C2%A3"
|
|
|
|
r = self.request_class(url="http://www.scrapy.org/price/£", encoding="latin1")
|
|
assert r.url == "http://www.scrapy.org/price/%C2%A3"
|
|
|
|
def test_url_encoding_query(self):
|
|
r1 = self.request_class(url="http://www.scrapy.org/price/£?unit=µ")
|
|
assert r1.url == "http://www.scrapy.org/price/%C2%A3?unit=%C2%B5"
|
|
|
|
# should be same as above
|
|
r2 = self.request_class(
|
|
url="http://www.scrapy.org/price/£?unit=µ", encoding="utf-8"
|
|
)
|
|
assert r2.url == "http://www.scrapy.org/price/%C2%A3?unit=%C2%B5"
|
|
|
|
def test_url_encoding_query_latin1(self):
|
|
# encoding is used for encoding query-string before percent-escaping;
|
|
# path is still UTF-8 encoded before percent-escaping
|
|
r3 = self.request_class(
|
|
url="http://www.scrapy.org/price/µ?currency=£", encoding="latin1"
|
|
)
|
|
assert r3.url == "http://www.scrapy.org/price/%C2%B5?currency=%A3"
|
|
|
|
def test_url_encoding_nonutf8_untouched(self):
|
|
# percent-escaping sequences that do not match valid UTF-8 sequences
|
|
# should be kept untouched (just upper-cased perhaps)
|
|
#
|
|
# See https://datatracker.ietf.org/doc/html/rfc3987#section-3.2
|
|
#
|
|
# "Conversions from URIs to IRIs MUST NOT use any character encoding
|
|
# other than UTF-8 in steps 3 and 4, even if it might be possible to
|
|
# guess from the context that another character encoding than UTF-8 was
|
|
# used in the URI. For example, the URI
|
|
# "http://www.example.org/r%E9sum%E9.html" might with some guessing be
|
|
# interpreted to contain two e-acute characters encoded as iso-8859-1.
|
|
# It must not be converted to an IRI containing these e-acute
|
|
# characters. Otherwise, in the future the IRI will be mapped to
|
|
# "http://www.example.org/r%C3%A9sum%C3%A9.html", which is a different
|
|
# URI from "http://www.example.org/r%E9sum%E9.html".
|
|
r1 = self.request_class(url="http://www.scrapy.org/price/%a3")
|
|
assert r1.url == "http://www.scrapy.org/price/%a3"
|
|
|
|
r2 = self.request_class(url="http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3")
|
|
assert r2.url == "http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3"
|
|
|
|
r3 = self.request_class(url="http://www.scrapy.org/résumé/%a3")
|
|
assert r3.url == "http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3"
|
|
|
|
r4 = self.request_class(url="http://www.example.org/r%E9sum%E9.html")
|
|
assert r4.url == "http://www.example.org/r%E9sum%E9.html"
|
|
|
|
def test_url_verbatim(self):
|
|
r = self.request_class(
|
|
url="http://www.scrapy.org/price/£",
|
|
meta={"verbatim_url": True},
|
|
)
|
|
assert r.url == "http://www.scrapy.org/price/£"
|
|
|
|
r = self.request_class(
|
|
url="http://www.scrapy.org/blank space",
|
|
meta={"verbatim_url": True},
|
|
)
|
|
assert r.url == "http://www.scrapy.org/blank space"
|
|
|
|
def test_body(self):
|
|
r1 = self.request_class(url="http://www.example.com/")
|
|
assert r1.body == b""
|
|
|
|
r2 = self.request_class(url="http://www.example.com/", body=b"")
|
|
assert isinstance(r2.body, bytes)
|
|
assert r2.encoding == "utf-8" # default encoding
|
|
|
|
r3 = self.request_class(
|
|
url="http://www.example.com/", body="Price: \xa3100", encoding="utf-8"
|
|
)
|
|
assert isinstance(r3.body, bytes)
|
|
assert r3.body == b"Price: \xc2\xa3100"
|
|
|
|
r4 = self.request_class(
|
|
url="http://www.example.com/", body="Price: \xa3100", encoding="latin1"
|
|
)
|
|
assert isinstance(r4.body, bytes)
|
|
assert r4.body == b"Price: \xa3100"
|
|
|
|
def test_copy(self):
|
|
"""Test Request copy"""
|
|
|
|
def somecallback():
|
|
pass
|
|
|
|
r1 = self.request_class(
|
|
"http://www.example.com",
|
|
flags=["f1", "f2"],
|
|
callback=somecallback,
|
|
errback=somecallback,
|
|
)
|
|
r1.meta["foo"] = "bar"
|
|
r1.cb_kwargs["key"] = "value"
|
|
r2 = r1.copy()
|
|
|
|
# make sure callbaclks are copied
|
|
assert r1.callback is somecallback
|
|
assert r1.errback is somecallback
|
|
assert r2.callback is r1.callback
|
|
assert r2.errback is r1.errback
|
|
|
|
# make sure flags list is shallow copied
|
|
assert r1.flags is not r2.flags, "flags must be a shallow copy, not identical"
|
|
assert r1.flags == r2.flags
|
|
|
|
# make sure cb_kwargs dict is shallow copied
|
|
assert r1.cb_kwargs is not r2.cb_kwargs, (
|
|
"cb_kwargs must be a shallow copy, not identical"
|
|
)
|
|
assert r1.cb_kwargs == r2.cb_kwargs
|
|
|
|
# make sure meta dict is shallow copied
|
|
assert r1.meta is not r2.meta, "meta must be a shallow copy, not identical"
|
|
assert r1.meta == r2.meta
|
|
|
|
# make sure headers attribute is shallow copied
|
|
assert r1.headers is not r2.headers, (
|
|
"headers must be a shallow copy, not identical"
|
|
)
|
|
assert r1.headers == r2.headers
|
|
assert r1.encoding == r2.encoding
|
|
assert r1.dont_filter == r2.dont_filter
|
|
|
|
# Request.body can be identical since it's an immutable object (str)
|
|
|
|
def test_copy_inherited_classes(self):
|
|
"""Test Request children copies preserve their class"""
|
|
|
|
class CustomRequest(self.request_class):
|
|
pass
|
|
|
|
r1 = CustomRequest("http://www.example.com")
|
|
r2 = r1.copy()
|
|
|
|
assert isinstance(r2, CustomRequest)
|
|
|
|
def test_replace(self):
|
|
"""Test Request.replace() method"""
|
|
r1 = self.request_class("http://www.example.com", method="GET")
|
|
hdrs = Headers(r1.headers)
|
|
hdrs[b"key"] = b"value"
|
|
r2 = r1.replace(method="POST", body="New body", headers=hdrs)
|
|
assert r1.url == r2.url
|
|
assert (r1.method, r2.method) == ("GET", "POST")
|
|
assert (r1.body, r2.body) == (b"", b"New body")
|
|
assert (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=b"", meta={}, dont_filter=False
|
|
)
|
|
assert r4.url == "http://www.example.com/2"
|
|
assert r4.body == b""
|
|
assert r4.meta == {}
|
|
assert r4.dont_filter is False
|
|
|
|
# the cls argument allows changing the resulting class
|
|
custom_request_cls = type("CustomRequest", (self.request_class,), {})
|
|
r5 = r1.replace(cls=custom_request_cls)
|
|
assert isinstance(r5, custom_request_cls)
|
|
assert r5.url == r1.url
|
|
|
|
def test_method_always_str(self):
|
|
r = self.request_class("http://www.example.com", method="POST")
|
|
assert isinstance(r.method, str)
|
|
|
|
def test_immutable_attributes(self):
|
|
r = self.request_class("http://example.com")
|
|
with pytest.raises(AttributeError):
|
|
r.url = "http://example2.com"
|
|
with pytest.raises(AttributeError):
|
|
r.body = "xxx"
|
|
|
|
def test_callback_and_errback(self):
|
|
def a_function():
|
|
pass
|
|
|
|
r1 = self.request_class("http://example.com")
|
|
assert r1.callback is None
|
|
assert r1.errback is None
|
|
|
|
r2 = self.request_class("http://example.com", callback=a_function)
|
|
assert r2.callback is a_function
|
|
assert r2.errback is None
|
|
|
|
r3 = self.request_class("http://example.com", errback=a_function)
|
|
assert r3.callback is None
|
|
assert r3.errback is a_function
|
|
|
|
r4 = self.request_class(
|
|
url="http://example.com",
|
|
callback=a_function,
|
|
errback=a_function,
|
|
)
|
|
assert r4.callback is a_function
|
|
assert r4.errback is a_function
|
|
|
|
r5 = self.request_class(
|
|
url="http://example.com",
|
|
callback=NO_CALLBACK,
|
|
errback=NO_CALLBACK,
|
|
)
|
|
assert r5.callback is NO_CALLBACK
|
|
assert r5.errback is NO_CALLBACK
|
|
|
|
def test_callback_and_errback_type(self):
|
|
with pytest.raises(TypeError):
|
|
self.request_class("http://example.com", callback="a_function")
|
|
with pytest.raises(TypeError):
|
|
self.request_class("http://example.com", errback="a_function")
|
|
with pytest.raises(TypeError):
|
|
self.request_class(
|
|
url="http://example.com",
|
|
callback="a_function",
|
|
errback="a_function",
|
|
)
|
|
|
|
def test_setters(self):
|
|
request = self.request_class("http://example.com")
|
|
|
|
request.flags = ["f1"]
|
|
assert request.flags == ["f1"]
|
|
|
|
request.cookies = {"sid": "1"}
|
|
assert request.cookies == {"sid": "1"}
|
|
|
|
headers = Headers({b"X-Test": b"1"})
|
|
request.headers = headers
|
|
assert request._headers is headers
|
|
request.headers = {b"A": b"b"}
|
|
assert isinstance(request.headers, Headers)
|
|
assert request._headers[b"A"] == b"b"
|
|
|
|
def test_setter_mutable_lazy_loading(self):
|
|
"""Mutable attributes are set internally to None only until they are
|
|
read, then they always return the same falsy instance of the
|
|
corresponding mutable structure.
|
|
|
|
Setting them to None causes the next read to return a different object.
|
|
"""
|
|
|
|
request = self.request_class("http://example.com")
|
|
|
|
assert request._flags is None
|
|
assert request.flags == []
|
|
assert request.flags is request.flags
|
|
assert request._flags == []
|
|
original_flags = request.flags
|
|
request.flags = None
|
|
assert request._flags is None
|
|
assert request.flags == []
|
|
assert request.flags is not original_flags
|
|
|
|
assert request._cookies is None
|
|
assert request.cookies == {}
|
|
assert request.cookies is request.cookies
|
|
assert request._cookies == {}
|
|
original_cookies = request.cookies
|
|
request.cookies = None
|
|
assert request._cookies is None
|
|
assert request.cookies == {}
|
|
assert request.cookies is not original_cookies
|
|
|
|
if self.default_headers:
|
|
assert request._headers == self.default_headers
|
|
assert request._headers is not self.default_headers
|
|
assert request.headers == self.default_headers
|
|
else:
|
|
assert request._headers is None
|
|
assert request.headers == {}
|
|
assert request.headers is request.headers
|
|
assert isinstance(request.headers, Headers)
|
|
assert isinstance(request._headers, Headers)
|
|
original_headers = request.headers
|
|
request.headers = None
|
|
assert request._headers is None
|
|
assert request.headers == {}
|
|
assert request._headers == {}
|
|
assert request.headers is not original_headers
|
|
|
|
def test_no_callback(self):
|
|
with pytest.raises(RuntimeError):
|
|
NO_CALLBACK()
|
|
|
|
def test_from_curl(self):
|
|
# Note: more curated tests regarding curl conversion are in
|
|
# `test_utils_curl.py`
|
|
curl_command = (
|
|
"curl 'http://httpbin.org/post' -X POST -H 'Cookie: _gauges_unique"
|
|
"_year=1; _gauges_unique=1; _gauges_unique_month=1; _gauges_unique"
|
|
"_hour=1; _gauges_unique_day=1' -H 'Origin: http://httpbin.org' -H"
|
|
" 'Accept-Encoding: gzip, deflate' -H 'Accept-Language: en-US,en;q"
|
|
"=0.9,ru;q=0.8,es;q=0.7' -H 'Upgrade-Insecure-Requests: 1' -H 'Use"
|
|
"r-Agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTM"
|
|
"L, like Gecko) Ubuntu Chromium/62.0.3202.75 Chrome/62.0.3202.75 S"
|
|
"afari/537.36' -H 'Content-Type: application /x-www-form-urlencode"
|
|
"d' -H 'Accept: text/html,application/xhtml+xml,application/xml;q="
|
|
"0.9,image/webp,image/apng,*/*;q=0.8' -H 'Cache-Control: max-age=0"
|
|
"' -H 'Referer: http://httpbin.org/forms/post' -H 'Connection: kee"
|
|
"p-alive' --data 'custname=John+Smith&custtel=500&custemail=jsmith"
|
|
"%40example.org&size=small&topping=cheese&topping=onion&delivery=1"
|
|
"2%3A15&comments=' --compressed"
|
|
)
|
|
r = self.request_class.from_curl(curl_command)
|
|
assert r.method == "POST"
|
|
assert r.url == "http://httpbin.org/post"
|
|
assert (
|
|
r.body == b"custname=John+Smith&custtel=500&custemail=jsmith%40"
|
|
b"example.org&size=small&topping=cheese&topping=onion"
|
|
b"&delivery=12%3A15&comments="
|
|
)
|
|
assert r.cookies == {
|
|
"_gauges_unique_year": "1",
|
|
"_gauges_unique": "1",
|
|
"_gauges_unique_month": "1",
|
|
"_gauges_unique_hour": "1",
|
|
"_gauges_unique_day": "1",
|
|
}
|
|
assert r.headers == {
|
|
b"Origin": [b"http://httpbin.org"],
|
|
b"Accept-Encoding": [b"gzip, deflate"],
|
|
b"Accept-Language": [b"en-US,en;q=0.9,ru;q=0.8,es;q=0.7"],
|
|
b"Upgrade-Insecure-Requests": [b"1"],
|
|
b"User-Agent": [
|
|
b"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537."
|
|
b"36 (KHTML, like Gecko) Ubuntu Chromium/62.0.3202"
|
|
b".75 Chrome/62.0.3202.75 Safari/537.36"
|
|
],
|
|
b"Content-Type": [b"application /x-www-form-urlencoded"],
|
|
b"Accept": [
|
|
b"text/html,application/xhtml+xml,application/xml;q=0."
|
|
b"9,image/webp,image/apng,*/*;q=0.8"
|
|
],
|
|
b"Cache-Control": [b"max-age=0"],
|
|
b"Referer": [b"http://httpbin.org/forms/post"],
|
|
b"Connection": [b"keep-alive"],
|
|
}
|
|
|
|
def test_from_curl_with_kwargs(self):
|
|
r = self.request_class.from_curl(
|
|
'curl -X PATCH "http://example.org"', method="POST", meta={"key": "value"}
|
|
)
|
|
assert r.method == "POST"
|
|
assert r.meta == {"key": "value"}
|
|
|
|
def test_from_curl_ignore_unknown_options(self):
|
|
# By default: it works and ignores the unknown options: --foo and -z
|
|
with warnings.catch_warnings(): # avoid warning when executing tests
|
|
warnings.filterwarnings(
|
|
"ignore", category=UserWarning, message="Unrecognized options:"
|
|
)
|
|
r = self.request_class.from_curl(
|
|
'curl -X DELETE "http://example.org" --foo -z',
|
|
)
|
|
assert r.method == "DELETE"
|
|
|
|
# If `ignore_unknown_options` is set to `False` it raises an error with
|
|
# the unknown options: --foo and -z
|
|
with pytest.raises(ValueError, match="Unrecognized options:"):
|
|
self.request_class.from_curl(
|
|
'curl -X PATCH "http://example.org" --foo -z',
|
|
ignore_unknown_options=False,
|
|
)
|
|
|
|
|
|
class TestXmlRpcRequest(TestRequest):
|
|
request_class = XmlRpcRequest
|
|
default_method = "POST"
|
|
default_headers = {b"Content-Type": [b"text/xml"]}
|
|
|
|
def _test_request(self, **kwargs):
|
|
r = self.request_class("http://scrapytest.org/rpc2", **kwargs)
|
|
assert r.headers[b"Content-Type"] == b"text/xml"
|
|
assert r.body == to_bytes(
|
|
xmlrpc.client.dumps(**kwargs), encoding=kwargs.get("encoding", "utf-8")
|
|
)
|
|
assert r.method == "POST"
|
|
assert r.encoding == kwargs.get("encoding", "utf-8")
|
|
assert r.dont_filter
|
|
|
|
def test_xmlrpc_dumps(self):
|
|
self._test_request(params=("value",))
|
|
self._test_request(params=("username", "password"), methodname="login")
|
|
self._test_request(params=("response",), methodresponse="login")
|
|
self._test_request(params=("pas£",), encoding="utf-8")
|
|
self._test_request(params=(None,), allow_none=1)
|
|
with pytest.raises(TypeError):
|
|
self._test_request()
|
|
with pytest.raises(TypeError):
|
|
self._test_request(params=(None,))
|
|
|
|
def test_latin1(self):
|
|
self._test_request(params=("pas£",), encoding="latin1")
|