RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Convert tests/test_utils* to plain asserts. (#6695)

This commit is contained in:
Andrey Rakhmatullin 2025-03-04 13:31:26 +05:00 committed by GitHub
parent 87db3f2fd6
commit d161d1d47d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
26 changed files with 835 additions and 1022 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/test_utils_asyncgen.py
View File

@ -4,15 +4,15 @@ from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
from scrapy.utils.defer import deferred_f_from_coro_f
class AsyncgenUtilsTest(unittest.TestCase):
class TestAsyncgenUtils(unittest.TestCase):
@deferred_f_from_coro_f
async def test_as_async_generator(self):
ag = as_async_generator(range(42))
results = [i async for i in ag]
self.assertEqual(results, list(range(42)))
assert results == list(range(42))
@deferred_f_from_coro_f
async def test_collect_asyncgen(self):
ag = as_async_generator(range(42))
results = await collect_asyncgen(ag)
self.assertEqual(results, list(range(42)))
assert results == list(range(42))

9
tests/test_utils_asyncio.py
View File

@ -2,7 +2,6 @@ import asyncio
import warnings
import pytest
from twisted.trial.unittest import TestCase
from scrapy.utils.defer import deferred_f_from_coro_f
from scrapy.utils.reactor import (
@ -13,19 +12,17 @@ from scrapy.utils.reactor import (
@pytest.mark.usefixtures("reactor_pytest")
class AsyncioTest(TestCase):
class TestAsyncio:
def test_is_asyncio_reactor_installed(self):
# the result should depend only on the pytest --reactor argument
self.assertEqual(
is_asyncio_reactor_installed(), self.reactor_pytest == "asyncio"
)
assert is_asyncio_reactor_installed() == (self.reactor_pytest == "asyncio")
def test_install_asyncio_reactor(self):
from twisted.internet import reactor as original_reactor
with warnings.catch_warnings(record=True) as w:
install_reactor("twisted.internet.asyncioreactor.AsyncioSelectorReactor")
self.assertEqual(len(w), 0)
assert len(w) == 0
from twisted.internet import reactor # pylint: disable=reimported
assert original_reactor == reactor

126
tests/test_utils_conf.py
View File

@ -1,5 +1,3 @@
import unittest
import pytest
from scrapy.exceptions import UsageError
@ -12,26 +10,24 @@ from scrapy.utils.conf import (
)
class BuildComponentListTest(unittest.TestCase):
class TestBuildComponentList:
def test_build_dict(self):
d = {"one": 1, "two": None, "three": 8, "four": 4}
self.assertEqual(
build_component_list(d, convert=lambda x: x), ["one", "four", "three"]
)
assert build_component_list(d, convert=lambda x: x) == ["one", "four", "three"]
def test_duplicate_components_in_basesettings(self):
# Higher priority takes precedence
duplicate_bs = BaseSettings({"one": 1, "two": 2}, priority=0)
duplicate_bs.set("ONE", 4, priority=10)
self.assertEqual(
build_component_list(duplicate_bs, convert=lambda x: x.lower()),
["two", "one"],
)
assert build_component_list(duplicate_bs, convert=lambda x: x.lower()) == [
"two",
"one",
]
duplicate_bs.set("one", duplicate_bs["one"], priority=20)
self.assertEqual(
build_component_list(duplicate_bs, convert=lambda x: x.lower()),
["one", "two"],
)
assert build_component_list(duplicate_bs, convert=lambda x: x.lower()) == [
"one",
"two",
]
# Same priority raises ValueError
duplicate_bs.set("ONE", duplicate_bs["ONE"], priority=20)
with pytest.raises(
@ -42,24 +38,24 @@ class BuildComponentListTest(unittest.TestCase):
def test_valid_numbers(self):
# work well with None and numeric values
d = {"a": 10, "b": None, "c": 15, "d": 5.0}
self.assertEqual(build_component_list(d, convert=lambda x: x), ["d", "a", "c"])
assert build_component_list(d, convert=lambda x: x) == ["d", "a", "c"]
d = {
"a": 33333333333333333333,
"b": 11111111111111111111,
"c": 22222222222222222222,
}
self.assertEqual(build_component_list(d, convert=lambda x: x), ["b", "c", "a"])
assert build_component_list(d, convert=lambda x: x) == ["b", "c", "a"]
class UtilsConfTestCase(unittest.TestCase):
class TestUtilsConf:
def test_arglist_to_dict(self):
self.assertEqual(
arglist_to_dict(["arg1=val1", "arg2=val2"]),
{"arg1": "val1", "arg2": "val2"},
)
assert arglist_to_dict(["arg1=val1", "arg2=val2"]) == {
"arg1": "val1",
"arg2": "val2",
}
class FeedExportConfigTestCase(unittest.TestCase):
class TestFeedExportConfig:
def test_feed_export_config_invalid_format(self):
settings = Settings()
with pytest.raises(UsageError):
@ -72,44 +68,36 @@ class FeedExportConfigTestCase(unittest.TestCase):
def test_feed_export_config_explicit_formats(self):
settings = Settings()
self.assertEqual(
{
"items_1.dat": {"format": "json"},
"items_2.dat": {"format": "xml"},
"items_3.dat": {"format": "csv"},
},
feed_process_params_from_cli(
settings, ["items_1.dat:json", "items_2.dat:xml", "items_3.dat:csv"]
),
assert {
"items_1.dat": {"format": "json"},
"items_2.dat": {"format": "xml"},
"items_3.dat": {"format": "csv"},
} == feed_process_params_from_cli(
settings, ["items_1.dat:json", "items_2.dat:xml", "items_3.dat:csv"]
)
def test_feed_export_config_implicit_formats(self):
settings = Settings()
self.assertEqual(
{
"items_1.json": {"format": "json"},
"items_2.xml": {"format": "xml"},
"items_3.csv": {"format": "csv"},
},
feed_process_params_from_cli(
settings, ["items_1.json", "items_2.xml", "items_3.csv"]
),
assert {
"items_1.json": {"format": "json"},
"items_2.xml": {"format": "xml"},
"items_3.csv": {"format": "csv"},
} == feed_process_params_from_cli(
settings, ["items_1.json", "items_2.xml", "items_3.csv"]
)
def test_feed_export_config_stdout(self):
settings = Settings()
self.assertEqual(
{"stdout:": {"format": "pickle"}},
feed_process_params_from_cli(settings, ["-:pickle"]),
assert {"stdout:": {"format": "pickle"}} == feed_process_params_from_cli(
settings, ["-:pickle"]
)
def test_feed_export_config_overwrite(self):
settings = Settings()
self.assertEqual(
{"output.json": {"format": "json", "overwrite": True}},
feed_process_params_from_cli(
settings, [], overwrite_output=["output.json"]
),
assert {
"output.json": {"format": "json", "overwrite": True}
} == feed_process_params_from_cli(
settings, [], overwrite_output=["output.json"]
)
def test_output_and_overwrite_output(self):
@ -131,18 +119,15 @@ class FeedExportConfigTestCase(unittest.TestCase):
}
)
new_feed = feed_complete_default_values_from_settings(feed, settings)
self.assertEqual(
new_feed,
{
"encoding": "custom encoding",
"fields": ["f1", "f2", "f3"],
"indent": 42,
"store_empty": True,
"uri_params": (1, 2, 3, 4),
"batch_item_count": 2,
"item_export_kwargs": {},
},
)
assert new_feed == {
"encoding": "custom encoding",
"fields": ["f1", "f2", "f3"],
"indent": 42,
"store_empty": True,
"uri_params": (1, 2, 3, 4),
"batch_item_count": 2,
"item_export_kwargs": {},
}
def test_feed_complete_default_values_from_settings_non_empty(self):
feed = {
@ -159,15 +144,12 @@ class FeedExportConfigTestCase(unittest.TestCase):
}
)
new_feed = feed_complete_default_values_from_settings(feed, settings)
self.assertEqual(
new_feed,
{
"encoding": "other encoding",
"fields": None,
"indent": 42,
"store_empty": True,
"uri_params": None,
"batch_item_count": 2,
"item_export_kwargs": {},
},
)
assert new_feed == {
"encoding": "other encoding",
"fields": None,
"indent": 42,
"store_empty": True,
"uri_params": None,
"batch_item_count": 2,
"item_export_kwargs": {},
}

20
tests/test_utils_console.py
View File

@ -1,4 +1,4 @@
import unittest
import pytest
from scrapy.utils.console import get_shell_embed_func
@ -18,23 +18,23 @@ except ImportError:
ipy = False
class UtilsConsoleTestCase(unittest.TestCase):
class TestUtilsConsole:
def test_get_shell_embed_func(self):
shell = get_shell_embed_func(["invalid"])
self.assertEqual(shell, None)
assert shell is None
shell = get_shell_embed_func(["invalid", "python"])
self.assertTrue(callable(shell))
self.assertEqual(shell.__name__, "_embed_standard_shell")
assert callable(shell)
assert shell.__name__ == "_embed_standard_shell"
@unittest.skipIf(not bpy, "bpython not available in testenv")
@pytest.mark.skipif(not bpy, reason="bpython not available in testenv")
def test_get_shell_embed_func2(self):
shell = get_shell_embed_func(["bpython"])
self.assertTrue(callable(shell))
self.assertEqual(shell.__name__, "_embed_bpython_shell")
assert callable(shell)
assert shell.__name__ == "_embed_bpython_shell"
@unittest.skipIf(not ipy, "IPython not available in testenv")
@pytest.mark.skipif(not ipy, reason="IPython not available in testenv")
def test_get_shell_embed_func3(self):
# default shell should be 'ipython'
shell = get_shell_embed_func()
self.assertEqual(shell.__name__, "_embed_ipython_shell")
assert shell.__name__ == "_embed_ipython_shell"

11
tests/test_utils_curl.py
View File

@ -1,4 +1,3 @@
import unittest
import warnings
import pytest
@ -8,16 +7,16 @@ from scrapy import Request
from scrapy.utils.curl import curl_to_request_kwargs
class CurlToRequestKwargsTest(unittest.TestCase):
class TestCurlToRequestKwargs:
maxDiff = 5000
def _test_command(self, curl_command, expected_result):
result = curl_to_request_kwargs(curl_command)
self.assertEqual(result, expected_result)
assert result == expected_result
try:
Request(**result)
except TypeError as e:
self.fail(f"Request kwargs are not correct {e}")
pytest.fail(f"Request kwargs are not correct {e}")
def test_get(self):
curl_command = "curl http://example.org/"
@ -203,7 +202,7 @@ class CurlToRequestKwargsTest(unittest.TestCase):
def test_get_silent(self):
curl_command = 'curl --silent "www.example.com"'
expected_result = {"method": "GET", "url": "http://www.example.com"}
self.assertEqual(curl_to_request_kwargs(curl_command), expected_result)
assert curl_to_request_kwargs(curl_command) == expected_result
def test_too_few_arguments_error(self):
with pytest.raises(
@ -218,7 +217,7 @@ class CurlToRequestKwargsTest(unittest.TestCase):
warnings.simplefilter("ignore")
curl_command = "curl --bar --baz http://www.example.com"
expected_result = {"method": "GET", "url": "http://www.example.com"}
self.assertEqual(curl_to_request_kwargs(curl_command), expected_result)
assert curl_to_request_kwargs(curl_command) == expected_result
# case 2: ignore_unknown_options=False (raise exception):
with pytest.raises(ValueError, match="Unrecognized options:.*--bar.*--baz"):

207
tests/test_utils_datatypes.py
View File

@ -1,5 +1,4 @@
import copy
import unittest
import warnings
from collections.abc import Iterator, Mapping, MutableMapping
@ -17,18 +16,18 @@ from scrapy.utils.datatypes import (
from scrapy.utils.python import garbage_collect
class CaseInsensitiveDictMixin:
class CaseInsensitiveDictBase:
def test_init_dict(self):
seq = {"red": 1, "black": 3}
d = self.dict_class(seq)
self.assertEqual(d["red"], 1)
self.assertEqual(d["black"], 3)
assert d["red"] == 1
assert d["black"] == 3
def test_init_pair_sequence(self):
seq = (("red", 1), ("black", 3))
d = self.dict_class(seq)
self.assertEqual(d["red"], 1)
self.assertEqual(d["black"], 3)
assert d["red"] == 1
assert d["black"] == 3
def test_init_mapping(self):
class MyMapping(Mapping):
@ -46,8 +45,8 @@ class CaseInsensitiveDictMixin:
seq = MyMapping(red=1, black=3)
d = self.dict_class(seq)
self.assertEqual(d["red"], 1)
self.assertEqual(d["black"], 3)
assert d["red"] == 1
assert d["black"] == 3
def test_init_mutable_mapping(self):
class MyMutableMapping(MutableMapping):
@ -71,18 +70,18 @@ class CaseInsensitiveDictMixin:
seq = MyMutableMapping(red=1, black=3)
d = self.dict_class(seq)
self.assertEqual(d["red"], 1)
self.assertEqual(d["black"], 3)
assert d["red"] == 1
assert d["black"] == 3
def test_caseless(self):
d = self.dict_class()
d["key_Lower"] = 1
self.assertEqual(d["KEy_loWer"], 1)
self.assertEqual(d.get("KEy_loWer"), 1)
assert d["KEy_loWer"] == 1
assert d.get("KEy_loWer") == 1
d["KEY_LOWER"] = 3
self.assertEqual(d["key_Lower"], 3)
self.assertEqual(d.get("key_Lower"), 3)
assert d["key_Lower"] == 3
assert d.get("key_Lower") == 3
def test_delete(self):
d = self.dict_class({"key_lower": 1})
@ -95,41 +94,41 @@ class CaseInsensitiveDictMixin:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_getdefault(self):
d = CaselessDict()
self.assertEqual(d.get("c", 5), 5)
assert d.get("c", 5) == 5
d["c"] = 10
self.assertEqual(d.get("c", 5), 10)
assert d.get("c", 5) == 10
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_setdefault(self):
d = CaselessDict({"a": 1, "b": 2})
r = d.setdefault("A", 5)
self.assertEqual(r, 1)
self.assertEqual(d["A"], 1)
assert r == 1
assert d["A"] == 1
r = d.setdefault("c", 5)
self.assertEqual(r, 5)
self.assertEqual(d["C"], 5)
assert r == 5
assert d["C"] == 5
def test_fromkeys(self):
keys = ("a", "b")
d = self.dict_class.fromkeys(keys)
self.assertEqual(d["A"], None)
self.assertEqual(d["B"], None)
assert d["A"] is None
assert d["B"] is None
d = self.dict_class.fromkeys(keys, 1)
self.assertEqual(d["A"], 1)
self.assertEqual(d["B"], 1)
assert d["A"] == 1
assert d["B"] == 1
instance = self.dict_class()
d = instance.fromkeys(keys)
self.assertEqual(d["A"], None)
self.assertEqual(d["B"], None)
assert d["A"] is None
assert d["B"] is None
d = instance.fromkeys(keys, 1)
self.assertEqual(d["A"], 1)
self.assertEqual(d["B"], 1)
assert d["A"] == 1
assert d["B"] == 1
def test_contains(self):
d = self.dict_class()
@ -139,7 +138,7 @@ class CaseInsensitiveDictMixin:
def test_pop(self):
d = self.dict_class()
d["a"] = 1
self.assertEqual(d.pop("A"), 1)
assert d.pop("A") == 1
with pytest.raises(KeyError):
d.pop("A")
@ -152,7 +151,7 @@ class CaseInsensitiveDictMixin:
d = MyDict()
d["key-one"] = 2
self.assertEqual(list(d.keys()), ["Key-One"])
assert list(d.keys()) == ["Key-One"]
def test_normvalue(self):
class MyDict(self.dict_class):
@ -164,62 +163,60 @@ class CaseInsensitiveDictMixin:
normvalue = _normvalue # deprecated CaselessDict class
d = MyDict({"key": 1})
self.assertEqual(d["key"], 2)
self.assertEqual(d.get("key"), 2)
assert d["key"] == 2
assert d.get("key") == 2
d = MyDict()
d["key"] = 1
self.assertEqual(d["key"], 2)
self.assertEqual(d.get("key"), 2)
assert d["key"] == 2
assert d.get("key") == 2
d = MyDict()
d.setdefault("key", 1)
self.assertEqual(d["key"], 2)
self.assertEqual(d.get("key"), 2)
assert d["key"] == 2
assert d.get("key") == 2
d = MyDict()
d.update({"key": 1})
self.assertEqual(d["key"], 2)
self.assertEqual(d.get("key"), 2)
assert d["key"] == 2
assert d.get("key") == 2
d = MyDict.fromkeys(("key",), 1)
self.assertEqual(d["key"], 2)
self.assertEqual(d.get("key"), 2)
assert d["key"] == 2
assert d.get("key") == 2
def test_copy(self):
h1 = self.dict_class({"header1": "value"})
h2 = copy.copy(h1)
assert isinstance(h2, self.dict_class)
self.assertEqual(h1, h2)
self.assertEqual(h1.get("header1"), h2.get("header1"))
self.assertEqual(h1.get("header1"), h2.get("HEADER1"))
assert h1 == h2
assert h1.get("header1") == h2.get("header1")
assert h1.get("header1") == h2.get("HEADER1")
h3 = h1.copy()
assert isinstance(h3, self.dict_class)
self.assertEqual(h1, h3)
self.assertEqual(h1.get("header1"), h3.get("header1"))
self.assertEqual(h1.get("header1"), h3.get("HEADER1"))
assert h1 == h3
assert h1.get("header1") == h3.get("header1")
assert h1.get("header1") == h3.get("HEADER1")
class CaseInsensitiveDictTest(CaseInsensitiveDictMixin, unittest.TestCase):
class TestCaseInsensitiveDict(CaseInsensitiveDictBase):
dict_class = CaseInsensitiveDict
def test_repr(self):
d1 = self.dict_class({"foo": "bar"})
self.assertEqual(repr(d1), "<CaseInsensitiveDict: {'foo': 'bar'}>")
assert repr(d1) == "<CaseInsensitiveDict: {'foo': 'bar'}>"
d2 = self.dict_class({"AsDf": "QwErTy", "FoO": "bAr"})
self.assertEqual(
repr(d2), "<CaseInsensitiveDict: {'AsDf': 'QwErTy', 'FoO': 'bAr'}>"
)
assert repr(d2) == "<CaseInsensitiveDict: {'AsDf': 'QwErTy', 'FoO': 'bAr'}>"
def test_iter(self):
d = self.dict_class({"AsDf": "QwErTy", "FoO": "bAr"})
iterkeys = iter(d)
self.assertIsInstance(iterkeys, Iterator)
self.assertEqual(list(iterkeys), ["AsDf", "FoO"])
assert isinstance(iterkeys, Iterator)
assert list(iterkeys) == ["AsDf", "FoO"]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
class CaselessDictTest(CaseInsensitiveDictMixin, unittest.TestCase):
class TestCaselessDict(CaseInsensitiveDictBase):
dict_class = CaselessDict
def test_deprecation_message(self):
@ -227,93 +224,93 @@ class CaselessDictTest(CaseInsensitiveDictMixin, unittest.TestCase):
warnings.filterwarnings("always", category=ScrapyDeprecationWarning)
self.dict_class({"foo": "bar"})
self.assertEqual(len(caught), 1)
self.assertTrue(issubclass(caught[0].category, ScrapyDeprecationWarning))
self.assertEqual(
"scrapy.utils.datatypes.CaselessDict is deprecated,"
" please use scrapy.utils.datatypes.CaseInsensitiveDict instead",
str(caught[0].message),
assert len(caught) == 1
assert issubclass(caught[0].category, ScrapyDeprecationWarning)
assert (
str(caught[0].message)
== "scrapy.utils.datatypes.CaselessDict is deprecated,"
" please use scrapy.utils.datatypes.CaseInsensitiveDict instead"
)
class SequenceExcludeTest(unittest.TestCase):
class TestSequenceExclude:
def test_list(self):
seq = [1, 2, 3]
d = SequenceExclude(seq)
self.assertIn(0, d)
self.assertIn(4, d)
self.assertNotIn(2, d)
assert 0 in d
assert 4 in d
assert 2 not in d
def test_range(self):
seq = range(10, 20)
d = SequenceExclude(seq)
self.assertIn(5, d)
self.assertIn(20, d)
self.assertNotIn(15, d)
assert 5 in d
assert 20 in d
assert 15 not in d
def test_range_step(self):
seq = range(10, 20, 3)
d = SequenceExclude(seq)
are_not_in = [v for v in range(10, 20, 3) if v in d]
self.assertEqual([], are_not_in)
assert are_not_in == []
are_not_in = [v for v in range(10, 20) if v in d]
self.assertEqual([11, 12, 14, 15, 17, 18], are_not_in)
assert are_not_in == [11, 12, 14, 15, 17, 18]
def test_string_seq(self):
seq = "cde"
d = SequenceExclude(seq)
chars = "".join(v for v in "abcdefg" if v in d)
self.assertEqual("abfg", chars)
assert chars == "abfg"
def test_stringset_seq(self):
seq = set("cde")
d = SequenceExclude(seq)
chars = "".join(v for v in "abcdefg" if v in d)
self.assertEqual("abfg", chars)
assert chars == "abfg"
def test_set(self):
"""Anything that is not in the supplied sequence will evaluate as 'in' the container."""
seq = {-3, "test", 1.1}
d = SequenceExclude(seq)
self.assertIn(0, d)
self.assertIn("foo", d)
self.assertIn(3.14, d)
self.assertIn(set("bar"), d)
assert 0 in d
assert "foo" in d
assert 3.14 in d
assert set("bar") in d
# supplied sequence is a set, so checking for list (non)inclusion fails
with pytest.raises(TypeError):
["a", "b", "c"] in d # noqa: B015
for v in [-3, "test", 1.1]:
self.assertNotIn(v, d)
assert v not in d
class LocalCacheTest(unittest.TestCase):
class TestLocalCache:
def test_cache_with_limit(self):
cache = LocalCache(limit=2)
cache["a"] = 1
cache["b"] = 2
cache["c"] = 3
self.assertEqual(len(cache), 2)
self.assertNotIn("a", cache)
self.assertIn("b", cache)
self.assertIn("c", cache)
self.assertEqual(cache["b"], 2)
self.assertEqual(cache["c"], 3)
assert len(cache) == 2
assert "a" not in cache
assert "b" in cache
assert "c" in cache
assert cache["b"] == 2
assert cache["c"] == 3
def test_cache_without_limit(self):
maximum = 10**4
cache = LocalCache()
for x in range(maximum):
cache[str(x)] = x
self.assertEqual(len(cache), maximum)
assert len(cache) == maximum
for x in range(maximum):
self.assertIn(str(x), cache)
self.assertEqual(cache[str(x)], x)
assert str(x) in cache
assert cache[str(x)] == x
class LocalWeakReferencedCacheTest(unittest.TestCase):
class TestLocalWeakReferencedCache:
def test_cache_with_limit(self):
cache = LocalWeakReferencedCache(limit=2)
r1 = Request("https://example.org")
@ -322,19 +319,19 @@ class LocalWeakReferencedCacheTest(unittest.TestCase):
cache[r1] = 1
cache[r2] = 2
cache[r3] = 3
self.assertEqual(len(cache), 2)
self.assertNotIn(r1, cache)
self.assertIn(r2, cache)
self.assertIn(r3, cache)
self.assertEqual(cache[r1], None)
self.assertEqual(cache[r2], 2)
self.assertEqual(cache[r3], 3)
assert len(cache) == 2
assert r1 not in cache
assert r2 in cache
assert r3 in cache
assert cache[r1] is None
assert cache[r2] == 2
assert cache[r3] == 3
del r2
# PyPy takes longer to collect dead references
garbage_collect()
self.assertEqual(len(cache), 1)
assert len(cache) == 1
def test_cache_non_weak_referenceable_objects(self):
cache = LocalWeakReferencedCache()
@ -344,10 +341,10 @@ class LocalWeakReferencedCacheTest(unittest.TestCase):
cache[k1] = 1
cache[k2] = 2
cache[k3] = 3
self.assertNotIn(k1, cache)
self.assertNotIn(k2, cache)
self.assertNotIn(k3, cache)
self.assertEqual(len(cache), 0)
assert k1 not in cache
assert k2 not in cache
assert k3 not in cache
assert len(cache) == 0
def test_cache_without_limit(self):
max = 10**4
@ -356,10 +353,10 @@ class LocalWeakReferencedCacheTest(unittest.TestCase):
for x in range(max):
refs.append(Request(f"https://example.org/{x}"))
cache[refs[-1]] = x
self.assertEqual(len(cache), max)
assert len(cache) == max
for i, r in enumerate(refs):
self.assertIn(r, cache)
self.assertEqual(cache[r], i)
assert r in cache
assert cache[r] == i
del r # delete reference to the last object in the list # pylint: disable=undefined-loop-variable
# delete half of the objects, make sure that is reflected in the cache
@ -369,7 +366,7 @@ class LocalWeakReferencedCacheTest(unittest.TestCase):
# PyPy takes longer to collect dead references
garbage_collect()
self.assertEqual(len(cache), max // 2)
assert len(cache) == max // 2
for i, r in enumerate(refs):
self.assertIn(r, cache)
self.assertEqual(cache[r], i)
assert r in cache
assert cache[r] == i

46
tests/test_utils_defer.py
View File

@ -18,7 +18,7 @@ from scrapy.utils.defer import (
)
class MustbeDeferredTest(unittest.TestCase):
class TestMustbeDeferred(unittest.TestCase):
def test_success_function(self):
steps = []
@ -66,23 +66,19 @@ def eb1(failure, arg1, arg2):
return f"(eb1 {failure.value.__class__.__name__} {arg1} {arg2})"
class DeferUtilsTest(unittest.TestCase):
class TestDeferUtils(unittest.TestCase):
@defer.inlineCallbacks
def test_process_chain(self):
x = yield process_chain([cb1, cb2, cb3], "res", "v1", "v2")
self.assertEqual(x, "(cb3 (cb2 (cb1 res v1 v2) v1 v2) v1 v2)")
assert x == "(cb3 (cb2 (cb1 res v1 v2) v1 v2) v1 v2)"
gotexc = False
try:
with pytest.raises(TypeError):
yield process_chain([cb1, cb_fail, cb3], "res", "v1", "v2")
except TypeError:
gotexc = True
self.assertTrue(gotexc)
@defer.inlineCallbacks
def test_process_parallel(self):
x = yield process_parallel([cb1, cb2, cb3], "res", "v1", "v2")
self.assertEqual(x, ["(cb1 res v1 v2)", "(cb2 res v1 v2)", "(cb3 res v1 v2)"])
assert x == ["(cb1 res v1 v2)", "(cb2 res v1 v2)", "(cb3 res v1 v2)"]
def test_process_parallel_failure(self):
d = process_parallel([cb1, cb_fail, cb3], "res", "v1", "v2")
@ -90,15 +86,15 @@ class DeferUtilsTest(unittest.TestCase):
return d
class IterErrbackTest(unittest.TestCase):
class TestIterErrback:
def test_iter_errback_good(self):
def itergood():
yield from range(10)
errors = []
out = list(iter_errback(itergood(), errors.append))
self.assertEqual(out, list(range(10)))
self.assertFalse(errors)
assert out == list(range(10))
assert not errors
def test_iter_errback_bad(self):
def iterbad():
@ -109,12 +105,12 @@ class IterErrbackTest(unittest.TestCase):
errors = []
out = list(iter_errback(iterbad(), errors.append))
self.assertEqual(out, [0, 1, 2, 3, 4])
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0].value, ZeroDivisionError)
assert out == [0, 1, 2, 3, 4]
assert len(errors) == 1
assert isinstance(errors[0].value, ZeroDivisionError)
class AiterErrbackTest(unittest.TestCase):
class TestAiterErrback(unittest.TestCase):
@deferred_f_from_coro_f
async def test_aiter_errback_good(self):
async def itergood():
@ -123,8 +119,8 @@ class AiterErrbackTest(unittest.TestCase):
errors = []
out = await collect_asyncgen(aiter_errback(itergood(), errors.append))
self.assertEqual(out, list(range(10)))
self.assertFalse(errors)
assert out == list(range(10))
assert not errors
@deferred_f_from_coro_f
async def test_iter_errback_bad(self):
@ -136,12 +132,12 @@ class AiterErrbackTest(unittest.TestCase):
errors = []
out = await collect_asyncgen(aiter_errback(iterbad(), errors.append))
self.assertEqual(out, [0, 1, 2, 3, 4])
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0].value, ZeroDivisionError)
assert out == [0, 1, 2, 3, 4]
assert len(errors) == 1
assert isinstance(errors[0].value, ZeroDivisionError)
class AsyncDefTestsuiteTest(unittest.TestCase):
class TestAsyncDefTestsuite(unittest.TestCase):
@deferred_f_from_coro_f
async def test_deferred_f_from_coro_f(self):
pass
@ -156,7 +152,7 @@ class AsyncDefTestsuiteTest(unittest.TestCase):
raise RuntimeError("This is expected to be raised")
class AsyncCooperatorTest(unittest.TestCase):
class TestAsyncCooperator(unittest.TestCase):
"""This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage.
parallel_async is called with the results of a callback (so an iterable of items, requests and None,
@ -207,7 +203,7 @@ class AsyncCooperatorTest(unittest.TestCase):
ait = self.get_async_iterable(length)
dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
yield dl
self.assertEqual(list(range(length)), sorted(results))
assert list(range(length)) == sorted(results)
@defer.inlineCallbacks
def test_delays(self):
@ -216,4 +212,4 @@ class AsyncCooperatorTest(unittest.TestCase):
ait = self.get_async_iterable_with_delays(length)
dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
yield dl
self.assertEqual(list(range(length)), sorted(results))
assert list(range(length)) == sorted(results)

83
tests/test_utils_deprecate.py
View File

@ -1,5 +1,4 @@
import inspect
import unittest
import warnings
from unittest import mock
@ -21,7 +20,7 @@ class NewName(SomeBaseClass):
pass
class WarnWhenSubclassedTest(unittest.TestCase):
class TestWarnWhenSubclassed:
def _mywarnings(self, w, category=MyWarning):
return [x for x in w if x.category is MyWarning]
@ -30,7 +29,7 @@ class WarnWhenSubclassedTest(unittest.TestCase):
create_deprecated_class("Deprecated", NewName)
w = self._mywarnings(w)
self.assertEqual(w, [])
assert w == []
def test_subclassing_warning_message(self):
Deprecated = create_deprecated_class(
@ -43,15 +42,14 @@ class WarnWhenSubclassedTest(unittest.TestCase):
pass
w = self._mywarnings(w)
self.assertEqual(len(w), 1)
self.assertEqual(
str(w[0].message),
"tests.test_utils_deprecate.UserClass inherits from "
assert len(w) == 1
assert (
str(w[0].message) == "tests.test_utils_deprecate.UserClass inherits from "
"deprecated class tests.test_utils_deprecate.Deprecated, "
"please inherit from tests.test_utils_deprecate.NewName."
" (warning only on first subclass, there may be others)",
" (warning only on first subclass, there may be others)"
)
self.assertEqual(w[0].lineno, inspect.getsourcelines(UserClass)[1])
assert w[0].lineno == inspect.getsourcelines(UserClass)[1]
def test_custom_class_paths(self):
Deprecated = create_deprecated_class(
@ -70,11 +68,11 @@ class WarnWhenSubclassedTest(unittest.TestCase):
_ = Deprecated()
w = self._mywarnings(w)
self.assertEqual(len(w), 2)
self.assertIn("foo.NewClass", str(w[0].message))
self.assertIn("bar.OldClass", str(w[0].message))
self.assertIn("foo.NewClass", str(w[1].message))
self.assertIn("bar.OldClass", str(w[1].message))
assert len(w) == 2
assert "foo.NewClass" in str(w[0].message)
assert "bar.OldClass" in str(w[0].message)
assert "foo.NewClass" in str(w[1].message)
assert "bar.OldClass" in str(w[1].message)
def test_subclassing_warns_only_on_direct_children(self):
Deprecated = create_deprecated_class(
@ -90,8 +88,8 @@ class WarnWhenSubclassedTest(unittest.TestCase):
pass
w = self._mywarnings(w)
self.assertEqual(len(w), 1)
self.assertIn("UserClass", str(w[0].message))
assert len(w) == 1
assert "UserClass" in str(w[0].message)
def test_subclassing_warns_once_by_default(self):
Deprecated = create_deprecated_class(
@ -110,8 +108,8 @@ class WarnWhenSubclassedTest(unittest.TestCase):
pass
w = self._mywarnings(w)
self.assertEqual(len(w), 1)
self.assertIn("UserClass", str(w[0].message))
assert len(w) == 1
assert "UserClass" in str(w[0].message)
def test_warning_on_instance(self):
Deprecated = create_deprecated_class(
@ -130,13 +128,12 @@ class WarnWhenSubclassedTest(unittest.TestCase):
_ = UserClass() # subclass instances don't warn
w = self._mywarnings(w)
self.assertEqual(len(w), 1)
self.assertEqual(
str(w[0].message),
"tests.test_utils_deprecate.Deprecated is deprecated, "
"instantiate tests.test_utils_deprecate.NewName instead.",
assert len(w) == 1
assert (
str(w[0].message) == "tests.test_utils_deprecate.Deprecated is deprecated, "
"instantiate tests.test_utils_deprecate.NewName instead."
)
self.assertEqual(w[0].lineno, lineno)
assert w[0].lineno == lineno
def test_warning_auto_message(self):
with warnings.catch_warnings(record=True) as w:
@ -146,8 +143,8 @@ class WarnWhenSubclassedTest(unittest.TestCase):
pass
msg = str(w[0].message)
self.assertIn("tests.test_utils_deprecate.NewName", msg)
self.assertIn("tests.test_utils_deprecate.Deprecated", msg)
assert "tests.test_utils_deprecate.NewName" in msg
assert "tests.test_utils_deprecate.Deprecated" in msg
def test_issubclass(self):
with warnings.catch_warnings():
@ -225,7 +222,7 @@ class WarnWhenSubclassedTest(unittest.TestCase):
warnings.simplefilter("ignore", ScrapyDeprecationWarning)
Deprecated = create_deprecated_class("Deprecated", NewName, {"foo": "bar"})
self.assertEqual(Deprecated.foo, "bar")
assert Deprecated.foo == "bar"
def test_deprecate_a_class_with_custom_metaclass(self):
Meta1 = type("Meta1", (type,), {})
@ -246,7 +243,7 @@ class WarnWhenSubclassedTest(unittest.TestCase):
)
w = self._mywarnings(w)
self.assertEqual(len(w), 0, str(map(str, w)))
assert len(w) == 0, str(map(str, w))
with warnings.catch_warnings(record=True) as w:
AlsoDeprecated()
@ -255,11 +252,11 @@ class WarnWhenSubclassedTest(unittest.TestCase):
pass
w = self._mywarnings(w)
self.assertEqual(len(w), 2)
self.assertIn("AlsoDeprecated", str(w[0].message))
self.assertIn("foo.Bar", str(w[0].message))
self.assertIn("AlsoDeprecated", str(w[1].message))
self.assertIn("foo.Bar", str(w[1].message))
assert len(w) == 2
assert "AlsoDeprecated" in str(w[0].message)
assert "foo.Bar" in str(w[0].message)
assert "AlsoDeprecated" in str(w[1].message)
assert "foo.Bar" in str(w[1].message)
def test_inspect_stack(self):
with (
@ -271,7 +268,7 @@ class WarnWhenSubclassedTest(unittest.TestCase):
class SubClass(DeprecatedName):
pass
self.assertIn("Error detecting parent module", str(w[0].message))
assert "Error detecting parent module" in str(w[0].message)
@mock.patch(
@ -281,27 +278,27 @@ class WarnWhenSubclassedTest(unittest.TestCase):
("scrapy.contrib.", "scrapy.extensions."),
],
)
class UpdateClassPathTest(unittest.TestCase):
class TestUpdateClassPath:
def test_old_path_gets_fixed(self):
with warnings.catch_warnings(record=True) as w:
output = update_classpath("scrapy.contrib.debug.Debug")
self.assertEqual(output, "scrapy.extensions.debug.Debug")
self.assertEqual(len(w), 1)
self.assertIn("scrapy.contrib.debug.Debug", str(w[0].message))
self.assertIn("scrapy.extensions.debug.Debug", str(w[0].message))
assert output == "scrapy.extensions.debug.Debug"
assert len(w) == 1
assert "scrapy.contrib.debug.Debug" in str(w[0].message)
assert "scrapy.extensions.debug.Debug" in str(w[0].message)
def test_sorted_replacement(self):
with warnings.catch_warnings():
warnings.simplefilter("ignore", ScrapyDeprecationWarning)
output = update_classpath("scrapy.contrib.pipeline.Pipeline")
self.assertEqual(output, "scrapy.pipelines.Pipeline")
assert output == "scrapy.pipelines.Pipeline"
def test_unmatched_path_stays_the_same(self):
with warnings.catch_warnings(record=True) as w:
output = update_classpath("scrapy.unmatched.Path")
self.assertEqual(output, "scrapy.unmatched.Path")
self.assertEqual(len(w), 0)
assert output == "scrapy.unmatched.Path"
assert len(w) == 0
def test_returns_nonstring(self):
for notastring in [None, True, [1, 2, 3], object()]:
self.assertEqual(update_classpath(notastring), notastring)
assert update_classpath(notastring) == notastring

20
tests/test_utils_display.py
View File

@ -1,10 +1,10 @@
from io import StringIO
from unittest import TestCase, mock
from unittest import mock
from scrapy.utils.display import pformat, pprint
class TestDisplay(TestCase):
class TestDisplay:
object = {"a": 1}
colorized_strings = {
(
@ -26,15 +26,15 @@ class TestDisplay(TestCase):
@mock.patch("sys.stdout.isatty")
def test_pformat(self, isatty):
isatty.return_value = True
self.assertIn(pformat(self.object), self.colorized_strings)
assert pformat(self.object) in self.colorized_strings
@mock.patch("sys.stdout.isatty")
def test_pformat_dont_colorize(self, isatty):
isatty.return_value = True
self.assertEqual(pformat(self.object, colorize=False), self.plain_string)
assert pformat(self.object, colorize=False) == self.plain_string
def test_pformat_not_tty(self):
self.assertEqual(pformat(self.object), self.plain_string)
assert pformat(self.object) == self.plain_string
@mock.patch("sys.platform", "win32")
@mock.patch("platform.version")
@ -42,7 +42,7 @@ class TestDisplay(TestCase):
def test_pformat_old_windows(self, isatty, version):
isatty.return_value = True
version.return_value = "10.0.14392"
self.assertIn(pformat(self.object), self.colorized_strings)
assert pformat(self.object) in self.colorized_strings
@mock.patch("sys.platform", "win32")
@mock.patch("scrapy.utils.display._enable_windows_terminal_processing")
@ -54,7 +54,7 @@ class TestDisplay(TestCase):
isatty.return_value = True
version.return_value = "10.0.14393"
terminal_processing.return_value = False
self.assertEqual(pformat(self.object), self.plain_string)
assert pformat(self.object) == self.plain_string
@mock.patch("sys.platform", "win32")
@mock.patch("scrapy.utils.display._enable_windows_terminal_processing")
@ -64,7 +64,7 @@ class TestDisplay(TestCase):
isatty.return_value = True
version.return_value = "10.0.14393"
terminal_processing.return_value = True
self.assertIn(pformat(self.object), self.colorized_strings)
assert pformat(self.object) in self.colorized_strings
@mock.patch("sys.platform", "linux")
@mock.patch("sys.stdout.isatty")
@ -81,10 +81,10 @@ class TestDisplay(TestCase):
return real_import(name, globals, locals, fromlist, level)
builtins.__import__ = mock_import
self.assertEqual(pformat(self.object), self.plain_string)
assert pformat(self.object) == self.plain_string
builtins.__import__ = real_import
def test_pprint(self):
with mock.patch("sys.stdout", new=StringIO()) as mock_out:
pprint(self.object)
self.assertEqual(mock_out.getvalue(), "{'a': 1}\n")
assert mock_out.getvalue() == "{'a': 1}\n"

19
tests/test_utils_gz.py
View File

@ -1,4 +1,3 @@
import unittest
from gzip import BadGzipFile
from pathlib import Path
@ -12,17 +11,17 @@ from tests import tests_datadir
SAMPLEDIR = Path(tests_datadir, "compressed")
class GunzipTest(unittest.TestCase):
class TestGunzip:
def test_gunzip_basic(self):
r1 = Response(
"http://www.example.com",
body=(SAMPLEDIR / "feed-sample1.xml.gz").read_bytes(),
)
self.assertTrue(gzip_magic_number(r1))
assert gzip_magic_number(r1)
r2 = Response("http://www.example.com", body=gunzip(r1.body))
self.assertFalse(gzip_magic_number(r2))
self.assertEqual(len(r2.body), 9950)
assert not gzip_magic_number(r2)
assert len(r2.body) == 9950
def test_gunzip_truncated(self):
text = gunzip((SAMPLEDIR / "truncated-crc-error.gz").read_bytes())
@ -37,15 +36,15 @@ class GunzipTest(unittest.TestCase):
"http://www.example.com",
body=(SAMPLEDIR / "truncated-crc-error-short.gz").read_bytes(),
)
self.assertTrue(gzip_magic_number(r1))
assert gzip_magic_number(r1)
r2 = Response("http://www.example.com", body=gunzip(r1.body))
assert r2.body.endswith(b"</html>")
self.assertFalse(gzip_magic_number(r2))
assert not gzip_magic_number(r2)
def test_is_gzipped_empty(self):
r1 = Response("http://www.example.com")
self.assertFalse(gzip_magic_number(r1))
assert not gzip_magic_number(r1)
def test_gunzip_illegal_eof(self):
text = html_to_unicode(
@ -54,5 +53,5 @@ class GunzipTest(unittest.TestCase):
expected_text = (SAMPLEDIR / "unexpected-eof-output.txt").read_text(
encoding="utf-8"
)
self.assertEqual(len(text), len(expected_text))
self.assertEqual(text, expected_text)
assert len(text) == len(expected_text)
assert text == expected_text

3
tests/test_utils_httpobj.py
View File

@ -1,11 +1,10 @@
import unittest
from urllib.parse import urlparse
from scrapy.http import Request
from scrapy.utils.httpobj import urlparse_cached
class HttpobjUtilsTest(unittest.TestCase):
class TestHttpobjUtils:
def test_urlparse_cached(self):
url = "http://www.example.com/index.html"
request1 = Request(url)

274
tests/test_utils_iterators.py
View File

@ -1,5 +1,4 @@
import pytest
from twisted.trial import unittest
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Response, TextResponse, XmlResponse
@ -7,7 +6,7 @@ from scrapy.utils.iterators import _body_or_str, csviter, xmliter, xmliter_lxml
from tests import get_testdata
class XmliterBaseTestCase:
class XmliterBase:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter(self):
body = b"""
@ -35,9 +34,10 @@ class XmliterBaseTestCase:
for x in self.xmliter(response, "product")
]
self.assertEqual(
attrs, [("001", ["Name 1"], ["Type 1"]), ("002", ["Name 2"], ["Type 2"])]
)
assert attrs == [
("001", ["Name 1"], ["Type 1"]),
("002", ["Name 2"], ["Type 2"]),
]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_unusual_node(self):
@ -51,7 +51,7 @@ class XmliterBaseTestCase:
nodenames = [
e.xpath("name()").getall() for e in self.xmliter(response, "matchme...")
]
self.assertEqual(nodenames, [["matchme..."]])
assert nodenames == [["matchme..."]]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_unicode(self):
@ -107,10 +107,11 @@ class XmliterBaseTestCase:
for x in self.xmliter(r, "þingflokkur")
]
self.assertEqual(
attrs,
[("26", ["-"], ["80"]), ("21", ["Ab"], ["76"]), ("27", ["A"], ["27"])],
)
assert attrs == [
("26", ["-"], ["80"]),
("21", ["Ab"], ["76"]),
("27", ["A"], ["27"]),
]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_text(self):
@ -119,10 +120,10 @@ class XmliterBaseTestCase:
"<products><product>one</product><product>two</product></products>"
)
self.assertEqual(
[x.xpath("text()").getall() for x in self.xmliter(body, "product")],
[["one"], ["two"]],
)
assert [x.xpath("text()").getall() for x in self.xmliter(body, "product")] == [
["one"],
["two"],
]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaces(self):
@ -148,21 +149,19 @@ class XmliterBaseTestCase:
my_iter = self.xmliter(response, "item")
node = next(my_iter)
node.register_namespace("g", "http://base.google.com/ns/1.0")
self.assertEqual(node.xpath("title/text()").getall(), ["Item 1"])
self.assertEqual(node.xpath("description/text()").getall(), ["This is item 1"])
self.assertEqual(
node.xpath("link/text()").getall(),
["http://www.mydummycompany.com/items/1"],
)
self.assertEqual(
node.xpath("g:image_link/text()").getall(),
["http://www.mydummycompany.com/images/item1.jpg"],
)
self.assertEqual(node.xpath("g:id/text()").getall(), ["ITEM_1"])
self.assertEqual(node.xpath("g:price/text()").getall(), ["400"])
self.assertEqual(node.xpath("image_link/text()").getall(), [])
self.assertEqual(node.xpath("id/text()").getall(), [])
self.assertEqual(node.xpath("price/text()").getall(), [])
assert node.xpath("title/text()").getall() == ["Item 1"]
assert node.xpath("description/text()").getall() == ["This is item 1"]
assert node.xpath("link/text()").getall() == [
"http://www.mydummycompany.com/items/1"
]
assert node.xpath("g:image_link/text()").getall() == [
"http://www.mydummycompany.com/images/item1.jpg"
]
assert node.xpath("g:id/text()").getall() == ["ITEM_1"]
assert node.xpath("g:price/text()").getall() == ["400"]
assert node.xpath("image_link/text()").getall() == []
assert node.xpath("id/text()").getall() == []
assert node.xpath("price/text()").getall() == []
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaced_nodename(self):
@ -188,10 +187,9 @@ class XmliterBaseTestCase:
my_iter = self.xmliter(response, "g:image_link")
node = next(my_iter)
node.register_namespace("g", "http://base.google.com/ns/1.0")
self.assertEqual(
node.xpath("text()").extract(),
["http://www.mydummycompany.com/images/item1.jpg"],
)
assert node.xpath("text()").extract() == [
"http://www.mydummycompany.com/images/item1.jpg"
]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaced_nodename_missing(self):
@ -246,13 +244,13 @@ class XmliterBaseTestCase:
b"</xml>\n\n"
)
response = XmlResponse("http://www.example.com", body=body)
self.assertEqual(
next(self.xmliter(response, "item")).get(),
"<item>Some Turkish Characters \xd6\xc7\u015e\u0130\u011e\xdc \xfc\u011f\u0131\u015f\xe7\xf6</item>",
assert (
next(self.xmliter(response, "item")).get()
== "<item>Some Turkish Characters \xd6\xc7\u015e\u0130\u011e\xdc \xfc\u011f\u0131\u015f\xe7\xf6</item>"
)
class XmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
class TestXmliter(XmliterBase):
xmliter = staticmethod(xmliter)
def test_deprecation(self):
@ -269,7 +267,7 @@ class XmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
next(self.xmliter(body, "product"))
class LxmlXmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
class TestLxmlXmliter(XmliterBase):
xmliter = staticmethod(xmliter_lxml)
def test_xmliter_iterate_namespace(self):
@ -293,21 +291,19 @@ class LxmlXmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
response = XmlResponse(url="http://mydummycompany.com", body=body)
no_namespace_iter = self.xmliter(response, "image_link")
self.assertEqual(len(list(no_namespace_iter)), 0)
assert len(list(no_namespace_iter)) == 0
namespace_iter = self.xmliter(
response, "image_link", "http://base.google.com/ns/1.0"
)
node = next(namespace_iter)
self.assertEqual(
node.xpath("text()").getall(),
["http://www.mydummycompany.com/images/item1.jpg"],
)
assert node.xpath("text()").getall() == [
"http://www.mydummycompany.com/images/item1.jpg"
]
node = next(namespace_iter)
self.assertEqual(
node.xpath("text()").getall(),
["http://www.mydummycompany.com/images/item2.jpg"],
)
assert node.xpath("text()").getall() == [
"http://www.mydummycompany.com/images/item2.jpg"
]
def test_xmliter_namespaces_prefix(self):
body = b"""
@ -332,16 +328,16 @@ class LxmlXmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
my_iter = self.xmliter(response, "table", "http://www.w3.org/TR/html4/", "h")
node = next(my_iter)
self.assertEqual(len(node.xpath("h:tr/h:td").getall()), 2)
self.assertEqual(node.xpath("h:tr/h:td[1]/text()").getall(), ["Apples"])
self.assertEqual(node.xpath("h:tr/h:td[2]/text()").getall(), ["Bananas"])
assert len(node.xpath("h:tr/h:td").getall()) == 2
assert node.xpath("h:tr/h:td[1]/text()").getall() == ["Apples"]
assert node.xpath("h:tr/h:td[2]/text()").getall() == ["Bananas"]
my_iter = self.xmliter(
response, "table", "http://www.w3schools.com/furniture", "f"
)
node = next(my_iter)
self.assertEqual(node.xpath("f:name/text()").getall(), ["African Coffee Table"])
assert node.xpath("f:name/text()").getall() == ["African Coffee Table"]
def test_xmliter_objtype_exception(self):
i = self.xmliter(42, "product")
@ -349,42 +345,36 @@ class LxmlXmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
next(i)
class UtilsCsvTestCase(unittest.TestCase):
class TestUtilsCsv:
def test_csviter_defaults(self):
body = get_testdata("feeds", "feed-sample3.csv")
response = TextResponse(url="http://example.com/", body=body)
csv = csviter(response)
result = list(csv)
self.assertEqual(
result,
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert result == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
# explicit type check cuz' we no like stinkin' autocasting! yarrr
for result_row in result:
self.assertTrue(all(isinstance(k, str) for k in result_row))
self.assertTrue(all(isinstance(v, str) for v in result_row.values()))
assert all(isinstance(k, str) for k in result_row)
assert all(isinstance(v, str) for v in result_row.values())
def test_csviter_delimiter(self):
body = get_testdata("feeds", "feed-sample3.csv").replace(b",", b"\t")
response = TextResponse(url="http://example.com/", body=body)
csv = csviter(response, delimiter="\t")
self.assertEqual(
list(csv),
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert list(csv) == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
def test_csviter_quotechar(self):
body1 = get_testdata("feeds", "feed-sample6.csv")
@ -393,62 +383,50 @@ class UtilsCsvTestCase(unittest.TestCase):
response1 = TextResponse(url="http://example.com/", body=body1)
csv1 = csviter(response1, quotechar="'")
self.assertEqual(
list(csv1),
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert list(csv1) == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
response2 = TextResponse(url="http://example.com/", body=body2)
csv2 = csviter(response2, delimiter="|", quotechar="'")
self.assertEqual(
list(csv2),
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert list(csv2) == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
def test_csviter_wrong_quotechar(self):
body = get_testdata("feeds", "feed-sample6.csv")
response = TextResponse(url="http://example.com/", body=body)
csv = csviter(response)
self.assertEqual(
list(csv),
[
{"'id'": "1", "'name'": "'alpha'", "'value'": "'foobar'"},
{
"'id'": "2",
"'name'": "'unicode'",
"'value'": "'\xfan\xedc\xf3d\xe9\u203d'",
},
{"'id'": "'3'", "'name'": "'multi'", "'value'": "'foo"},
{"'id'": "4", "'name'": "'empty'", "'value'": ""},
],
)
assert list(csv) == [
{"'id'": "1", "'name'": "'alpha'", "'value'": "'foobar'"},
{
"'id'": "2",
"'name'": "'unicode'",
"'value'": "'\xfan\xedc\xf3d\xe9\u203d'",
},
{"'id'": "'3'", "'name'": "'multi'", "'value'": "'foo"},
{"'id'": "4", "'name'": "'empty'", "'value'": ""},
]
def test_csviter_delimiter_binary_response_assume_utf8_encoding(self):
body = get_testdata("feeds", "feed-sample3.csv").replace(b",", b"\t")
response = Response(url="http://example.com/", body=body)
csv = csviter(response, delimiter="\t")
self.assertEqual(
list(csv),
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert list(csv) == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
def test_csviter_headers(self):
sample = get_testdata("feeds", "feed-sample3.csv").splitlines()
@ -457,15 +435,12 @@ class UtilsCsvTestCase(unittest.TestCase):
response = TextResponse(url="http://example.com/", body=body)
csv = csviter(response, headers=[h.decode("utf-8") for h in headers])
self.assertEqual(
list(csv),
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert list(csv) == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
def test_csviter_falserow(self):
body = get_testdata("feeds", "feed-sample3.csv")
@ -474,15 +449,12 @@ class UtilsCsvTestCase(unittest.TestCase):
response = TextResponse(url="http://example.com/", body=body)
csv = csviter(response)
self.assertEqual(
list(csv),
[
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
],
)
assert list(csv) == [
{"id": "1", "name": "alpha", "value": "foobar"},
{"id": "2", "name": "unicode", "value": "\xfan\xedc\xf3d\xe9\u203d"},
{"id": "3", "name": "multi", "value": "foo\nbar"},
{"id": "4", "name": "empty", "value": ""},
]
def test_csviter_exception(self):
body = get_testdata("feeds", "feed-sample3.csv")
@ -504,30 +476,24 @@ class UtilsCsvTestCase(unittest.TestCase):
url="http://example.com/", body=body1, encoding="latin1"
)
csv = csviter(response)
self.assertEqual(
list(csv),
[
{"id": "1", "name": "latin1", "value": "test"},
{"id": "2", "name": "something", "value": "\xf1\xe1\xe9\xf3"},
],
)
assert list(csv) == [
{"id": "1", "name": "latin1", "value": "test"},
{"id": "2", "name": "something", "value": "\xf1\xe1\xe9\xf3"},
]
response = TextResponse(url="http://example.com/", body=body2, encoding="cp852")
csv = csviter(response)
self.assertEqual(
list(csv),
[
{"id": "1", "name": "cp852", "value": "test"},
{
"id": "2",
"name": "something",
"value": "\u255a\u2569\u2569\u2569\u2550\u2550\u2557",
},
],
)
assert list(csv) == [
{"id": "1", "name": "cp852", "value": "test"},
{
"id": "2",
"name": "something",
"value": "\u255a\u2569\u2569\u2569\u2550\u2550\u2557",
},
]
class TestHelper(unittest.TestCase):
class TestHelper:
bbody = b"utf8-body"
ubody = bbody.decode("utf8")
txtresponse = TextResponse(url="http://example.org/", body=bbody, encoding="utf-8")
@ -541,11 +507,9 @@ class TestHelper(unittest.TestCase):
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)
self.assertTrue(type(r1) is type(r2))
self.assertTrue(type(r1) is not type(r3))
assert type(r1) is type(r2)
assert type(r1) is not type(r3)
def _assert_type_and_value(self, a, b, obj):
self.assertTrue(
type(a) is type(b), f"Got {type(a)}, expected {type(b)} for {obj!r}"
)
self.assertEqual(a, b)
assert type(a) is type(b), f"Got {type(a)}, expected {type(b)} for {obj!r}"
assert a == b

43
tests/test_utils_log.py
View File

@ -7,7 +7,6 @@ import sys
import unittest
from io import StringIO
from typing import TYPE_CHECKING, Any
from unittest import TestCase
import pytest
from testfixtures import LogCapture
@ -27,7 +26,7 @@ if TYPE_CHECKING:
from collections.abc import Mapping, MutableMapping
class FailureToExcInfoTest(unittest.TestCase):
class TestFailureToExcInfo:
def test_failure(self):
try:
0 / 0
@ -35,14 +34,14 @@ class FailureToExcInfoTest(unittest.TestCase):
exc_info = sys.exc_info()
failure = Failure()
self.assertTupleEqual(exc_info, failure_to_exc_info(failure))
assert exc_info == failure_to_exc_info(failure)
def test_non_failure(self):
self.assertIsNone(failure_to_exc_info("test"))
assert failure_to_exc_info("test") is None
class TopLevelFormatterTest(unittest.TestCase):
def setUp(self):
class TestTopLevelFormatter:
def setup_method(self):
self.handler = LogCapture()
self.handler.addFilter(TopLevelFormatter(["test"]))
@ -71,8 +70,8 @@ class TopLevelFormatterTest(unittest.TestCase):
log.check(("different", "WARNING", "test log msg"))
class LogCounterHandlerTest(unittest.TestCase):
def setUp(self):
class TestLogCounterHandler:
def setup_method(self):
settings = {"LOG_LEVEL": "WARNING"}
self.logger = logging.getLogger("test")
self.logger.setLevel(logging.NOTSET)
@ -81,24 +80,24 @@ class LogCounterHandlerTest(unittest.TestCase):
self.handler = LogCounterHandler(self.crawler)
self.logger.addHandler(self.handler)
def tearDown(self):
def teardown_method(self):
self.logger.propagate = True
self.logger.removeHandler(self.handler)
def test_init(self):
self.assertIsNone(self.crawler.stats.get_value("log_count/DEBUG"))
self.assertIsNone(self.crawler.stats.get_value("log_count/INFO"))
self.assertIsNone(self.crawler.stats.get_value("log_count/WARNING"))
self.assertIsNone(self.crawler.stats.get_value("log_count/ERROR"))
self.assertIsNone(self.crawler.stats.get_value("log_count/CRITICAL"))
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):
self.logger.error("test log msg")
self.assertEqual(self.crawler.stats.get_value("log_count/ERROR"), 1)
assert self.crawler.stats.get_value("log_count/ERROR") == 1
def test_filtered_out_level(self):
self.logger.debug("test log msg")
self.assertIsNone(self.crawler.stats.get_value("log_count/INFO"))
assert self.crawler.stats.get_value("log_count/INFO") is None
class StreamLoggerTest(unittest.TestCase):
@ -152,8 +151,8 @@ def test_spider_logger_adapter_process(
assert result_kwargs == expected_extra
class LoggingTestCase(TestCase):
def setUp(self):
class TestLogging:
def setup_method(self):
self.log_stream = StringIO()
handler = logging.StreamHandler(self.log_stream)
logger = logging.getLogger("log_spider")
@ -163,7 +162,7 @@ class LoggingTestCase(TestCase):
self.logger = logger
self.spider = LogSpider()
def tearDown(self):
def teardown_method(self):
self.logger.removeHandler(self.handler)
def test_debug_logging(self):
@ -202,8 +201,8 @@ class LoggingTestCase(TestCase):
assert log_contents == f"{log_message}\n"
class LoggingWithExtraTestCase(TestCase):
def setUp(self):
class TestLoggingWithExtra:
def setup_method(self):
self.log_stream = StringIO()
handler = logging.StreamHandler(self.log_stream)
formatter = logging.Formatter(
@ -218,7 +217,7 @@ class LoggingWithExtraTestCase(TestCase):
self.spider = LogSpider()
self.regex_pattern = re.compile(r"^<LogSpider\s'log_spider'\sat\s[^>]+>$")
def tearDown(self):
def teardown_method(self):
self.logger.removeHandler(self.handler)
def test_debug_logging(self):

51
tests/test_utils_misc/__init__.py
View File

@ -1,6 +1,5 @@
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
@ -18,18 +17,18 @@ from scrapy.utils.misc import (
)
class UtilsMiscTestCase(unittest.TestCase):
class TestUtilsMisc:
def test_load_object_class(self):
obj = load_object(Field)
self.assertIs(obj, Field)
assert obj is Field
obj = load_object("scrapy.item.Field")
self.assertIs(obj, Field)
assert obj is Field
def test_load_object_function(self):
obj = load_object(load_object)
self.assertIs(obj, load_object)
assert obj is load_object
obj = load_object("scrapy.utils.misc.load_object")
self.assertIs(obj, load_object)
assert obj is load_object
def test_load_object_exceptions(self):
with pytest.raises(ImportError):
@ -47,20 +46,20 @@ class UtilsMiscTestCase(unittest.TestCase):
"tests.test_utils_misc.test_walk_modules.mod.mod0",
"tests.test_utils_misc.test_walk_modules.mod1",
]
self.assertEqual({m.__name__ for m in mods}, set(expected))
assert {m.__name__ for m in mods} == set(expected)
mods = walk_modules("tests.test_utils_misc.test_walk_modules.mod")
expected = [
"tests.test_utils_misc.test_walk_modules.mod",
"tests.test_utils_misc.test_walk_modules.mod.mod0",
]
self.assertEqual({m.__name__ for m in mods}, set(expected))
assert {m.__name__ for m in mods} == set(expected)
mods = walk_modules("tests.test_utils_misc.test_walk_modules.mod1")
expected = [
"tests.test_utils_misc.test_walk_modules.mod1",
]
self.assertEqual({m.__name__ for m in mods}, set(expected))
assert {m.__name__ for m in mods} == set(expected)
with pytest.raises(ImportError):
walk_modules("nomodule999")
@ -76,7 +75,7 @@ class UtilsMiscTestCase(unittest.TestCase):
"testegg.spiders.b",
"testegg",
]
self.assertEqual({m.__name__ for m in mods}, set(expected))
assert {m.__name__ for m in mods} == set(expected)
finally:
sys.path.remove(egg)
@ -90,15 +89,13 @@ class UtilsMiscTestCase(unittest.TestCase):
assert hasattr(arg_to_iter([1, 2, 3]), "__iter__")
assert hasattr(arg_to_iter(c for c in "abcd"), "__iter__")
self.assertEqual(list(arg_to_iter(None)), [])
self.assertEqual(list(arg_to_iter("lala")), ["lala"])
self.assertEqual(list(arg_to_iter(100)), [100])
self.assertEqual(list(arg_to_iter(c for c in "abc")), ["a", "b", "c"])
self.assertEqual(list(arg_to_iter([1, 2, 3])), [1, 2, 3])
self.assertEqual(list(arg_to_iter({"a": 1})), [{"a": 1}])
self.assertEqual(
list(arg_to_iter(TestItem(name="john"))), [TestItem(name="john")]
)
assert not list(arg_to_iter(None))
assert list(arg_to_iter("lala")) == ["lala"]
assert list(arg_to_iter(100)) == [100]
assert list(arg_to_iter(c for c in "abc")) == ["a", "b", "c"]
assert list(arg_to_iter([1, 2, 3])) == [1, 2, 3]
assert list(arg_to_iter({"a": 1})) == [{"a": 1}]
assert list(arg_to_iter(TestItem(name="john"))) == [TestItem(name="john")]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_create_instance(self):
@ -110,10 +107,10 @@ class UtilsMiscTestCase(unittest.TestCase):
def _test_with_settings(mock, settings):
create_instance(mock, settings, None, *args, **kwargs)
if hasattr(mock, "from_crawler"):
self.assertEqual(mock.from_crawler.call_count, 0)
assert mock.from_crawler.call_count == 0
if hasattr(mock, "from_settings"):
mock.from_settings.assert_called_once_with(settings, *args, **kwargs)
self.assertEqual(mock.call_count, 0)
assert mock.call_count == 0
else:
mock.assert_called_once_with(*args, **kwargs)
@ -122,11 +119,11 @@ class UtilsMiscTestCase(unittest.TestCase):
if hasattr(mock, "from_crawler"):
mock.from_crawler.assert_called_once_with(crawler, *args, **kwargs)
if hasattr(mock, "from_settings"):
self.assertEqual(mock.from_settings.call_count, 0)
self.assertEqual(mock.call_count, 0)
assert mock.from_settings.call_count == 0
assert mock.call_count == 0
elif hasattr(mock, "from_settings"):
mock.from_settings.assert_called_once_with(settings, *args, **kwargs)
self.assertEqual(mock.call_count, 0)
assert mock.call_count == 0
else:
mock.assert_called_once_with(*args, **kwargs)
@ -172,11 +169,11 @@ class UtilsMiscTestCase(unittest.TestCase):
if hasattr(mock, "from_crawler"):
mock.from_crawler.assert_called_once_with(crawler, *args, **kwargs)
if hasattr(mock, "from_settings"):
self.assertEqual(mock.from_settings.call_count, 0)
self.assertEqual(mock.call_count, 0)
assert mock.from_settings.call_count == 0
assert mock.call_count == 0
elif hasattr(mock, "from_settings"):
mock.from_settings.assert_called_once_with(settings, *args, **kwargs)
self.assertEqual(mock.call_count, 0)
assert mock.call_count == 0
else:
mock.assert_called_once_with(*args, **kwargs)

63
tests/test_utils_misc/test_return_with_argument_inside_generator.py
View File

@ -1,4 +1,3 @@
import unittest
import warnings
from functools import partial
from unittest import mock
@ -40,7 +39,7 @@ def generator_that_returns_stuff():
return 3
class UtilsMiscPy3TestCase(unittest.TestCase):
class TestUtilsMisc:
def test_generators_return_something(self):
def f1():
yield 1
@ -77,27 +76,27 @@ https://example.org
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, top_level_return_something)
self.assertEqual(len(w), 1)
self.assertIn(
'The "NoneType.top_level_return_something" method is a generator',
str(w[0].message),
assert len(w) == 1
assert (
'The "NoneType.top_level_return_something" method is a generator'
in str(w[0].message)
)
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, f1)
self.assertEqual(len(w), 1)
self.assertIn('The "NoneType.f1" method is a generator', str(w[0].message))
assert len(w) == 1
assert 'The "NoneType.f1" method is a generator' in str(w[0].message)
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, g1)
self.assertEqual(len(w), 1)
self.assertIn('The "NoneType.g1" method is a generator', str(w[0].message))
assert len(w) == 1
assert 'The "NoneType.g1" method is a generator' in str(w[0].message)
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, h1)
self.assertEqual(len(w), 1)
self.assertIn('The "NoneType.h1" method is a generator', str(w[0].message))
assert len(w) == 1
assert 'The "NoneType.h1" method is a generator' in str(w[0].message)
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, i1)
self.assertEqual(len(w), 1)
self.assertIn('The "NoneType.i1" method is a generator', str(w[0].message))
assert len(w) == 1
assert 'The "NoneType.i1" method is a generator' in str(w[0].message)
def test_generators_return_none(self):
def f2():
@ -144,28 +143,28 @@ https://example.org
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, top_level_return_none)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, f2)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, g2)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, h2)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, i2)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, j2)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, k2)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, l2)
self.assertEqual(len(w), 0)
assert len(w) == 0
def test_generators_return_none_with_decorator(self):
def decorator(func):
@ -225,28 +224,28 @@ https://example.org
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, top_level_return_none)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, f3)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, g3)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, h3)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, i3)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, j3)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, k3)
self.assertEqual(len(w), 0)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, l3)
self.assertEqual(len(w), 0)
assert len(w) == 0
@mock.patch(
"scrapy.utils.misc.is_generator_with_return_value", new=_indentation_error
@ -254,8 +253,8 @@ https://example.org
def test_indentation_error(self):
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(None, top_level_return_none)
self.assertEqual(len(w), 1)
self.assertIn("Unable to determine", str(w[0].message))
assert len(w) == 1
assert "Unable to determine" in str(w[0].message)
def test_partial(self):
def cb(arg1, arg2):

13
tests/test_utils_project.py
View File

@ -2,7 +2,6 @@ import contextlib
import os
import shutil
import tempfile
import unittest
import warnings
from pathlib import Path
@ -25,21 +24,21 @@ def inside_a_project():
shutil.rmtree(project_dir)
class ProjectUtilsTest(unittest.TestCase):
class TestProjectUtils:
def test_data_path_outside_project(self):
self.assertEqual(str(Path(".scrapy", "somepath")), data_path("somepath"))
assert str(Path(".scrapy", "somepath")) == data_path("somepath")
abspath = str(Path(os.path.sep, "absolute", "path"))
self.assertEqual(abspath, data_path(abspath))
assert abspath == data_path(abspath)
def test_data_path_inside_project(self):
with inside_a_project() as proj_path:
expected = Path(proj_path, ".scrapy", "somepath")
self.assertEqual(expected.resolve(), Path(data_path("somepath")).resolve())
assert expected.resolve() == Path(data_path("somepath")).resolve()
abspath = str(Path(os.path.sep, "absolute", "path").resolve())
self.assertEqual(abspath, data_path(abspath))
assert abspath == data_path(abspath)
class GetProjectSettingsTestCase(unittest.TestCase):
class TestGetProjectSettings:
def test_valid_envvar(self):
value = "tests.test_cmdline.settings"
envvars = {

113
tests/test_utils_python.py
View File

@ -21,18 +21,18 @@ from scrapy.utils.python import (
)
class MutableChainTest(unittest.TestCase):
class TestMutableChain:
def test_mutablechain(self):
m = MutableChain(range(2), [2, 3], (4, 5))
m.extend(range(6, 7))
m.extend([7, 8])
m.extend([9, 10], (11, 12))
self.assertEqual(next(m), 0)
self.assertEqual(m.__next__(), 1)
self.assertEqual(list(m), list(range(2, 13)))
assert next(m) == 0
assert m.__next__() == 1
assert list(m) == list(range(2, 13))
class MutableAsyncChainTest(unittest.TestCase):
class TestMutableAsyncChain(unittest.TestCase):
@staticmethod
async def g1():
for i in range(3):
@ -62,9 +62,9 @@ class MutableAsyncChainTest(unittest.TestCase):
m.extend(self.g2())
m.extend(self.g3())
self.assertEqual(await m.__anext__(), 0)
assert await m.__anext__() == 0
results = await collect_asyncgen(m)
self.assertEqual(results, list(range(1, 10)))
assert results == list(range(1, 10))
@deferred_f_from_coro_f
async def test_mutableasyncchain_exc(self):
@ -73,46 +73,46 @@ class MutableAsyncChainTest(unittest.TestCase):
m.extend(self.g3())
results = await collect_asyncgen(aiter_errback(m, lambda _: None))
self.assertEqual(results, list(range(5)))
assert results == list(range(5))
class ToUnicodeTest(unittest.TestCase):
class TestToUnicode:
def test_converting_an_utf8_encoded_string_to_unicode(self):
self.assertEqual(to_unicode(b"lel\xc3\xb1e"), "lel\xf1e")
assert to_unicode(b"lel\xc3\xb1e") == "lel\xf1e"
def test_converting_a_latin_1_encoded_string_to_unicode(self):
self.assertEqual(to_unicode(b"lel\xf1e", "latin-1"), "lel\xf1e")
assert to_unicode(b"lel\xf1e", "latin-1") == "lel\xf1e"
def test_converting_a_unicode_to_unicode_should_return_the_same_object(self):
self.assertEqual(to_unicode("\xf1e\xf1e\xf1e"), "\xf1e\xf1e\xf1e")
assert to_unicode("\xf1e\xf1e\xf1e") == "\xf1e\xf1e\xf1e"
def test_converting_a_strange_object_should_raise_TypeError(self):
def test_converting_a_strange_object_should_raise_type_error(self):
with pytest.raises(TypeError):
to_unicode(423)
def test_errors_argument(self):
self.assertEqual(to_unicode(b"a\xedb", "utf-8", errors="replace"), "a\ufffdb")
assert to_unicode(b"a\xedb", "utf-8", errors="replace") == "a\ufffdb"
class ToBytesTest(unittest.TestCase):
class TestToBytes:
def test_converting_a_unicode_object_to_an_utf_8_encoded_string(self):
self.assertEqual(to_bytes("\xa3 49"), b"\xc2\xa3 49")
assert to_bytes("\xa3 49") == b"\xc2\xa3 49"
def test_converting_a_unicode_object_to_a_latin_1_encoded_string(self):
self.assertEqual(to_bytes("\xa3 49", "latin-1"), b"\xa3 49")
assert to_bytes("\xa3 49", "latin-1") == b"\xa3 49"
def test_converting_a_regular_bytes_to_bytes_should_return_the_same_object(self):
self.assertEqual(to_bytes(b"lel\xf1e"), b"lel\xf1e")
assert to_bytes(b"lel\xf1e") == b"lel\xf1e"
def test_converting_a_strange_object_should_raise_TypeError(self):
def test_converting_a_strange_object_should_raise_type_error(self):
with pytest.raises(TypeError):
to_bytes(pytest)
def test_errors_argument(self):
self.assertEqual(to_bytes("a\ufffdb", "latin-1", errors="replace"), b"a?b")
assert to_bytes("a\ufffdb", "latin-1", errors="replace") == b"a?b"
class MemoizedMethodTest(unittest.TestCase):
class TestMemoizedMethod:
def test_memoizemethod_noargs(self):
class A:
@memoizemethod_noargs
@ -130,7 +130,7 @@ class MemoizedMethodTest(unittest.TestCase):
assert one is not three
class BinaryIsTextTest(unittest.TestCase):
class TestBinaryIsText:
def test_binaryistext(self):
assert binary_is_text(b"hello")
@ -144,7 +144,7 @@ class BinaryIsTextTest(unittest.TestCase):
assert not binary_is_text(b"\x02\xa3")
class UtilsPythonTestCase(unittest.TestCase):
class TestUtilsPython:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_equal_attributes(self):
class Obj:
@ -153,31 +153,31 @@ class UtilsPythonTestCase(unittest.TestCase):
a = Obj()
b = Obj()
# no attributes given return False
self.assertFalse(equal_attributes(a, b, []))
assert not equal_attributes(a, b, [])
# nonexistent attributes
self.assertFalse(equal_attributes(a, b, ["x", "y"]))
assert not equal_attributes(a, b, ["x", "y"])
a.x = 1
b.x = 1
# equal attribute
self.assertTrue(equal_attributes(a, b, ["x"]))
assert equal_attributes(a, b, ["x"])
b.y = 2
# obj1 has no attribute y
self.assertFalse(equal_attributes(a, b, ["x", "y"]))
assert not equal_attributes(a, b, ["x", "y"])
a.y = 2
# equal attributes
self.assertTrue(equal_attributes(a, b, ["x", "y"]))
assert equal_attributes(a, b, ["x", "y"])
a.y = 1
# different attributes
self.assertFalse(equal_attributes(a, b, ["x", "y"]))
assert not equal_attributes(a, b, ["x", "y"])
# test callable
a.meta = {}
b.meta = {}
self.assertTrue(equal_attributes(a, b, ["meta"]))
assert equal_attributes(a, b, ["meta"])
# compare ['meta']['a']
a.meta["z"] = 1
@ -189,10 +189,10 @@ class UtilsPythonTestCase(unittest.TestCase):
def compare_z(obj):
return get_z(get_meta(obj))
self.assertTrue(equal_attributes(a, b, [compare_z, "x"]))
assert equal_attributes(a, b, [compare_z, "x"])
# fail z equality
a.meta["z"] = 2
self.assertFalse(equal_attributes(a, b, [compare_z, "x"]))
assert not equal_attributes(a, b, [compare_z, "x"])
def test_get_func_args(self):
def f1(a, b, c):
@ -221,36 +221,35 @@ class UtilsPythonTestCase(unittest.TestCase):
partial_f2 = functools.partial(f1, b=None)
partial_f3 = functools.partial(partial_f2, None)
self.assertEqual(get_func_args(f1), ["a", "b", "c"])
self.assertEqual(get_func_args(f2), ["a", "b", "c"])
self.assertEqual(get_func_args(f3), ["a", "b", "c"])
self.assertEqual(get_func_args(A), ["a", "b", "c"])
self.assertEqual(get_func_args(a.method), ["a", "b", "c"])
self.assertEqual(get_func_args(partial_f1), ["b", "c"])
self.assertEqual(get_func_args(partial_f2), ["a", "c"])
self.assertEqual(get_func_args(partial_f3), ["c"])
self.assertEqual(get_func_args(cal), ["a", "b", "c"])
self.assertEqual(get_func_args(object), [])
self.assertEqual(get_func_args(str.split, stripself=True), ["sep", "maxsplit"])
self.assertEqual(get_func_args(" ".join, stripself=True), ["iterable"])
assert get_func_args(f1) == ["a", "b", "c"]
assert get_func_args(f2) == ["a", "b", "c"]
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"]
if sys.version_info >= (3, 13) or platform.python_implementation() == "PyPy":
# the correct and correctly extracted signature
self.assertEqual(
get_func_args(operator.itemgetter(2), stripself=True), ["obj"]
)
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)
self.assertIn(
get_func_args(operator.itemgetter(2), stripself=True),
[[], ["args", "kwargs"]],
)
assert get_func_args(operator.itemgetter(2), stripself=True) in [
[],
["args", "kwargs"],
]
def test_without_none_values(self):
self.assertEqual(without_none_values([1, None, 3, 4]), [1, 3, 4])
self.assertEqual(without_none_values((1, None, 3, 4)), (1, 3, 4))
self.assertEqual(
without_none_values({"one": 1, "none": None, "three": 3, "four": 4}),
{"one": 1, "three": 3, "four": 4},
)
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,
}

106
tests/test_utils_request.py
View File

@ -1,7 +1,6 @@
from __future__ import annotations
import json
import unittest
import warnings
from hashlib import sha1
from weakref import WeakKeyDictionary
@ -21,23 +20,23 @@ from scrapy.utils.request import (
from scrapy.utils.test import get_crawler
class UtilsRequestTest(unittest.TestCase):
class TestUtilsRequest:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_request_authenticate(self):
r = Request("http://www.example.com")
request_authenticate(r, "someuser", "somepass")
self.assertEqual(r.headers["Authorization"], b"Basic c29tZXVzZXI6c29tZXBhc3M=")
assert r.headers["Authorization"] == b"Basic c29tZXVzZXI6c29tZXBhc3M="
def test_request_httprepr(self):
r1 = Request("http://www.example.com")
self.assertEqual(
request_httprepr(r1), b"GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n"
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")
self.assertEqual(
request_httprepr(r1),
b"GET /some/page.html?arg=1 HTTP/1.1\r\nHost: www.example.com\r\n\r\n",
assert (
request_httprepr(r1)
== b"GET /some/page.html?arg=1 HTTP/1.1\r\nHost: www.example.com\r\n\r\n"
)
r1 = Request(
@ -46,9 +45,9 @@ class UtilsRequestTest(unittest.TestCase):
headers={"Content-type": b"text/html"},
body=b"Some body",
)
self.assertEqual(
request_httprepr(r1),
b"POST / HTTP/1.1\r\nHost: www.example.com\r\nContent-Type: text/html\r\n\r\nSome 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):
@ -57,7 +56,7 @@ class UtilsRequestTest(unittest.TestCase):
request_httprepr(Request("ftp://localhost/tmp/foo.txt"))
class FingerprintTest(unittest.TestCase):
class TestFingerprint:
maxDiff = None
function: staticmethod = staticmethod(fingerprint)
@ -147,23 +146,23 @@ class FingerprintTest(unittest.TestCase):
def test_query_string_key_order(self):
r1 = Request("http://www.example.com/query?id=111&cat=222")
r2 = Request("http://www.example.com/query?cat=222&id=111")
self.assertEqual(self.function(r1), self.function(r1))
self.assertEqual(self.function(r1), self.function(r2))
assert self.function(r1) == self.function(r1)
assert self.function(r1) == self.function(r2)
def test_query_string_key_without_value(self):
r1 = Request("http://www.example.com/hnnoticiaj1.aspx?78132,199")
r2 = Request("http://www.example.com/hnnoticiaj1.aspx?78160,199")
self.assertNotEqual(self.function(r1), self.function(r2))
assert self.function(r1) != self.function(r2)
def test_caching(self):
r1 = Request("http://www.example.com/hnnoticiaj1.aspx?78160,199")
self.assertEqual(self.function(r1), self.cache[r1][self.default_cache_key])
assert self.function(r1) == self.cache[r1][self.default_cache_key]
def test_header(self):
r1 = Request("http://www.example.com/members/offers.html")
r2 = Request("http://www.example.com/members/offers.html")
r2.headers["SESSIONID"] = b"somehash"
self.assertEqual(self.function(r1), self.function(r2))
assert self.function(r1) == self.function(r2)
def test_headers(self):
r1 = Request("http://www.example.com/")
@ -173,36 +172,35 @@ class FingerprintTest(unittest.TestCase):
r3.headers["Accept-Language"] = b"en"
r3.headers["SESSIONID"] = b"somehash"
self.assertEqual(self.function(r1), self.function(r2), self.function(r3))
assert self.function(r1) == self.function(r2) == self.function(r3)
self.assertEqual(
self.function(r1), self.function(r1, include_headers=["Accept-Language"])
assert self.function(r1) == self.function(
r1, include_headers=["Accept-Language"]
)
self.assertNotEqual(
self.function(r1), self.function(r2, include_headers=["Accept-Language"])
assert self.function(r1) != self.function(
r2, include_headers=["Accept-Language"]
)
self.assertEqual(
self.function(r3, include_headers=["accept-language", "sessionid"]),
self.function(r3, include_headers=["SESSIONID", "Accept-Language"]),
)
assert self.function(
r3, include_headers=["accept-language", "sessionid"]
) == self.function(r3, include_headers=["SESSIONID", "Accept-Language"])
def test_fragment(self):
r1 = Request("http://www.example.com/test.html")
r2 = Request("http://www.example.com/test.html#fragment")
self.assertEqual(self.function(r1), self.function(r2))
self.assertEqual(self.function(r1), self.function(r1, keep_fragments=True))
self.assertNotEqual(self.function(r2), self.function(r2, keep_fragments=True))
self.assertNotEqual(self.function(r1), self.function(r2, keep_fragments=True))
assert self.function(r1) == self.function(r2)
assert self.function(r1) == self.function(r1, keep_fragments=True)
assert self.function(r2) != self.function(r2, keep_fragments=True)
assert self.function(r1) != self.function(r2, keep_fragments=True)
def test_method_and_body(self):
r1 = Request("http://www.example.com")
r2 = Request("http://www.example.com", method="POST")
r3 = Request("http://www.example.com", method="POST", body=b"request body")
self.assertNotEqual(self.function(r1), self.function(r2))
self.assertNotEqual(self.function(r2), self.function(r3))
assert self.function(r1) != self.function(r2)
assert self.function(r2) != self.function(r3)
def test_request_replace(self):
# cached fingerprint must be cleared on request copy
@ -210,7 +208,7 @@ class FingerprintTest(unittest.TestCase):
fp1 = self.function(r1)
r2 = r1.replace(url="http://www.example.com/other")
fp2 = self.function(r2)
self.assertNotEqual(fp1, fp2)
assert fp1 != fp2
def test_part_separation(self):
# An old implementation used to serialize request data in a way that
@ -219,7 +217,7 @@ class FingerprintTest(unittest.TestCase):
fp1 = self.function(r1)
r2 = Request("http://www.example.com/f", body=b"oo")
fp2 = self.function(r2)
self.assertNotEqual(fp1, fp2)
assert fp1 != fp2
def test_hashes(self):
"""Test hardcoded hashes, to make sure future changes to not introduce
@ -228,7 +226,7 @@ class FingerprintTest(unittest.TestCase):
self.function(request, **kwargs) for request, _, kwargs in self.known_hashes
]
expected = [_fingerprint for _, _fingerprint, _ in self.known_hashes]
self.assertEqual(actual, expected)
assert actual == expected
REQUEST_OBJECTS_TO_TEST = (
@ -260,13 +258,12 @@ REQUEST_OBJECTS_TO_TEST = (
)
class RequestFingerprinterTestCase(unittest.TestCase):
class TestRequestFingerprinter:
def test_default_implementation(self):
crawler = get_crawler()
request = Request("https://example.com")
self.assertEqual(
crawler.request_fingerprinter.fingerprint(request),
fingerprint(request),
assert crawler.request_fingerprinter.fingerprint(request) == fingerprint(
request
)
def test_deprecated_implementation(self):
@ -276,14 +273,13 @@ class RequestFingerprinterTestCase(unittest.TestCase):
with warnings.catch_warnings(record=True) as logged_warnings:
crawler = get_crawler(settings_dict=settings)
request = Request("https://example.com")
self.assertEqual(
crawler.request_fingerprinter.fingerprint(request),
fingerprint(request),
assert crawler.request_fingerprinter.fingerprint(request) == fingerprint(
request
)
self.assertTrue(logged_warnings)
assert logged_warnings
class CustomRequestFingerprinterTestCase(unittest.TestCase):
class TestCustomRequestFingerprinter:
def test_include_headers(self):
class RequestFingerprinter:
def fingerprint(self, request):
@ -298,7 +294,7 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
fp1 = crawler.request_fingerprinter.fingerprint(r1)
r2 = Request("http://www.example.com", headers={"X-ID": "2"})
fp2 = crawler.request_fingerprinter.fingerprint(r2)
self.assertNotEqual(fp1, fp2)
assert fp1 != fp2
def test_dont_canonicalize(self):
class RequestFingerprinter:
@ -320,7 +316,7 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
fp1 = crawler.request_fingerprinter.fingerprint(r1)
r2 = Request("http://www.example.com?a=2&a=1")
fp2 = crawler.request_fingerprinter.fingerprint(r2)
self.assertNotEqual(fp1, fp2)
assert fp1 != fp2
def test_meta(self):
class RequestFingerprinter:
@ -342,10 +338,10 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
fp3 = crawler.request_fingerprinter.fingerprint(r3)
r4 = Request("http://www.example.com", meta={"fingerprint": "b"})
fp4 = crawler.request_fingerprinter.fingerprint(r4)
self.assertNotEqual(fp1, fp2)
self.assertNotEqual(fp1, fp4)
self.assertNotEqual(fp2, fp4)
self.assertEqual(fp2, fp3)
assert fp1 != fp2
assert fp1 != fp4
assert fp2 != fp4
assert fp2 == fp3
def test_from_crawler(self):
class RequestFingerprinter:
@ -367,7 +363,7 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
request = Request("http://www.example.com")
fingerprint = crawler.request_fingerprinter.fingerprint(request)
self.assertEqual(fingerprint, settings["FINGERPRINT"])
assert fingerprint == settings["FINGERPRINT"]
def test_from_settings(self):
class RequestFingerprinter:
@ -391,7 +387,7 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
request = Request("http://www.example.com")
fingerprint = crawler.request_fingerprinter.fingerprint(request)
self.assertEqual(fingerprint, settings["FINGERPRINT"])
assert fingerprint == settings["FINGERPRINT"]
def test_from_crawler_and_settings(self):
class RequestFingerprinter:
@ -418,13 +414,13 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
request = Request("http://www.example.com")
fingerprint = crawler.request_fingerprinter.fingerprint(request)
self.assertEqual(fingerprint, settings["FINGERPRINT"])
assert fingerprint == settings["FINGERPRINT"]
class RequestToCurlTest(unittest.TestCase):
class TestRequestToCurl:
def _test_request(self, request_object, expected_curl_command):
curl_command = request_to_curl(request_object)
self.assertEqual(curl_command, expected_curl_command)
assert curl_command == expected_curl_command
def test_get(self):
request_object = Request("https://www.example.com")

27
tests/test_utils_response.py
View File

@ -1,4 +1,3 @@
import unittest
from pathlib import Path
from time import process_time
from urllib.parse import urlparse
@ -16,7 +15,7 @@ from scrapy.utils.response import (
)
class ResponseUtilsTest(unittest.TestCase):
class TestResponseUtils:
dummy_response = TextResponse(url="http://example.org/", body=b"dummy_response")
def test_open_in_browser(self):
@ -28,7 +27,7 @@ class ResponseUtilsTest(unittest.TestCase):
if not path or not Path(path).exists():
path = burl.replace("file://", "")
bbody = Path(path).read_bytes()
self.assertIn(b'<base href="' + to_bytes(url) + b'">', bbody)
assert b'<base href="' + to_bytes(url) + b'">' in bbody
return True
response = HtmlResponse(url, body=body)
@ -68,9 +67,9 @@ class ResponseUtilsTest(unittest.TestCase):
</script>
""",
)
self.assertEqual(get_meta_refresh(r1), (5.0, "http://example.org/newpage"))
self.assertEqual(get_meta_refresh(r2), (None, None))
self.assertEqual(get_meta_refresh(r3), (None, None))
assert get_meta_refresh(r1) == (5.0, "http://example.org/newpage")
assert get_meta_refresh(r2) == (None, None)
assert get_meta_refresh(r3) == (None, None)
def test_get_base_url(self):
resp = HtmlResponse(
@ -81,19 +80,19 @@ class ResponseUtilsTest(unittest.TestCase):
<body>blahablsdfsal&amp;</body>
</html>""",
)
self.assertEqual(get_base_url(resp), "http://www.example.com/img/")
assert get_base_url(resp) == "http://www.example.com/img/"
resp2 = HtmlResponse(
"http://www.example.com",
body=b"""
<html><body>blahablsdfsal&amp;</body></html>""",
)
self.assertEqual(get_base_url(resp2), "http://www.example.com")
assert get_base_url(resp2) == "http://www.example.com"
def test_response_status_message(self):
self.assertEqual(response_status_message(200), "200 OK")
self.assertEqual(response_status_message(404), "404 Not Found")
self.assertEqual(response_status_message(573), "573 Unknown Status")
assert response_status_message(200) == "200 OK"
assert response_status_message(404) == "404 Not Found"
assert response_status_message(573) == "573 Unknown Status"
def test_inject_base_url(self):
url = "http://www.example.com"
@ -103,7 +102,7 @@ class ResponseUtilsTest(unittest.TestCase):
if not path or not Path(path).exists():
path = burl.replace("file://", "")
bbody = Path(path).read_bytes()
self.assertEqual(bbody.count(b'<base href="' + to_bytes(url) + b'">'), 1)
assert bbody.count(b'<base href="' + to_bytes(url) + b'">') == 1
return True
r1 = HtmlResponse(
@ -185,7 +184,7 @@ class ResponseUtilsTest(unittest.TestCase):
open_in_browser(response, lambda url: True)
end_time = process_time()
self.assertLess(end_time - start_time, MAX_CPU_TIME)
assert end_time - start_time < MAX_CPU_TIME
def test_open_in_browser_redos_head(self):
MAX_CPU_TIME = 0.02
@ -202,7 +201,7 @@ class ResponseUtilsTest(unittest.TestCase):
open_in_browser(response, lambda url: True)
end_time = process_time()
self.assertLess(end_time - start_time, MAX_CPU_TIME)
assert end_time - start_time < MAX_CPU_TIME
@pytest.mark.parametrize(

27
tests/test_utils_serialize.py
View File

@ -1,7 +1,6 @@
import dataclasses
import datetime
import json
import unittest
from decimal import Decimal
import attr
@ -11,8 +10,8 @@ from scrapy.http import Request, Response
from scrapy.utils.serialize import ScrapyJSONEncoder
class JsonEncoderTestCase(unittest.TestCase):
def setUp(self):
class TestJsonEncoder:
def setup_method(self):
self.encoder = ScrapyJSONEncoder(sort_keys=True)
def test_encode_decode(self):
@ -39,24 +38,22 @@ class JsonEncoderTestCase(unittest.TestCase):
(s, ss),
(dt_set, dt_sets),
]:
self.assertEqual(
self.encoder.encode(input), json.dumps(output, sort_keys=True)
)
assert self.encoder.encode(input) == json.dumps(output, sort_keys=True)
def test_encode_deferred(self):
self.assertIn("Deferred", self.encoder.encode(defer.Deferred()))
assert "Deferred" in self.encoder.encode(defer.Deferred())
def test_encode_request(self):
r = Request("http://www.example.com/lala")
rs = self.encoder.encode(r)
self.assertIn(r.method, rs)
self.assertIn(r.url, rs)
assert r.method in rs
assert r.url in rs
def test_encode_response(self):
r = Response("http://www.example.com/lala")
rs = self.encoder.encode(r)
self.assertIn(r.url, rs)
self.assertIn(str(r.status), rs)
assert r.url in rs
assert str(r.status) in rs
def test_encode_dataclass_item(self) -> None:
@dataclasses.dataclass
@ -67,9 +64,7 @@ class JsonEncoderTestCase(unittest.TestCase):
item = TestDataClass(name="Product", url="http://product.org", price=1)
encoded = self.encoder.encode(item)
self.assertEqual(
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):
@attr.s
@ -80,6 +75,4 @@ class JsonEncoderTestCase(unittest.TestCase):
item = AttrsItem(name="Product", url="http://product.org", price=1)
encoded = self.encoder.encode(item)
self.assertEqual(
encoded, '{"name": "Product", "price": 1, "url": "http://product.org"}'
)
assert encoded == '{"name": "Product", "price": 1, "url": "http://product.org"}'

22
tests/test_utils_signal.py
View File

@ -11,7 +11,7 @@ from scrapy.utils.signal import send_catch_log, send_catch_log_deferred
from scrapy.utils.test import get_from_asyncio_queue
class SendCatchLogTest(unittest.TestCase):
class TestSendCatchLog(unittest.TestCase):
@defer.inlineCallbacks
def test_send_catch_log(self):
test_signal = object()
@ -29,13 +29,13 @@ class SendCatchLogTest(unittest.TestCase):
assert self.error_handler in handlers_called
assert self.ok_handler in handlers_called
self.assertEqual(len(log.records), 1)
assert len(log.records) == 1
record = log.records[0]
self.assertIn("error_handler", record.getMessage())
self.assertEqual(record.levelname, "ERROR")
self.assertEqual(result[0][0], self.error_handler)
self.assertIsInstance(result[0][1], Failure)
self.assertEqual(result[1], (self.ok_handler, "OK"))
assert "error_handler" in record.getMessage()
assert record.levelname == "ERROR"
assert result[0][0] == self.error_handler # pylint: disable=comparison-with-callable
assert isinstance(result[0][1], Failure)
assert result[1] == (self.ok_handler, "OK")
dispatcher.disconnect(self.error_handler, signal=test_signal)
dispatcher.disconnect(self.ok_handler, signal=test_signal)
@ -53,7 +53,7 @@ class SendCatchLogTest(unittest.TestCase):
return "OK"
class SendCatchLogDeferredTest(SendCatchLogTest):
class SendCatchLogDeferredTest(TestSendCatchLog):
def _get_result(self, signal, *a, **kw):
return send_catch_log_deferred(signal, *a, **kw)
@ -85,7 +85,7 @@ class SendCatchLogDeferredAsyncioTest(SendCatchLogDeferredTest):
return await get_from_asyncio_queue("OK")
class SendCatchLogTest2(unittest.TestCase):
class TestSendCatchLog2:
def test_error_logged_if_deferred_not_supported(self):
def test_handler():
return defer.Deferred()
@ -94,6 +94,6 @@ class SendCatchLogTest2(unittest.TestCase):
dispatcher.connect(test_handler, test_signal)
with LogCapture() as log:
send_catch_log(test_signal)
self.assertEqual(len(log.records), 1)
self.assertIn("Cannot return deferreds from signal handler", str(log))
assert len(log.records) == 1
assert "Cannot return deferreds from signal handler" in str(log)
dispatcher.disconnect(test_handler, test_signal)

178
tests/test_utils_sitemap.py
View File

@ -1,9 +1,7 @@
import unittest
from scrapy.utils.sitemap import Sitemap, sitemap_urls_from_robots
class SitemapTest(unittest.TestCase):
class TestSitemap:
def test_sitemap(self):
s = Sitemap(
b"""<?xml version="1.0" encoding="UTF-8"?>
@ -23,23 +21,20 @@ class SitemapTest(unittest.TestCase):
</urlset>"""
)
assert s.type == "urlset"
self.assertEqual(
list(s),
[
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{
"priority": "0.8",
"loc": "http://www.example.com/Special-Offers.html",
"lastmod": "2009-08-16",
"changefreq": "weekly",
},
],
)
assert list(s) == [
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{
"priority": "0.8",
"loc": "http://www.example.com/Special-Offers.html",
"lastmod": "2009-08-16",
"changefreq": "weekly",
},
]
def test_sitemap_index(self):
s = Sitemap(
@ -56,19 +51,16 @@ class SitemapTest(unittest.TestCase):
</sitemapindex>"""
)
assert s.type == "sitemapindex"
self.assertEqual(
list(s),
[
{
"loc": "http://www.example.com/sitemap1.xml.gz",
"lastmod": "2004-10-01T18:23:17+00:00",
},
{
"loc": "http://www.example.com/sitemap2.xml.gz",
"lastmod": "2005-01-01",
},
],
)
assert list(s) == [
{
"loc": "http://www.example.com/sitemap1.xml.gz",
"lastmod": "2004-10-01T18:23:17+00:00",
},
{
"loc": "http://www.example.com/sitemap2.xml.gz",
"lastmod": "2005-01-01",
},
]
def test_sitemap_strip(self):
"""Assert we can deal with trailing spaces inside <loc> tags - we've
@ -90,18 +82,15 @@ class SitemapTest(unittest.TestCase):
</urlset>
"""
)
self.assertEqual(
list(s),
[
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{"loc": "http://www.example.com/2", "lastmod": ""},
],
)
assert list(s) == [
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{"loc": "http://www.example.com/2", "lastmod": ""},
]
def test_sitemap_wrong_ns(self):
"""We have seen sitemaps with wrongs ns. Presumably, Google still works
@ -122,18 +111,15 @@ class SitemapTest(unittest.TestCase):
</urlset>
"""
)
self.assertEqual(
list(s),
[
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{"loc": "http://www.example.com/2", "lastmod": ""},
],
)
assert list(s) == [
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{"loc": "http://www.example.com/2", "lastmod": ""},
]
def test_sitemap_wrong_ns2(self):
"""We have seen sitemaps with wrongs ns. Presumably, Google still works
@ -155,18 +141,15 @@ class SitemapTest(unittest.TestCase):
"""
)
assert s.type == "urlset"
self.assertEqual(
list(s),
[
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{"loc": "http://www.example.com/2", "lastmod": ""},
],
)
assert list(s) == [
{
"priority": "1",
"loc": "http://www.example.com/",
"lastmod": "2009-08-16",
"changefreq": "daily",
},
{"loc": "http://www.example.com/2", "lastmod": ""},
]
def test_sitemap_urls_from_robots(self):
robots = """User-agent: *
@ -187,15 +170,14 @@ Sitemap: /sitemap-relative-url.xml
Disallow: /forum/search/
Disallow: /forum/active/
"""
self.assertEqual(
list(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-uppercase.xml",
"http://example.com/sitemap-relative-url.xml",
],
)
assert list(
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-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"""
@ -224,14 +206,11 @@ Disallow: /forum/active/
</sitemapindex>
"""
)
self.assertEqual(
list(s),
[
{"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/sitemap3.xml"},
],
)
assert list(s) == [
{"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/sitemap3.xml"},
]
def test_comment(self):
s = Sitemap(
@ -245,7 +224,7 @@ Disallow: /forum/active/
</urlset>"""
)
self.assertEqual(list(s), [{"loc": "http://www.example.com/"}])
assert list(s) == [{"loc": "http://www.example.com/"}]
def test_alternate(self):
s = Sitemap(
@ -265,19 +244,16 @@ Disallow: /forum/active/
</urlset>"""
)
self.assertEqual(
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/",
],
}
],
)
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(
@ -294,4 +270,4 @@ Disallow: /forum/active/
"""
)
self.assertEqual(list(s), [{"loc": "http://127.0.0.1:8000/"}])
assert list(s) == [{"loc": "http://127.0.0.1:8000/"}]

14
tests/test_utils_spider.py
View File

@ -1,5 +1,3 @@
import unittest
from scrapy import Spider
from scrapy.http import Request
from scrapy.item import Item
@ -14,19 +12,19 @@ class MySpider2(Spider):
name = "myspider2"
class UtilsSpidersTestCase(unittest.TestCase):
class TestUtilsSpiders:
def test_iterate_spider_output(self):
i = Item()
r = Request("http://scrapytest.org")
o = object()
self.assertEqual(list(iterate_spider_output(i)), [i])
self.assertEqual(list(iterate_spider_output(r)), [r])
self.assertEqual(list(iterate_spider_output(o)), [o])
self.assertEqual(list(iterate_spider_output([r, i, o])), [r, i, o])
assert list(iterate_spider_output(i)) == [i]
assert list(iterate_spider_output(r)) == [r]
assert list(iterate_spider_output(o)) == [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)
self.assertEqual(set(it), {MySpider1, MySpider2})
assert set(it) == {MySpider1, MySpider2}

11
tests/test_utils_template.py
View File

@ -1,4 +1,3 @@
import unittest
from pathlib import Path
from shutil import rmtree
from tempfile import mkdtemp
@ -6,11 +5,11 @@ from tempfile import mkdtemp
from scrapy.utils.template import render_templatefile
class UtilsRenderTemplateFileTestCase(unittest.TestCase):
def setUp(self):
class TestUtilsRenderTemplateFile:
def setup_method(self):
self.tmp_path = mkdtemp()
def tearDown(self):
def teardown_method(self):
rmtree(self.tmp_path)
def test_simple_render(self):
@ -26,8 +25,8 @@ class UtilsRenderTemplateFileTestCase(unittest.TestCase):
render_templatefile(template_path, **context)
self.assertFalse(template_path.exists())
self.assertEqual(render_path.read_text(encoding="utf8"), rendered)
assert not template_path.exists()
assert render_path.read_text(encoding="utf8") == rendered
render_path.unlink()
assert not render_path.exists() # Failure of test itself

42
tests/test_utils_trackref.py
View File

@ -1,4 +1,3 @@
import unittest
from io import StringIO
from time import sleep, time
from unittest import mock
@ -16,48 +15,48 @@ class Bar(trackref.object_ref):
pass
class TrackrefTestCase(unittest.TestCase):
def setUp(self):
class TestTrackref:
def setup_method(self):
trackref.live_refs.clear()
def test_format_live_refs(self):
o1 = Foo() # noqa: F841
o2 = Bar() # noqa: F841
o3 = Foo() # noqa: F841
self.assertEqual(
trackref.format_live_refs(),
"""\
assert (
trackref.format_live_refs()
== """\
Live References
Bar 1 oldest: 0s ago
Foo 2 oldest: 0s ago
""",
"""
)
self.assertEqual(
trackref.format_live_refs(ignore=Foo),
"""\
assert (
trackref.format_live_refs(ignore=Foo)
== """\
Live References
Bar 1 oldest: 0s ago
""",
"""
)
@mock.patch("sys.stdout", new_callable=StringIO)
def test_print_live_refs_empty(self, stdout):
trackref.print_live_refs()
self.assertEqual(stdout.getvalue(), "Live References\n\n\n")
assert stdout.getvalue() == "Live References\n\n\n"
@mock.patch("sys.stdout", new_callable=StringIO)
def test_print_live_refs_with_objects(self, stdout):
o1 = Foo() # noqa: F841
trackref.print_live_refs()
self.assertEqual(
stdout.getvalue(),
"""\
assert (
stdout.getvalue()
== """\
Live References
Foo 1 oldest: 0s ago\n\n""",
Foo 1 oldest: 0s ago\n\n"""
)
def test_get_oldest(self):
@ -75,15 +74,12 @@ Foo 1 oldest: 0s ago\n\n""",
raise SkipTest("time.time is not precise enough")
o3 = Foo() # noqa: F841
self.assertIs(trackref.get_oldest("Foo"), o1)
self.assertIs(trackref.get_oldest("Bar"), o2)
self.assertIsNone(trackref.get_oldest("XXX"))
assert trackref.get_oldest("Foo") is o1
assert trackref.get_oldest("Bar") is o2
assert trackref.get_oldest("XXX") is None
def test_iter_all(self):
o1 = Foo()
o2 = Bar() # noqa: F841
o3 = Foo()
self.assertEqual(
set(trackref.iter_all("Foo")),
{o1, o3},
)
assert set(trackref.iter_all("Foo")) == {o1, o3}

323
tests/test_utils_url.py
View File

@ -18,301 +18,240 @@ from scrapy.utils.url import ( # type: ignore[attr-defined]
)
class UrlUtilsTest(unittest.TestCase):
class TestUrlUtils:
def test_url_is_from_any_domain(self):
url = "http://www.wheele-bin-art.co.uk/get/product/123"
self.assertTrue(url_is_from_any_domain(url, ["wheele-bin-art.co.uk"]))
self.assertFalse(url_is_from_any_domain(url, ["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"])
url = "http://wheele-bin-art.co.uk/get/product/123"
self.assertTrue(url_is_from_any_domain(url, ["wheele-bin-art.co.uk"]))
self.assertFalse(url_is_from_any_domain(url, ["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"])
url = "http://www.Wheele-Bin-Art.co.uk/get/product/123"
self.assertTrue(url_is_from_any_domain(url, ["wheele-bin-art.CO.UK"]))
self.assertTrue(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"
self.assertTrue(url_is_from_any_domain(url, ["192.169.0.15:8080"]))
self.assertFalse(url_is_from_any_domain(url, ["192.169.0.15"]))
assert url_is_from_any_domain(url, ["192.169.0.15:8080"])
assert not url_is_from_any_domain(url, ["192.169.0.15"])
url = (
"javascript:%20document.orderform_2581_1190810811.mode.value=%27add%27;%20"
"javascript:%20document.orderform_2581_1190810811.submit%28%29"
)
self.assertFalse(url_is_from_any_domain(url, ["testdomain.com"]))
self.assertFalse(
url_is_from_any_domain(url + ".testdomain.com", ["testdomain.com"])
)
assert not url_is_from_any_domain(url, ["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")
self.assertTrue(
url_is_from_spider("http://www.example.com/some/page.html", spider)
)
self.assertTrue(
url_is_from_spider("http://sub.example.com/some/page.html", spider)
)
self.assertFalse(
url_is_from_spider("http://www.example.org/some/page.html", spider)
)
self.assertFalse(
url_is_from_spider("http://www.example.net/some/page.html", spider)
)
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"
self.assertTrue(
url_is_from_spider("http://www.example.com/some/page.html", MySpider)
)
self.assertTrue(
url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
)
self.assertFalse(
url_is_from_spider("http://www.example.org/some/page.html", MySpider)
)
self.assertFalse(
url_is_from_spider("http://www.example.net/some/page.html", MySpider)
)
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"]
)
self.assertTrue(
url_is_from_spider("http://www.example.com/some/page.html", spider)
)
self.assertTrue(
url_is_from_spider("http://sub.example.com/some/page.html", spider)
)
self.assertTrue(url_is_from_spider("http://example.com/some/page.html", spider))
self.assertTrue(
url_is_from_spider("http://www.example.org/some/page.html", spider)
)
self.assertTrue(
url_is_from_spider("http://www.example.net/some/page.html", spider)
)
self.assertFalse(
url_is_from_spider("http://www.example.us/some/page.html", spider)
)
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"}
)
self.assertTrue(
url_is_from_spider("http://www.example.com/some/page.html", spider)
)
assert url_is_from_spider("http://www.example.com/some/page.html", spider)
spider = Spider(
name="example.com", allowed_domains=("example.com", "example.net")
)
self.assertTrue(
url_is_from_spider("http://www.example.com/some/page.html", spider)
)
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")
self.assertTrue(
url_is_from_spider("http://www.example.com/some/page.html", MySpider)
)
self.assertTrue(
url_is_from_spider("http://sub.example.com/some/page.html", MySpider)
)
self.assertTrue(
url_is_from_spider("http://example.com/some/page.html", MySpider)
)
self.assertTrue(
url_is_from_spider("http://www.example.org/some/page.html", MySpider)
)
self.assertTrue(
url_is_from_spider("http://www.example.net/some/page.html", MySpider)
)
self.assertFalse(
url_is_from_spider("http://www.example.us/some/page.html", MySpider)
)
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)}
self.assertTrue(
url_has_any_extension(
"http://www.example.com/archive.tar.gz", deny_extensions
)
assert url_has_any_extension(
"http://www.example.com/archive.tar.gz", deny_extensions
)
self.assertTrue(
url_has_any_extension("http://www.example.com/page.doc", 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
)
self.assertTrue(
url_has_any_extension("http://www.example.com/page.pdf", deny_extensions)
)
self.assertFalse(
url_has_any_extension("http://www.example.com/page.htm", deny_extensions)
)
self.assertFalse(
url_has_any_extension("http://www.example.com/", deny_extensions)
)
self.assertFalse(
url_has_any_extension(
"http://www.example.com/page.doc.html", 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 AddHttpIfNoScheme(unittest.TestCase):
class TestAddHttpIfNoScheme:
def test_add_scheme(self):
self.assertEqual(
add_http_if_no_scheme("www.example.com"), "http://www.example.com"
)
assert add_http_if_no_scheme("www.example.com") == "http://www.example.com"
def test_without_subdomain(self):
self.assertEqual(add_http_if_no_scheme("example.com"), "http://example.com")
assert add_http_if_no_scheme("example.com") == "http://example.com"
def test_path(self):
self.assertEqual(
add_http_if_no_scheme("www.example.com/some/page.html"),
"http://www.example.com/some/page.html",
assert (
add_http_if_no_scheme("www.example.com/some/page.html")
== "http://www.example.com/some/page.html"
)
def test_port(self):
self.assertEqual(
add_http_if_no_scheme("www.example.com:80"), "http://www.example.com:80"
assert (
add_http_if_no_scheme("www.example.com:80") == "http://www.example.com:80"
)
def test_fragment(self):
self.assertEqual(
add_http_if_no_scheme("www.example.com/some/page#frag"),
"http://www.example.com/some/page#frag",
assert (
add_http_if_no_scheme("www.example.com/some/page#frag")
== "http://www.example.com/some/page#frag"
)
def test_query(self):
self.assertEqual(
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",
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):
self.assertEqual(
add_http_if_no_scheme("username:password@www.example.com"),
"http://username:password@www.example.com",
assert (
add_http_if_no_scheme("username:password@www.example.com")
== "http://username:password@www.example.com"
)
def test_complete_url(self):
self.assertEqual(
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",
)
== "http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
def test_preserve_http(self):
self.assertEqual(
add_http_if_no_scheme("http://www.example.com"), "http://www.example.com"
assert (
add_http_if_no_scheme("http://www.example.com") == "http://www.example.com"
)
def test_preserve_http_without_subdomain(self):
self.assertEqual(
add_http_if_no_scheme("http://example.com"), "http://example.com"
)
assert add_http_if_no_scheme("http://example.com") == "http://example.com"
def test_preserve_http_path(self):
self.assertEqual(
add_http_if_no_scheme("http://www.example.com/some/page.html"),
"http://www.example.com/some/page.html",
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):
self.assertEqual(
add_http_if_no_scheme("http://www.example.com:80"),
"http://www.example.com:80",
assert (
add_http_if_no_scheme("http://www.example.com:80")
== "http://www.example.com:80"
)
def test_preserve_http_fragment(self):
self.assertEqual(
add_http_if_no_scheme("http://www.example.com/some/page#frag"),
"http://www.example.com/some/page#frag",
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):
self.assertEqual(
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",
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):
self.assertEqual(
add_http_if_no_scheme("http://username:password@www.example.com"),
"http://username:password@www.example.com",
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):
self.assertEqual(
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",
)
== "http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
def test_protocol_relative(self):
self.assertEqual(
add_http_if_no_scheme("//www.example.com"), "http://www.example.com"
)
assert add_http_if_no_scheme("//www.example.com") == "http://www.example.com"
def test_protocol_relative_without_subdomain(self):
self.assertEqual(add_http_if_no_scheme("//example.com"), "http://example.com")
assert add_http_if_no_scheme("//example.com") == "http://example.com"
def test_protocol_relative_path(self):
self.assertEqual(
add_http_if_no_scheme("//www.example.com/some/page.html"),
"http://www.example.com/some/page.html",
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):
self.assertEqual(
add_http_if_no_scheme("//www.example.com:80"), "http://www.example.com:80"
assert (
add_http_if_no_scheme("//www.example.com:80") == "http://www.example.com:80"
)
def test_protocol_relative_fragment(self):
self.assertEqual(
add_http_if_no_scheme("//www.example.com/some/page#frag"),
"http://www.example.com/some/page#frag",
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):
self.assertEqual(
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",
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):
self.assertEqual(
add_http_if_no_scheme("//username:password@www.example.com"),
"http://username:password@www.example.com",
assert (
add_http_if_no_scheme("//username:password@www.example.com")
== "http://username:password@www.example.com"
)
def test_protocol_relative_complete_url(self):
self.assertEqual(
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",
)
== "http://username:password@www.example.com:80/some/page/do?a=1&b=2&c=3#frag"
)
def test_preserve_https(self):
self.assertEqual(
add_http_if_no_scheme("https://www.example.com"), "https://www.example.com"
assert (
add_http_if_no_scheme("https://www.example.com")
== "https://www.example.com"
)
def test_preserve_ftp(self):
self.assertEqual(
add_http_if_no_scheme("ftp://www.example.com"), "ftp://www.example.com"
)
assert add_http_if_no_scheme("ftp://www.example.com") == "ftp://www.example.com"
class GuessSchemeTest(unittest.TestCase):
class TestGuessScheme:
pass
@ -361,7 +300,7 @@ for k, args in enumerate(
):
t_method = create_guess_scheme_t(args)
t_method.__name__ = f"test_uri_{k:03}"
setattr(GuessSchemeTest, t_method.__name__, t_method)
setattr(TestGuessScheme, t_method.__name__, t_method)
# TODO: the following tests do not pass with current implementation
for k, skip_args in enumerate(
@ -376,29 +315,29 @@ for k, skip_args in enumerate(
):
t_method = create_skipped_scheme_t(skip_args)
t_method.__name__ = f"test_uri_skipped_{k:03}"
setattr(GuessSchemeTest, t_method.__name__, t_method)
setattr(TestGuessScheme, t_method.__name__, t_method)
class StripUrl(unittest.TestCase):
class TestStripUrl:
def test_noop(self):
self.assertEqual(
strip_url("http://www.example.com/index.html"),
"http://www.example.com/index.html",
assert (
strip_url("http://www.example.com/index.html")
== "http://www.example.com/index.html"
)
def test_noop_query_string(self):
self.assertEqual(
strip_url("http://www.example.com/index.html?somekey=somevalue"),
"http://www.example.com/index.html?somekey=somevalue",
assert (
strip_url("http://www.example.com/index.html?somekey=somevalue")
== "http://www.example.com/index.html?somekey=somevalue"
)
def test_fragments(self):
self.assertEqual(
assert (
strip_url(
"http://www.example.com/index.html?somekey=somevalue#section",
strip_fragment=False,
),
"http://www.example.com/index.html?somekey=somevalue#section",
)
== "http://www.example.com/index.html?somekey=somevalue#section"
)
def test_path(self):
@ -407,7 +346,7 @@ class StripUrl(unittest.TestCase):
("http://www.example.com", False, "http://www.example.com"),
("http://www.example.com", True, "http://www.example.com/"),
]:
self.assertEqual(strip_url(input_url, origin_only=origin), output_url)
assert strip_url(input_url, origin_only=origin) == output_url
def test_credentials(self):
for i, o in [
@ -424,7 +363,7 @@ class StripUrl(unittest.TestCase):
"ftp://www.example.com/index.html?somekey=somevalue",
),
]:
self.assertEqual(strip_url(i, strip_credentials=True), o)
assert strip_url(i, strip_credentials=True) == o
def test_credentials_encoded_delims(self):
for i, o in [
@ -447,7 +386,7 @@ class StripUrl(unittest.TestCase):
"ftp://www.example.com/index.html?somekey=somevalue",
),
]:
self.assertEqual(strip_url(i, strip_credentials=True), o)
assert strip_url(i, strip_credentials=True) == o
def test_default_ports_creds_off(self):
for i, o in [
@ -484,7 +423,7 @@ class StripUrl(unittest.TestCase):
"ftp://www.example.com:221/file.txt",
),
]:
self.assertEqual(strip_url(i), o)
assert strip_url(i) == o
def test_default_ports(self):
for i, o in [
@ -521,9 +460,7 @@ class StripUrl(unittest.TestCase):
"ftp://username:password@www.example.com:221/file.txt",
),
]:
self.assertEqual(
strip_url(i, strip_default_port=True, strip_credentials=False), o
)
assert strip_url(i, strip_default_port=True, strip_credentials=False) == o
def test_default_ports_keep(self):
for i, o in [
@ -560,9 +497,7 @@ class StripUrl(unittest.TestCase):
"ftp://username:password@www.example.com:221/file.txt",
),
]:
self.assertEqual(
strip_url(i, strip_default_port=False, strip_credentials=False), o
)
assert strip_url(i, strip_default_port=False, strip_credentials=False) == o
def test_origin_only(self):
for i, o in [
@ -583,10 +518,10 @@ class StripUrl(unittest.TestCase):
"https://www.example.com/",
),
]:
self.assertEqual(strip_url(i, origin_only=True), o)
assert strip_url(i, origin_only=True) == o
class IsPathTestCase(unittest.TestCase):
class TestIsPath:
def test_path(self):
for input_value, output_value in (
# https://en.wikipedia.org/wiki/Path_(computing)#Representations_of_paths_by_operating_system_and_shell
@ -604,9 +539,7 @@ class IsPathTestCase(unittest.TestCase):
(r"C:\user\docs\somefile.ext:alternate_stream_name", True),
(r"https://example.com", False),
):
self.assertEqual(
_is_filesystem_path(input_value), output_value, input_value
)
assert _is_filesystem_path(input_value) == output_value, input_value
@pytest.mark.parametrize(