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

Converting tests to plain asserts, part 6. (#6709)

This commit is contained in:
Andrey Rakhmatullin 2025-03-09 23:20:24 +04:00 committed by GitHub
parent 044c3f69ed
commit 02ed71d887
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 730 additions and 928 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
tests/test_item.py
View File

@ -1,4 +1,3 @@
import unittest
from abc import ABCMeta
from unittest import mock
@ -7,9 +6,9 @@ import pytest
from scrapy.item import Field, Item, ItemMeta
class ItemTest(unittest.TestCase):
class TestItem:
def assertSortedEqual(self, first, second, msg=None):
return self.assertEqual(sorted(first), sorted(second), msg)
assert sorted(first) == sorted(second), msg
def test_simple(self):
class TestItem(Item):
@ -17,7 +16,7 @@ class ItemTest(unittest.TestCase):
i = TestItem()
i["name"] = "name"
self.assertEqual(i["name"], "name")
assert i["name"] == "name"
def test_init(self):
class TestItem(Item):
@ -28,13 +27,13 @@ class ItemTest(unittest.TestCase):
i["name"]
i2 = TestItem(name="john doe")
self.assertEqual(i2["name"], "john doe")
assert i2["name"] == "john doe"
i3 = TestItem({"name": "john doe"})
self.assertEqual(i3["name"], "john doe")
assert i3["name"] == "john doe"
i4 = TestItem(i3)
self.assertEqual(i4["name"], "john doe")
assert i4["name"] == "john doe"
with pytest.raises(KeyError):
TestItem({"name": "john doe", "other": "foo"})
@ -59,11 +58,11 @@ class ItemTest(unittest.TestCase):
i["number"] = 123
itemrepr = repr(i)
self.assertEqual(itemrepr, "{'name': 'John Doe', 'number': 123}")
assert itemrepr == "{'name': 'John Doe', 'number': 123}"
i2 = eval(itemrepr) # pylint: disable=eval-used
self.assertEqual(i2["name"], "John Doe")
self.assertEqual(i2["number"], 123)
assert i2["name"] == "John Doe"
assert i2["number"] == 123
def test_private_attr(self):
class TestItem(Item):
@ -71,7 +70,7 @@ class ItemTest(unittest.TestCase):
i = TestItem()
i._private = "test"
self.assertEqual(i._private, "test")
assert i._private == "test"
def test_raise_getattr(self):
class TestItem(Item):
@ -103,9 +102,9 @@ class ItemTest(unittest.TestCase):
with pytest.raises(KeyError):
i.get_name()
i["name"] = "lala"
self.assertEqual(i.get_name(), "lala")
assert i.get_name() == "lala"
i.change_name("other")
self.assertEqual(i.get_name(), "other")
assert i.get_name() == "other"
def test_metaclass(self):
class TestItem(Item):
@ -115,8 +114,8 @@ class ItemTest(unittest.TestCase):
i = TestItem()
i["name"] = "John"
self.assertEqual(list(i.keys()), ["name"])
self.assertEqual(list(i.values()), ["John"])
assert list(i.keys()) == ["name"]
assert list(i.values()) == ["John"]
i["keys"] = "Keys"
i["values"] = "Values"
@ -142,8 +141,8 @@ class ItemTest(unittest.TestCase):
i = TestItem()
i["keys"] = 3
self.assertEqual(list(i.keys()), ["keys"])
self.assertEqual(list(i.values()), [3])
assert list(i.keys()) == ["keys"]
assert list(i.values()) == [3]
def test_metaclass_multiple_inheritance_simple(self):
class A(Item):
@ -161,17 +160,17 @@ class ItemTest(unittest.TestCase):
pass
item = D(save="X", load="Y")
self.assertEqual(item["save"], "X")
self.assertEqual(item["load"], "Y")
self.assertEqual(D.fields, {"load": {"default": "A"}, "save": {"default": "A"}})
assert item["save"] == "X"
assert item["load"] == "Y"
assert D.fields == {"load": {"default": "A"}, "save": {"default": "A"}}
# D class inverted
class E(C, B):
pass
self.assertEqual(E(save="X")["save"], "X")
self.assertEqual(E(load="X")["load"], "X")
self.assertEqual(E.fields, {"load": {"default": "C"}, "save": {"default": "C"}})
assert E(save="X")["save"] == "X"
assert E(load="X")["load"] == "X"
assert E.fields == {"load": {"default": "C"}, "save": {"default": "C"}}
def test_metaclass_multiple_inheritance_diamond(self):
class A(Item):
@ -190,31 +189,25 @@ class ItemTest(unittest.TestCase):
fields = {"update": Field(default="D")}
load = Field(default="D")
self.assertEqual(D(save="X")["save"], "X")
self.assertEqual(D(load="X")["load"], "X")
self.assertEqual(
D.fields,
{
"save": {"default": "C"},
"load": {"default": "D"},
"update": {"default": "D"},
},
)
assert D(save="X")["save"] == "X"
assert D(load="X")["load"] == "X"
assert D.fields == {
"save": {"default": "C"},
"load": {"default": "D"},
"update": {"default": "D"},
}
# D class inverted
class E(C, B):
load = Field(default="E")
self.assertEqual(E(save="X")["save"], "X")
self.assertEqual(E(load="X")["load"], "X")
self.assertEqual(
E.fields,
{
"save": {"default": "C"},
"load": {"default": "E"},
"update": {"default": "C"},
},
)
assert E(save="X")["save"] == "X"
assert E(load="X")["load"] == "X"
assert E.fields == {
"save": {"default": "C"},
"load": {"default": "E"},
"update": {"default": "C"},
}
def test_metaclass_multiple_inheritance_without_metaclass(self):
class A(Item):
@ -234,8 +227,8 @@ class ItemTest(unittest.TestCase):
with pytest.raises(KeyError):
D(not_allowed="value")
self.assertEqual(D(save="X")["save"], "X")
self.assertEqual(D.fields, {"save": {"default": "A"}, "load": {"default": "A"}})
assert D(save="X")["save"] == "X"
assert D.fields == {"save": {"default": "A"}, "load": {"default": "A"}}
# D class inverted
class E(C, B):
@ -243,8 +236,8 @@ class ItemTest(unittest.TestCase):
with pytest.raises(KeyError):
E(not_allowed="value")
self.assertEqual(E(save="X")["save"], "X")
self.assertEqual(E.fields, {"save": {"default": "A"}, "load": {"default": "A"}})
assert E(save="X")["save"] == "X"
assert E.fields == {"save": {"default": "A"}, "load": {"default": "A"}}
def test_to_dict(self):
class TestItem(Item):
@ -252,7 +245,7 @@ class ItemTest(unittest.TestCase):
i = TestItem()
i["name"] = "John"
self.assertEqual(dict(i), {"name": "John"})
assert dict(i) == {"name": "John"}
def test_copy(self):
class TestItem(Item):
@ -260,9 +253,9 @@ class ItemTest(unittest.TestCase):
item = TestItem({"name": "lower"})
copied_item = item.copy()
self.assertNotEqual(id(item), id(copied_item))
assert id(item) != id(copied_item)
copied_item["name"] = copied_item["name"].upper()
self.assertNotEqual(item["name"], copied_item["name"])
assert item["name"] != copied_item["name"]
def test_deepcopy(self):
class TestItem(Item):
@ -274,7 +267,7 @@ class ItemTest(unittest.TestCase):
assert item["tags"] != copied_item["tags"]
class ItemMetaTest(unittest.TestCase):
class TestItemMeta:
def test_new_method_propagates_classcell(self):
new_mock = mock.Mock(side_effect=ABCMeta.__new__)
base = ItemMeta.__bases__[0]
@ -297,7 +290,7 @@ class ItemMetaTest(unittest.TestCase):
assert "__classcell__" in attrs
class ItemMetaClassCellRegression(unittest.TestCase):
class TestItemMetaClassCellRegression:
def test_item_meta_classcell_regression(self):
class MyItem(Item, metaclass=ItemMeta):
def __init__(self, *args, **kwargs): # pylint: disable=useless-parent-delegation

12
tests/test_link.py
View File

@ -1,18 +1,16 @@
import unittest
import pytest
from scrapy.link import Link
class LinkTest(unittest.TestCase):
class TestLink:
def _assert_same_links(self, link1, link2):
self.assertEqual(link1, link2)
self.assertEqual(hash(link1), hash(link2))
assert link1 == link2
assert hash(link1) == hash(link2)
def _assert_different_links(self, link1, link2):
self.assertNotEqual(link1, link2)
self.assertNotEqual(hash(link1), hash(link2))
assert link1 != link2
assert hash(link1) != hash(link2)
def test_eq_and_hash(self):
l1 = Link("http://www.example.com")

937
tests/test_linkextractors.py
View File

File diff suppressed because it is too large Load Diff

46
tests/test_pipeline_crawl.py
View File

@ -53,7 +53,7 @@ class RedirectedMediaDownloadSpider(MediaDownloadSpider):
)
class FileDownloadCrawlTestCase(TestCase):
class TestFileDownloadCrawl(TestCase):
pipeline_class = "scrapy.pipelines.files.FilesPipeline"
store_setting_key = "FILES_STORE"
media_key = "files"
@ -98,52 +98,46 @@ class FileDownloadCrawlTestCase(TestCase):
return crawler
def _assert_files_downloaded(self, items, logs):
self.assertEqual(len(items), 1)
self.assertIn(self.media_key, items[0])
assert len(items) == 1
assert self.media_key in items[0]
# check that logs show the expected number of successful file downloads
file_dl_success = "File (downloaded): Downloaded file from"
self.assertEqual(logs.count(file_dl_success), 3)
assert logs.count(file_dl_success) == 3
# check that the images/files status is `downloaded`
for item in items:
for i in item[self.media_key]:
self.assertEqual(i["status"], "downloaded")
assert i["status"] == "downloaded"
# check that the images/files checksums are what we know they should be
if self.expected_checksums is not None:
checksums = {i["checksum"] for item in items for i in item[self.media_key]}
self.assertEqual(checksums, self.expected_checksums)
assert checksums == self.expected_checksums
# check that the image files where actually written to the media store
for item in items:
for i in item[self.media_key]:
self.assertTrue((self.tmpmediastore / i["path"]).exists())
assert (self.tmpmediastore / i["path"]).exists()
def _assert_files_download_failure(self, crawler, items, code, logs):
# check that the item does NOT have the "images/files" field populated
self.assertEqual(len(items), 1)
self.assertIn(self.media_key, items[0])
self.assertFalse(items[0][self.media_key])
assert len(items) == 1
assert self.media_key in items[0]
assert not items[0][self.media_key]
# check that there was 1 successful fetch and 3 other responses with non-200 code
self.assertEqual(
crawler.stats.get_value("downloader/request_method_count/GET"), 4
)
self.assertEqual(crawler.stats.get_value("downloader/response_count"), 4)
self.assertEqual(
crawler.stats.get_value("downloader/response_status_count/200"), 1
)
self.assertEqual(
crawler.stats.get_value(f"downloader/response_status_count/{code}"), 3
)
assert crawler.stats.get_value("downloader/request_method_count/GET") == 4
assert crawler.stats.get_value("downloader/response_count") == 4
assert crawler.stats.get_value("downloader/response_status_count/200") == 1
assert crawler.stats.get_value(f"downloader/response_status_count/{code}") == 3
# check that logs do show the failure on the file downloads
file_dl_failure = f"File (code: {code}): Error downloading file from"
self.assertEqual(logs.count(file_dl_failure), 3)
assert logs.count(file_dl_failure) == 3
# check that no files were written to the media store
self.assertEqual(list(self.tmpmediastore.iterdir()), [])
assert not list(self.tmpmediastore.iterdir())
@defer.inlineCallbacks
def test_download_media(self):
@ -193,9 +187,7 @@ class FileDownloadCrawlTestCase(TestCase):
mockserver=self.mockserver,
)
self._assert_files_downloaded(self.items, str(log))
self.assertEqual(
crawler.stats.get_value("downloader/response_status_count/302"), 3
)
assert crawler.stats.get_value("downloader/response_status_count/302") == 3
@defer.inlineCallbacks
def test_download_media_file_path_error(self):
@ -218,7 +210,7 @@ class FileDownloadCrawlTestCase(TestCase):
media_urls_key=self.media_urls_key,
mockserver=self.mockserver,
)
self.assertIn("ZeroDivisionError", str(log))
assert "ZeroDivisionError" in str(log)
skip_pillow: str | None
@ -230,7 +222,7 @@ else:
skip_pillow = None
class ImageDownloadCrawlTestCase(FileDownloadCrawlTestCase):
class ImageDownloadCrawlTestCase(TestFileDownloadCrawl):
skip = skip_pillow
pipeline_class = "scrapy.pipelines.images.ImagesPipeline"

213
tests/test_pipeline_files.py
View File

@ -77,7 +77,7 @@ def get_ftp_content_and_delete(
return b"".join(ftp_data)
class FilesPipelineTestCase(unittest.TestCase):
class TestFilesPipeline(unittest.TestCase):
def setUp(self):
self.tempdir = mkdtemp()
settings_dict = {"FILES_STORE": self.tempdir}
@ -91,73 +91,73 @@ class FilesPipelineTestCase(unittest.TestCase):
def test_file_path(self):
file_path = self.pipeline.file_path
self.assertEqual(
file_path(Request("https://dev.mydeco.com/mydeco.pdf")),
"full/c9b564df929f4bc635bdd19fde4f3d4847c757c5.pdf",
assert (
file_path(Request("https://dev.mydeco.com/mydeco.pdf"))
== "full/c9b564df929f4bc635bdd19fde4f3d4847c757c5.pdf"
)
self.assertEqual(
assert (
file_path(
Request(
"http://www.maddiebrown.co.uk///catalogue-items//image_54642_12175_95307.txt"
)
),
"full/4ce274dd83db0368bafd7e406f382ae088e39219.txt",
)
== "full/4ce274dd83db0368bafd7e406f382ae088e39219.txt"
)
self.assertEqual(
assert (
file_path(
Request("https://dev.mydeco.com/two/dirs/with%20spaces%2Bsigns.doc")
),
"full/94ccc495a17b9ac5d40e3eabf3afcb8c2c9b9e1a.doc",
)
== "full/94ccc495a17b9ac5d40e3eabf3afcb8c2c9b9e1a.doc"
)
self.assertEqual(
assert (
file_path(
Request(
"http://www.dfsonline.co.uk/get_prod_image.php?img=status_0907_mdm.jpg"
)
),
"full/4507be485f38b0da8a0be9eb2e1dfab8a19223f2.jpg",
)
== "full/4507be485f38b0da8a0be9eb2e1dfab8a19223f2.jpg"
)
self.assertEqual(
file_path(Request("http://www.dorma.co.uk/images/product_details/2532/")),
"full/97ee6f8a46cbbb418ea91502fd24176865cf39b2",
assert (
file_path(Request("http://www.dorma.co.uk/images/product_details/2532/"))
== "full/97ee6f8a46cbbb418ea91502fd24176865cf39b2"
)
self.assertEqual(
file_path(Request("http://www.dorma.co.uk/images/product_details/2532")),
"full/244e0dd7d96a3b7b01f54eded250c9e272577aa1",
assert (
file_path(Request("http://www.dorma.co.uk/images/product_details/2532"))
== "full/244e0dd7d96a3b7b01f54eded250c9e272577aa1"
)
self.assertEqual(
assert (
file_path(
Request("http://www.dorma.co.uk/images/product_details/2532"),
response=Response("http://www.dorma.co.uk/images/product_details/2532"),
info=object(),
),
"full/244e0dd7d96a3b7b01f54eded250c9e272577aa1",
)
== "full/244e0dd7d96a3b7b01f54eded250c9e272577aa1"
)
self.assertEqual(
assert (
file_path(
Request(
"http://www.dfsonline.co.uk/get_prod_image.php?img=status_0907_mdm.jpg.bohaha"
)
),
"full/76c00cef2ef669ae65052661f68d451162829507",
)
== "full/76c00cef2ef669ae65052661f68d451162829507"
)
self.assertEqual(
assert (
file_path(
Request(
"data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAR0AAACxCAMAAADOHZloAAACClBMVEX/\
//+F0tzCwMK76ZKQ21AMqr7oAAC96JvD5aWM2kvZ78J0N7fmAAC46Y4Ap7y"
)
),
"full/178059cbeba2e34120a67f2dc1afc3ecc09b61cb.png",
)
== "full/178059cbeba2e34120a67f2dc1afc3ecc09b61cb.png"
)
def test_fs_store(self):
assert isinstance(self.pipeline.store, FSFilesStore)
self.assertEqual(self.pipeline.store.basedir, self.tempdir)
assert self.pipeline.store.basedir == self.tempdir
path = "some/image/key.jpg"
fullpath = Path(self.tempdir, "some", "image", "key.jpg")
self.assertEqual(self.pipeline.store._get_filesystem_path(path), fullpath)
assert self.pipeline.store._get_filesystem_path(path) == fullpath
@defer.inlineCallbacks
def test_file_not_expired(self):
@ -180,8 +180,8 @@ class FilesPipelineTestCase(unittest.TestCase):
p.start()
result = yield self.pipeline.process_item(item, None)
self.assertEqual(result["files"][0]["checksum"], "abc")
self.assertEqual(result["files"][0]["status"], "uptodate")
assert result["files"][0]["checksum"] == "abc"
assert result["files"][0]["status"] == "uptodate"
for p in patchers:
p.stop()
@ -211,8 +211,8 @@ class FilesPipelineTestCase(unittest.TestCase):
p.start()
result = yield self.pipeline.process_item(item, None)
self.assertNotEqual(result["files"][0]["checksum"], "abc")
self.assertEqual(result["files"][0]["status"], "downloaded")
assert result["files"][0]["checksum"] != "abc"
assert result["files"][0]["status"] == "downloaded"
for p in patchers:
p.stop()
@ -242,8 +242,8 @@ class FilesPipelineTestCase(unittest.TestCase):
p.start()
result = yield self.pipeline.process_item(item, None)
self.assertNotEqual(result["files"][0]["checksum"], "abc")
self.assertEqual(result["files"][0]["status"], "cached")
assert result["files"][0]["checksum"] != "abc"
assert result["files"][0]["status"] == "cached"
for p in patchers:
p.stop()
@ -262,14 +262,14 @@ class FilesPipelineTestCase(unittest.TestCase):
).file_path
item = {"path": "path-to-store-file"}
request = Request("http://example.com")
self.assertEqual(file_path(request, item=item), "full/path-to-store-file")
assert file_path(request, item=item) == "full/path-to-store-file"
class FilesPipelineTestCaseFieldsMixin:
def setUp(self):
def setup_method(self):
self.tempdir = mkdtemp()
def tearDown(self):
def teardown_method(self):
rmtree(self.tempdir)
def test_item_fields_default(self):
@ -279,12 +279,12 @@ class FilesPipelineTestCaseFieldsMixin:
get_crawler(None, {"FILES_STORE": self.tempdir})
)
requests = list(pipeline.get_media_requests(item, None))
self.assertEqual(requests[0].url, url)
assert requests[0].url == url
results = [(True, {"url": url})]
item = pipeline.item_completed(results, item, None)
files = ItemAdapter(item).get("files")
self.assertEqual(files, [results[0][1]])
self.assertIsInstance(item, self.item_class)
assert files == [results[0][1]]
assert isinstance(item, self.item_class)
def test_item_fields_override_settings(self):
url = "http://www.example.com/files/1.txt"
@ -300,17 +300,15 @@ class FilesPipelineTestCaseFieldsMixin:
)
)
requests = list(pipeline.get_media_requests(item, None))
self.assertEqual(requests[0].url, url)
assert requests[0].url == url
results = [(True, {"url": url})]
item = pipeline.item_completed(results, item, None)
custom_files = ItemAdapter(item).get("custom_files")
self.assertEqual(custom_files, [results[0][1]])
self.assertIsInstance(item, self.item_class)
assert custom_files == [results[0][1]]
assert isinstance(item, self.item_class)
class FilesPipelineTestCaseFieldsDict(
FilesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestFilesPipelineFieldsDict(FilesPipelineTestCaseFieldsMixin):
item_class = dict
@ -324,9 +322,7 @@ class FilesPipelineTestItem(Item):
custom_files = Field()
class FilesPipelineTestCaseFieldsItem(
FilesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestFilesPipelineFieldsItem(FilesPipelineTestCaseFieldsMixin):
item_class = FilesPipelineTestItem
@ -341,9 +337,7 @@ class FilesPipelineTestDataClass:
custom_files: list = dataclasses.field(default_factory=list)
class FilesPipelineTestCaseFieldsDataClass(
FilesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestFilesPipelineFieldsDataClass(FilesPipelineTestCaseFieldsMixin):
item_class = FilesPipelineTestDataClass
@ -358,13 +352,11 @@ class FilesPipelineTestAttrsItem:
custom_files: list[dict[str, str]] = attr.ib(default=list)
class FilesPipelineTestCaseFieldsAttrsItem(
FilesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestFilesPipelineFieldsAttrsItem(FilesPipelineTestCaseFieldsMixin):
item_class = FilesPipelineTestAttrsItem
class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
class TestFilesPipelineCustomSettings:
default_cls_settings = {
"EXPIRES": 90,
"FILES_URLS_FIELD": "file_urls",
@ -376,10 +368,10 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
("FILES_RESULT_FIELD", "FILES_RESULT_FIELD", "files_result_field"),
}
def setUp(self):
def setup_method(self):
self.tempdir = mkdtemp()
def tearDown(self):
def teardown_method(self):
rmtree(self.tempdir)
def _generate_fake_settings(self, prefix=None):
@ -420,10 +412,10 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
one_pipeline = FilesPipeline(self.tempdir, crawler=get_crawler(None))
for pipe_attr, settings_attr, pipe_ins_attr in self.file_cls_attr_settings_map:
default_value = self.default_cls_settings[pipe_attr]
self.assertEqual(getattr(one_pipeline, pipe_attr), default_value)
assert getattr(one_pipeline, pipe_attr) == default_value
custom_value = custom_settings[settings_attr]
self.assertNotEqual(default_value, custom_value)
self.assertEqual(getattr(another_pipeline, pipe_ins_attr), custom_value)
assert default_value != custom_value
assert getattr(another_pipeline, pipe_ins_attr) == custom_value
def test_subclass_attributes_preserved_if_no_settings(self):
"""
@ -433,8 +425,8 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
pipe = pipe_cls.from_crawler(get_crawler(None, {"FILES_STORE": self.tempdir}))
for pipe_attr, settings_attr, pipe_ins_attr in self.file_cls_attr_settings_map:
custom_value = getattr(pipe, pipe_ins_attr)
self.assertNotEqual(custom_value, self.default_cls_settings[pipe_attr])
self.assertEqual(getattr(pipe, pipe_ins_attr), getattr(pipe, pipe_attr))
assert custom_value != self.default_cls_settings[pipe_attr]
assert getattr(pipe, pipe_ins_attr) == getattr(pipe, pipe_attr)
def test_subclass_attrs_preserved_custom_settings(self):
"""
@ -447,8 +439,8 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr, pipe_ins_attr in self.file_cls_attr_settings_map:
value = getattr(pipeline, pipe_ins_attr)
setting_value = settings.get(settings_attr)
self.assertNotEqual(value, self.default_cls_settings[pipe_attr])
self.assertEqual(value, setting_value)
assert value != self.default_cls_settings[pipe_attr]
assert value == setting_value
def test_no_custom_settings_for_subclasses(self):
"""
@ -465,7 +457,7 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr, pipe_ins_attr in self.file_cls_attr_settings_map:
# Values from settings for custom pipeline should be set on pipeline instance.
custom_value = self.default_cls_settings.get(pipe_attr.upper())
self.assertEqual(getattr(user_pipeline, pipe_ins_attr), custom_value)
assert getattr(user_pipeline, pipe_ins_attr) == custom_value
def test_custom_settings_for_subclasses(self):
"""
@ -484,8 +476,8 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr, pipe_inst_attr in self.file_cls_attr_settings_map:
# Values from settings for custom pipeline should be set on pipeline instance.
custom_value = settings.get(prefix + "_" + settings_attr)
self.assertNotEqual(custom_value, self.default_cls_settings[pipe_attr])
self.assertEqual(getattr(user_pipeline, pipe_inst_attr), custom_value)
assert custom_value != self.default_cls_settings[pipe_attr]
assert getattr(user_pipeline, pipe_inst_attr) == custom_value
def test_custom_settings_and_class_attrs_for_subclasses(self):
"""
@ -502,8 +494,8 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
pipe_inst_attr,
) in self.file_cls_attr_settings_map:
custom_value = settings.get(prefix + "_" + settings_attr)
self.assertNotEqual(custom_value, self.default_cls_settings[pipe_cls_attr])
self.assertEqual(getattr(user_pipeline, pipe_inst_attr), custom_value)
assert custom_value != self.default_cls_settings[pipe_cls_attr]
assert getattr(user_pipeline, pipe_inst_attr) == custom_value
def test_cls_attrs_with_DEFAULT_prefix(self):
class UserDefinedFilesPipeline(FilesPipeline):
@ -513,12 +505,13 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
pipeline = UserDefinedFilesPipeline.from_crawler(
get_crawler(None, {"FILES_STORE": self.tempdir})
)
self.assertEqual(
pipeline.files_result_field,
UserDefinedFilesPipeline.DEFAULT_FILES_RESULT_FIELD,
assert (
pipeline.files_result_field
== UserDefinedFilesPipeline.DEFAULT_FILES_RESULT_FIELD
)
self.assertEqual(
pipeline.files_urls_field, UserDefinedFilesPipeline.DEFAULT_FILES_URLS_FIELD
assert (
pipeline.files_urls_field
== UserDefinedFilesPipeline.DEFAULT_FILES_URLS_FIELD
)
def test_user_defined_subclass_default_key_names(self):
@ -535,7 +528,7 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr, pipe_inst_attr in self.file_cls_attr_settings_map:
expected_value = settings.get(settings_attr)
self.assertEqual(getattr(pipeline_cls, pipe_inst_attr), expected_value)
assert getattr(pipeline_cls, pipe_inst_attr) == expected_value
def test_file_pipeline_using_pathlike_objects(self):
class CustomFilesPipelineWithPathLikeDir(FilesPipeline):
@ -546,12 +539,12 @@ class FilesPipelineTestCaseCustomSettings(unittest.TestCase):
get_crawler(None, {"FILES_STORE": Path("./Temp")})
)
request = Request("http://example.com/image01.jpg")
self.assertEqual(pipeline.file_path(request), Path("subdir/image01.jpg"))
assert pipeline.file_path(request) == Path("subdir/image01.jpg")
def test_files_store_constructor_with_pathlike_object(self):
path = Path("./FileDir")
fs_store = FSFilesStore(path)
self.assertEqual(fs_store.basedir, str(path))
assert fs_store.basedir == str(path)
@pytest.mark.requires_botocore
@ -593,13 +586,8 @@ class TestS3FilesStore(unittest.TestCase):
)
stub.assert_no_pending_responses()
self.assertEqual(
buffer.method_calls,
[
mock.call.seek(0),
# The call to read does not happen with Stubber
],
)
# The call to read does not happen with Stubber
assert buffer.method_calls == [mock.call.seek(0)]
@defer.inlineCallbacks
def test_stat(self):
@ -626,13 +614,10 @@ class TestS3FilesStore(unittest.TestCase):
)
file_stats = yield store.stat_file("", info=None)
self.assertEqual(
file_stats,
{
"checksum": checksum,
"last_modified": last_modified.timestamp(),
},
)
assert file_stats == {
"checksum": checksum,
"last_modified": last_modified.timestamp(),
}
stub.assert_no_pending_responses()
@ -655,16 +640,16 @@ class TestGCSFilesStore(unittest.TestCase):
expected_policy = {"role": "READER", "entity": "allAuthenticatedUsers"}
yield store.persist_file(path, buf, info=None, meta=meta, headers=None)
s = yield store.stat_file(path, info=None)
self.assertIn("last_modified", s)
self.assertIn("checksum", s)
self.assertEqual(s["checksum"], "cdcda85605e46d0af6110752770dce3c")
assert "last_modified" in s
assert "checksum" in s
assert s["checksum"] == "cdcda85605e46d0af6110752770dce3c"
u = urlparse(uri)
content, acl, blob = get_gcs_content_and_delete(u.hostname, u.path[1:] + path)
self.assertEqual(content, data)
self.assertEqual(blob.metadata, {"foo": "bar"})
self.assertEqual(blob.cache_control, GCSFilesStore.CACHE_CONTROL)
self.assertEqual(blob.content_type, "application/octet-stream")
self.assertIn(expected_policy, acl)
assert content == data
assert blob.metadata == {"foo": "bar"}
assert blob.cache_control == GCSFilesStore.CACHE_CONTROL
assert blob.content_type == "application/octet-stream"
assert expected_policy in acl
@defer.inlineCallbacks
def test_blob_path_consistency(self):
@ -702,12 +687,12 @@ class TestFTPFileStore(unittest.TestCase):
with MockFTPServer() as ftp_server:
store = FTPFilesStore(ftp_server.url("/"))
empty_dict = yield store.stat_file(path, info=None)
self.assertEqual(empty_dict, {})
assert empty_dict == {}
yield store.persist_file(path, buf, info=None, meta=meta, headers=None)
stat = yield store.stat_file(path, info=None)
self.assertIn("last_modified", stat)
self.assertIn("checksum", stat)
self.assertEqual(stat["checksum"], "d113d66b2ec7258724a268bd88eef6b6")
assert "last_modified" in stat
assert "checksum" in stat
assert stat["checksum"] == "d113d66b2ec7258724a268bd88eef6b6"
path = f"{store.basedir}/{path}"
content = get_ftp_content_and_delete(
path,
@ -717,7 +702,7 @@ class TestFTPFileStore(unittest.TestCase):
store.password,
store.USE_ACTIVE_MODE,
)
self.assertEqual(data, content)
assert data == content
class ItemWithFiles(Item):
@ -739,12 +724,12 @@ def _prepare_request_object(item_url, flags=None):
# this is separate from the one in test_pipeline_media.py to specifically test FilesPipeline subclasses
class BuildFromCrawlerTestCase(unittest.TestCase):
def setUp(self):
class TestBuildFromCrawler:
def setup_method(self):
self.tempdir = mkdtemp()
self.crawler = get_crawler(None, {"FILES_STORE": self.tempdir})
def tearDown(self):
def teardown_method(self):
rmtree(self.tempdir)
def test_simple(self):
@ -755,7 +740,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 0)
assert len(w) == 0
assert pipe.store
def test_has_old_init(self):
@ -768,7 +753,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 2)
assert len(w) == 2
assert pipe._init_called
def test_has_from_settings(self):
@ -785,7 +770,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 3)
assert len(w) == 3
assert pipe.store
assert pipe._from_settings_called
@ -805,6 +790,6 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 0)
assert len(w) == 0
assert pipe.store
assert pipe._from_crawler_called

204
tests/test_pipeline_images.py
View File

@ -9,109 +9,106 @@ from tempfile import mkdtemp
import attr
import pytest
from itemadapter import ItemAdapter
from twisted.trial import unittest
from scrapy.http import Request, Response
from scrapy.item import Field, Item
from scrapy.pipelines.images import ImageException, ImagesPipeline
from scrapy.utils.test import get_crawler
skip_pillow: str | None
try:
from PIL import Image
except ImportError:
skip_pillow = "Missing Python Imaging Library, install https://pypi.org/pypi/Pillow"
pytest.skip(
"Missing Python Imaging Library, install https://pypi.org/pypi/Pillow",
allow_module_level=True,
)
else:
encoders = {"jpeg_encoder", "jpeg_decoder"}
if not encoders.issubset(set(Image.core.__dict__)): # type: ignore[attr-defined]
skip_pillow = "Missing JPEG encoders"
else:
skip_pillow = None
pytest.skip("Missing JPEG encoders", allow_module_level=True)
class ImagesPipelineTestCase(unittest.TestCase):
skip = skip_pillow
def setUp(self):
class TestImagesPipeline:
def setup_method(self):
self.tempdir = mkdtemp()
crawler = get_crawler()
self.pipeline = ImagesPipeline(self.tempdir, crawler=crawler)
def tearDown(self):
def teardown_method(self):
rmtree(self.tempdir)
def test_file_path(self):
file_path = self.pipeline.file_path
self.assertEqual(
file_path(Request("https://dev.mydeco.com/mydeco.gif")),
"full/3fd165099d8e71b8a48b2683946e64dbfad8b52d.jpg",
assert (
file_path(Request("https://dev.mydeco.com/mydeco.gif"))
== "full/3fd165099d8e71b8a48b2683946e64dbfad8b52d.jpg"
)
self.assertEqual(
assert (
file_path(
Request(
"http://www.maddiebrown.co.uk///catalogue-items//image_54642_12175_95307.jpg"
)
),
"full/0ffcd85d563bca45e2f90becd0ca737bc58a00b2.jpg",
)
== "full/0ffcd85d563bca45e2f90becd0ca737bc58a00b2.jpg"
)
self.assertEqual(
assert (
file_path(
Request("https://dev.mydeco.com/two/dirs/with%20spaces%2Bsigns.gif")
),
"full/b250e3a74fff2e4703e310048a5b13eba79379d2.jpg",
)
== "full/b250e3a74fff2e4703e310048a5b13eba79379d2.jpg"
)
self.assertEqual(
assert (
file_path(
Request(
"http://www.dfsonline.co.uk/get_prod_image.php?img=status_0907_mdm.jpg"
)
),
"full/4507be485f38b0da8a0be9eb2e1dfab8a19223f2.jpg",
)
== "full/4507be485f38b0da8a0be9eb2e1dfab8a19223f2.jpg"
)
self.assertEqual(
file_path(Request("http://www.dorma.co.uk/images/product_details/2532/")),
"full/97ee6f8a46cbbb418ea91502fd24176865cf39b2.jpg",
assert (
file_path(Request("http://www.dorma.co.uk/images/product_details/2532/"))
== "full/97ee6f8a46cbbb418ea91502fd24176865cf39b2.jpg"
)
self.assertEqual(
file_path(Request("http://www.dorma.co.uk/images/product_details/2532")),
"full/244e0dd7d96a3b7b01f54eded250c9e272577aa1.jpg",
assert (
file_path(Request("http://www.dorma.co.uk/images/product_details/2532"))
== "full/244e0dd7d96a3b7b01f54eded250c9e272577aa1.jpg"
)
self.assertEqual(
assert (
file_path(
Request("http://www.dorma.co.uk/images/product_details/2532"),
response=Response("http://www.dorma.co.uk/images/product_details/2532"),
info=object(),
),
"full/244e0dd7d96a3b7b01f54eded250c9e272577aa1.jpg",
)
== "full/244e0dd7d96a3b7b01f54eded250c9e272577aa1.jpg"
)
def test_thumbnail_name(self):
thumb_path = self.pipeline.thumb_path
name = "50"
self.assertEqual(
thumb_path(Request("file:///tmp/foo.jpg"), name),
"thumbs/50/38a86208c36e59d4404db9e37ce04be863ef0335.jpg",
assert (
thumb_path(Request("file:///tmp/foo.jpg"), name)
== "thumbs/50/38a86208c36e59d4404db9e37ce04be863ef0335.jpg"
)
self.assertEqual(
thumb_path(Request("file://foo.png"), name),
"thumbs/50/e55b765eba0ec7348e50a1df496040449071b96a.jpg",
assert (
thumb_path(Request("file://foo.png"), name)
== "thumbs/50/e55b765eba0ec7348e50a1df496040449071b96a.jpg"
)
self.assertEqual(
thumb_path(Request("file:///tmp/foo"), name),
"thumbs/50/0329ad83ebb8e93ea7c7906d46e9ed55f7349a50.jpg",
assert (
thumb_path(Request("file:///tmp/foo"), name)
== "thumbs/50/0329ad83ebb8e93ea7c7906d46e9ed55f7349a50.jpg"
)
self.assertEqual(
thumb_path(Request("file:///tmp/some.name/foo"), name),
"thumbs/50/850233df65a5b83361798f532f1fc549cd13cbe9.jpg",
assert (
thumb_path(Request("file:///tmp/some.name/foo"), name)
== "thumbs/50/850233df65a5b83361798f532f1fc549cd13cbe9.jpg"
)
self.assertEqual(
assert (
thumb_path(
Request("file:///tmp/some.name/foo"),
name,
response=Response("file:///tmp/some.name/foo"),
info=object(),
),
"thumbs/50/850233df65a5b83361798f532f1fc549cd13cbe9.jpg",
)
== "thumbs/50/850233df65a5b83361798f532f1fc549cd13cbe9.jpg"
)
def test_thumbnail_name_from_item(self):
@ -130,8 +127,8 @@ class ImagesPipelineTestCase(unittest.TestCase):
).thumb_path
item = {"path": "path-to-store-file"}
request = Request("http://example.com")
self.assertEqual(
thumb_path(request, "small", item=item), "thumb/small/path-to-store-file"
assert (
thumb_path(request, "small", item=item) == "thumb/small/path-to-store-file"
)
def test_get_images_exception(self):
@ -169,16 +166,13 @@ class ImagesPipelineTestCase(unittest.TestCase):
)
path, new_im, new_buf = next(get_images_gen)
self.assertEqual(path, "full/3fd165099d8e71b8a48b2683946e64dbfad8b52d.jpg")
self.assertEqual(orig_im, new_im)
self.assertEqual(buf.getvalue(), new_buf.getvalue())
assert path == "full/3fd165099d8e71b8a48b2683946e64dbfad8b52d.jpg"
assert orig_im == new_im
assert buf.getvalue() == new_buf.getvalue()
thumb_path, thumb_img, thumb_buf = next(get_images_gen)
self.assertEqual(
thumb_path, "thumbs/small/3fd165099d8e71b8a48b2683946e64dbfad8b52d.jpg"
)
self.assertEqual(thumb_img, thumb_img)
self.assertEqual(orig_thumb_buf.getvalue(), thumb_buf.getvalue())
assert thumb_path == "thumbs/small/3fd165099d8e71b8a48b2683946e64dbfad8b52d.jpg"
assert orig_thumb_buf.getvalue() == thumb_buf.getvalue()
def test_convert_image(self):
SIZE = (100, 100)
@ -186,37 +180,35 @@ class ImagesPipelineTestCase(unittest.TestCase):
COLOUR = (0, 127, 255)
im, buf = _create_image("JPEG", "RGB", SIZE, COLOUR)
converted, converted_buf = self.pipeline.convert_image(im, response_body=buf)
self.assertEqual(converted.mode, "RGB")
self.assertEqual(converted.getcolors(), [(10000, COLOUR)])
assert converted.mode == "RGB"
assert converted.getcolors() == [(10000, COLOUR)]
# check that we don't convert JPEGs again
self.assertEqual(converted_buf, buf)
assert converted_buf == buf
# check that thumbnail keep image ratio
thumbnail, _ = self.pipeline.convert_image(
converted, size=(10, 25), response_body=converted_buf
)
self.assertEqual(thumbnail.mode, "RGB")
self.assertEqual(thumbnail.size, (10, 10))
assert thumbnail.mode == "RGB"
assert thumbnail.size == (10, 10)
# transparency case: RGBA and PNG
COLOUR = (0, 127, 255, 50)
im, buf = _create_image("PNG", "RGBA", SIZE, COLOUR)
converted, _ = self.pipeline.convert_image(im, response_body=buf)
self.assertEqual(converted.mode, "RGB")
self.assertEqual(converted.getcolors(), [(10000, (205, 230, 255))])
assert converted.mode == "RGB"
assert converted.getcolors() == [(10000, (205, 230, 255))]
# transparency case with palette: P and PNG
COLOUR = (0, 127, 255, 50)
im, buf = _create_image("PNG", "RGBA", SIZE, COLOUR)
im = im.convert("P")
converted, _ = self.pipeline.convert_image(im, response_body=buf)
self.assertEqual(converted.mode, "RGB")
self.assertEqual(converted.getcolors(), [(10000, (205, 230, 255))])
assert converted.mode == "RGB"
assert converted.getcolors() == [(10000, (205, 230, 255))]
class ImagesPipelineTestCaseFieldsMixin:
skip = skip_pillow
def test_item_fields_default(self):
url = "http://www.example.com/images/1.jpg"
item = self.item_class(name="item1", image_urls=[url])
@ -224,12 +216,12 @@ class ImagesPipelineTestCaseFieldsMixin:
get_crawler(None, {"IMAGES_STORE": "s3://example/images/"})
)
requests = list(pipeline.get_media_requests(item, None))
self.assertEqual(requests[0].url, url)
assert requests[0].url == url
results = [(True, {"url": url})]
item = pipeline.item_completed(results, item, None)
images = ItemAdapter(item).get("images")
self.assertEqual(images, [results[0][1]])
self.assertIsInstance(item, self.item_class)
assert images == [results[0][1]]
assert isinstance(item, self.item_class)
def test_item_fields_override_settings(self):
url = "http://www.example.com/images/1.jpg"
@ -245,17 +237,15 @@ class ImagesPipelineTestCaseFieldsMixin:
)
)
requests = list(pipeline.get_media_requests(item, None))
self.assertEqual(requests[0].url, url)
assert requests[0].url == url
results = [(True, {"url": url})]
item = pipeline.item_completed(results, item, None)
custom_images = ItemAdapter(item).get("custom_images")
self.assertEqual(custom_images, [results[0][1]])
self.assertIsInstance(item, self.item_class)
assert custom_images == [results[0][1]]
assert isinstance(item, self.item_class)
class ImagesPipelineTestCaseFieldsDict(
ImagesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestImagesPipelineFieldsDict(ImagesPipelineTestCaseFieldsMixin):
item_class = dict
@ -269,9 +259,7 @@ class ImagesPipelineTestItem(Item):
custom_images = Field()
class ImagesPipelineTestCaseFieldsItem(
ImagesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestImagesPipelineFieldsItem(ImagesPipelineTestCaseFieldsMixin):
item_class = ImagesPipelineTestItem
@ -286,9 +274,7 @@ class ImagesPipelineTestDataClass:
custom_images: list = dataclasses.field(default_factory=list)
class ImagesPipelineTestCaseFieldsDataClass(
ImagesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestImagesPipelineFieldsDataClass(ImagesPipelineTestCaseFieldsMixin):
item_class = ImagesPipelineTestDataClass
@ -303,15 +289,11 @@ class ImagesPipelineTestAttrsItem:
custom_images: list[dict[str, str]] = attr.ib(default=list)
class ImagesPipelineTestCaseFieldsAttrsItem(
ImagesPipelineTestCaseFieldsMixin, unittest.TestCase
):
class TestImagesPipelineFieldsAttrsItem(ImagesPipelineTestCaseFieldsMixin):
item_class = ImagesPipelineTestAttrsItem
class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
skip = skip_pillow
class TestImagesPipelineCustomSettings:
img_cls_attribute_names = [
# Pipeline attribute names with corresponding setting names.
("EXPIRES", "IMAGES_EXPIRES"),
@ -332,10 +314,10 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
"IMAGES_RESULT_FIELD": "images",
}
def setUp(self):
def setup_method(self):
self.tempdir = mkdtemp()
def tearDown(self):
def teardown_method(self):
rmtree(self.tempdir)
def _generate_fake_settings(self, prefix=None):
@ -397,11 +379,11 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr in self.img_cls_attribute_names:
expected_default_value = self.default_pipeline_settings.get(pipe_attr)
custom_value = custom_settings.get(settings_attr)
self.assertNotEqual(expected_default_value, custom_value)
self.assertEqual(
getattr(default_sts_pipe, pipe_attr.lower()), expected_default_value
assert expected_default_value != custom_value
assert (
getattr(default_sts_pipe, pipe_attr.lower()) == expected_default_value
)
self.assertEqual(getattr(user_sts_pipe, pipe_attr.lower()), custom_value)
assert getattr(user_sts_pipe, pipe_attr.lower()) == custom_value
def test_subclass_attrs_preserved_default_settings(self):
"""
@ -415,8 +397,8 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr in self.img_cls_attribute_names:
# Instance attribute (lowercase) must be equal to class attribute (uppercase).
attr_value = getattr(pipeline, pipe_attr.lower())
self.assertNotEqual(attr_value, self.default_pipeline_settings[pipe_attr])
self.assertEqual(attr_value, getattr(pipeline, pipe_attr))
assert attr_value != self.default_pipeline_settings[pipe_attr]
assert attr_value == getattr(pipeline, pipe_attr)
def test_subclass_attrs_preserved_custom_settings(self):
"""
@ -430,9 +412,9 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
# Instance attribute (lowercase) must be equal to
# value defined in settings.
value = getattr(pipeline, pipe_attr.lower())
self.assertNotEqual(value, self.default_pipeline_settings[pipe_attr])
assert value != self.default_pipeline_settings[pipe_attr]
setings_value = settings.get(settings_attr)
self.assertEqual(value, setings_value)
assert value == setings_value
def test_no_custom_settings_for_subclasses(self):
"""
@ -449,7 +431,7 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr in self.img_cls_attribute_names:
# Values from settings for custom pipeline should be set on pipeline instance.
custom_value = self.default_pipeline_settings.get(pipe_attr.upper())
self.assertEqual(getattr(user_pipeline, pipe_attr.lower()), custom_value)
assert getattr(user_pipeline, pipe_attr.lower()) == custom_value
def test_custom_settings_for_subclasses(self):
"""
@ -468,8 +450,8 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr in self.img_cls_attribute_names:
# Values from settings for custom pipeline should be set on pipeline instance.
custom_value = settings.get(prefix + "_" + settings_attr)
self.assertNotEqual(custom_value, self.default_pipeline_settings[pipe_attr])
self.assertEqual(getattr(user_pipeline, pipe_attr.lower()), custom_value)
assert custom_value != self.default_pipeline_settings[pipe_attr]
assert getattr(user_pipeline, pipe_attr.lower()) == custom_value
def test_custom_settings_and_class_attrs_for_subclasses(self):
"""
@ -482,8 +464,8 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
user_pipeline = pipeline_cls.from_crawler(get_crawler(None, settings))
for pipe_attr, settings_attr in self.img_cls_attribute_names:
custom_value = settings.get(prefix + "_" + settings_attr)
self.assertNotEqual(custom_value, self.default_pipeline_settings[pipe_attr])
self.assertEqual(getattr(user_pipeline, pipe_attr.lower()), custom_value)
assert custom_value != self.default_pipeline_settings[pipe_attr]
assert getattr(user_pipeline, pipe_attr.lower()) == custom_value
def test_cls_attrs_with_DEFAULT_prefix(self):
class UserDefinedImagePipeline(ImagesPipeline):
@ -493,13 +475,13 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
pipeline = UserDefinedImagePipeline.from_crawler(
get_crawler(None, {"IMAGES_STORE": self.tempdir})
)
self.assertEqual(
pipeline.images_result_field,
UserDefinedImagePipeline.DEFAULT_IMAGES_RESULT_FIELD,
assert (
pipeline.images_result_field
== UserDefinedImagePipeline.DEFAULT_IMAGES_RESULT_FIELD
)
self.assertEqual(
pipeline.images_urls_field,
UserDefinedImagePipeline.DEFAULT_IMAGES_URLS_FIELD,
assert (
pipeline.images_urls_field
== UserDefinedImagePipeline.DEFAULT_IMAGES_URLS_FIELD
)
def test_user_defined_subclass_default_key_names(self):
@ -516,7 +498,7 @@ class ImagesPipelineTestCaseCustomSettings(unittest.TestCase):
for pipe_attr, settings_attr in self.img_cls_attribute_names:
expected_value = settings.get(settings_attr)
self.assertEqual(getattr(pipeline_cls, pipe_attr.lower()), expected_value)
assert getattr(pipeline_cls, pipe_attr.lower()) == expected_value
def _create_image(format, *a, **kw):

149
tests/test_pipeline_media.py
View File

@ -2,6 +2,7 @@ from __future__ import annotations
import warnings
import pytest
from testfixtures import LogCapture
from twisted.internet import reactor
from twisted.internet.defer import Deferred, inlineCallbacks
@ -18,15 +19,6 @@ from scrapy.utils.log import failure_to_exc_info
from scrapy.utils.signal import disconnect_all
from scrapy.utils.test import get_crawler
try:
from PIL import Image # noqa: F401
except ImportError:
skip_pillow: str | None = (
"Missing Python Imaging Library, install https://pypi.org/pypi/Pillow"
)
else:
skip_pillow = None
def _mocked_download_func(request, info):
assert request.callback is NO_CALLBACK
@ -51,7 +43,7 @@ class UserDefinedPipeline(MediaPipeline):
return ""
class BaseMediaPipelineTestCase(unittest.TestCase):
class TestBaseMediaPipeline(unittest.TestCase):
pipeline_class = UserDefinedPipeline
settings = None
@ -123,9 +115,9 @@ class BaseMediaPipelineTestCase(unittest.TestCase):
failure = Failure(file_exc)
# The Failure should encapsulate a FileException ...
self.assertEqual(failure.value, file_exc)
assert failure.value == file_exc
# ... and it should have the StopIteration exception set as its context
self.assertEqual(failure.value.__context__, def_gen_return_exc)
assert failure.value.__context__ == def_gen_return_exc
# Let's calculate the request fingerprint and fake some runtime data...
fp = self.fingerprint(request)
@ -136,12 +128,12 @@ class BaseMediaPipelineTestCase(unittest.TestCase):
# When calling the method that caches the Request's result ...
self.pipe._cache_result_and_execute_waiters(failure, fp, info)
# ... it should store the Twisted Failure ...
self.assertEqual(info.downloaded[fp], failure)
assert info.downloaded[fp] == failure
# ... encapsulating the original FileException ...
self.assertEqual(info.downloaded[fp].value, file_exc)
assert info.downloaded[fp].value == file_exc
# ... but it should not store the StopIteration exception on its context
context = getattr(info.downloaded[fp].value, "__context__", None)
self.assertIsNone(context)
assert context is None
def test_default_item_completed(self):
item = {"name": "name"}
@ -158,7 +150,7 @@ class BaseMediaPipelineTestCase(unittest.TestCase):
assert len(log.records) == 1
record = log.records[0]
assert record.levelname == "ERROR"
self.assertTupleEqual(record.exc_info, failure_to_exc_info(fail))
assert record.exc_info == failure_to_exc_info(fail)
# disable failure logging and check again
self.pipe.LOG_FAILED_RESULTS = False
@ -208,7 +200,7 @@ class MockedMediaPipeline(UserDefinedPipeline):
return item
class MediaPipelineTestCase(BaseMediaPipelineTestCase):
class TestMediaPipeline(TestBaseMediaPipeline):
pipeline_class = MockedMediaPipeline
def _errback(self, result):
@ -225,16 +217,13 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
)
item = {"requests": req}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertEqual(new_item["results"], [(True, {})])
self.assertEqual(
self.pipe._mockcalled,
[
"get_media_requests",
"media_to_download",
"media_downloaded",
"item_completed",
],
)
assert new_item["results"] == [(True, {})]
assert self.pipe._mockcalled == [
"get_media_requests",
"media_to_download",
"media_downloaded",
"item_completed",
]
@inlineCallbacks
def test_result_failure(self):
@ -247,17 +236,14 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
)
item = {"requests": req}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertEqual(new_item["results"], [(False, fail)])
self.assertEqual(
self.pipe._mockcalled,
[
"get_media_requests",
"media_to_download",
"media_failed",
"request_errback",
"item_completed",
],
)
assert new_item["results"] == [(False, fail)]
assert self.pipe._mockcalled == [
"get_media_requests",
"media_to_download",
"media_failed",
"request_errback",
"item_completed",
]
@inlineCallbacks
def test_mix_of_success_and_failure(self):
@ -268,18 +254,18 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
req2 = Request("http://url2", meta={"response": fail})
item = {"requests": [req1, req2]}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertEqual(new_item["results"], [(True, {}), (False, fail)])
assert new_item["results"] == [(True, {}), (False, fail)]
m = self.pipe._mockcalled
# only once
self.assertEqual(m[0], "get_media_requests") # first hook called
self.assertEqual(m.count("get_media_requests"), 1)
self.assertEqual(m.count("item_completed"), 1)
self.assertEqual(m[-1], "item_completed") # last hook called
assert m[0] == "get_media_requests" # first hook called
assert m.count("get_media_requests") == 1
assert m.count("item_completed") == 1
assert m[-1] == "item_completed" # last hook called
# twice, one per request
self.assertEqual(m.count("media_to_download"), 2)
assert m.count("media_to_download") == 2
# one to handle success and other for failure
self.assertEqual(m.count("media_downloaded"), 1)
self.assertEqual(m.count("media_failed"), 1)
assert m.count("media_downloaded") == 1
assert m.count("media_failed") == 1
@inlineCallbacks
def test_get_media_requests(self):
@ -288,7 +274,7 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
item = {"requests": req} # pass a single item
new_item = yield self.pipe.process_item(item, self.spider)
assert new_item is item
self.assertIn(self.fingerprint(req), self.info.downloaded)
assert self.fingerprint(req) in self.info.downloaded
# returns iterable of Requests
req1 = Request("http://url1")
@ -305,8 +291,8 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
req1 = Request("http://url1", meta={"response": rsp1})
item = {"requests": req1}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertTrue(new_item is item)
self.assertEqual(new_item["results"], [(True, {})])
assert new_item is item
assert new_item["results"] == [(True, {})]
# rsp2 is ignored, rsp1 must be in results because request fingerprints are the same
req2 = Request(
@ -314,9 +300,9 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
)
item = {"requests": req2}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertTrue(new_item is item)
self.assertEqual(self.fingerprint(req1), self.fingerprint(req2))
self.assertEqual(new_item["results"], [(True, {})])
assert new_item is item
assert self.fingerprint(req1) == self.fingerprint(req2)
assert new_item["results"] == [(True, {})]
@inlineCallbacks
def test_results_are_cached_for_requests_of_single_item(self):
@ -327,17 +313,17 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
)
item = {"requests": [req1, req2]}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertTrue(new_item is item)
self.assertEqual(new_item["results"], [(True, {}), (True, {})])
assert new_item is item
assert new_item["results"] == [(True, {}), (True, {})]
@inlineCallbacks
def test_wait_if_request_is_downloading(self):
def _check_downloading(response):
fp = self.fingerprint(req1)
self.assertTrue(fp in self.info.downloading)
self.assertTrue(fp in self.info.waiting)
self.assertTrue(fp not in self.info.downloaded)
self.assertEqual(len(self.info.waiting[fp]), 2)
assert fp in self.info.downloading
assert fp in self.info.waiting
assert fp not in self.info.downloaded
assert len(self.info.waiting[fp]) == 2
return response
rsp1 = Response("http://url")
@ -348,39 +334,40 @@ class MediaPipelineTestCase(BaseMediaPipelineTestCase):
return dfd
def rsp2_func():
self.fail("it must cache rsp1 result and must not try to redownload")
pytest.fail("it must cache rsp1 result and must not try to redownload")
req1 = Request("http://url", meta={"response": rsp1_func})
req2 = Request(req1.url, meta={"response": rsp2_func})
item = {"requests": [req1, req2]}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertEqual(new_item["results"], [(True, {}), (True, {})])
assert new_item["results"] == [(True, {}), (True, {})]
@inlineCallbacks
def test_use_media_to_download_result(self):
req = Request("http://url", meta={"result": "ITSME", "response": self.fail})
item = {"requests": req}
new_item = yield self.pipe.process_item(item, self.spider)
self.assertEqual(new_item["results"], [(True, "ITSME")])
self.assertEqual(
self.pipe._mockcalled,
["get_media_requests", "media_to_download", "item_completed"],
)
assert new_item["results"] == [(True, "ITSME")]
assert self.pipe._mockcalled == [
"get_media_requests",
"media_to_download",
"item_completed",
]
def test_key_for_pipe(self):
self.assertEqual(
self.pipe._key_for_pipe("IMAGES", base_class_name="MediaPipeline"),
"MOCKEDMEDIAPIPELINE_IMAGES",
assert (
self.pipe._key_for_pipe("IMAGES", base_class_name="MediaPipeline")
== "MOCKEDMEDIAPIPELINE_IMAGES"
)
class MediaPipelineAllowRedirectSettingsTestCase(unittest.TestCase):
class TestMediaPipelineAllowRedirectSettings:
def _assert_request_no3xx(self, pipeline_class, settings):
pipe = pipeline_class(crawler=get_crawler(None, settings))
request = Request("http://url")
pipe._modify_media_request(request)
self.assertIn("handle_httpstatus_list", request.meta)
assert "handle_httpstatus_list" in request.meta
for status, check in [
(200, True),
# These are the status codes we want
@ -396,9 +383,9 @@ class MediaPipelineAllowRedirectSettingsTestCase(unittest.TestCase):
(500, True),
]:
if check:
self.assertIn(status, request.meta["handle_httpstatus_list"])
assert status in request.meta["handle_httpstatus_list"]
else:
self.assertNotIn(status, request.meta["handle_httpstatus_list"])
assert status not in request.meta["handle_httpstatus_list"]
def test_subclass_standard_setting(self):
self._assert_request_no3xx(UserDefinedPipeline, {"MEDIA_ALLOW_REDIRECTS": True})
@ -409,8 +396,8 @@ class MediaPipelineAllowRedirectSettingsTestCase(unittest.TestCase):
)
class BuildFromCrawlerTestCase(unittest.TestCase):
def setUp(self):
class TestBuildFromCrawler:
def setup_method(self):
self.crawler = get_crawler(None, {"FILES_STORE": "/foo"})
def test_simple(self):
@ -421,7 +408,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 0)
assert len(w) == 0
def test_has_old_init(self):
class Pipeline(UserDefinedPipeline):
@ -433,7 +420,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 2)
assert len(w) == 2
assert pipe._init_called
def test_has_from_settings(self):
@ -450,7 +437,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 2)
assert len(w) == 2
assert pipe._from_settings_called
def test_has_from_settings_and_from_crawler(self):
@ -474,7 +461,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 2)
assert len(w) == 2
assert pipe._from_settings_called
assert pipe._from_crawler_called
@ -497,7 +484,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 2)
assert len(w) == 2
assert pipe._from_settings_called
assert pipe._init_called
@ -521,7 +508,7 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 0)
assert len(w) == 0
assert pipe._from_crawler_called
assert pipe._init_called
@ -542,5 +529,5 @@ class BuildFromCrawlerTestCase(unittest.TestCase):
# this and the next assert will fail as MediaPipeline.from_crawler() wasn't called
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
self.assertEqual(len(w), 0)
assert len(w) == 0
assert pipe._from_crawler_called