Refactoring of test_utils_*. (#6905)

This commit is contained in:
Andrey Rakhmatullin 2025-06-23 23:39:24 +05:00 committed by GitHub
parent 0d75355b41
commit d70f8a3f14
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 1251 additions and 1318 deletions

View File

@ -106,8 +106,6 @@ def get_permissions_dict(
class TestStartprojectTemplates(TestProjectBase): class TestStartprojectTemplates(TestProjectBase):
maxDiff = None
def setup_method(self): def setup_method(self):
super().setup_method() super().setup_method()
self.tmpl = str(Path(self.temp_path, "templates")) self.tmpl = str(Path(self.temp_path, "templates"))

View File

@ -150,8 +150,6 @@ class KeywordArgumentsSpider(MockServerSpider):
class TestCallbackKeywordArguments(TestCase): class TestCallbackKeywordArguments(TestCase):
maxDiff = None
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
cls.mockserver = MockServer() cls.mockserver = MockServer()

View File

@ -1,31 +1,35 @@
import scrapy import scrapy
class TestToplevel: def test_version():
def test_version(self): assert isinstance(scrapy.__version__, str)
assert isinstance(scrapy.__version__, str)
def test_version_info(self):
assert isinstance(scrapy.version_info, tuple)
def test_request_shortcut(self): def test_version_info():
from scrapy.http import FormRequest, Request assert isinstance(scrapy.version_info, tuple)
assert scrapy.Request is Request
assert scrapy.FormRequest is FormRequest
def test_spider_shortcut(self): def test_request_shortcut():
from scrapy.spiders import Spider from scrapy.http import FormRequest, Request
assert scrapy.Spider is Spider assert scrapy.Request is Request
assert scrapy.FormRequest is FormRequest
def test_selector_shortcut(self):
from scrapy.selector import Selector
assert scrapy.Selector is Selector def test_spider_shortcut():
from scrapy.spiders import Spider
def test_item_shortcut(self): assert scrapy.Spider is Spider
from scrapy.item import Field, Item
assert scrapy.Item is Item
assert scrapy.Field is Field def test_selector_shortcut():
from scrapy.selector import Selector
assert scrapy.Selector is Selector
def test_item_shortcut():
from scrapy.item import Field, Item
assert scrapy.Item is Item
assert scrapy.Field is Field

View File

@ -1,10 +0,0 @@
from urllib.parse import urlparse
class TestUrlparse:
def test_s3_url(self):
p = urlparse("s3://bucket/key/name?param=value")
assert p.scheme == "s3"
assert p.hostname == "bucket"
assert p.path == "/key/name"
assert p.query == "param=value"

View File

@ -47,12 +47,11 @@ class TestBuildComponentList:
assert build_component_list(d, convert=lambda x: x) == ["b", "c", "a"] assert build_component_list(d, convert=lambda x: x) == ["b", "c", "a"]
class TestUtilsConf: def test_arglist_to_dict():
def test_arglist_to_dict(self): assert arglist_to_dict(["arg1=val1", "arg2=val2"]) == {
assert arglist_to_dict(["arg1=val1", "arg2=val2"]) == { "arg1": "val1",
"arg1": "val1", "arg2": "val2",
"arg2": "val2", }
}
class TestFeedExportConfig: class TestFeedExportConfig:

View File

@ -18,23 +18,24 @@ except ImportError:
ipy = False ipy = False
class TestUtilsConsole: def test_get_shell_embed_func():
def test_get_shell_embed_func(self): shell = get_shell_embed_func(["invalid"])
shell = get_shell_embed_func(["invalid"]) assert shell is None
assert shell is None
shell = get_shell_embed_func(["invalid", "python"]) shell = get_shell_embed_func(["invalid", "python"])
assert callable(shell) assert callable(shell)
assert shell.__name__ == "_embed_standard_shell" assert shell.__name__ == "_embed_standard_shell"
@pytest.mark.skipif(not bpy, reason="bpython not available in testenv")
def test_get_shell_embed_func2(self):
shell = get_shell_embed_func(["bpython"])
assert callable(shell)
assert shell.__name__ == "_embed_bpython_shell"
@pytest.mark.skipif(not ipy, reason="IPython not available in testenv") @pytest.mark.skipif(not bpy, reason="bpython not available in testenv")
def test_get_shell_embed_func3(self): def test_get_shell_embed_func_bpython():
# default shell should be 'ipython' shell = get_shell_embed_func(["bpython"])
shell = get_shell_embed_func() assert callable(shell)
assert shell.__name__ == "_embed_ipython_shell" assert shell.__name__ == "_embed_bpython_shell"
@pytest.mark.skipif(not ipy, reason="IPython not available in testenv")
def test_get_shell_embed_func_ipython():
# default shell should be 'ipython'
shell = get_shell_embed_func()
assert shell.__name__ == "_embed_ipython_shell"

View File

@ -1,4 +1,5 @@
import warnings import warnings
from typing import Any
import pytest import pytest
from w3lib.http import basic_auth_header from w3lib.http import basic_auth_header
@ -8,9 +9,8 @@ from scrapy.utils.curl import curl_to_request_kwargs
class TestCurlToRequestKwargs: class TestCurlToRequestKwargs:
maxDiff = 5000 @staticmethod
def _test_command(curl_command: str, expected_result: dict[str, Any]) -> None:
def _test_command(self, curl_command, expected_result):
result = curl_to_request_kwargs(curl_command) result = curl_to_request_kwargs(curl_command)
assert result == expected_result assert result == expected_result
try: try:

View File

@ -1,5 +1,6 @@
import copy import copy
import warnings import warnings
from abc import ABC, abstractmethod
from collections.abc import Iterator, Mapping, MutableMapping from collections.abc import Iterator, Mapping, MutableMapping
import pytest import pytest
@ -16,7 +17,12 @@ from scrapy.utils.datatypes import (
from scrapy.utils.python import garbage_collect from scrapy.utils.python import garbage_collect
class CaseInsensitiveDictBase: class TestCaseInsensitiveDictBase(ABC):
@property
@abstractmethod
def dict_class(self) -> type[MutableMapping]:
raise NotImplementedError
def test_init_dict(self): def test_init_dict(self):
seq = {"red": 1, "black": 3} seq = {"red": 1, "black": 3}
d = self.dict_class(seq) d = self.dict_class(seq)
@ -199,7 +205,7 @@ class CaseInsensitiveDictBase:
assert h1.get("header1") == h3.get("HEADER1") assert h1.get("header1") == h3.get("HEADER1")
class TestCaseInsensitiveDict(CaseInsensitiveDictBase): class TestCaseInsensitiveDict(TestCaseInsensitiveDictBase):
dict_class = CaseInsensitiveDict dict_class = CaseInsensitiveDict
def test_repr(self): def test_repr(self):
@ -216,7 +222,7 @@ class TestCaseInsensitiveDict(CaseInsensitiveDictBase):
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
class TestCaselessDict(CaseInsensitiveDictBase): class TestCaselessDict(TestCaseInsensitiveDictBase):
dict_class = CaselessDict dict_class = CaselessDict
def test_deprecation_message(self): def test_deprecation_message(self):

View File

@ -1,6 +1,7 @@
import inspect import inspect
import warnings import warnings
from unittest import mock from unittest import mock
from warnings import WarningMessage
import pytest import pytest
@ -21,7 +22,9 @@ class NewName(SomeBaseClass):
class TestWarnWhenSubclassed: class TestWarnWhenSubclassed:
def _mywarnings(self, w, category=MyWarning): def _mywarnings(
self, w: list[WarningMessage], category: type[Warning] = MyWarning
) -> list[WarningMessage]:
return [x for x in w if x.category is MyWarning] return [x for x in w if x.category is MyWarning]
def test_no_warning_on_definition(self): def test_no_warning_on_definition(self):

View File

@ -3,88 +3,92 @@ from unittest import mock
from scrapy.utils.display import pformat, pprint from scrapy.utils.display import pformat, pprint
value = {"a": 1}
class TestDisplay: colorized_strings = {
object = {"a": 1} (
colorized_strings = {
( (
( "{\x1b[33m'\x1b[39;49;00m\x1b[33ma\x1b[39;49;00m\x1b[33m'"
"{\x1b[33m'\x1b[39;49;00m\x1b[33ma\x1b[39;49;00m\x1b[33m'" "\x1b[39;49;00m: \x1b[34m1\x1b[39;49;00m}"
"\x1b[39;49;00m: \x1b[34m1\x1b[39;49;00m}"
)
+ suffix
) )
for suffix in ( + suffix
# https://github.com/pygments/pygments/issues/2313 )
"\n", # pygments ≤ 2.13 for suffix in (
"\x1b[37m\x1b[39;49;00m\n", # pygments ≥ 2.14 # https://github.com/pygments/pygments/issues/2313
) "\n", # pygments ≤ 2.13
} "\x1b[37m\x1b[39;49;00m\n", # pygments ≥ 2.14
plain_string = "{'a': 1}" )
}
plain_string = "{'a': 1}"
@mock.patch("sys.platform", "linux")
@mock.patch("sys.stdout.isatty")
def test_pformat(self, isatty):
isatty.return_value = True
assert pformat(self.object) in self.colorized_strings
@mock.patch("sys.stdout.isatty") @mock.patch("sys.platform", "linux")
def test_pformat_dont_colorize(self, isatty): @mock.patch("sys.stdout.isatty")
isatty.return_value = True def test_pformat(isatty):
assert pformat(self.object, colorize=False) == self.plain_string isatty.return_value = True
assert pformat(value) in colorized_strings
def test_pformat_not_tty(self):
assert pformat(self.object) == self.plain_string
@mock.patch("sys.platform", "win32") @mock.patch("sys.stdout.isatty")
@mock.patch("platform.version") def test_pformat_dont_colorize(isatty):
@mock.patch("sys.stdout.isatty") isatty.return_value = True
def test_pformat_old_windows(self, isatty, version): assert pformat(value, colorize=False) == plain_string
isatty.return_value = True
version.return_value = "10.0.14392"
assert pformat(self.object) in self.colorized_strings
@mock.patch("sys.platform", "win32")
@mock.patch("scrapy.utils.display._enable_windows_terminal_processing")
@mock.patch("platform.version")
@mock.patch("sys.stdout.isatty")
def test_pformat_windows_no_terminal_processing(
self, isatty, version, terminal_processing
):
isatty.return_value = True
version.return_value = "10.0.14393"
terminal_processing.return_value = False
assert pformat(self.object) == self.plain_string
@mock.patch("sys.platform", "win32") def test_pformat_not_tty():
@mock.patch("scrapy.utils.display._enable_windows_terminal_processing") assert pformat(value) == plain_string
@mock.patch("platform.version")
@mock.patch("sys.stdout.isatty")
def test_pformat_windows(self, isatty, version, terminal_processing):
isatty.return_value = True
version.return_value = "10.0.14393"
terminal_processing.return_value = True
assert pformat(self.object) in self.colorized_strings
@mock.patch("sys.platform", "linux")
@mock.patch("sys.stdout.isatty")
def test_pformat_no_pygments(self, isatty):
isatty.return_value = True
import builtins @mock.patch("sys.platform", "win32")
@mock.patch("platform.version")
@mock.patch("sys.stdout.isatty")
def test_pformat_old_windows(isatty, version):
isatty.return_value = True
version.return_value = "10.0.14392"
assert pformat(value) in colorized_strings
real_import = builtins.__import__
def mock_import(name, globals, locals, fromlist, level): @mock.patch("sys.platform", "win32")
if "pygments" in name: @mock.patch("scrapy.utils.display._enable_windows_terminal_processing")
raise ImportError @mock.patch("platform.version")
return real_import(name, globals, locals, fromlist, level) @mock.patch("sys.stdout.isatty")
def test_pformat_windows_no_terminal_processing(isatty, version, terminal_processing):
isatty.return_value = True
version.return_value = "10.0.14393"
terminal_processing.return_value = False
assert pformat(value) == plain_string
builtins.__import__ = mock_import
assert pformat(self.object) == self.plain_string
builtins.__import__ = real_import
def test_pprint(self): @mock.patch("sys.platform", "win32")
with mock.patch("sys.stdout", new=StringIO()) as mock_out: @mock.patch("scrapy.utils.display._enable_windows_terminal_processing")
pprint(self.object) @mock.patch("platform.version")
assert mock_out.getvalue() == "{'a': 1}\n" @mock.patch("sys.stdout.isatty")
def test_pformat_windows(isatty, version, terminal_processing):
isatty.return_value = True
version.return_value = "10.0.14393"
terminal_processing.return_value = True
assert pformat(value) in colorized_strings
@mock.patch("sys.platform", "linux")
@mock.patch("sys.stdout.isatty")
def test_pformat_no_pygments(isatty):
isatty.return_value = True
import builtins
real_import = builtins.__import__
def mock_import(name, globals, locals, fromlist, level):
if "pygments" in name:
raise ImportError
return real_import(name, globals, locals, fromlist, level)
builtins.__import__ = mock_import
assert pformat(value) == plain_string
builtins.__import__ = real_import
def test_pprint():
with mock.patch("sys.stdout", new=StringIO()) as mock_out:
pprint(value)
assert mock_out.getvalue() == "{'a': 1}\n"

View File

@ -11,47 +11,51 @@ from tests import tests_datadir
SAMPLEDIR = Path(tests_datadir, "compressed") SAMPLEDIR = Path(tests_datadir, "compressed")
class TestGunzip: def test_gunzip_basic():
def test_gunzip_basic(self): r1 = Response(
r1 = Response( "http://www.example.com",
"http://www.example.com", body=(SAMPLEDIR / "feed-sample1.xml.gz").read_bytes(),
body=(SAMPLEDIR / "feed-sample1.xml.gz").read_bytes(), )
) assert gzip_magic_number(r1)
assert gzip_magic_number(r1)
r2 = Response("http://www.example.com", body=gunzip(r1.body)) r2 = Response("http://www.example.com", body=gunzip(r1.body))
assert not gzip_magic_number(r2) assert not gzip_magic_number(r2)
assert len(r2.body) == 9950 assert len(r2.body) == 9950
def test_gunzip_truncated(self):
text = gunzip((SAMPLEDIR / "truncated-crc-error.gz").read_bytes())
assert text.endswith(b"</html")
def test_gunzip_no_gzip_file_raises(self): def test_gunzip_truncated():
with pytest.raises(BadGzipFile): text = gunzip((SAMPLEDIR / "truncated-crc-error.gz").read_bytes())
gunzip((SAMPLEDIR / "feed-sample1.xml").read_bytes()) assert text.endswith(b"</html")
def test_gunzip_truncated_short(self):
r1 = Response(
"http://www.example.com",
body=(SAMPLEDIR / "truncated-crc-error-short.gz").read_bytes(),
)
assert gzip_magic_number(r1)
r2 = Response("http://www.example.com", body=gunzip(r1.body)) def test_gunzip_no_gzip_file_raises():
assert r2.body.endswith(b"</html>") with pytest.raises(BadGzipFile):
assert not gzip_magic_number(r2) gunzip((SAMPLEDIR / "feed-sample1.xml").read_bytes())
def test_is_gzipped_empty(self):
r1 = Response("http://www.example.com")
assert not gzip_magic_number(r1)
def test_gunzip_illegal_eof(self): def test_gunzip_truncated_short():
text = html_to_unicode( r1 = Response(
"charset=cp1252", gunzip((SAMPLEDIR / "unexpected-eof.gz").read_bytes()) "http://www.example.com",
)[1] body=(SAMPLEDIR / "truncated-crc-error-short.gz").read_bytes(),
expected_text = (SAMPLEDIR / "unexpected-eof-output.txt").read_text( )
encoding="utf-8" assert gzip_magic_number(r1)
)
assert len(text) == len(expected_text) r2 = Response("http://www.example.com", body=gunzip(r1.body))
assert text == expected_text assert r2.body.endswith(b"</html>")
assert not gzip_magic_number(r2)
def test_is_gzipped_empty():
r1 = Response("http://www.example.com")
assert not gzip_magic_number(r1)
def test_gunzip_illegal_eof():
text = html_to_unicode(
"charset=cp1252", gunzip((SAMPLEDIR / "unexpected-eof.gz").read_bytes())
)[1]
expected_text = (SAMPLEDIR / "unexpected-eof-output.txt").read_text(
encoding="utf-8"
)
assert len(text) == len(expected_text)
assert text == expected_text

View File

@ -4,18 +4,17 @@ from scrapy.http import Request
from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.httpobj import urlparse_cached
class TestHttpobjUtils: def test_urlparse_cached():
def test_urlparse_cached(self): url = "http://www.example.com/index.html"
url = "http://www.example.com/index.html" request1 = Request(url)
request1 = Request(url) request2 = Request(url)
request2 = Request(url) req1a = urlparse_cached(request1)
req1a = urlparse_cached(request1) req1b = urlparse_cached(request1)
req1b = urlparse_cached(request1) req2 = urlparse_cached(request2)
req2 = urlparse_cached(request2) urlp = urlparse(url)
urlp = urlparse(url)
assert req1a == req2 assert req1a == req2
assert req1a == urlp assert req1a == urlp
assert req1a is req1b assert req1a is req1b
assert req1a is not req2 assert req1a is not req2
assert req1a is not req2 assert req1a is not req2

View File

@ -1,3 +1,8 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
import pytest import pytest
from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.exceptions import ScrapyDeprecationWarning
@ -5,9 +10,19 @@ from scrapy.http import Response, TextResponse, XmlResponse
from scrapy.utils.iterators import _body_or_str, csviter, xmliter, xmliter_lxml from scrapy.utils.iterators import _body_or_str, csviter, xmliter, xmliter_lxml
from tests import get_testdata from tests import get_testdata
if TYPE_CHECKING:
from collections.abc import Iterator
from scrapy import Selector
class TestXmliterBase(ABC):
@abstractmethod
def xmliter(
self, obj: Response | str | bytes, nodename: str, *args: Any
) -> Iterator[Selector]:
raise NotImplementedError
class XmliterBase:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter(self): def test_xmliter(self):
body = b""" body = b"""
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
@ -39,7 +54,6 @@ class XmliterBase:
("002", ["Name 2"], ["Type 2"]), ("002", ["Name 2"], ["Type 2"]),
] ]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_unusual_node(self): def test_xmliter_unusual_node(self):
body = b"""<?xml version="1.0" encoding="UTF-8"?> body = b"""<?xml version="1.0" encoding="UTF-8"?>
<root> <root>
@ -53,7 +67,6 @@ class XmliterBase:
] ]
assert nodenames == [["matchme..."]] assert nodenames == [["matchme..."]]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_unicode(self): def test_xmliter_unicode(self):
# example taken from https://github.com/scrapy/scrapy/issues/1665 # example taken from https://github.com/scrapy/scrapy/issues/1665
body = """<?xml version="1.0" encoding="UTF-8"?> body = """<?xml version="1.0" encoding="UTF-8"?>
@ -113,7 +126,6 @@ class XmliterBase:
("27", ["A"], ["27"]), ("27", ["A"], ["27"]),
] ]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_text(self): def test_xmliter_text(self):
body = ( body = (
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
@ -125,7 +137,6 @@ class XmliterBase:
["two"], ["two"],
] ]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaces(self): def test_xmliter_namespaces(self):
body = b""" body = b"""
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
@ -163,7 +174,6 @@ class XmliterBase:
assert node.xpath("id/text()").getall() == [] assert node.xpath("id/text()").getall() == []
assert node.xpath("price/text()").getall() == [] assert node.xpath("price/text()").getall() == []
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaced_nodename(self): def test_xmliter_namespaced_nodename(self):
body = b""" body = b"""
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
@ -191,7 +201,6 @@ class XmliterBase:
"http://www.mydummycompany.com/images/item1.jpg" "http://www.mydummycompany.com/images/item1.jpg"
] ]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaced_nodename_missing(self): def test_xmliter_namespaced_nodename_missing(self):
body = b""" body = b"""
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
@ -216,7 +225,6 @@ class XmliterBase:
with pytest.raises(StopIteration): with pytest.raises(StopIteration):
next(my_iter) next(my_iter)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_exception(self): def test_xmliter_exception(self):
body = ( body = (
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
@ -229,13 +237,11 @@ class XmliterBase:
with pytest.raises(StopIteration): with pytest.raises(StopIteration):
next(iter) next(iter)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_objtype_exception(self): def test_xmliter_objtype_exception(self):
i = self.xmliter(42, "product") i = self.xmliter(42, "product")
with pytest.raises(TypeError): with pytest.raises(TypeError):
next(i) next(i)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_encoding(self): def test_xmliter_encoding(self):
body = ( body = (
b'<?xml version="1.0" encoding="ISO-8859-9"?>\n' b'<?xml version="1.0" encoding="ISO-8859-9"?>\n'
@ -250,8 +256,12 @@ class XmliterBase:
) )
class TestXmliter(XmliterBase): @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
xmliter = staticmethod(xmliter) class TestXmliter(TestXmliterBase):
def xmliter(
self, obj: Response | str | bytes, nodename: str, *args: Any
) -> Iterator[Selector]:
return xmliter(obj, nodename)
def test_deprecation(self): def test_deprecation(self):
body = b""" body = b"""
@ -267,8 +277,11 @@ class TestXmliter(XmliterBase):
next(self.xmliter(body, "product")) next(self.xmliter(body, "product"))
class TestLxmlXmliter(XmliterBase): class TestLxmlXmliter(TestXmliterBase):
xmliter = staticmethod(xmliter_lxml) def xmliter(
self, obj: Response | str | bytes, nodename: str, *args: Any
) -> Iterator[Selector]:
return xmliter_lxml(obj, nodename, *args)
def test_xmliter_iterate_namespace(self): def test_xmliter_iterate_namespace(self):
body = b""" body = b"""
@ -493,23 +506,32 @@ class TestUtilsCsv:
] ]
class TestHelper: class TestBodyOrStr:
bbody = b"utf8-body" bbody = b"utf8-body"
ubody = bbody.decode("utf8") ubody = bbody.decode("utf8")
txtresponse = TextResponse(url="http://example.org/", body=bbody, encoding="utf-8")
response = Response(url="http://example.org/", body=bbody)
def test_body_or_str(self): @pytest.mark.parametrize(
for obj in (self.bbody, self.ubody, self.txtresponse, self.response): "obj",
r1 = _body_or_str(obj) [
self._assert_type_and_value(r1, self.ubody, obj) bbody,
r2 = _body_or_str(obj, unicode=True) ubody,
self._assert_type_and_value(r2, self.ubody, obj) TextResponse(url="http://example.org/", body=bbody, encoding="utf-8"),
r3 = _body_or_str(obj, unicode=False) Response(url="http://example.org/", body=bbody),
self._assert_type_and_value(r3, self.bbody, obj) ],
assert type(r1) is type(r2) )
assert type(r1) is not type(r3) def test_body_or_str(self, obj: Response | str | bytes) -> None:
r1 = _body_or_str(obj)
self._assert_type_and_value(r1, self.ubody, obj)
r2 = _body_or_str(obj, unicode=True)
self._assert_type_and_value(r2, self.ubody, obj)
r3 = _body_or_str(obj, unicode=False)
self._assert_type_and_value(r3, self.bbody, obj)
assert type(r1) is type(r2)
assert type(r1) is not type(r3)
def _assert_type_and_value(self, a, b, obj): @staticmethod
def _assert_type_and_value(
a: str | bytes, b: str | bytes, obj: Response | str | bytes
) -> None:
assert type(a) is type(b), f"Got {type(a)}, expected {type(b)} for {obj!r}" assert type(a) is type(b), f"Got {type(a)}, expected {type(b)} for {obj!r}"
assert a == b assert a == b

View File

@ -22,7 +22,9 @@ from scrapy.utils.test import get_crawler
from tests.spiders import LogSpider from tests.spiders import LogSpider
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Mapping, MutableMapping from collections.abc import Generator, Mapping, MutableMapping
from scrapy.crawler import Crawler
class TestFailureToExcInfo: class TestFailureToExcInfo:
@ -70,33 +72,41 @@ class TestTopLevelFormatter:
class TestLogCounterHandler: class TestLogCounterHandler:
def setup_method(self): @pytest.fixture
def crawler(self) -> Crawler:
settings = {"LOG_LEVEL": "WARNING"} settings = {"LOG_LEVEL": "WARNING"}
self.logger = logging.getLogger("test") return get_crawler(settings_dict=settings)
self.logger.setLevel(logging.NOTSET)
self.logger.propagate = False
self.crawler = get_crawler(settings_dict=settings)
self.handler = LogCounterHandler(self.crawler)
self.logger.addHandler(self.handler)
def teardown_method(self): @pytest.fixture
self.logger.propagate = True def logger(self, crawler: Crawler) -> Generator[logging.Logger]:
self.logger.removeHandler(self.handler) logger = logging.getLogger("test")
logger.setLevel(logging.NOTSET)
logger.propagate = False
handler = LogCounterHandler(crawler)
logger.addHandler(handler)
def test_init(self): yield logger
assert self.crawler.stats.get_value("log_count/DEBUG") is None
assert self.crawler.stats.get_value("log_count/INFO") is None
assert self.crawler.stats.get_value("log_count/WARNING") is None
assert self.crawler.stats.get_value("log_count/ERROR") is None
assert self.crawler.stats.get_value("log_count/CRITICAL") is None
def test_accepted_level(self): logger.propagate = True
self.logger.error("test log msg") logger.removeHandler(handler)
assert self.crawler.stats.get_value("log_count/ERROR") == 1
def test_filtered_out_level(self): def test_init(self, crawler: Crawler, logger: logging.Logger) -> None:
self.logger.debug("test log msg") assert crawler.stats
assert self.crawler.stats.get_value("log_count/INFO") is None assert crawler.stats.get_value("log_count/DEBUG") is None
assert crawler.stats.get_value("log_count/INFO") is None
assert crawler.stats.get_value("log_count/WARNING") is None
assert crawler.stats.get_value("log_count/ERROR") is None
assert crawler.stats.get_value("log_count/CRITICAL") is None
def test_accepted_level(self, crawler: Crawler, logger: logging.Logger) -> None:
logger.error("test log msg")
assert crawler.stats
assert crawler.stats.get_value("log_count/ERROR") == 1
def test_filtered_out_level(self, crawler: Crawler, logger: logging.Logger) -> None:
logger.debug("test log msg")
assert crawler.stats
assert crawler.stats.get_value("log_count/INFO") is None
class TestStreamLogger: class TestStreamLogger:
@ -135,7 +145,7 @@ class TestStreamLogger:
) )
def test_spider_logger_adapter_process( def test_spider_logger_adapter_process(
base_extra: Mapping[str, Any], log_extra: MutableMapping, expected_extra: dict base_extra: Mapping[str, Any], log_extra: MutableMapping, expected_extra: dict
): ) -> None:
logger = logging.getLogger("test") logger = logging.getLogger("test")
spider_logger_adapter = SpiderLoggerAdapter(logger, base_extra) spider_logger_adapter = SpiderLoggerAdapter(logger, base_extra)
@ -149,59 +159,75 @@ def test_spider_logger_adapter_process(
class TestLogging: class TestLogging:
def setup_method(self): @pytest.fixture
self.log_stream = StringIO() def log_stream(self) -> StringIO:
handler = logging.StreamHandler(self.log_stream) return StringIO()
@pytest.fixture
def spider(self) -> LogSpider:
return LogSpider()
@pytest.fixture(autouse=True)
def logger(self, log_stream: StringIO) -> Generator[logging.Logger]:
handler = logging.StreamHandler(log_stream)
logger = logging.getLogger("log_spider") logger = logging.getLogger("log_spider")
logger.addHandler(handler) logger.addHandler(handler)
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
self.handler = handler
self.logger = logger
self.spider = LogSpider()
def teardown_method(self): yield logger
self.logger.removeHandler(self.handler)
def test_debug_logging(self): logger.removeHandler(handler)
def test_debug_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Foo message" log_message = "Foo message"
self.spider.log_debug(log_message) spider.log_debug(log_message)
log_contents = self.log_stream.getvalue() log_contents = log_stream.getvalue()
assert log_contents == f"{log_message}\n" assert log_contents == f"{log_message}\n"
def test_info_logging(self): def test_info_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Bar message" log_message = "Bar message"
self.spider.log_info(log_message) spider.log_info(log_message)
log_contents = self.log_stream.getvalue() log_contents = log_stream.getvalue()
assert log_contents == f"{log_message}\n" assert log_contents == f"{log_message}\n"
def test_warning_logging(self): def test_warning_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Baz message" log_message = "Baz message"
self.spider.log_warning(log_message) spider.log_warning(log_message)
log_contents = self.log_stream.getvalue() log_contents = log_stream.getvalue()
assert log_contents == f"{log_message}\n" assert log_contents == f"{log_message}\n"
def test_error_logging(self): def test_error_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Foo bar message" log_message = "Foo bar message"
self.spider.log_error(log_message) spider.log_error(log_message)
log_contents = self.log_stream.getvalue() log_contents = log_stream.getvalue()
assert log_contents == f"{log_message}\n" assert log_contents == f"{log_message}\n"
def test_critical_logging(self): def test_critical_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Foo bar baz message" log_message = "Foo bar baz message"
self.spider.log_critical(log_message) spider.log_critical(log_message)
log_contents = self.log_stream.getvalue() log_contents = log_stream.getvalue()
assert log_contents == f"{log_message}\n" assert log_contents == f"{log_message}\n"
class TestLoggingWithExtra: class TestLoggingWithExtra:
def setup_method(self): regex_pattern = re.compile(r"^<LogSpider\s'log_spider'\sat\s[^>]+>$")
self.log_stream = StringIO()
handler = logging.StreamHandler(self.log_stream) @pytest.fixture
def log_stream(self) -> StringIO:
return StringIO()
@pytest.fixture
def spider(self) -> LogSpider:
return LogSpider()
@pytest.fixture(autouse=True)
def logger(self, log_stream: StringIO) -> Generator[logging.Logger]:
handler = logging.StreamHandler(log_stream)
formatter = logging.Formatter( formatter = logging.Formatter(
'{"levelname": "%(levelname)s", "message": "%(message)s", "spider": "%(spider)s", "important_info": "%(important_info)s"}' '{"levelname": "%(levelname)s", "message": "%(message)s", "spider": "%(spider)s", "important_info": "%(important_info)s"}'
) )
@ -209,80 +235,79 @@ class TestLoggingWithExtra:
logger = logging.getLogger("log_spider") logger = logging.getLogger("log_spider")
logger.addHandler(handler) logger.addHandler(handler)
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
self.handler = handler
self.logger = logger
self.spider = LogSpider()
self.regex_pattern = re.compile(r"^<LogSpider\s'log_spider'\sat\s[^>]+>$")
def teardown_method(self): yield logger
self.logger.removeHandler(self.handler)
def test_debug_logging(self): logger.removeHandler(handler)
def test_debug_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Foo message" log_message = "Foo message"
extra = {"important_info": "foo"} extra = {"important_info": "foo"}
self.spider.log_debug(log_message, extra) spider.log_debug(log_message, extra)
log_contents = self.log_stream.getvalue() log_contents_str = log_stream.getvalue()
log_contents = json.loads(log_contents) log_contents = json.loads(log_contents_str)
assert log_contents["levelname"] == "DEBUG" assert log_contents["levelname"] == "DEBUG"
assert log_contents["message"] == log_message assert log_contents["message"] == log_message
assert self.regex_pattern.match(log_contents["spider"]) assert self.regex_pattern.match(log_contents["spider"])
assert log_contents["important_info"] == extra["important_info"] assert log_contents["important_info"] == extra["important_info"]
def test_info_logging(self): def test_info_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Bar message" log_message = "Bar message"
extra = {"important_info": "bar"} extra = {"important_info": "bar"}
self.spider.log_info(log_message, extra) spider.log_info(log_message, extra)
log_contents = self.log_stream.getvalue() log_contents_str = log_stream.getvalue()
log_contents = json.loads(log_contents) log_contents = json.loads(log_contents_str)
assert log_contents["levelname"] == "INFO" assert log_contents["levelname"] == "INFO"
assert log_contents["message"] == log_message assert log_contents["message"] == log_message
assert self.regex_pattern.match(log_contents["spider"]) assert self.regex_pattern.match(log_contents["spider"])
assert log_contents["important_info"] == extra["important_info"] assert log_contents["important_info"] == extra["important_info"]
def test_warning_logging(self): def test_warning_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Baz message" log_message = "Baz message"
extra = {"important_info": "baz"} extra = {"important_info": "baz"}
self.spider.log_warning(log_message, extra) spider.log_warning(log_message, extra)
log_contents = self.log_stream.getvalue() log_contents_str = log_stream.getvalue()
log_contents = json.loads(log_contents) log_contents = json.loads(log_contents_str)
assert log_contents["levelname"] == "WARNING" assert log_contents["levelname"] == "WARNING"
assert log_contents["message"] == log_message assert log_contents["message"] == log_message
assert self.regex_pattern.match(log_contents["spider"]) assert self.regex_pattern.match(log_contents["spider"])
assert log_contents["important_info"] == extra["important_info"] assert log_contents["important_info"] == extra["important_info"]
def test_error_logging(self): def test_error_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Foo bar message" log_message = "Foo bar message"
extra = {"important_info": "foo bar"} extra = {"important_info": "foo bar"}
self.spider.log_error(log_message, extra) spider.log_error(log_message, extra)
log_contents = self.log_stream.getvalue() log_contents_str = log_stream.getvalue()
log_contents = json.loads(log_contents) log_contents = json.loads(log_contents_str)
assert log_contents["levelname"] == "ERROR" assert log_contents["levelname"] == "ERROR"
assert log_contents["message"] == log_message assert log_contents["message"] == log_message
assert self.regex_pattern.match(log_contents["spider"]) assert self.regex_pattern.match(log_contents["spider"])
assert log_contents["important_info"] == extra["important_info"] assert log_contents["important_info"] == extra["important_info"]
def test_critical_logging(self): def test_critical_logging(self, log_stream: StringIO, spider: LogSpider) -> None:
log_message = "Foo bar baz message" log_message = "Foo bar baz message"
extra = {"important_info": "foo bar baz"} extra = {"important_info": "foo bar baz"}
self.spider.log_critical(log_message, extra) spider.log_critical(log_message, extra)
log_contents = self.log_stream.getvalue() log_contents_str = log_stream.getvalue()
log_contents = json.loads(log_contents) log_contents = json.loads(log_contents_str)
assert log_contents["levelname"] == "CRITICAL" assert log_contents["levelname"] == "CRITICAL"
assert log_contents["message"] == log_message assert log_contents["message"] == log_message
assert self.regex_pattern.match(log_contents["spider"]) assert self.regex_pattern.match(log_contents["spider"])
assert log_contents["important_info"] == extra["important_info"] assert log_contents["important_info"] == extra["important_info"]
def test_overwrite_spider_extra(self): def test_overwrite_spider_extra(
self, log_stream: StringIO, spider: LogSpider
) -> None:
log_message = "Foo message" log_message = "Foo message"
extra = {"important_info": "foo", "spider": "shouldn't change"} extra = {"important_info": "foo", "spider": "shouldn't change"}
self.spider.log_error(log_message, extra) spider.log_error(log_message, extra)
log_contents = self.log_stream.getvalue() log_contents_str = log_stream.getvalue()
log_contents = json.loads(log_contents) log_contents = json.loads(log_contents_str)
assert log_contents["levelname"] == "ERROR" assert log_contents["levelname"] == "ERROR"
assert log_contents["message"] == log_message assert log_contents["message"] == log_message

View File

@ -1,18 +1,17 @@
import contextlib
import os import os
import shutil
import tempfile
import warnings import warnings
from pathlib import Path from pathlib import Path
import pytest
from scrapy.utils.misc import set_environ from scrapy.utils.misc import set_environ
from scrapy.utils.project import data_path, get_project_settings from scrapy.utils.project import data_path, get_project_settings
@contextlib.contextmanager @pytest.fixture
def inside_a_project(): def proj_path(tmp_path):
prev_dir = Path.cwd() prev_dir = Path.cwd()
project_dir = tempfile.mkdtemp() project_dir = tmp_path
try: try:
os.chdir(project_dir) os.chdir(project_dir)
@ -21,21 +20,19 @@ def inside_a_project():
yield project_dir yield project_dir
finally: finally:
os.chdir(prev_dir) os.chdir(prev_dir)
shutil.rmtree(project_dir)
class TestProjectUtils: def test_data_path_outside_project():
def test_data_path_outside_project(self): assert str(Path(".scrapy", "somepath")) == data_path("somepath")
assert str(Path(".scrapy", "somepath")) == data_path("somepath") abspath = str(Path(os.path.sep, "absolute", "path"))
abspath = str(Path(os.path.sep, "absolute", "path")) assert abspath == data_path(abspath)
assert abspath == data_path(abspath)
def test_data_path_inside_project(self):
with inside_a_project() as proj_path: def test_data_path_inside_project(proj_path: Path) -> None:
expected = Path(proj_path, ".scrapy", "somepath") expected = proj_path / ".scrapy" / "somepath"
assert expected.resolve() == Path(data_path("somepath")).resolve() assert expected.resolve() == Path(data_path("somepath")).resolve()
abspath = str(Path(os.path.sep, "absolute", "path").resolve()) abspath = str(Path(os.path.sep, "absolute", "path").resolve())
assert abspath == data_path(abspath) assert abspath == data_path(abspath)
class TestGetProjectSettings: class TestGetProjectSettings:

View File

@ -1,7 +1,10 @@
from __future__ import annotations
import functools import functools
import operator import operator
import platform import platform
import sys import sys
from typing import TYPE_CHECKING, TypeVar
import pytest import pytest
from twisted.trial import unittest from twisted.trial import unittest
@ -20,16 +23,22 @@ from scrapy.utils.python import (
without_none_values, without_none_values,
) )
if TYPE_CHECKING:
from collections.abc import Iterable, Mapping
class TestMutableChain:
def test_mutablechain(self): _KT = TypeVar("_KT")
m = MutableChain(range(2), [2, 3], (4, 5)) _VT = TypeVar("_VT")
m.extend(range(6, 7))
m.extend([7, 8])
m.extend([9, 10], (11, 12)) def test_mutablechain():
assert next(m) == 0 m = MutableChain(range(2), [2, 3], (4, 5))
assert m.__next__() == 1 m.extend(range(6, 7))
assert list(m) == list(range(2, 13)) m.extend([7, 8])
m.extend([9, 10], (11, 12))
assert next(m) == 0
assert m.__next__() == 1
assert list(m) == list(range(2, 13))
class TestMutableAsyncChain(unittest.TestCase): class TestMutableAsyncChain(unittest.TestCase):
@ -112,144 +121,150 @@ class TestToBytes:
assert to_bytes("a\ufffdb", "latin-1", errors="replace") == b"a?b" assert to_bytes("a\ufffdb", "latin-1", errors="replace") == b"a?b"
class TestMemoizedMethod: def test_memoizemethod_noargs():
def test_memoizemethod_noargs(self): class A:
class A: @memoizemethod_noargs
@memoizemethod_noargs def cached(self):
def cached(self): return object()
return object()
def noncached(self): def noncached(self):
return object() return object()
a = A() a = A()
one = a.cached() one = a.cached()
two = a.cached() two = a.cached()
three = a.noncached() three = a.noncached()
assert one is two assert one is two
assert one is not three assert one is not three
class TestBinaryIsText: @pytest.mark.parametrize(
def test_binaryistext(self): ("value", "expected"),
assert binary_is_text(b"hello") [
(b"hello", True),
def test_utf_16_strings_contain_null_bytes(self): ("hello".encode("utf-16"), True),
assert binary_is_text("hello".encode("utf-16")) (b"<div>Price \xa3</div>", True),
(b"\x02\xa3", False),
def test_one_with_encoding(self): ],
assert binary_is_text(b"<div>Price \xa3</div>") )
def test_binaryistext(value: bytes, expected: bool) -> None:
def test_real_binary_bytes(self): assert binary_is_text(value) is expected
assert not binary_is_text(b"\x02\xa3")
class TestUtilsPython: @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") def test_equal_attributes():
def test_equal_attributes(self): class Obj:
class Obj: pass
a = Obj()
b = Obj()
# no attributes given return False
assert not equal_attributes(a, b, [])
# nonexistent attributes
assert not equal_attributes(a, b, ["x", "y"])
a.x = 1
b.x = 1
# equal attribute
assert equal_attributes(a, b, ["x"])
b.y = 2
# obj1 has no attribute y
assert not equal_attributes(a, b, ["x", "y"])
a.y = 2
# equal attributes
assert equal_attributes(a, b, ["x", "y"])
a.y = 1
# different attributes
assert not equal_attributes(a, b, ["x", "y"])
# test callable
a.meta = {}
b.meta = {}
assert equal_attributes(a, b, ["meta"])
# compare ['meta']['a']
a.meta["z"] = 1
b.meta["z"] = 1
get_z = operator.itemgetter("z")
get_meta = operator.attrgetter("meta")
def compare_z(obj):
return get_z(get_meta(obj))
assert equal_attributes(a, b, [compare_z, "x"])
# fail z equality
a.meta["z"] = 2
assert not equal_attributes(a, b, [compare_z, "x"])
def test_get_func_args():
def f1(a, b, c):
pass
def f2(a, b=None, c=None):
pass
def f3(a, b=None, *, c=None):
pass
class A:
def __init__(self, a, b, c):
pass pass
a = Obj() def method(self, a, b, c):
b = Obj()
# no attributes given return False
assert not equal_attributes(a, b, [])
# nonexistent attributes
assert not equal_attributes(a, b, ["x", "y"])
a.x = 1
b.x = 1
# equal attribute
assert equal_attributes(a, b, ["x"])
b.y = 2
# obj1 has no attribute y
assert not equal_attributes(a, b, ["x", "y"])
a.y = 2
# equal attributes
assert equal_attributes(a, b, ["x", "y"])
a.y = 1
# different attributes
assert not equal_attributes(a, b, ["x", "y"])
# test callable
a.meta = {}
b.meta = {}
assert equal_attributes(a, b, ["meta"])
# compare ['meta']['a']
a.meta["z"] = 1
b.meta["z"] = 1
get_z = operator.itemgetter("z")
get_meta = operator.attrgetter("meta")
def compare_z(obj):
return get_z(get_meta(obj))
assert equal_attributes(a, b, [compare_z, "x"])
# fail z equality
a.meta["z"] = 2
assert not equal_attributes(a, b, [compare_z, "x"])
def test_get_func_args(self):
def f1(a, b, c):
pass pass
def f2(a, b=None, c=None): class Callable:
def __call__(self, a, b, c):
pass pass
def f3(a, b=None, *, c=None): a = A(1, 2, 3)
pass cal = Callable()
partial_f1 = functools.partial(f1, None)
partial_f2 = functools.partial(f1, b=None)
partial_f3 = functools.partial(partial_f2, None)
class A: assert get_func_args(f1) == ["a", "b", "c"]
def __init__(self, a, b, c): assert get_func_args(f2) == ["a", "b", "c"]
pass assert get_func_args(f3) == ["a", "b", "c"]
assert get_func_args(A) == ["a", "b", "c"]
assert get_func_args(a.method) == ["a", "b", "c"]
assert get_func_args(partial_f1) == ["b", "c"]
assert get_func_args(partial_f2) == ["a", "c"]
assert get_func_args(partial_f3) == ["c"]
assert get_func_args(cal) == ["a", "b", "c"]
assert get_func_args(object) == []
assert get_func_args(str.split, stripself=True) == ["sep", "maxsplit"]
assert get_func_args(" ".join, stripself=True) == ["iterable"]
def method(self, a, b, c): if sys.version_info >= (3, 13) or platform.python_implementation() == "PyPy":
pass # the correct and correctly extracted signature
assert get_func_args(operator.itemgetter(2), stripself=True) == ["obj"]
elif platform.python_implementation() == "CPython":
# ["args", "kwargs"] is a correct result for the pre-3.13 incorrect function signature
# [] is an incorrect result on even older CPython (https://github.com/python/cpython/issues/86951)
assert get_func_args(operator.itemgetter(2), stripself=True) in [
[],
["args", "kwargs"],
]
class Callable:
def __call__(self, a, b, c):
pass
a = A(1, 2, 3) @pytest.mark.parametrize(
cal = Callable() ("value", "expected"),
partial_f1 = functools.partial(f1, None) [
partial_f2 = functools.partial(f1, b=None) ([1, None, 3, 4], [1, 3, 4]),
partial_f3 = functools.partial(partial_f2, None) ((1, None, 3, 4), (1, 3, 4)),
(
assert get_func_args(f1) == ["a", "b", "c"] {"one": 1, "none": None, "three": 3, "four": 4},
assert get_func_args(f2) == ["a", "b", "c"] {"one": 1, "three": 3, "four": 4},
assert get_func_args(f3) == ["a", "b", "c"] ),
assert get_func_args(A) == ["a", "b", "c"] ],
assert get_func_args(a.method) == ["a", "b", "c"] )
assert get_func_args(partial_f1) == ["b", "c"] def test_without_none_values(
assert get_func_args(partial_f2) == ["a", "c"] value: Mapping[_KT, _VT] | Iterable[_KT], expected: dict[_KT, _VT] | Iterable[_KT]
assert get_func_args(partial_f3) == ["c"] ) -> None:
assert get_func_args(cal) == ["a", "b", "c"] assert without_none_values(value) == expected
assert get_func_args(object) == []
assert get_func_args(str.split, stripself=True) == ["sep", "maxsplit"]
assert get_func_args(" ".join, stripself=True) == ["iterable"]
if sys.version_info >= (3, 13) or platform.python_implementation() == "PyPy":
# the correct and correctly extracted signature
assert get_func_args(operator.itemgetter(2), stripself=True) == ["obj"]
elif platform.python_implementation() == "CPython":
# ["args", "kwargs"] is a correct result for the pre-3.13 incorrect function signature
# [] is an incorrect result on even older CPython (https://github.com/python/cpython/issues/86951)
assert get_func_args(operator.itemgetter(2), stripself=True) in [
[],
["args", "kwargs"],
]
def test_without_none_values(self):
assert without_none_values([1, None, 3, 4]) == [1, 3, 4]
assert without_none_values((1, None, 3, 4)) == (1, 3, 4)
assert without_none_values({"one": 1, "none": None, "three": 3, "four": 4}) == {
"one": 1,
"three": 3,
"four": 4,
}

View File

@ -20,45 +20,52 @@ from scrapy.utils.request import (
from scrapy.utils.test import get_crawler from scrapy.utils.test import get_crawler
class TestUtilsRequest: @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") def test_request_authenticate():
def test_request_authenticate(self): r = Request("http://www.example.com")
r = Request("http://www.example.com") request_authenticate(r, "someuser", "somepass")
request_authenticate(r, "someuser", "somepass") assert r.headers["Authorization"] == b"Basic c29tZXVzZXI6c29tZXBhc3M="
assert r.headers["Authorization"] == b"Basic c29tZXVzZXI6c29tZXBhc3M="
def test_request_httprepr(self):
r1 = Request("http://www.example.com")
assert (
request_httprepr(r1) == b"GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n"
)
r1 = Request("http://www.example.com/some/page.html?arg=1") @pytest.mark.parametrize(
assert ( ("r", "expected"),
request_httprepr(r1) [
== b"GET /some/page.html?arg=1 HTTP/1.1\r\nHost: www.example.com\r\n\r\n" (
) Request("http://www.example.com"),
b"GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n",
),
(
Request("http://www.example.com/some/page.html?arg=1"),
b"GET /some/page.html?arg=1 HTTP/1.1\r\nHost: www.example.com\r\n\r\n",
),
(
Request(
"http://www.example.com",
method="POST",
headers={"Content-type": b"text/html"},
body=b"Some body",
),
b"POST / HTTP/1.1\r\nHost: www.example.com\r\nContent-Type: text/html\r\n\r\nSome body",
),
],
)
def test_request_httprepr(r: Request, expected: bytes) -> None:
assert request_httprepr(r) == expected
r1 = Request(
"http://www.example.com",
method="POST",
headers={"Content-type": b"text/html"},
body=b"Some body",
)
assert (
request_httprepr(r1)
== b"POST / HTTP/1.1\r\nHost: www.example.com\r\nContent-Type: text/html\r\n\r\nSome body"
)
def test_request_httprepr_for_non_http_request(self): @pytest.mark.parametrize(
# the representation is not important but it must not fail. "r",
request_httprepr(Request("file:///tmp/foo.txt")) [
request_httprepr(Request("ftp://localhost/tmp/foo.txt")) Request("file:///tmp/foo.txt"),
Request("ftp://localhost/tmp/foo.txt"),
],
)
def test_request_httprepr_for_non_http_request(r: Request) -> None:
# the representation is not important but it must not fail.
request_httprepr(r)
class TestFingerprint: class TestFingerprint:
maxDiff = None
function: staticmethod = staticmethod(fingerprint) function: staticmethod = staticmethod(fingerprint)
cache: ( cache: (
WeakKeyDictionary[Request, dict[tuple[tuple[bytes, ...] | None, bool], bytes]] WeakKeyDictionary[Request, dict[tuple[tuple[bytes, ...] | None, bool], bytes]]
@ -229,35 +236,6 @@ class TestFingerprint:
assert actual == expected assert actual == expected
REQUEST_OBJECTS_TO_TEST = (
Request("http://www.example.com/"),
Request("http://www.example.com/query?id=111&cat=222"),
Request("http://www.example.com/query?cat=222&id=111"),
Request("http://www.example.com/hnnoticiaj1.aspx?78132,199"),
Request("http://www.example.com/hnnoticiaj1.aspx?78160,199"),
Request("http://www.example.com/members/offers.html"),
Request(
"http://www.example.com/members/offers.html",
headers={"SESSIONID": b"somehash"},
),
Request(
"http://www.example.com/",
headers={"Accept-Language": b"en"},
),
Request(
"http://www.example.com/",
headers={
"Accept-Language": b"en",
"SESSIONID": b"somehash",
},
),
Request("http://www.example.com/test.html"),
Request("http://www.example.com/test.html#fragment"),
Request("http://www.example.com", method="POST"),
Request("http://www.example.com", method="POST", body=b"request body"),
)
class TestRequestFingerprinter: class TestRequestFingerprinter:
def test_default_implementation(self): def test_default_implementation(self):
crawler = get_crawler() crawler = get_crawler()

View File

@ -4,7 +4,7 @@ from urllib.parse import urlparse
import pytest import pytest
from scrapy.http import HtmlResponse, Response, TextResponse from scrapy.http import HtmlResponse, Response
from scrapy.utils.python import to_bytes from scrapy.utils.python import to_bytes
from scrapy.utils.response import ( from scrapy.utils.response import (
_remove_html_comments, _remove_html_comments,
@ -15,229 +15,203 @@ from scrapy.utils.response import (
) )
class TestResponseUtils: def test_open_in_browser():
dummy_response = TextResponse(url="http://example.org/", body=b"dummy_response") url = "http:///www.example.com/some/page.html"
body = (
b"<html> <head> <title>test page</title> </head> <body>test body</body> </html>"
)
def test_open_in_browser(self): def browser_open(burl: str) -> bool:
url = "http:///www.example.com/some/page.html" path = urlparse(burl).path
body = b"<html> <head> <title>test page</title> </head> <body>test body</body> </html>" if not path or not Path(path).exists():
path = burl.replace("file://", "")
bbody = Path(path).read_bytes()
assert b'<base href="' + to_bytes(url) + b'">' in bbody
return True
def browser_open(burl): response = HtmlResponse(url, body=body)
path = urlparse(burl).path assert open_in_browser(response, _openfunc=browser_open), "Browser not called"
if not path or not Path(path).exists():
path = burl.replace("file://", "")
bbody = Path(path).read_bytes()
assert b'<base href="' + to_bytes(url) + b'">' in bbody
return True
response = HtmlResponse(url, body=body) resp = Response(url, body=body)
assert open_in_browser(response, _openfunc=browser_open), "Browser not called" with pytest.raises(TypeError):
open_in_browser(resp, debug=True) # pylint: disable=unexpected-keyword-arg
resp = Response(url, body=body)
with pytest.raises(TypeError):
open_in_browser(resp, debug=True) # pylint: disable=unexpected-keyword-arg
def test_get_meta_refresh(self): def test_get_meta_refresh():
r1 = HtmlResponse( r1 = HtmlResponse(
"http://www.example.com", "http://www.example.com",
body=b""" body=b"""
<html> <html>
<head><title>Dummy</title><meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head> <head><title>Dummy</title><meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head>
<body>blahablsdfsal&amp;</body> <body>blahablsdfsal&amp;</body>
</html>""", </html>""",
) )
r2 = HtmlResponse( r2 = HtmlResponse(
"http://www.example.com", "http://www.example.com",
body=b""" body=b"""
<html> <html>
<head><title>Dummy</title><noScript> <head><title>Dummy</title><noScript>
<meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head> <meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head>
</noSCRIPT> </noSCRIPT>
<body>blahablsdfsal&amp;</body> <body>blahablsdfsal&amp;</body>
</html>""", </html>""",
) )
r3 = HtmlResponse( r3 = HtmlResponse(
"http://www.example.com", "http://www.example.com",
body=b""" body=b"""
<noscript><meta http-equiv="REFRESH" content="0;url=http://www.example.com/newpage</noscript> <noscript><meta http-equiv="REFRESH" content="0;url=http://www.example.com/newpage</noscript>
<script type="text/javascript"> <script type="text/javascript">
if(!checkCookies()){ if(!checkCookies()){
document.write('<meta http-equiv="REFRESH" content="0;url=http://www.example.com/newpage">'); document.write('<meta http-equiv="REFRESH" content="0;url=http://www.example.com/newpage">');
} }
</script> </script>
""", """,
) )
assert get_meta_refresh(r1) == (5.0, "http://example.org/newpage") assert get_meta_refresh(r1) == (5.0, "http://example.org/newpage")
assert get_meta_refresh(r2) == (None, None) assert get_meta_refresh(r2) == (None, None)
assert get_meta_refresh(r3) == (None, None) assert get_meta_refresh(r3) == (None, None)
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>""",
)
assert get_base_url(resp) == "http://www.example.com/img/"
resp2 = HtmlResponse( def test_get_base_url():
"http://www.example.com", resp = HtmlResponse(
body=b""" "http://www.example.com",
<html><body>blahablsdfsal&amp;</body></html>""", body=b"""
) <html>
assert get_base_url(resp2) == "http://www.example.com" <head><base href="http://www.example.com/img/" target="_blank"></head>
<body>blahablsdfsal&amp;</body>
</html>""",
)
assert get_base_url(resp) == "http://www.example.com/img/"
def test_response_status_message(self): resp2 = HtmlResponse(
assert response_status_message(200) == "200 OK" "http://www.example.com",
assert response_status_message(404) == "404 Not Found" body=b"""
assert response_status_message(573) == "573 Unknown Status" <html><body>blahablsdfsal&amp;</body></html>""",
)
assert get_base_url(resp2) == "http://www.example.com"
def test_inject_base_url(self):
url = "http://www.example.com"
def check_base_url(burl): def test_response_status_message():
path = urlparse(burl).path assert response_status_message(200) == "200 OK"
if not path or not Path(path).exists(): assert response_status_message(404) == "404 Not Found"
path = burl.replace("file://", "") assert response_status_message(573) == "573 Unknown Status"
bbody = Path(path).read_bytes()
assert bbody.count(b'<base href="' + to_bytes(url) + b'">') == 1
return True
r1 = HtmlResponse(
url,
body=b"""
<html>
<head><title>Dummy</title></head>
<body><p>Hello world.</p></body>
</html>""",
)
r2 = HtmlResponse(
url,
body=b"""
<html>
<head id="foo"><title>Dummy</title></head>
<body>Hello world.</body>
</html>""",
)
r3 = HtmlResponse(
url,
body=b"""
<html>
<head><title>Dummy</title></head>
<body>
<header>Hello header</header>
<p>Hello world.</p>
</body>
</html>""",
)
r4 = HtmlResponse(
url,
body=b"""
<html>
<!-- <head>Dummy comment</head> -->
<head><title>Dummy</title></head>
<body><p>Hello world.</p></body>
</html>""",
)
r5 = HtmlResponse(
url,
body=b"""
<html>
<!--[if IE]>
<head><title>IE head</title></head>
<![endif]-->
<!--[if !IE]>-->
<head><title>Standard head</title></head>
<!--<![endif]-->
<body><p>Hello world.</p></body>
</html>""",
)
assert open_in_browser(r1, _openfunc=check_base_url), "Inject base url" def test_inject_base_url():
assert open_in_browser(r2, _openfunc=check_base_url), ( url = "http://www.example.com"
"Inject base url with argumented head"
)
assert open_in_browser(r3, _openfunc=check_base_url), (
"Inject unique base url with misleading tag"
)
assert open_in_browser(r4, _openfunc=check_base_url), (
"Inject unique base url with misleading comment"
)
assert open_in_browser(r5, _openfunc=check_base_url), (
"Inject unique base url with conditional comment"
)
def test_open_in_browser_redos_comment(self): def check_base_url(burl):
MAX_CPU_TIME = 0.02 path = urlparse(burl).path
if not path or not Path(path).exists():
path = burl.replace("file://", "")
bbody = Path(path).read_bytes()
assert bbody.count(b'<base href="' + to_bytes(url) + b'">') == 1
return True
# Exploit input from r1 = HtmlResponse(
# https://makenowjust-labs.github.io/recheck/playground/ url,
# for /<!--.*?-->/ (old pattern to remove comments). body=b"""
body = b"-><!--\x00" * 25_000 + b"->\n<!---->" <html>
<head><title>Dummy</title></head>
<body><p>Hello world.</p></body>
</html>""",
)
r2 = HtmlResponse(
url,
body=b"""
<html>
<head id="foo"><title>Dummy</title></head>
<body>Hello world.</body>
</html>""",
)
r3 = HtmlResponse(
url,
body=b"""
<html>
<head><title>Dummy</title></head>
<body>
<header>Hello header</header>
<p>Hello world.</p>
</body>
</html>""",
)
r4 = HtmlResponse(
url,
body=b"""
<html>
<!-- <head>Dummy comment</head> -->
<head><title>Dummy</title></head>
<body><p>Hello world.</p></body>
</html>""",
)
r5 = HtmlResponse(
url,
body=b"""
<html>
<!--[if IE]>
<head><title>IE head</title></head>
<![endif]-->
<!--[if !IE]>-->
<head><title>Standard head</title></head>
<!--<![endif]-->
<body><p>Hello world.</p></body>
</html>""",
)
response = HtmlResponse("https://example.com", body=body) assert open_in_browser(r1, _openfunc=check_base_url), "Inject base url"
assert open_in_browser(r2, _openfunc=check_base_url), (
"Inject base url with argumented head"
)
assert open_in_browser(r3, _openfunc=check_base_url), (
"Inject unique base url with misleading tag"
)
assert open_in_browser(r4, _openfunc=check_base_url), (
"Inject unique base url with misleading comment"
)
assert open_in_browser(r5, _openfunc=check_base_url), (
"Inject unique base url with conditional comment"
)
start_time = process_time()
open_in_browser(response, lambda url: True) def test_open_in_browser_redos_comment():
MAX_CPU_TIME = 0.02
end_time = process_time() # Exploit input from
assert end_time - start_time < MAX_CPU_TIME # https://makenowjust-labs.github.io/recheck/playground/
# for /<!--.*?-->/ (old pattern to remove comments).
body = b"-><!--\x00" * 25_000 + b"->\n<!---->"
response = HtmlResponse("https://example.com", body=body)
start_time = process_time()
open_in_browser(response, lambda url: True)
end_time = process_time()
assert end_time - start_time < MAX_CPU_TIME
def test_open_in_browser_redos_head(self):
MAX_CPU_TIME = 0.02
# Exploit input from def test_open_in_browser_redos_head():
# https://makenowjust-labs.github.io/recheck/playground/ MAX_CPU_TIME = 0.02
# for /(<head(?:>|\s.*?>))/ (old pattern to find the head element).
body = b"<head\t" * 8_000
response = HtmlResponse("https://example.com", body=body) # Exploit input from
# https://makenowjust-labs.github.io/recheck/playground/
start_time = process_time() # for /(<head(?:>|\s.*?>))/ (old pattern to find the head element).
body = b"<head\t" * 8_000
open_in_browser(response, lambda url: True) response = HtmlResponse("https://example.com", body=body)
start_time = process_time()
end_time = process_time() open_in_browser(response, lambda url: True)
assert end_time - start_time < MAX_CPU_TIME end_time = process_time()
assert end_time - start_time < MAX_CPU_TIME
@pytest.mark.parametrize( @pytest.mark.parametrize(
("input_body", "output_body"), ("input_body", "output_body"),
[ [
( (b"a<!--", b"a"),
b"a<!--", (b"a<!---->b", b"ab"),
b"a", (b"a<!--b-->c", b"ac"),
), (b"a<!--b-->c<!--", b"ac"),
( (b"a<!--b-->c<!--d", b"ac"),
b"a<!---->b", (b"a<!--b-->c<!---->d", b"acd"),
b"ab", (b"a<!--b--><!--c-->d", b"ad"),
),
(
b"a<!--b-->c",
b"ac",
),
(
b"a<!--b-->c<!--",
b"ac",
),
(
b"a<!--b-->c<!--d",
b"ac",
),
(
b"a<!--b-->c<!---->d",
b"acd",
),
(
b"a<!--b--><!--c-->d",
b"ad",
),
], ],
) )
def test_remove_html_comments(input_body, output_body): def test_remove_html_comments(input_body, output_body):
assert _remove_html_comments(input_body) == output_body, ( assert _remove_html_comments(input_body) == output_body
f"{_remove_html_comments(input_body)=} == {output_body=}"
)

View File

@ -4,6 +4,7 @@ import json
from decimal import Decimal from decimal import Decimal
import attr import attr
import pytest
from twisted.internet import defer from twisted.internet import defer
from scrapy.http import Request, Response from scrapy.http import Request, Response
@ -11,10 +12,11 @@ from scrapy.utils.serialize import ScrapyJSONEncoder
class TestJsonEncoder: class TestJsonEncoder:
def setup_method(self): @pytest.fixture
self.encoder = ScrapyJSONEncoder(sort_keys=True) def encoder(self) -> ScrapyJSONEncoder:
return ScrapyJSONEncoder(sort_keys=True)
def test_encode_decode(self): def test_encode_decode(self, encoder: ScrapyJSONEncoder) -> None:
dt = datetime.datetime(2010, 1, 2, 10, 11, 12) dt = datetime.datetime(2010, 1, 2, 10, 11, 12)
dts = "2010-01-02 10:11:12" dts = "2010-01-02 10:11:12"
d = datetime.date(2010, 1, 2) d = datetime.date(2010, 1, 2)
@ -38,24 +40,24 @@ class TestJsonEncoder:
(s, ss), (s, ss),
(dt_set, dt_sets), (dt_set, dt_sets),
]: ]:
assert self.encoder.encode(input) == json.dumps(output, sort_keys=True) assert encoder.encode(input) == json.dumps(output, sort_keys=True)
def test_encode_deferred(self): def test_encode_deferred(self, encoder: ScrapyJSONEncoder) -> None:
assert "Deferred" in self.encoder.encode(defer.Deferred()) assert "Deferred" in encoder.encode(defer.Deferred())
def test_encode_request(self): def test_encode_request(self, encoder: ScrapyJSONEncoder) -> None:
r = Request("http://www.example.com/lala") r = Request("http://www.example.com/lala")
rs = self.encoder.encode(r) rs = encoder.encode(r)
assert r.method in rs assert r.method in rs
assert r.url in rs assert r.url in rs
def test_encode_response(self): def test_encode_response(self, encoder: ScrapyJSONEncoder) -> None:
r = Response("http://www.example.com/lala") r = Response("http://www.example.com/lala")
rs = self.encoder.encode(r) rs = encoder.encode(r)
assert r.url in rs assert r.url in rs
assert str(r.status) in rs assert str(r.status) in rs
def test_encode_dataclass_item(self) -> None: def test_encode_dataclass_item(self, encoder: ScrapyJSONEncoder) -> None:
@dataclasses.dataclass @dataclasses.dataclass
class TestDataClass: class TestDataClass:
name: str name: str
@ -63,10 +65,10 @@ class TestJsonEncoder:
price: int price: int
item = TestDataClass(name="Product", url="http://product.org", price=1) item = TestDataClass(name="Product", url="http://product.org", price=1)
encoded = self.encoder.encode(item) encoded = encoder.encode(item)
assert encoded == '{"name": "Product", "price": 1, "url": "http://product.org"}' assert encoded == '{"name": "Product", "price": 1, "url": "http://product.org"}'
def test_encode_attrs_item(self): def test_encode_attrs_item(self, encoder: ScrapyJSONEncoder) -> None:
@attr.s @attr.s
class AttrsItem: class AttrsItem:
name = attr.ib(type=str) name = attr.ib(type=str)
@ -74,5 +76,5 @@ class TestJsonEncoder:
price = attr.ib(type=int) price = attr.ib(type=int)
item = AttrsItem(name="Product", url="http://product.org", price=1) item = AttrsItem(name="Product", url="http://product.org", price=1)
encoded = self.encoder.encode(item) encoded = encoder.encode(item)
assert encoded == '{"name": "Product", "price": 1, "url": "http://product.org"}' assert encoded == '{"name": "Product", "price": 1, "url": "http://product.org"}'

View File

@ -1,158 +1,162 @@
from scrapy.utils.sitemap import Sitemap, sitemap_urls_from_robots from scrapy.utils.sitemap import Sitemap, sitemap_urls_from_robots
class TestSitemap: def test_sitemap():
def test_sitemap(self): s = Sitemap(
s = Sitemap( b"""<?xml version="1.0" encoding="UTF-8"?>
b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.google.com/schemas/sitemap/0.84"> <urlset xmlns="http://www.google.com/schemas/sitemap/0.84">
<url> <url>
<loc>http://www.example.com/</loc> <loc>http://www.example.com/</loc>
<lastmod>2009-08-16</lastmod> <lastmod>2009-08-16</lastmod>
<changefreq>daily</changefreq> <changefreq>daily</changefreq>
<priority>1</priority> <priority>1</priority>
</url> </url>
<url> <url>
<loc>http://www.example.com/Special-Offers.html</loc> <loc>http://www.example.com/Special-Offers.html</loc>
<lastmod>2009-08-16</lastmod> <lastmod>2009-08-16</lastmod>
<changefreq>weekly</changefreq> <changefreq>weekly</changefreq>
<priority>0.8</priority> <priority>0.8</priority>
</url> </url>
</urlset>""" </urlset>"""
) )
assert s.type == "urlset" assert s.type == "urlset"
assert list(s) == [ assert list(s) == [
{ {
"priority": "1", "priority": "1",
"loc": "http://www.example.com/", "loc": "http://www.example.com/",
"lastmod": "2009-08-16", "lastmod": "2009-08-16",
"changefreq": "daily", "changefreq": "daily",
}, },
{ {
"priority": "0.8", "priority": "0.8",
"loc": "http://www.example.com/Special-Offers.html", "loc": "http://www.example.com/Special-Offers.html",
"lastmod": "2009-08-16", "lastmod": "2009-08-16",
"changefreq": "weekly", "changefreq": "weekly",
}, },
] ]
def test_sitemap_index(self):
s = Sitemap( def test_sitemap_index():
b"""<?xml version="1.0" encoding="UTF-8"?> s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"> <sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
<sitemap> <sitemap>
<loc>http://www.example.com/sitemap1.xml.gz</loc> <loc>http://www.example.com/sitemap1.xml.gz</loc>
<lastmod>2004-10-01T18:23:17+00:00</lastmod> <lastmod>2004-10-01T18:23:17+00:00</lastmod>
</sitemap> </sitemap>
<sitemap> <sitemap>
<loc>http://www.example.com/sitemap2.xml.gz</loc> <loc>http://www.example.com/sitemap2.xml.gz</loc>
<lastmod>2005-01-01</lastmod> <lastmod>2005-01-01</lastmod>
</sitemap> </sitemap>
</sitemapindex>""" </sitemapindex>"""
) )
assert s.type == "sitemapindex" assert s.type == "sitemapindex"
assert list(s) == [ assert list(s) == [
{ {
"loc": "http://www.example.com/sitemap1.xml.gz", "loc": "http://www.example.com/sitemap1.xml.gz",
"lastmod": "2004-10-01T18:23:17+00:00", "lastmod": "2004-10-01T18:23:17+00:00",
}, },
{ {
"loc": "http://www.example.com/sitemap2.xml.gz", "loc": "http://www.example.com/sitemap2.xml.gz",
"lastmod": "2005-01-01", "lastmod": "2005-01-01",
}, },
] ]
def test_sitemap_strip(self):
"""Assert we can deal with trailing spaces inside <loc> tags - we've def test_sitemap_strip():
seen those """Assert we can deal with trailing spaces inside <loc> tags - we've
""" seen those
s = Sitemap( """
b"""<?xml version="1.0" encoding="UTF-8"?> s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.google.com/schemas/sitemap/0.84"> <urlset xmlns="http://www.google.com/schemas/sitemap/0.84">
<url> <url>
<loc> http://www.example.com/</loc> <loc> http://www.example.com/</loc>
<lastmod>2009-08-16</lastmod> <lastmod>2009-08-16</lastmod>
<changefreq>daily</changefreq> <changefreq>daily</changefreq>
<priority>1</priority> <priority>1</priority>
</url> </url>
<url> <url>
<loc> http://www.example.com/2</loc> <loc> http://www.example.com/2</loc>
<lastmod /> <lastmod />
</url> </url>
</urlset> </urlset>
""" """
) )
assert list(s) == [ assert list(s) == [
{ {
"priority": "1", "priority": "1",
"loc": "http://www.example.com/", "loc": "http://www.example.com/",
"lastmod": "2009-08-16", "lastmod": "2009-08-16",
"changefreq": "daily", "changefreq": "daily",
}, },
{"loc": "http://www.example.com/2", "lastmod": ""}, {"loc": "http://www.example.com/2", "lastmod": ""},
] ]
def test_sitemap_wrong_ns(self):
"""We have seen sitemaps with wrongs ns. Presumably, Google still works def test_sitemap_wrong_ns():
with these, though is not 100% confirmed""" """We have seen sitemaps with wrongs ns. Presumably, Google still works
s = Sitemap( with these, though is not 100% confirmed"""
b"""<?xml version="1.0" encoding="UTF-8"?> s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.google.com/schemas/sitemap/0.84"> <urlset xmlns="http://www.google.com/schemas/sitemap/0.84">
<url xmlns=""> <url xmlns="">
<loc> http://www.example.com/</loc> <loc> http://www.example.com/</loc>
<lastmod>2009-08-16</lastmod> <lastmod>2009-08-16</lastmod>
<changefreq>daily</changefreq> <changefreq>daily</changefreq>
<priority>1</priority> <priority>1</priority>
</url> </url>
<url xmlns=""> <url xmlns="">
<loc> http://www.example.com/2</loc> <loc> http://www.example.com/2</loc>
<lastmod /> <lastmod />
</url> </url>
</urlset> </urlset>
""" """
) )
assert list(s) == [ assert list(s) == [
{ {
"priority": "1", "priority": "1",
"loc": "http://www.example.com/", "loc": "http://www.example.com/",
"lastmod": "2009-08-16", "lastmod": "2009-08-16",
"changefreq": "daily", "changefreq": "daily",
}, },
{"loc": "http://www.example.com/2", "lastmod": ""}, {"loc": "http://www.example.com/2", "lastmod": ""},
] ]
def test_sitemap_wrong_ns2(self):
"""We have seen sitemaps with wrongs ns. Presumably, Google still works def test_sitemap_wrong_ns2():
with these, though is not 100% confirmed""" """We have seen sitemaps with wrongs ns. Presumably, Google still works
s = Sitemap( with these, though is not 100% confirmed"""
b"""<?xml version="1.0" encoding="UTF-8"?> s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset> <urlset>
<url xmlns=""> <url xmlns="">
<loc> http://www.example.com/</loc> <loc> http://www.example.com/</loc>
<lastmod>2009-08-16</lastmod> <lastmod>2009-08-16</lastmod>
<changefreq>daily</changefreq> <changefreq>daily</changefreq>
<priority>1</priority> <priority>1</priority>
</url> </url>
<url xmlns=""> <url xmlns="">
<loc> http://www.example.com/2</loc> <loc> http://www.example.com/2</loc>
<lastmod /> <lastmod />
</url> </url>
</urlset> </urlset>
""" """
) )
assert s.type == "urlset" assert s.type == "urlset"
assert list(s) == [ assert list(s) == [
{ {
"priority": "1", "priority": "1",
"loc": "http://www.example.com/", "loc": "http://www.example.com/",
"lastmod": "2009-08-16", "lastmod": "2009-08-16",
"changefreq": "daily", "changefreq": "daily",
}, },
{"loc": "http://www.example.com/2", "lastmod": ""}, {"loc": "http://www.example.com/2", "lastmod": ""},
] ]
def test_sitemap_urls_from_robots(self):
robots = """User-agent: * def test_sitemap_urls_from_robots():
robots = """User-agent: *
Disallow: /aff/ Disallow: /aff/
Disallow: /wl/ Disallow: /wl/
@ -170,19 +174,18 @@ Sitemap: /sitemap-relative-url.xml
Disallow: /forum/search/ Disallow: /forum/search/
Disallow: /forum/active/ Disallow: /forum/active/
""" """
assert list( assert list(sitemap_urls_from_robots(robots, base_url="http://example.com")) == [
sitemap_urls_from_robots(robots, base_url="http://example.com") "http://example.com/sitemap.xml",
) == [ "http://example.com/sitemap-product-index.xml",
"http://example.com/sitemap.xml", "http://example.com/sitemap-uppercase.xml",
"http://example.com/sitemap-product-index.xml", "http://example.com/sitemap-relative-url.xml",
"http://example.com/sitemap-uppercase.xml", ]
"http://example.com/sitemap-relative-url.xml",
]
def test_sitemap_blanklines(self):
"""Assert we can deal with starting blank lines before <xml> tag""" def test_sitemap_blanklines():
s = Sitemap( """Assert we can deal with starting blank lines before <xml> tag"""
b""" s = Sitemap(
b"""
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"> <sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
@ -205,69 +208,69 @@ Disallow: /forum/active/
<!-- end cache --> <!-- end cache -->
</sitemapindex> </sitemapindex>
""" """
) )
assert list(s) == [ assert list(s) == [
{"lastmod": "2013-07-15", "loc": "http://www.example.com/sitemap1.xml"}, {"lastmod": "2013-07-15", "loc": "http://www.example.com/sitemap1.xml"},
{"lastmod": "2013-07-15", "loc": "http://www.example.com/sitemap2.xml"}, {"lastmod": "2013-07-15", "loc": "http://www.example.com/sitemap2.xml"},
{"lastmod": "2013-07-15", "loc": "http://www.example.com/sitemap3.xml"}, {"lastmod": "2013-07-15", "loc": "http://www.example.com/sitemap3.xml"},
] ]
def test_comment(self):
s = Sitemap( def test_comment():
b"""<?xml version="1.0" encoding="UTF-8"?> s = Sitemap(
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" b"""<?xml version="1.0" encoding="UTF-8"?>
xmlns:xhtml="http://www.w3.org/1999/xhtml"> <urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
xmlns:xhtml="http://www.w3.org/1999/xhtml">
<url>
<loc>http://www.example.com/</loc>
<!-- this is a comment on which the parser might raise an exception if implemented incorrectly -->
</url>
</urlset>"""
)
assert list(s) == [{"loc": "http://www.example.com/"}]
def test_alternate():
s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
xmlns:xhtml="http://www.w3.org/1999/xhtml">
<url>
<loc>http://www.example.com/english/</loc>
<xhtml:link rel="alternate" hreflang="de"
href="http://www.example.com/deutsch/"/>
<xhtml:link rel="alternate" hreflang="de-ch"
href="http://www.example.com/schweiz-deutsch/"/>
<xhtml:link rel="alternate" hreflang="en"
href="http://www.example.com/english/"/>
<xhtml:link rel="alternate" hreflang="en"/><!-- wrong tag without href -->
</url>
</urlset>"""
)
assert list(s) == [
{
"loc": "http://www.example.com/english/",
"alternate": [
"http://www.example.com/deutsch/",
"http://www.example.com/schweiz-deutsch/",
"http://www.example.com/english/",
],
}
]
def test_xml_entity_expansion():
s = Sitemap(
b"""<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE foo [
<!ELEMENT foo ANY >
<!ENTITY xxe SYSTEM "file:///etc/passwd" >
]>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
<url> <url>
<loc>http://www.example.com/</loc> <loc>http://127.0.0.1:8000/&xxe;</loc>
<!-- this is a comment on which the parser might raise an exception if implemented incorrectly -->
</url> </url>
</urlset>""" </urlset>
) """
)
assert list(s) == [{"loc": "http://www.example.com/"}] assert list(s) == [{"loc": "http://127.0.0.1:8000/"}]
def test_alternate(self):
s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
xmlns:xhtml="http://www.w3.org/1999/xhtml">
<url>
<loc>http://www.example.com/english/</loc>
<xhtml:link rel="alternate" hreflang="de"
href="http://www.example.com/deutsch/"/>
<xhtml:link rel="alternate" hreflang="de-ch"
href="http://www.example.com/schweiz-deutsch/"/>
<xhtml:link rel="alternate" hreflang="en"
href="http://www.example.com/english/"/>
<xhtml:link rel="alternate" hreflang="en"/><!-- wrong tag without href -->
</url>
</urlset>"""
)
assert list(s) == [
{
"loc": "http://www.example.com/english/",
"alternate": [
"http://www.example.com/deutsch/",
"http://www.example.com/schweiz-deutsch/",
"http://www.example.com/english/",
],
}
]
def test_xml_entity_expansion(self):
s = Sitemap(
b"""<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE foo [
<!ELEMENT foo ANY >
<!ENTITY xxe SYSTEM "file:///etc/passwd" >
]>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
<url>
<loc>http://127.0.0.1:8000/&xxe;</loc>
</url>
</urlset>
"""
)
assert list(s) == [{"loc": "http://127.0.0.1:8000/"}]

View File

@ -12,19 +12,19 @@ class MySpider2(Spider):
name = "myspider2" name = "myspider2"
class TestUtilsSpiders: def test_iterate_spider_output():
def test_iterate_spider_output(self): i = Item()
i = Item() r = Request("http://scrapytest.org")
r = Request("http://scrapytest.org") o = object()
o = object()
assert list(iterate_spider_output(i)) == [i] assert list(iterate_spider_output(i)) == [i]
assert list(iterate_spider_output(r)) == [r] assert list(iterate_spider_output(r)) == [r]
assert list(iterate_spider_output(o)) == [o] assert list(iterate_spider_output(o)) == [o]
assert list(iterate_spider_output([r, i, o])) == [r, i, o] assert list(iterate_spider_output([r, i, o])) == [r, i, o]
def test_iter_spider_classes(self):
import tests.test_utils_spider # noqa: PLW0406 # pylint: disable=import-self
it = iter_spider_classes(tests.test_utils_spider) def test_iter_spider_classes():
assert set(it) == {MySpider1, MySpider2} import tests.test_utils_spider # noqa: PLW0406 # pylint: disable=import-self
it = iter_spider_classes(tests.test_utils_spider)
assert set(it) == {MySpider1, MySpider2}

View File

@ -1,22 +1,21 @@
from scrapy.utils.template import render_templatefile from scrapy.utils.template import render_templatefile
class TestUtilsRenderTemplateFile: def test_simple_render(tmp_path):
def test_simple_render(self, tmp_path): context = {"project_name": "proj", "name": "spi", "classname": "TheSpider"}
context = {"project_name": "proj", "name": "spi", "classname": "TheSpider"} template = "from ${project_name}.spiders.${name} import ${classname}"
template = "from ${project_name}.spiders.${name} import ${classname}" rendered = "from proj.spiders.spi import TheSpider"
rendered = "from proj.spiders.spi import TheSpider"
template_path = tmp_path / "templ.py.tmpl" template_path = tmp_path / "templ.py.tmpl"
render_path = tmp_path / "templ.py" render_path = tmp_path / "templ.py"
template_path.write_text(template, encoding="utf8") template_path.write_text(template, encoding="utf8")
assert template_path.is_file() # Failure of test itself assert template_path.is_file() # Failure of test itself
render_templatefile(template_path, **context) render_templatefile(template_path, **context)
assert not template_path.exists() assert not template_path.exists()
assert render_path.read_text(encoding="utf8") == rendered assert render_path.read_text(encoding="utf8") == rendered
render_path.unlink() render_path.unlink()
assert not render_path.exists() # Failure of test itself assert not render_path.exists() # Failure of test itself

View File

@ -15,71 +15,76 @@ class Bar(trackref.object_ref):
pass pass
class TestTrackref: @pytest.fixture(autouse=True)
def setup_method(self): def clear_refs() -> None:
trackref.live_refs.clear() trackref.live_refs.clear()
def test_format_live_refs(self):
o1 = Foo() # noqa: F841 def test_format_live_refs():
o2 = Bar() # noqa: F841 o1 = Foo() # noqa: F841
o3 = Foo() # noqa: F841 o2 = Bar() # noqa: F841
assert ( o3 = Foo() # noqa: F841
trackref.format_live_refs() assert (
== """\ trackref.format_live_refs()
== """\
Live References Live References
Bar 1 oldest: 0s ago Bar 1 oldest: 0s ago
Foo 2 oldest: 0s ago Foo 2 oldest: 0s ago
""" """
) )
assert ( assert (
trackref.format_live_refs(ignore=Foo) trackref.format_live_refs(ignore=Foo)
== """\ == """\
Live References Live References
Bar 1 oldest: 0s ago Bar 1 oldest: 0s ago
""" """
) )
@mock.patch("sys.stdout", new_callable=StringIO)
def test_print_live_refs_empty(self, stdout):
trackref.print_live_refs()
assert stdout.getvalue() == "Live References\n\n\n"
@mock.patch("sys.stdout", new_callable=StringIO) @mock.patch("sys.stdout", new_callable=StringIO)
def test_print_live_refs_with_objects(self, stdout): def test_print_live_refs_empty(stdout):
o1 = Foo() # noqa: F841 trackref.print_live_refs()
trackref.print_live_refs() assert stdout.getvalue() == "Live References\n\n\n"
assert (
stdout.getvalue()
== """\ @mock.patch("sys.stdout", new_callable=StringIO)
def test_print_live_refs_with_objects(stdout):
o1 = Foo() # noqa: F841
trackref.print_live_refs()
assert (
stdout.getvalue()
== """\
Live References Live References
Foo 1 oldest: 0s ago\n\n""" Foo 1 oldest: 0s ago\n\n"""
) )
def test_get_oldest(self):
o1 = Foo()
o1_time = time() def test_get_oldest():
o1 = Foo()
o2 = Bar() o1_time = time()
o2 = Bar()
o3_time = time()
if o3_time <= o1_time:
sleep(0.01)
o3_time = time() o3_time = time()
if o3_time <= o1_time: if o3_time <= o1_time:
sleep(0.01) pytest.skip("time.time is not precise enough")
o3_time = time()
if o3_time <= o1_time:
pytest.skip("time.time is not precise enough")
o3 = Foo() # noqa: F841 o3 = Foo() # noqa: F841
assert trackref.get_oldest("Foo") is o1 assert trackref.get_oldest("Foo") is o1
assert trackref.get_oldest("Bar") is o2 assert trackref.get_oldest("Bar") is o2
assert trackref.get_oldest("XXX") is None assert trackref.get_oldest("XXX") is None
def test_iter_all(self):
o1 = Foo() def test_iter_all():
o2 = Bar() # noqa: F841 o1 = Foo()
o3 = Foo() o2 = Bar() # noqa: F841
assert set(trackref.iter_all("Foo")) == {o1, o3} o3 = Foo()
assert set(trackref.iter_all("Foo")) == {o1, o3}

View File

@ -4,7 +4,6 @@ import pytest
from scrapy.linkextractors import IGNORED_EXTENSIONS from scrapy.linkextractors import IGNORED_EXTENSIONS
from scrapy.spiders import Spider from scrapy.spiders import Spider
from scrapy.utils.misc import arg_to_iter
from scrapy.utils.url import ( # type: ignore[attr-defined] from scrapy.utils.url import ( # type: ignore[attr-defined]
_is_filesystem_path, _is_filesystem_path,
_public_w3lib_objects, _public_w3lib_objects,
@ -17,261 +16,156 @@ from scrapy.utils.url import ( # type: ignore[attr-defined]
) )
class TestUrlUtils: def test_url_is_from_any_domain():
def test_url_is_from_any_domain(self): url = "http://www.wheele-bin-art.co.uk/get/product/123"
url = "http://www.wheele-bin-art.co.uk/get/product/123" assert url_is_from_any_domain(url, ["wheele-bin-art.co.uk"])
assert url_is_from_any_domain(url, ["wheele-bin-art.co.uk"]) assert not url_is_from_any_domain(url, ["art.co.uk"])
assert not url_is_from_any_domain(url, ["art.co.uk"])
url = "http://wheele-bin-art.co.uk/get/product/123" url = "http://wheele-bin-art.co.uk/get/product/123"
assert url_is_from_any_domain(url, ["wheele-bin-art.co.uk"]) assert url_is_from_any_domain(url, ["wheele-bin-art.co.uk"])
assert not url_is_from_any_domain(url, ["art.co.uk"]) assert not url_is_from_any_domain(url, ["art.co.uk"])
url = "http://www.Wheele-Bin-Art.co.uk/get/product/123" url = "http://www.Wheele-Bin-Art.co.uk/get/product/123"
assert url_is_from_any_domain(url, ["wheele-bin-art.CO.UK"]) assert url_is_from_any_domain(url, ["wheele-bin-art.CO.UK"])
assert url_is_from_any_domain(url, ["WHEELE-BIN-ART.CO.UK"]) assert url_is_from_any_domain(url, ["WHEELE-BIN-ART.CO.UK"])
url = "http://192.169.0.15:8080/mypage.html" url = "http://192.169.0.15:8080/mypage.html"
assert url_is_from_any_domain(url, ["192.169.0.15:8080"]) assert url_is_from_any_domain(url, ["192.169.0.15:8080"])
assert not url_is_from_any_domain(url, ["192.169.0.15"]) assert not url_is_from_any_domain(url, ["192.169.0.15"])
url = ( url = (
"javascript:%20document.orderform_2581_1190810811.mode.value=%27add%27;%20" "javascript:%20document.orderform_2581_1190810811.mode.value=%27add%27;%20"
"javascript:%20document.orderform_2581_1190810811.submit%28%29" "javascript:%20document.orderform_2581_1190810811.submit%28%29"
) )
assert not url_is_from_any_domain(url, ["testdomain.com"]) assert not url_is_from_any_domain(url, ["testdomain.com"])
assert not url_is_from_any_domain(url + ".testdomain.com", ["testdomain.com"]) assert not url_is_from_any_domain(url + ".testdomain.com", ["testdomain.com"])
def test_url_is_from_spider(self):
spider = Spider(name="example.com")
assert url_is_from_spider("http://www.example.com/some/page.html", spider)
assert url_is_from_spider("http://sub.example.com/some/page.html", spider)
assert not url_is_from_spider("http://www.example.org/some/page.html", spider)
assert not url_is_from_spider("http://www.example.net/some/page.html", spider)
def test_url_is_from_spider_class_attributes(self):
class MySpider(Spider):
name = "example.com"
assert url_is_from_spider("http://www.example.com/some/page.html", MySpider)
assert url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.org/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.net/some/page.html", MySpider)
def test_url_is_from_spider_with_allowed_domains(self):
spider = Spider(
name="example.com", allowed_domains=["example.org", "example.net"]
)
assert url_is_from_spider("http://www.example.com/some/page.html", spider)
assert url_is_from_spider("http://sub.example.com/some/page.html", spider)
assert url_is_from_spider("http://example.com/some/page.html", spider)
assert url_is_from_spider("http://www.example.org/some/page.html", spider)
assert url_is_from_spider("http://www.example.net/some/page.html", spider)
assert not url_is_from_spider("http://www.example.us/some/page.html", spider)
spider = Spider(
name="example.com", allowed_domains={"example.com", "example.net"}
)
assert url_is_from_spider("http://www.example.com/some/page.html", spider)
spider = Spider(
name="example.com", allowed_domains=("example.com", "example.net")
)
assert url_is_from_spider("http://www.example.com/some/page.html", spider)
def test_url_is_from_spider_with_allowed_domains_class_attributes(self):
class MySpider(Spider):
name = "example.com"
allowed_domains = ("example.org", "example.net")
assert url_is_from_spider("http://www.example.com/some/page.html", MySpider)
assert url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
assert url_is_from_spider("http://example.com/some/page.html", MySpider)
assert url_is_from_spider("http://www.example.org/some/page.html", MySpider)
assert url_is_from_spider("http://www.example.net/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.us/some/page.html", MySpider)
def test_url_has_any_extension(self):
deny_extensions = {"." + e for e in arg_to_iter(IGNORED_EXTENSIONS)}
assert url_has_any_extension(
"http://www.example.com/archive.tar.gz", deny_extensions
)
assert url_has_any_extension("http://www.example.com/page.doc", deny_extensions)
assert url_has_any_extension("http://www.example.com/page.pdf", deny_extensions)
assert not url_has_any_extension(
"http://www.example.com/page.htm", deny_extensions
)
assert not url_has_any_extension("http://www.example.com/", deny_extensions)
assert not url_has_any_extension(
"http://www.example.com/page.doc.html", deny_extensions
)
class TestAddHttpIfNoScheme: def test_url_is_from_spider():
def test_add_scheme(self): class MySpider(Spider):
assert add_http_if_no_scheme("www.example.com") == "http://www.example.com" name = "example.com"
def test_without_subdomain(self): assert url_is_from_spider("http://www.example.com/some/page.html", MySpider)
assert add_http_if_no_scheme("example.com") == "http://example.com" assert url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.org/some/page.html", MySpider)
def test_path(self): assert not url_is_from_spider("http://www.example.net/some/page.html", MySpider)
assert (
add_http_if_no_scheme("www.example.com/some/page.html")
== "http://www.example.com/some/page.html"
)
def test_port(self):
assert (
add_http_if_no_scheme("www.example.com:80") == "http://www.example.com:80"
)
def test_fragment(self):
assert (
add_http_if_no_scheme("www.example.com/some/page#frag")
== "http://www.example.com/some/page#frag"
)
def test_query(self):
assert (
add_http_if_no_scheme("www.example.com/do?a=1&b=2&c=3")
== "http://www.example.com/do?a=1&b=2&c=3"
)
def test_username_password(self):
assert (
add_http_if_no_scheme("username:password@www.example.com")
== "http://username:password@www.example.com"
)
def test_complete_url(self):
assert (
add_http_if_no_scheme(
"username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
== "http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
def test_preserve_http(self):
assert (
add_http_if_no_scheme("http://www.example.com") == "http://www.example.com"
)
def test_preserve_http_without_subdomain(self):
assert add_http_if_no_scheme("http://example.com") == "http://example.com"
def test_preserve_http_path(self):
assert (
add_http_if_no_scheme("http://www.example.com/some/page.html")
== "http://www.example.com/some/page.html"
)
def test_preserve_http_port(self):
assert (
add_http_if_no_scheme("http://www.example.com:80")
== "http://www.example.com:80"
)
def test_preserve_http_fragment(self):
assert (
add_http_if_no_scheme("http://www.example.com/some/page#frag")
== "http://www.example.com/some/page#frag"
)
def test_preserve_http_query(self):
assert (
add_http_if_no_scheme("http://www.example.com/do?a=1&b=2&c=3")
== "http://www.example.com/do?a=1&b=2&c=3"
)
def test_preserve_http_username_password(self):
assert (
add_http_if_no_scheme("http://username:password@www.example.com")
== "http://username:password@www.example.com"
)
def test_preserve_http_complete_url(self):
assert (
add_http_if_no_scheme(
"http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
== "http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
def test_protocol_relative(self):
assert add_http_if_no_scheme("//www.example.com") == "http://www.example.com"
def test_protocol_relative_without_subdomain(self):
assert add_http_if_no_scheme("//example.com") == "http://example.com"
def test_protocol_relative_path(self):
assert (
add_http_if_no_scheme("//www.example.com/some/page.html")
== "http://www.example.com/some/page.html"
)
def test_protocol_relative_port(self):
assert (
add_http_if_no_scheme("//www.example.com:80") == "http://www.example.com:80"
)
def test_protocol_relative_fragment(self):
assert (
add_http_if_no_scheme("//www.example.com/some/page#frag")
== "http://www.example.com/some/page#frag"
)
def test_protocol_relative_query(self):
assert (
add_http_if_no_scheme("//www.example.com/do?a=1&b=2&c=3")
== "http://www.example.com/do?a=1&b=2&c=3"
)
def test_protocol_relative_username_password(self):
assert (
add_http_if_no_scheme("//username:password@www.example.com")
== "http://username:password@www.example.com"
)
def test_protocol_relative_complete_url(self):
assert (
add_http_if_no_scheme(
"//username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
== "http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
def test_preserve_https(self):
assert (
add_http_if_no_scheme("https://www.example.com")
== "https://www.example.com"
)
def test_preserve_ftp(self):
assert add_http_if_no_scheme("ftp://www.example.com") == "ftp://www.example.com"
class TestGuessScheme: def test_url_is_from_spider_class_attributes():
pass class MySpider(Spider):
name = "example.com"
assert url_is_from_spider("http://www.example.com/some/page.html", MySpider)
assert url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.org/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.net/some/page.html", MySpider)
def create_guess_scheme_t(args): def test_url_is_from_spider_with_allowed_domains():
def do_expected(self): class MySpider(Spider):
url = guess_scheme(args[0]) name = "example.com"
assert url.startswith(args[1]), ( allowed_domains = ["example.org", "example.net"]
f"Wrong scheme guessed: for `{args[0]}` got `{url}`, expected `{args[1]}...`"
)
return do_expected assert url_is_from_spider("http://www.example.com/some/page.html", MySpider)
assert url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
assert url_is_from_spider("http://example.com/some/page.html", MySpider)
assert url_is_from_spider("http://www.example.org/some/page.html", MySpider)
assert url_is_from_spider("http://www.example.net/some/page.html", MySpider)
assert not url_is_from_spider("http://www.example.us/some/page.html", MySpider)
class MySpider2(Spider):
name = "example.com"
allowed_domains = {"example.com", "example.net"}
assert url_is_from_spider("http://www.example.com/some/page.html", MySpider2)
class MySpider3(Spider):
name = "example.com"
allowed_domains = ("example.com", "example.net")
assert url_is_from_spider("http://www.example.com/some/page.html", MySpider3)
def create_skipped_scheme_t(args): @pytest.mark.parametrize(
def do_expected(self): ("url", "expected"),
pytest.skip(args[2]) [
("http://www.example.com/archive.tar.gz", True),
return do_expected ("http://www.example.com/page.doc", True),
("http://www.example.com/page.pdf", True),
("http://www.example.com/page.htm", False),
("http://www.example.com/", False),
("http://www.example.com/page.doc.html", False),
],
)
def test_url_has_any_extension(url: str, expected: bool) -> None:
deny_extensions = {"." + e for e in IGNORED_EXTENSIONS}
assert url_has_any_extension(url, deny_extensions) is expected
for k, args in enumerate( @pytest.mark.parametrize(
("url", "expected"),
[
("www.example.com", "http://www.example.com"),
("example.com", "http://example.com"),
("www.example.com/some/page.html", "http://www.example.com/some/page.html"),
("www.example.com:80", "http://www.example.com:80"),
("www.example.com/some/page#frag", "http://www.example.com/some/page#frag"),
("www.example.com/do?a=1&b=2&c=3", "http://www.example.com/do?a=1&b=2&c=3"),
(
"username:password@www.example.com",
"http://username:password@www.example.com",
),
(
"username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag",
"http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag",
),
("http://www.example.com", "http://www.example.com"),
("http://example.com", "http://example.com"),
(
"http://www.example.com/some/page.html",
"http://www.example.com/some/page.html",
),
("http://www.example.com:80", "http://www.example.com:80"),
(
"http://www.example.com/some/page#frag",
"http://www.example.com/some/page#frag",
),
(
"http://www.example.com/do?a=1&b=2&c=3",
"http://www.example.com/do?a=1&b=2&c=3",
),
(
"http://username:password@www.example.com",
"http://username:password@www.example.com",
),
(
"http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag",
"http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag",
),
("//www.example.com", "http://www.example.com"),
("//example.com", "http://example.com"),
("//www.example.com/some/page.html", "http://www.example.com/some/page.html"),
("//www.example.com:80", "http://www.example.com:80"),
("//www.example.com/some/page#frag", "http://www.example.com/some/page#frag"),
("//www.example.com/do?a=1&b=2&c=3", "http://www.example.com/do?a=1&b=2&c=3"),
(
"//username:password@www.example.com",
"http://username:password@www.example.com",
),
(
"//username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag",
"http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag",
),
("https://www.example.com", "https://www.example.com"),
("ftp://www.example.com", "ftp://www.example.com"),
],
)
def test_add_http_if_no_scheme(url: str, expected: str) -> None:
assert add_http_if_no_scheme(url) == expected
@pytest.mark.parametrize(
("url", "expected"),
[ [
("/index", "file://"), ("/index", "file://"),
("/index.html", "file://"), ("/index.html", "file://"),
@ -295,14 +189,13 @@ for k, args in enumerate(
("/", "http://"), ("/", "http://"),
(".../test", "http://"), (".../test", "http://"),
], ],
start=1, )
): def test_guess_scheme(url: str, expected: str):
t_method = create_guess_scheme_t(args) assert guess_scheme(url).startswith(expected)
t_method.__name__ = f"test_uri_{k:03}"
setattr(TestGuessScheme, t_method.__name__, t_method)
# TODO: the following tests do not pass with current implementation
for k, skip_args in enumerate( @pytest.mark.parametrize(
("url", "expected", "reason"),
[ [
( (
r"C:\absolute\path\to\a\file.html", r"C:\absolute\path\to\a\file.html",
@ -310,25 +203,21 @@ for k, skip_args in enumerate(
"Windows filepath are not supported for scrapy shell", "Windows filepath are not supported for scrapy shell",
), ),
], ],
start=1, )
): def test_guess_scheme_skipped(url: str, expected: str, reason: str):
t_method = create_skipped_scheme_t(skip_args) pytest.skip(reason)
t_method.__name__ = f"test_uri_skipped_{k:03}"
setattr(TestGuessScheme, t_method.__name__, t_method)
class TestStripUrl: class TestStripUrl:
def test_noop(self): @pytest.mark.parametrize(
assert ( "url",
strip_url("http://www.example.com/index.html") [
== "http://www.example.com/index.html" "http://www.example.com/index.html",
) "http://www.example.com/index.html?somekey=somevalue",
],
def test_noop_query_string(self): )
assert ( def test_noop(self, url: str) -> None:
strip_url("http://www.example.com/index.html?somekey=somevalue") assert strip_url(url) == url
== "http://www.example.com/index.html?somekey=somevalue"
)
def test_fragments(self): def test_fragments(self):
assert ( assert (
@ -339,16 +228,20 @@ class TestStripUrl:
== "http://www.example.com/index.html?somekey=somevalue#section" == "http://www.example.com/index.html?somekey=somevalue#section"
) )
def test_path(self): @pytest.mark.parametrize(
for input_url, origin, output_url in [ ("url", "origin", "expected"),
[
("http://www.example.com/", False, "http://www.example.com/"), ("http://www.example.com/", False, "http://www.example.com/"),
("http://www.example.com", False, "http://www.example.com"), ("http://www.example.com", False, "http://www.example.com"),
("http://www.example.com", True, "http://www.example.com/"), ("http://www.example.com", True, "http://www.example.com/"),
]: ],
assert strip_url(input_url, origin_only=origin) == output_url )
def test_path(self, url: str, origin: bool, expected: str) -> None:
assert strip_url(url, origin_only=origin) == expected
def test_credentials(self): @pytest.mark.parametrize(
for i, o in [ ("url", "expected"),
[
( (
"http://username@www.example.com/index.html?somekey=somevalue#section", "http://username@www.example.com/index.html?somekey=somevalue#section",
"http://www.example.com/index.html?somekey=somevalue", "http://www.example.com/index.html?somekey=somevalue",
@ -361,34 +254,29 @@ class TestStripUrl:
"ftp://username:password@www.example.com/index.html?somekey=somevalue#section", "ftp://username:password@www.example.com/index.html?somekey=somevalue#section",
"ftp://www.example.com/index.html?somekey=somevalue", "ftp://www.example.com/index.html?somekey=somevalue",
), ),
]: # user: "username@", password: none
assert strip_url(i, strip_credentials=True) == o
def test_credentials_encoded_delims(self):
for i, o in [
# user: "username@"
# password: none
( (
"http://username%40@www.example.com/index.html?somekey=somevalue#section", "http://username%40@www.example.com/index.html?somekey=somevalue#section",
"http://www.example.com/index.html?somekey=somevalue", "http://www.example.com/index.html?somekey=somevalue",
), ),
# user: "username:pass" # user: "username:pass", password: ""
# password: ""
( (
"https://username%3Apass:@www.example.com/index.html?somekey=somevalue#section", "https://username%3Apass:@www.example.com/index.html?somekey=somevalue#section",
"https://www.example.com/index.html?somekey=somevalue", "https://www.example.com/index.html?somekey=somevalue",
), ),
# user: "me" # user: "me", password: "user@domain.com"
# password: "user@domain.com"
( (
"ftp://me:user%40domain.com@www.example.com/index.html?somekey=somevalue#section", "ftp://me:user%40domain.com@www.example.com/index.html?somekey=somevalue#section",
"ftp://www.example.com/index.html?somekey=somevalue", "ftp://www.example.com/index.html?somekey=somevalue",
), ),
]: ],
assert strip_url(i, strip_credentials=True) == o )
def test_credentials(self, url: str, expected: str) -> None:
assert strip_url(url, strip_credentials=True) == expected
def test_default_ports_creds_off(self): @pytest.mark.parametrize(
for i, o in [ ("url", "expected"),
[
( (
"http://username:password@www.example.com:80/index.html?somekey=somevalue#section", "http://username:password@www.example.com:80/index.html?somekey=somevalue#section",
"http://www.example.com/index.html?somekey=somevalue", "http://www.example.com/index.html?somekey=somevalue",
@ -421,11 +309,14 @@ class TestStripUrl:
"ftp://username:password@www.example.com:221/file.txt", "ftp://username:password@www.example.com:221/file.txt",
"ftp://www.example.com:221/file.txt", "ftp://www.example.com:221/file.txt",
), ),
]: ],
assert strip_url(i) == o )
def test_default_ports_creds_off(self, url: str, expected: str) -> None:
assert strip_url(url) == expected
def test_default_ports(self): @pytest.mark.parametrize(
for i, o in [ ("url", "expected"),
[
( (
"http://username:password@www.example.com:80/index.html", "http://username:password@www.example.com:80/index.html",
"http://username:password@www.example.com/index.html", "http://username:password@www.example.com/index.html",
@ -458,11 +349,16 @@ class TestStripUrl:
"ftp://username:password@www.example.com:221/file.txt", "ftp://username:password@www.example.com:221/file.txt",
"ftp://username:password@www.example.com:221/file.txt", "ftp://username:password@www.example.com:221/file.txt",
), ),
]: ],
assert strip_url(i, strip_default_port=True, strip_credentials=False) == o )
def test_default_ports(self, url: str, expected: str) -> None:
assert (
strip_url(url, strip_default_port=True, strip_credentials=False) == expected
)
def test_default_ports_keep(self): @pytest.mark.parametrize(
for i, o in [ ("url", "expected"),
[
( (
"http://username:password@www.example.com:80/index.html?somekey=somevalue&someotherkey=sov#section", "http://username:password@www.example.com:80/index.html?somekey=somevalue&someotherkey=sov#section",
"http://username:password@www.example.com:80/index.html?somekey=somevalue&someotherkey=sov", "http://username:password@www.example.com:80/index.html?somekey=somevalue&someotherkey=sov",
@ -495,11 +391,17 @@ class TestStripUrl:
"ftp://username:password@www.example.com:221/file.txt", "ftp://username:password@www.example.com:221/file.txt",
"ftp://username:password@www.example.com:221/file.txt", "ftp://username:password@www.example.com:221/file.txt",
), ),
]: ],
assert strip_url(i, strip_default_port=False, strip_credentials=False) == o )
def test_default_ports_keep(self, url: str, expected: str) -> None:
assert (
strip_url(url, strip_default_port=False, strip_credentials=False)
== expected
)
def test_origin_only(self): @pytest.mark.parametrize(
for i, o in [ ("url", "expected"),
[
( (
"http://username:password@www.example.com/index.html", "http://username:password@www.example.com/index.html",
"http://www.example.com/", "http://www.example.com/",
@ -516,29 +418,33 @@ class TestStripUrl:
"https://username:password@www.example.com:443/index.html", "https://username:password@www.example.com:443/index.html",
"https://www.example.com/", "https://www.example.com/",
), ),
]: ],
assert strip_url(i, origin_only=True) == o )
def test_origin_only(self, url: str, expected: str) -> None:
assert strip_url(url, origin_only=True) == expected
class TestIsPath: @pytest.mark.parametrize(
def test_path(self): ("path", "expected"),
for input_value, output_value in ( [
# https://en.wikipedia.org/wiki/Path_(computing)#Representations_of_paths_by_operating_system_and_shell # https://en.wikipedia.org/wiki/Path_(computing)#Representations_of_paths_by_operating_system_and_shell
# Unix-like OS, Microsoft Windows / cmd.exe # Unix-like OS, Microsoft Windows / cmd.exe
("/home/user/docs/Letter.txt", True), ("/home/user/docs/Letter.txt", True),
("./inthisdir", True), ("./inthisdir", True),
("../../greatgrandparent", True), ("../../greatgrandparent", True),
("~/.rcinfo", True), ("~/.rcinfo", True),
(r"C:\user\docs\Letter.txt", True), (r"C:\user\docs\Letter.txt", True),
("/user/docs/Letter.txt", True), ("/user/docs/Letter.txt", True),
(r"C:\Letter.txt", True), (r"C:\Letter.txt", True),
(r"\\Server01\user\docs\Letter.txt", True), (r"\\Server01\user\docs\Letter.txt", True),
(r"\\?\UNC\Server01\user\docs\Letter.txt", True), (r"\\?\UNC\Server01\user\docs\Letter.txt", True),
(r"\\?\C:\user\docs\Letter.txt", True), (r"\\?\C:\user\docs\Letter.txt", True),
(r"C:\user\docs\somefile.ext:alternate_stream_name", True), (r"C:\user\docs\somefile.ext:alternate_stream_name", True),
(r"https://example.com", False), (r"https://example.com", False),
): ],
assert _is_filesystem_path(input_value) == output_value, input_value )
def test__is_filesystem_path(path: str, expected: bool) -> None:
assert _is_filesystem_path(path) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -550,7 +456,7 @@ class TestIsPath:
*_public_w3lib_objects, *_public_w3lib_objects,
], ],
) )
def test_deprecated_imports_from_w3lib(obj_name): def test_deprecated_imports_from_w3lib(obj_name: str) -> None:
with warnings.catch_warnings(record=True) as warns: with warnings.catch_warnings(record=True) as warns:
obj_type = "attribute" if obj_name == "_safe_chars" else "function" obj_type = "attribute" if obj_name == "_safe_chars" else "function"
message = f"The scrapy.utils.url.{obj_name} {obj_type} is deprecated, use w3lib.url.{obj_name} instead." message = f"The scrapy.utils.url.{obj_name} {obj_type} is deprecated, use w3lib.url.{obj_name} instead."
@ -559,4 +465,5 @@ def test_deprecated_imports_from_w3lib(obj_name):
getattr(import_module("scrapy.utils.url"), obj_name) getattr(import_module("scrapy.utils.url"), obj_name)
assert isinstance(warns[0].message, Warning)
assert message in warns[0].message.args assert message in warns[0].message.args