mirror of https://github.com/scrapy/scrapy.git
703 lines
24 KiB
Python
703 lines
24 KiB
Python
import dataclasses
|
|
import json
|
|
import marshal
|
|
import pickle
|
|
import re
|
|
from abc import ABC, abstractmethod
|
|
from datetime import datetime
|
|
from io import BytesIO
|
|
from typing import Any
|
|
|
|
import lxml.etree
|
|
import pytest
|
|
from itemadapter import ItemAdapter
|
|
|
|
from scrapy.exporters import (
|
|
BaseItemExporter,
|
|
CsvItemExporter,
|
|
JsonItemExporter,
|
|
JsonLinesItemExporter,
|
|
MarshalItemExporter,
|
|
PickleItemExporter,
|
|
PprintItemExporter,
|
|
PythonItemExporter,
|
|
XmlItemExporter,
|
|
)
|
|
from scrapy.item import Field, Item
|
|
from scrapy.utils.python import to_unicode
|
|
|
|
|
|
def custom_serializer(value):
|
|
return str(int(value) + 2)
|
|
|
|
|
|
class MyItem(Item):
|
|
name = Field()
|
|
age = Field()
|
|
|
|
|
|
class CustomFieldItem(Item):
|
|
name = Field()
|
|
age = Field(serializer=custom_serializer)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class MyDataClass:
|
|
name: str
|
|
age: int
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class CustomFieldDataclass:
|
|
name: str
|
|
age: int = dataclasses.field(metadata={"serializer": custom_serializer})
|
|
|
|
|
|
class TestBaseItemExporter(ABC):
|
|
item_class: type = MyItem
|
|
custom_field_item_class: type = CustomFieldItem
|
|
|
|
def setup_method(self):
|
|
self.i = self.item_class(name="John\xa3", age="22")
|
|
self.output = BytesIO()
|
|
self.ie = self._get_exporter()
|
|
|
|
@abstractmethod
|
|
def _get_exporter(self, **kwargs) -> BaseItemExporter:
|
|
raise NotImplementedError
|
|
|
|
def _check_output(self): # noqa: B027
|
|
pass
|
|
|
|
def _assert_expected_item(self, exported_dict):
|
|
for k, v in exported_dict.items():
|
|
exported_dict[k] = to_unicode(v)
|
|
assert self.i == self.item_class(**exported_dict)
|
|
|
|
def _get_nonstring_types_item(self):
|
|
return {
|
|
"boolean": False,
|
|
"number": 22,
|
|
"time": datetime(2015, 1, 1, 1, 1, 1),
|
|
"float": 3.14,
|
|
}
|
|
|
|
def assertItemExportWorks(self, item):
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(item)
|
|
self.ie.finish_exporting()
|
|
# Delete the item exporter object, so that if it causes the output
|
|
# file handle to be closed, which should not be the case, follow-up
|
|
# interactions with the output file handle will surface the issue.
|
|
del self.ie
|
|
self._check_output()
|
|
|
|
def test_export_item(self):
|
|
self.assertItemExportWorks(self.i)
|
|
|
|
def test_export_dict_item(self):
|
|
self.assertItemExportWorks(ItemAdapter(self.i).asdict())
|
|
|
|
def test_serialize_field(self):
|
|
a = ItemAdapter(self.i)
|
|
res = self.ie.serialize_field(a.get_field_meta("name"), "name", a["name"])
|
|
assert res == "John\xa3"
|
|
|
|
res = self.ie.serialize_field(a.get_field_meta("age"), "age", a["age"])
|
|
assert res == "22"
|
|
|
|
def test_fields_to_export(self):
|
|
ie = self._get_exporter(fields_to_export=["name"])
|
|
assert list(ie._get_serialized_fields(self.i)) == [("name", "John\xa3")]
|
|
|
|
ie = self._get_exporter(fields_to_export=["name"], encoding="latin-1")
|
|
_, name = next(iter(ie._get_serialized_fields(self.i)))
|
|
assert isinstance(name, str)
|
|
assert name == "John\xa3"
|
|
|
|
ie = self._get_exporter(fields_to_export={"name": "名稱"})
|
|
assert list(ie._get_serialized_fields(self.i)) == [("名稱", "John\xa3")]
|
|
|
|
def test_field_custom_serializer(self):
|
|
i = self.custom_field_item_class(name="John\xa3", age="22")
|
|
a = ItemAdapter(i)
|
|
ie = self._get_exporter()
|
|
assert (
|
|
ie.serialize_field(a.get_field_meta("name"), "name", a["name"])
|
|
== "John\xa3"
|
|
)
|
|
assert ie.serialize_field(a.get_field_meta("age"), "age", a["age"]) == "24"
|
|
|
|
|
|
class TestPythonItemExporter(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
return PythonItemExporter(**kwargs)
|
|
|
|
def test_invalid_option(self):
|
|
with pytest.raises(TypeError, match="Unexpected options: invalid_option"):
|
|
PythonItemExporter(invalid_option="something")
|
|
|
|
def test_nested_item(self):
|
|
i1 = self.item_class(name="Joseph", age="22")
|
|
i2 = {"name": "Maria", "age": i1}
|
|
i3 = self.item_class(name="Jesus", age=i2)
|
|
ie = self._get_exporter()
|
|
exported = ie.export_item(i3)
|
|
assert isinstance(exported, dict)
|
|
assert exported == {
|
|
"age": {"age": {"age": "22", "name": "Joseph"}, "name": "Maria"},
|
|
"name": "Jesus",
|
|
}
|
|
assert isinstance(exported["age"], dict)
|
|
assert isinstance(exported["age"]["age"], dict)
|
|
|
|
def test_export_list(self):
|
|
i1 = self.item_class(name="Joseph", age="22")
|
|
i2 = self.item_class(name="Maria", age=[i1])
|
|
i3 = self.item_class(name="Jesus", age=[i2])
|
|
ie = self._get_exporter()
|
|
exported = ie.export_item(i3)
|
|
assert exported == {
|
|
"age": [{"age": [{"age": "22", "name": "Joseph"}], "name": "Maria"}],
|
|
"name": "Jesus",
|
|
}
|
|
assert isinstance(exported["age"][0], dict)
|
|
assert isinstance(exported["age"][0]["age"][0], dict)
|
|
|
|
def test_export_item_dict_list(self):
|
|
i1 = self.item_class(name="Joseph", age="22")
|
|
i2 = {"name": "Maria", "age": [i1]}
|
|
i3 = self.item_class(name="Jesus", age=[i2])
|
|
ie = self._get_exporter()
|
|
exported = ie.export_item(i3)
|
|
assert exported == {
|
|
"age": [{"age": [{"age": "22", "name": "Joseph"}], "name": "Maria"}],
|
|
"name": "Jesus",
|
|
}
|
|
assert isinstance(exported["age"][0], dict)
|
|
assert isinstance(exported["age"][0]["age"][0], dict)
|
|
|
|
def test_nonstring_types_item(self):
|
|
item = self._get_nonstring_types_item()
|
|
ie = self._get_exporter()
|
|
exported = ie.export_item(item)
|
|
assert exported == item
|
|
|
|
|
|
class TestPythonItemExporterDataclass(TestPythonItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestPprintItemExporter(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
return PprintItemExporter(self.output, **kwargs)
|
|
|
|
def _check_output(self):
|
|
self._assert_expected_item(eval(self.output.getvalue()))
|
|
|
|
|
|
class TestPprintItemExporterDataclass(TestPprintItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestPickleItemExporter(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
return PickleItemExporter(self.output, **kwargs)
|
|
|
|
def _check_output(self):
|
|
self._assert_expected_item(pickle.loads(self.output.getvalue()))
|
|
|
|
def test_export_multiple_items(self):
|
|
i1 = self.item_class(name="hello", age="world")
|
|
i2 = self.item_class(name="bye", age="world")
|
|
f = BytesIO()
|
|
ie = PickleItemExporter(f)
|
|
ie.start_exporting()
|
|
ie.export_item(i1)
|
|
ie.export_item(i2)
|
|
ie.finish_exporting()
|
|
del ie # See the first “del self.ie” in this file for context.
|
|
f.seek(0)
|
|
assert self.item_class(**pickle.load(f)) == i1
|
|
assert self.item_class(**pickle.load(f)) == i2
|
|
|
|
def test_nonstring_types_item(self):
|
|
item = self._get_nonstring_types_item()
|
|
fp = BytesIO()
|
|
ie = PickleItemExporter(fp)
|
|
ie.start_exporting()
|
|
ie.export_item(item)
|
|
ie.finish_exporting()
|
|
del ie # See the first “del self.ie” in this file for context.
|
|
assert pickle.loads(fp.getvalue()) == item
|
|
|
|
|
|
class TestPickleItemExporterDataclass(TestPickleItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestMarshalItemExporter(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
return MarshalItemExporter(self.output, **kwargs)
|
|
|
|
def _check_output(self):
|
|
self.output.seek(0)
|
|
self._assert_expected_item(marshal.load(self.output))
|
|
|
|
def test_nonstring_types_item(self):
|
|
item = self._get_nonstring_types_item()
|
|
item.pop("time") # datetime is not marshallable
|
|
fp = BytesIO()
|
|
ie = MarshalItemExporter(fp)
|
|
ie.start_exporting()
|
|
ie.export_item(item)
|
|
ie.finish_exporting()
|
|
del ie # See the first “del self.ie” in this file for context.
|
|
fp.seek(0)
|
|
assert marshal.load(fp) == item
|
|
fp.close()
|
|
|
|
|
|
class TestMarshalItemExporterDataclass(TestMarshalItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestCsvItemExporter(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
# We need a fresh instance for each exporter, because
|
|
# CsvItemExporter.stream.__del__() closes the underlying file
|
|
# (CsvItemExporter.finish_exporting() calls detach() but not all tests
|
|
# call it).
|
|
self.output = BytesIO()
|
|
return CsvItemExporter(self.output, **kwargs)
|
|
|
|
def assertCsvEqual(self, first, second, msg=None):
|
|
def split_csv(csv):
|
|
return [
|
|
sorted(re.split(r"(,|\s+)", line))
|
|
for line in to_unicode(csv).splitlines(True)
|
|
]
|
|
|
|
assert split_csv(first) == split_csv(second), msg
|
|
|
|
def _check_output(self):
|
|
self.output.seek(0)
|
|
self.assertCsvEqual(
|
|
to_unicode(self.output.read()), "age,name\r\n22,John\xa3\r\n"
|
|
)
|
|
|
|
def assertExportResult(self, item, expected, **kwargs):
|
|
fp = BytesIO()
|
|
ie = CsvItemExporter(fp, **kwargs)
|
|
ie.start_exporting()
|
|
ie.export_item(item)
|
|
ie.finish_exporting()
|
|
del ie # See the first “del self.ie” in this file for context.
|
|
self.assertCsvEqual(fp.getvalue(), expected)
|
|
|
|
def test_header_export_all(self):
|
|
self.assertExportResult(
|
|
item=self.i,
|
|
fields_to_export=ItemAdapter(self.i).field_names(),
|
|
expected=b"age,name\r\n22,John\xc2\xa3\r\n",
|
|
)
|
|
|
|
def test_header_export_all_dict(self):
|
|
self.assertExportResult(
|
|
item=ItemAdapter(self.i).asdict(),
|
|
expected=b"age,name\r\n22,John\xc2\xa3\r\n",
|
|
)
|
|
|
|
def test_header_export_single_field(self):
|
|
for item in [self.i, ItemAdapter(self.i).asdict()]:
|
|
self.assertExportResult(
|
|
item=item,
|
|
fields_to_export=["age"],
|
|
expected=b"age\r\n22\r\n",
|
|
)
|
|
|
|
def test_header_export_two_items(self):
|
|
for item in [self.i, ItemAdapter(self.i).asdict()]:
|
|
output = BytesIO()
|
|
ie = CsvItemExporter(output)
|
|
ie.start_exporting()
|
|
ie.export_item(item)
|
|
ie.export_item(item)
|
|
ie.finish_exporting()
|
|
del ie # See the first “del self.ie” in this file for context.
|
|
self.assertCsvEqual(
|
|
output.getvalue(), b"age,name\r\n22,John\xc2\xa3\r\n22,John\xc2\xa3\r\n"
|
|
)
|
|
|
|
def test_header_no_header_line(self):
|
|
for item in [self.i, ItemAdapter(self.i).asdict()]:
|
|
self.assertExportResult(
|
|
item=item,
|
|
include_headers_line=False,
|
|
expected=b"22,John\xc2\xa3\r\n",
|
|
)
|
|
|
|
def test_join_multivalue(self):
|
|
class TestItem2(Item):
|
|
name = Field()
|
|
friends = Field()
|
|
|
|
for cls in TestItem2, dict:
|
|
self.assertExportResult(
|
|
item=cls(name="John", friends=["Mary", "Paul"]),
|
|
include_headers_line=False,
|
|
expected='"Mary,Paul",John\r\n',
|
|
)
|
|
|
|
def test_join_multivalue_not_strings(self):
|
|
self.assertExportResult(
|
|
item={"name": "John", "friends": [4, 8]},
|
|
include_headers_line=False,
|
|
expected='"[4, 8]",John\r\n',
|
|
)
|
|
|
|
def test_nonstring_types_item(self):
|
|
self.assertExportResult(
|
|
item=self._get_nonstring_types_item(),
|
|
include_headers_line=False,
|
|
expected="22,False,3.14,2015-01-01 01:01:01\r\n",
|
|
)
|
|
|
|
def test_errors_default(self):
|
|
with pytest.raises(UnicodeEncodeError):
|
|
self.assertExportResult(
|
|
item={"text": "W\u0275\u200brd"},
|
|
expected=None,
|
|
encoding="windows-1251",
|
|
)
|
|
|
|
def test_errors_xmlcharrefreplace(self):
|
|
self.assertExportResult(
|
|
item={"text": "W\u0275\u200brd"},
|
|
include_headers_line=False,
|
|
expected="Wɵ​rd\r\n",
|
|
encoding="windows-1251",
|
|
errors="xmlcharrefreplace",
|
|
)
|
|
|
|
def test_csv_dropped_fields_warning(self, caplog):
|
|
out = BytesIO()
|
|
exporter = CsvItemExporter(out)
|
|
exporter.start_exporting()
|
|
|
|
exporter.export_item({"name": "Apple"})
|
|
|
|
with caplog.at_level("WARNING", logger="scrapy.exporters"):
|
|
exporter.export_item({"name": "Banana", "price": 2.00})
|
|
|
|
assert len(caplog.records) == 1
|
|
assert "CSVExporter dropped fields" in caplog.text
|
|
assert "price" in caplog.text
|
|
|
|
|
|
class TestCsvItemExporterDataclass(TestCsvItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestXmlItemExporter(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
# We need a fresh instance for each exporter, because
|
|
# XmlItemExporter.stream.__del__() closes the underlying file
|
|
# (XmlItemExporter.finish_exporting() calls detach() but not all tests
|
|
# call it).
|
|
self.output = BytesIO()
|
|
return XmlItemExporter(self.output, **kwargs)
|
|
|
|
def assertXmlEquivalent(self, first, second, msg=None):
|
|
def xmltuple(elem):
|
|
children = list(elem.iterchildren())
|
|
if children:
|
|
return [(child.tag, sorted(xmltuple(child))) for child in children]
|
|
return [(elem.tag, [(elem.text, ())])]
|
|
|
|
def xmlsplit(xmlcontent):
|
|
doc = lxml.etree.fromstring(xmlcontent)
|
|
return xmltuple(doc)
|
|
|
|
assert xmlsplit(first) == xmlsplit(second), msg
|
|
|
|
def assertExportResult(self, item, expected_value):
|
|
fp = BytesIO()
|
|
ie = XmlItemExporter(fp)
|
|
ie.start_exporting()
|
|
ie.export_item(item)
|
|
ie.finish_exporting()
|
|
del ie # See the first “del self.ie” in this file for context.
|
|
self.assertXmlEquivalent(fp.getvalue(), expected_value)
|
|
|
|
def _check_output(self):
|
|
expected_value = (
|
|
b'<?xml version="1.0" encoding="utf-8"?>\n'
|
|
b"<items><item><age>22</age><name>John\xc2\xa3</name></item></items>"
|
|
)
|
|
self.assertXmlEquivalent(self.output.getvalue(), expected_value)
|
|
|
|
def test_multivalued_fields(self):
|
|
self.assertExportResult(
|
|
self.item_class(name=["John\xa3", "Doe"], age=[1, 2, 3]),
|
|
b"""<?xml version="1.0" encoding="utf-8"?>\n
|
|
<items>
|
|
<item>
|
|
<name><value>John\xc2\xa3</value><value>Doe</value></name>
|
|
<age><value>1</value><value>2</value><value>3</value></age>
|
|
</item>
|
|
</items>
|
|
""",
|
|
)
|
|
|
|
def test_nested_item(self):
|
|
i1 = {"name": "foo\xa3hoo", "age": "22"}
|
|
i2 = {"name": "bar", "age": i1}
|
|
i3 = self.item_class(name="buz", age=i2)
|
|
|
|
self.assertExportResult(
|
|
i3,
|
|
b"""<?xml version="1.0" encoding="utf-8"?>\n
|
|
<items>
|
|
<item>
|
|
<age>
|
|
<age>
|
|
<age>22</age>
|
|
<name>foo\xc2\xa3hoo</name>
|
|
</age>
|
|
<name>bar</name>
|
|
</age>
|
|
<name>buz</name>
|
|
</item>
|
|
</items>
|
|
""",
|
|
)
|
|
|
|
def test_nested_list_item(self):
|
|
i1 = {"name": "foo"}
|
|
i2 = {"name": "bar", "v2": {"egg": ["spam"]}}
|
|
i3 = self.item_class(name="buz", age=[i1, i2])
|
|
|
|
self.assertExportResult(
|
|
i3,
|
|
b"""<?xml version="1.0" encoding="utf-8"?>\n
|
|
<items>
|
|
<item>
|
|
<age>
|
|
<value><name>foo</name></value>
|
|
<value><name>bar</name><v2><egg><value>spam</value></egg></v2></value>
|
|
</age>
|
|
<name>buz</name>
|
|
</item>
|
|
</items>
|
|
""",
|
|
)
|
|
|
|
def test_nonstring_types_item(self):
|
|
item = self._get_nonstring_types_item()
|
|
self.assertExportResult(
|
|
item,
|
|
b"""<?xml version="1.0" encoding="utf-8"?>\n
|
|
<items>
|
|
<item>
|
|
<float>3.14</float>
|
|
<boolean>False</boolean>
|
|
<number>22</number>
|
|
<time>2015-01-01 01:01:01</time>
|
|
</item>
|
|
</items>
|
|
""",
|
|
)
|
|
|
|
|
|
class TestXmlItemExporterDataclass(TestXmlItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestJsonLinesItemExporter(TestBaseItemExporter):
|
|
_expected_nested: Any = {
|
|
"name": "Jesus",
|
|
"age": {"name": "Maria", "age": {"name": "Joseph", "age": "22"}},
|
|
}
|
|
|
|
def _get_exporter(self, **kwargs):
|
|
return JsonLinesItemExporter(self.output, **kwargs)
|
|
|
|
def _check_output(self):
|
|
exported = json.loads(to_unicode(self.output.getvalue().strip()))
|
|
assert exported == ItemAdapter(self.i).asdict()
|
|
|
|
def test_nested_item(self):
|
|
i1 = self.item_class(name="Joseph", age="22")
|
|
i2 = {"name": "Maria", "age": i1}
|
|
i3 = self.item_class(name="Jesus", age=i2)
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(i3)
|
|
self.ie.finish_exporting()
|
|
del self.ie # See the first “del self.ie” in this file for context.
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
assert exported == self._expected_nested
|
|
|
|
def test_extra_keywords(self):
|
|
self.ie = self._get_exporter(sort_keys=True)
|
|
self.test_export_item()
|
|
self._check_output()
|
|
with pytest.raises(TypeError):
|
|
self._get_exporter(foo_unknown_keyword_bar=True)
|
|
|
|
def test_nonstring_types_item(self):
|
|
item = self._get_nonstring_types_item()
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(item)
|
|
self.ie.finish_exporting()
|
|
del self.ie # See the first “del self.ie” in this file for context.
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
item["time"] = str(item["time"])
|
|
assert exported == item
|
|
|
|
|
|
class TestJsonLinesItemExporterDataclass(TestJsonLinesItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestJsonItemExporter(TestJsonLinesItemExporter):
|
|
_expected_nested = [TestJsonLinesItemExporter._expected_nested]
|
|
|
|
def _get_exporter(self, **kwargs):
|
|
return JsonItemExporter(self.output, **kwargs)
|
|
|
|
def _check_output(self):
|
|
exported = json.loads(to_unicode(self.output.getvalue().strip()))
|
|
assert exported == [ItemAdapter(self.i).asdict()]
|
|
|
|
def assertTwoItemsExported(self, item):
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(item)
|
|
self.ie.export_item(item)
|
|
self.ie.finish_exporting()
|
|
del self.ie # See the first “del self.ie” in this file for context.
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
assert exported == [ItemAdapter(item).asdict(), ItemAdapter(item).asdict()]
|
|
|
|
def test_two_items(self):
|
|
self.assertTwoItemsExported(self.i)
|
|
|
|
def test_two_dict_items(self):
|
|
self.assertTwoItemsExported(ItemAdapter(self.i).asdict())
|
|
|
|
def test_two_items_with_failure_between(self):
|
|
i1 = MyItem(name="Joseph\xa3", age="22")
|
|
i2 = MyItem(
|
|
name="Maria", age=1j
|
|
) # Invalid datetimes didn't consistently fail between Python versions
|
|
i3 = MyItem(name="Jesus", age="44")
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(i1)
|
|
with pytest.raises(TypeError):
|
|
self.ie.export_item(i2)
|
|
self.ie.export_item(i3)
|
|
self.ie.finish_exporting()
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
assert exported == [dict(i1), dict(i3)]
|
|
|
|
def test_nested_item(self):
|
|
i1 = self.item_class(name="Joseph\xa3", age="22")
|
|
i2 = self.item_class(name="Maria", age=i1)
|
|
i3 = self.item_class(name="Jesus", age=i2)
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(i3)
|
|
self.ie.finish_exporting()
|
|
del self.ie # See the first “del self.ie” in this file for context.
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
expected = {
|
|
"name": "Jesus",
|
|
"age": {"name": "Maria", "age": ItemAdapter(i1).asdict()},
|
|
}
|
|
assert exported == [expected]
|
|
|
|
def test_nested_dict_item(self):
|
|
i1 = {"name": "Joseph\xa3", "age": "22"}
|
|
i2 = self.item_class(name="Maria", age=i1)
|
|
i3 = {"name": "Jesus", "age": i2}
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(i3)
|
|
self.ie.finish_exporting()
|
|
del self.ie # See the first “del self.ie” in this file for context.
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
expected = {"name": "Jesus", "age": {"name": "Maria", "age": i1}}
|
|
assert exported == [expected]
|
|
|
|
def test_nonstring_types_item(self):
|
|
item = self._get_nonstring_types_item()
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(item)
|
|
self.ie.finish_exporting()
|
|
del self.ie # See the first “del self.ie” in this file for context.
|
|
exported = json.loads(to_unicode(self.output.getvalue()))
|
|
item["time"] = str(item["time"])
|
|
assert exported == [item]
|
|
|
|
|
|
class TestJsonItemExporterToBytes(TestBaseItemExporter):
|
|
def _get_exporter(self, **kwargs):
|
|
kwargs["encoding"] = "latin"
|
|
return JsonItemExporter(self.output, **kwargs)
|
|
|
|
def test_two_items_with_failure_between(self):
|
|
i1 = MyItem(name="Joseph", age="22")
|
|
i2 = MyItem(name="\u263a", age="11")
|
|
i3 = MyItem(name="Jesus", age="44")
|
|
self.ie.start_exporting()
|
|
self.ie.export_item(i1)
|
|
with pytest.raises(UnicodeEncodeError):
|
|
self.ie.export_item(i2)
|
|
self.ie.export_item(i3)
|
|
self.ie.finish_exporting()
|
|
exported = json.loads(to_unicode(self.output.getvalue(), encoding="latin"))
|
|
assert exported == [dict(i1), dict(i3)]
|
|
|
|
|
|
class TestJsonItemExporterDataclass(TestJsonItemExporter):
|
|
item_class = MyDataClass
|
|
custom_field_item_class = CustomFieldDataclass
|
|
|
|
|
|
class TestCustomExporterItem:
|
|
item_class: type = MyItem
|
|
|
|
def setup_method(self):
|
|
if self.item_class is None:
|
|
pytest.skip("item class is None")
|
|
|
|
def test_exporter_custom_serializer(self):
|
|
class CustomItemExporter(BaseItemExporter):
|
|
def serialize_field(self, field, name, value):
|
|
if name == "age":
|
|
return str(int(value) + 1)
|
|
return super().serialize_field(field, name, value)
|
|
|
|
def export_item(self, item: Any) -> None:
|
|
pass
|
|
|
|
i = self.item_class(name="John", age="22")
|
|
a = ItemAdapter(i)
|
|
ie = CustomItemExporter()
|
|
|
|
assert ie.serialize_field(a.get_field_meta("name"), "name", a["name"]) == "John"
|
|
assert ie.serialize_field(a.get_field_meta("age"), "age", a["age"]) == "23"
|
|
|
|
i2 = {"name": "John", "age": "22"}
|
|
assert ie.serialize_field({}, "name", i2["name"]) == "John"
|
|
assert ie.serialize_field({}, "age", i2["age"]) == "23"
|
|
|
|
|
|
class TestCustomExporterDataclass(TestCustomExporterItem):
|
|
item_class = MyDataClass
|