import json import re import warnings import xmlrpc.client from typing import Any from unittest import mock from urllib.parse import parse_qs, unquote_to_bytes import pytest from scrapy.http import ( FormRequest, Headers, HtmlResponse, JsonRequest, Request, XmlRpcRequest, ) from scrapy.http.request import NO_CALLBACK from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.python import to_bytes, to_unicode 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) r = self.request_class("http://www.example.com") 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_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 copy does not propagate callbacks assert r1.callback is somecallback assert r1.errback is somecallback assert r2.callback is r1.callback assert r2.errback is r2.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 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.cb_kwargs = {"a": 1} assert request.cb_kwargs == {"a": 1} request.meta = {"k": "v"} assert request.meta == {"k": "v"} 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._cb_kwargs is None assert request.cb_kwargs == {} assert request.cb_kwargs is request.cb_kwargs assert request._cb_kwargs == {} original_cb_kwargs = request.cb_kwargs request.cb_kwargs = None assert request.cb_kwargs == {} assert request.cb_kwargs is not original_cb_kwargs assert request._meta is None assert request.meta == {} assert request.meta is request.meta assert request._meta == {} original_meta = request.meta request.meta = None assert request.meta == {} assert request.meta is not original_meta 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 == [] 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 == {} 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 == {} 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.simplefilter("ignore") 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 TestFormRequest(TestRequest): request_class = FormRequest def assertQueryEqual(self, first, second, msg=None): first = to_unicode(first).split("&") second = to_unicode(second).split("&") assert sorted(first) == sorted(second), msg def test_empty_formdata(self): r1 = self.request_class("http://www.example.com", formdata={}) assert r1.body == b"" def test_formdata_overrides_querystring(self): data = (("a", "one"), ("a", "two"), ("b", "2")) url = self.request_class( "http://www.example.com/?a=0&b=1&c=3#fragment", method="GET", formdata=data ).url.split("#", maxsplit=1)[0] fs = _qs(self.request_class(url, method="GET", formdata=data)) assert set(fs[b"a"]) == {b"one", b"two"} assert fs[b"b"] == [b"2"] assert fs.get(b"c") is None data = {"a": "1", "b": "2"} fs = _qs( self.request_class("http://www.example.com/", method="GET", formdata=data) ) assert fs[b"a"] == [b"1"] assert fs[b"b"] == [b"2"] def test_default_encoding_bytes(self): # using default encoding (utf-8) data = {b"one": b"two", b"price": b"\xc2\xa3 100"} r2 = self.request_class("http://www.example.com", formdata=data) assert r2.method == "POST" assert r2.encoding == "utf-8" self.assertQueryEqual(r2.body, b"price=%C2%A3+100&one=two") assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_default_encoding_textual_data(self): # using default encoding (utf-8) data = {"µ one": "two", "price": "£ 100"} r2 = self.request_class("http://www.example.com", formdata=data) assert r2.method == "POST" assert r2.encoding == "utf-8" self.assertQueryEqual(r2.body, b"price=%C2%A3+100&%C2%B5+one=two") assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_default_encoding_mixed_data(self): # using default encoding (utf-8) data = {"\u00b5one": b"two", b"price\xc2\xa3": "\u00a3 100"} r2 = self.request_class("http://www.example.com", formdata=data) assert r2.method == "POST" assert r2.encoding == "utf-8" self.assertQueryEqual(r2.body, b"%C2%B5one=two&price%C2%A3=%C2%A3+100") assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_custom_encoding_bytes(self): data = {b"\xb5 one": b"two", b"price": b"\xa3 100"} r2 = self.request_class( "http://www.example.com", formdata=data, encoding="latin1" ) assert r2.method == "POST" assert r2.encoding == "latin1" self.assertQueryEqual(r2.body, b"price=%A3+100&%B5+one=two") assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_custom_encoding_textual_data(self): data = {"price": "£ 100"} r3 = self.request_class( "http://www.example.com", formdata=data, encoding="latin1" ) assert r3.encoding == "latin1" assert r3.body == b"price=%A3+100" def test_multi_key_values(self): # using multiples values for a single key data = {"price": "\xa3 100", "colours": ["red", "blue", "green"]} r3 = self.request_class("http://www.example.com", formdata=data) self.assertQueryEqual( r3.body, b"colours=red&colours=blue&colours=green&price=%C2%A3+100" ) def test_from_response_post(self): response = _buildresponse( b"""
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response( response, formdata={"one": ["two", "three"], "six": "seven"} ) assert req.method == "POST" assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req) assert set(fs[b"test"]) == {b"val1", b"val2"} assert set(fs[b"one"]) == {b"two", b"three"} assert fs[b"test2"] == [b"xxx"] assert fs[b"six"] == [b"seven"] def test_from_response_post_nonascii_bytes_utf8(self): response = _buildresponse( b"""
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response( response, formdata={"one": ["two", "three"], "six": "seven"} ) assert req.method == "POST" assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req, to_unicode=True) assert set(fs["test £"]) == {"val1", "val2"} assert set(fs["one"]) == {"two", "three"} assert fs["test2"] == ["xxx µ"] assert fs["six"] == ["seven"] def test_from_response_post_nonascii_bytes_latin1(self): response = _buildresponse( b"""
""", url="http://www.example.com/this/list.html", encoding="latin1", ) req = self.request_class.from_response( response, formdata={"one": ["two", "three"], "six": "seven"} ) assert req.method == "POST" assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req, to_unicode=True, encoding="latin1") assert set(fs["test £"]) == {"val1", "val2"} assert set(fs["one"]) == {"two", "three"} assert fs["test2"] == ["xxx µ"] assert fs["six"] == ["seven"] def test_from_response_post_nonascii_unicode(self): response = _buildresponse( """
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response( response, formdata={"one": ["two", "three"], "six": "seven"} ) assert req.method == "POST" assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req, to_unicode=True) assert set(fs["test £"]) == {"val1", "val2"} assert set(fs["one"]) == {"two", "three"} assert fs["test2"] == ["xxx µ"] assert fs["six"] == ["seven"] def test_from_response_duplicate_form_key(self): response = _buildresponse("
", url="http://www.example.com") req = self.request_class.from_response( response=response, method="GET", formdata=(("foo", "bar"), ("foo", "baz")), ) assert urlparse_cached(req).hostname == "www.example.com" assert urlparse_cached(req).query == "foo=bar&foo=baz" def test_from_response_override_duplicate_form_key(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response, formdata=(("two", "2"), ("two", "4")) ) fs = _qs(req) assert fs[b"one"] == [b"1"] assert fs[b"two"] == [b"2", b"4"] def test_from_response_extra_headers(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response=response, formdata={"one": ["two", "three"], "six": "seven"}, headers={"Accept-Encoding": "gzip,deflate"}, ) assert req.method == "POST" assert req.headers["Content-type"] == b"application/x-www-form-urlencoded" assert req.headers["Accept-Encoding"] == b"gzip,deflate" def test_from_response_get(self): response = _buildresponse( """
""", url="http://www.example.com/this/list.html", ) r1 = self.request_class.from_response( response, formdata={"one": ["two", "three"], "six": "seven"} ) assert r1.method == "GET" assert urlparse_cached(r1).hostname == "www.example.com" assert urlparse_cached(r1).path == "/this/get.php" fs = _qs(r1) assert set(fs[b"test"]) == {b"val1", b"val2"} assert set(fs[b"one"]) == {b"two", b"three"} assert fs[b"test2"] == [b"xxx"] assert fs[b"six"] == [b"seven"] def test_from_response_override_params(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response, formdata={"two": "2"}) fs = _qs(req) assert fs[b"one"] == [b"1"] assert fs[b"two"] == [b"2"] def test_from_response_drop_params(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response, formdata={"two": None}) fs = _qs(req) assert fs[b"one"] == [b"1"] assert b"two" not in fs def test_from_response_override_method(self): response = _buildresponse( """
""" ) request = FormRequest.from_response(response) assert request.method == "GET" request = FormRequest.from_response(response, method="POST") assert request.method == "POST" def test_from_response_override_url(self): response = _buildresponse( """
""" ) request = FormRequest.from_response(response) assert request.url == "http://example.com/app" request = FormRequest.from_response(response, url="http://foo.bar/absolute") assert request.url == "http://foo.bar/absolute" request = FormRequest.from_response(response, url="/relative") assert request.url == "http://example.com/relative" def test_from_response_case_insensitive(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response) fs = _qs(req) assert fs[b"clickable1"] == [b"clicked1"] assert b"i1" not in fs, fs # xpath in _get_inputs() assert b"clickable2" not in fs, fs # xpath in _get_clickable() def test_from_response_submit_first_clickable(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response, formdata={"two": "2"}) fs = _qs(req) assert fs[b"clickable1"] == [b"clicked1"] assert b"clickable2" not in fs, fs assert fs[b"one"] == [b"1"] assert fs[b"two"] == [b"2"] def test_from_response_submit_not_first_clickable(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response, formdata={"two": "2"}, clickdata={"name": "clickable2"} ) fs = _qs(req) assert fs[b"clickable2"] == [b"clicked2"] assert b"clickable1" not in fs, fs assert fs[b"one"] == [b"1"] assert fs[b"two"] == [b"2"] def test_from_response_dont_submit_image_as_input(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response, dont_click=True) fs = _qs(req) assert fs == {b"i1": [b"i1v"]} def test_from_response_dont_submit_reset_as_input(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response, dont_click=True) fs = _qs(req) assert fs == {b"i1": [b"i1v"], b"i2": [b"i2v"]} def test_from_response_clickdata_does_not_ignore_image(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response) fs = _qs(req) assert fs == {b"i1": [b"i1v"], b"i2": [b"i2v"]} def test_from_response_multiple_clickdata(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response, clickdata={"name": "clickable", "value": "clicked2"} ) fs = _qs(req) assert fs[b"clickable"] == [b"clicked2"] assert fs[b"one"] == [b"clicked1"] assert fs[b"two"] == [b"clicked2"] def test_from_response_unicode_clickdata(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response, clickdata={"name": "price in \u00a3"} ) fs = _qs(req, to_unicode=True) assert fs["price in \u00a3"] def test_from_response_unicode_clickdata_latin1(self): response = _buildresponse( """
""", encoding="latin1", ) req = self.request_class.from_response( response, clickdata={"name": "price in \u00a5"} ) fs = _qs(req, to_unicode=True, encoding="latin1") assert fs["price in \u00a5"] def test_from_response_multiple_forms_clickdata(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response, formname="form2", clickdata={"name": "clickable"} ) fs = _qs(req) assert fs[b"clickable"] == [b"clicked2"] assert fs[b"field2"] == [b"value2"] assert b"field1" not in fs, fs def test_from_response_override_clickable(self): response = _buildresponse( """
""" ) req = self.request_class.from_response( response, formdata={"clickme": "two"}, clickdata={"name": "clickme"} ) fs = _qs(req) assert fs[b"clickme"] == [b"two"] def test_from_response_dont_click(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response(response, dont_click=True) fs = _qs(r1) assert b"clickable1" not in fs, fs assert b"clickable2" not in fs, fs def test_from_response_ambiguous_clickdata(self): response = _buildresponse( """
""" ) with pytest.raises( ValueError, match=r"Multiple elements found .* matching the criteria in clickdata", ): self.request_class.from_response(response, clickdata={"type": "submit"}) def test_from_response_non_matching_clickdata(self): response = _buildresponse( """
""" ) with pytest.raises( ValueError, match="No clickable element matching clickdata:" ): self.request_class.from_response( response, clickdata={"nonexistent": "notme"} ) def test_from_response_nr_index_clickdata(self): response = _buildresponse( """
""" ) req = self.request_class.from_response(response, clickdata={"nr": 1}) fs = _qs(req) assert b"clickable2" in fs assert b"clickable1" not in fs def test_from_response_invalid_nr_index_clickdata(self): response = _buildresponse( """
""" ) with pytest.raises( ValueError, match="No clickable element matching clickdata:" ): self.request_class.from_response(response, clickdata={"nr": 1}) def test_from_response_errors_noform(self): response = _buildresponse("""""") with pytest.raises(ValueError, match="No
element found in"): self.request_class.from_response(response) def test_from_response_invalid_html5(self): response = _buildresponse( """""" """""" """
""" ) req = self.request_class.from_response(response, formdata={"bar": "buz"}) fs = _qs(req) assert fs == {b"foo": [b"xxx"], b"bar": [b"buz"]} def test_from_response_errors_formnumber(self): response = _buildresponse( """
""" ) with pytest.raises(IndexError): self.request_class.from_response(response, formnumber=1) def test_from_response_noformname(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response(response, formdata={"two": "3"}) assert r1.method == "POST" assert r1.headers["Content-type"] == b"application/x-www-form-urlencoded" fs = _qs(r1) assert fs == {b"one": [b"1"], b"two": [b"3"]} def test_from_response_formname_exists(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response(response, formname="form2") assert r1.method == "POST" fs = _qs(r1) assert fs == {b"four": [b"4"], b"three": [b"3"]} def test_from_response_formname_nonexistent(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response(response, formname="form3") assert r1.method == "POST" fs = _qs(r1) assert fs == {b"one": [b"1"]} def test_from_response_formname_errors_formnumber(self): response = _buildresponse( """
""" ) with pytest.raises(IndexError): self.request_class.from_response(response, formname="form3", formnumber=2) def test_from_response_formid_exists(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response(response, formid="form2") assert r1.method == "POST" fs = _qs(r1) assert fs == {b"four": [b"4"], b"three": [b"3"]} def test_from_response_formname_nonexistent_fallback_formid(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response( response, formname="form3", formid="form2" ) assert r1.method == "POST" fs = _qs(r1) assert fs == {b"four": [b"4"], b"three": [b"3"]} def test_from_response_formid_nonexistent(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response(response, formid="form3") assert r1.method == "POST" fs = _qs(r1) assert fs == {b"one": [b"1"]} def test_from_response_formid_errors_formnumber(self): response = _buildresponse( """
""" ) with pytest.raises(IndexError): self.request_class.from_response(response, formid="form3", formnumber=2) def test_from_response_select(self): res = _buildresponse( """
""" ) req = self.request_class.from_response(res) fs = _qs(req) assert fs == {b"i1": [b"iv2"], b"i2": [b"on"]} def test_from_response_checkbox(self): res = _buildresponse( """
""" ) req = self.request_class.from_response(res) fs = _qs(req) assert fs == {b"i1": [b"iv2"], b"i2": [b"on"]} def test_from_response_input_text(self): res = _buildresponse( """
""" ) req = self.request_class.from_response(res) fs = _qs(req) assert fs == {b"i1": [b"i1v1"], b"i2": [b""], b"i4": [b"i4v1"]} def test_from_response_input_hidden(self): res = _buildresponse( """
""" ) req = self.request_class.from_response(res) fs = _qs(req) assert fs == {b"i1": [b"i1v1"], b"i2": [b""]} def test_from_response_input_textarea(self): res = _buildresponse( """
""" ) req = self.request_class.from_response(res) fs = _qs(req) assert fs == {b"i1": [b"i1v"], b"i2": [b""], b"i3": [b""]} def test_from_response_descendants(self): res = _buildresponse( """
""" ) req = self.request_class.from_response(res) fs = _qs(req) assert set(fs) == {b"h2", b"i2", b"i1", b"i3", b"h1", b"i5", b"i4"} def test_from_response_xpath(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response( response, formxpath="//form[@action='post.php']" ) fs = _qs(r1) assert fs[b"one"] == [b"1"] r1 = self.request_class.from_response( response, formxpath="//form/input[@name='four']" ) fs = _qs(r1) assert fs[b"three"] == [b"3"] with pytest.raises(ValueError, match="No
element found with"): self.request_class.from_response( response, formxpath="//form/input[@name='abc']" ) def test_from_response_unicode_xpath(self): response = _buildresponse(b'
') r = self.request_class.from_response( response, formxpath="//form[@name='\u044a']" ) fs = _qs(r) assert not fs xpath = "//form[@name='\u03b1']" with pytest.raises(ValueError, match=re.escape(xpath)): self.request_class.from_response(response, formxpath=xpath) def test_from_response_button_submit(self): response = _buildresponse( """
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response(response) assert req.method == "POST" assert req.headers["Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req) assert fs[b"test1"] == [b"val1"] assert fs[b"test2"] == [b"val2"] assert fs[b"button1"] == [b"submit1"] def test_from_response_button_notype(self): response = _buildresponse( """
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response(response) assert req.method == "POST" assert req.headers["Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req) assert fs[b"test1"] == [b"val1"] assert fs[b"test2"] == [b"val2"] assert fs[b"button1"] == [b"submit1"] def test_from_response_submit_novalue(self): response = _buildresponse( """
Submit
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response(response) assert req.method == "POST" assert req.headers["Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req) assert fs[b"test1"] == [b"val1"] assert fs[b"test2"] == [b"val2"] assert fs[b"button1"] == [b""] def test_from_response_button_novalue(self): response = _buildresponse( """
""", url="http://www.example.com/this/list.html", ) req = self.request_class.from_response(response) assert req.method == "POST" assert req.headers["Content-type"] == b"application/x-www-form-urlencoded" assert req.url == "http://www.example.com/this/post.php" fs = _qs(req) assert fs[b"test1"] == [b"val1"] assert fs[b"test2"] == [b"val2"] assert fs[b"button1"] == [b""] def test_html_base_form_action(self): response = _buildresponse( """
""", url="http://a.com/", ) req = self.request_class.from_response(response) assert req.url == "http://b.com/test_form" def test_spaces_in_action(self): resp = _buildresponse('
') req = self.request_class.from_response(resp) assert req.url == "http://example.com/path" def test_from_response_css(self): response = _buildresponse( """
""" ) r1 = self.request_class.from_response( response, formcss="form[action='post.php']" ) fs = _qs(r1) assert fs[b"one"] == [b"1"] r1 = self.request_class.from_response(response, formcss="input[name='four']") fs = _qs(r1) assert fs[b"three"] == [b"3"] with pytest.raises(ValueError, match="No
element found with"): self.request_class.from_response(response, formcss="input[name='abc']") def test_from_response_valid_form_methods(self): form_methods = [ [method, method] for method in self.request_class.valid_form_methods ] form_methods.append(["UNKNOWN", "GET"]) for method, expected in form_methods: response = _buildresponse( f'' '' "
" ) r = self.request_class.from_response(response) assert r.method == expected def test_form_response_with_invalid_formdata_type_error(self): """Test that a ValueError is raised for non-iterable and non-dict formdata input""" response = _buildresponse( """
""" ) with pytest.raises( ValueError, match="formdata should be a dict or iterable of tuples" ): FormRequest.from_response(response, formdata=123) def test_form_response_with_custom_invalid_formdata_value_error(self): """Test that a ValueError is raised for fault-inducing iterable formdata input""" response = _buildresponse( """
""" ) with pytest.raises( ValueError, match="formdata should be a dict or iterable of tuples" ): FormRequest.from_response(response, formdata=("a",)) def test_get_form_with_xpath_no_form_parent(self): """Test that _get_from raised a ValueError when an XPath selects an element not nested within a
and no parent is found""" response = _buildresponse( """

This paragraph is not inside a form.

""" ) with pytest.raises(ValueError, match="No
element found with"): FormRequest.from_response(response, formxpath='//div[@id="outside-form"]/p') def _buildresponse(body, **kwargs): kwargs.setdefault("body", body) kwargs.setdefault("url", "http://example.com") kwargs.setdefault("encoding", "utf-8") return HtmlResponse(**kwargs) def _qs(req, encoding="utf-8", to_unicode=False): qs = req.body if req.method == "POST" else req.url.partition("?")[2] uqs = unquote_to_bytes(qs) if to_unicode: uqs = uqs.decode(encoding) return parse_qs(uqs, True) 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") class TestJsonRequest(TestRequest): request_class = JsonRequest default_method = "GET" default_headers = { b"Content-Type": [b"application/json"], b"Accept": [b"application/json, text/javascript, */*; q=0.01"], } def test_data(self): r1 = self.request_class(url="http://www.example.com/") assert r1.body == b"" body = b"body" r2 = self.request_class(url="http://www.example.com/", body=body) assert r2.body == body data = { "name": "value", } r3 = self.request_class(url="http://www.example.com/", data=data) assert r3.body == to_bytes(json.dumps(data)) # empty data r4 = self.request_class(url="http://www.example.com/", data=[]) assert r4.body == to_bytes(json.dumps([])) def test_data_method(self): # data is not passed r1 = self.request_class(url="http://www.example.com/") assert r1.method == "GET" body = b"body" r2 = self.request_class(url="http://www.example.com/", body=body) assert r2.method == "GET" data = { "name": "value", } r3 = self.request_class(url="http://www.example.com/", data=data) assert r3.method == "POST" # method passed explicitly r4 = self.request_class(url="http://www.example.com/", data=data, method="GET") assert r4.method == "GET" r5 = self.request_class(url="http://www.example.com/", data=[]) assert r5.method == "POST" def test_body_data(self): """passing both body and data should result a warning""" body = b"body" data = { "name": "value", } with warnings.catch_warnings(record=True) as _warnings: r5 = self.request_class(url="http://www.example.com/", body=body, data=data) assert r5.body == body assert r5.method == "GET" assert len(_warnings) == 1 assert "data will be ignored" in str(_warnings[0].message) def test_empty_body_data(self): """passing any body value and data should result a warning""" data = { "name": "value", } with warnings.catch_warnings(record=True) as _warnings: r6 = self.request_class(url="http://www.example.com/", body=b"", data=data) assert r6.body == b"" assert r6.method == "GET" assert len(_warnings) == 1 assert "data will be ignored" in str(_warnings[0].message) def test_body_none_data(self): data = { "name": "value", } with warnings.catch_warnings(record=True) as _warnings: r7 = self.request_class(url="http://www.example.com/", body=None, data=data) assert r7.body == to_bytes(json.dumps(data)) assert r7.method == "POST" assert len(_warnings) == 0 def test_body_data_none(self): with warnings.catch_warnings(record=True) as _warnings: r8 = self.request_class(url="http://www.example.com/", body=None, data=None) assert r8.method == "GET" assert len(_warnings) == 0 def test_dumps_sort_keys(self): """Test that sort_keys=True is passed to json.dumps by default""" data = { "name": "value", } with mock.patch("json.dumps", return_value=b"") as mock_dumps: self.request_class(url="http://www.example.com/", data=data) kwargs = mock_dumps.call_args[1] assert kwargs["sort_keys"] is True def test_dumps_kwargs(self): """Test that dumps_kwargs are passed to json.dumps""" data = { "name": "value", } dumps_kwargs = { "ensure_ascii": True, "allow_nan": True, } with mock.patch("json.dumps", return_value=b"") as mock_dumps: self.request_class( url="http://www.example.com/", data=data, dumps_kwargs=dumps_kwargs ) kwargs = mock_dumps.call_args[1] assert kwargs["ensure_ascii"] is True assert kwargs["allow_nan"] is True def test_replace_data(self): data1 = { "name1": "value1", } data2 = { "name2": "value2", } r1 = self.request_class(url="http://www.example.com/", data=data1) r2 = r1.replace(data=data2) assert r2.body == to_bytes(json.dumps(data2)) def test_replace_sort_keys(self): """Test that replace provides sort_keys=True to json.dumps""" data1 = { "name1": "value1", } data2 = { "name2": "value2", } r1 = self.request_class(url="http://www.example.com/", data=data1) with mock.patch("json.dumps", return_value=b"") as mock_dumps: r1.replace(data=data2) kwargs = mock_dumps.call_args[1] assert kwargs["sort_keys"] is True def test_replace_dumps_kwargs(self): """Test that dumps_kwargs are provided to json.dumps when replace is called""" data1 = { "name1": "value1", } data2 = { "name2": "value2", } dumps_kwargs = { "ensure_ascii": True, "allow_nan": True, } r1 = self.request_class( url="http://www.example.com/", data=data1, dumps_kwargs=dumps_kwargs ) with mock.patch("json.dumps", return_value=b"") as mock_dumps: r1.replace(data=data2) kwargs = mock_dumps.call_args[1] assert kwargs["ensure_ascii"] is True assert kwargs["allow_nan"] is True def test_replacement_both_body_and_data_warns(self): """Test that we get a warning if both body and data are passed""" body1 = None body2 = b"body" data1 = { "name1": "value1", } data2 = { "name2": "value2", } r1 = self.request_class(url="http://www.example.com/", data=data1, body=body1) with warnings.catch_warnings(record=True) as _warnings: r1.replace(data=data2, body=body2) assert "Both body and data passed. data will be ignored" in str( _warnings[0].message )