1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
import pytest
from scrapy.exceptions import UsageError
from scrapy.settings import BaseSettings, Settings
from scrapy.utils.conf import (
arglist_to_dict,
build_component_list,
feed_complete_default_values_from_settings,
feed_process_params_from_cli,
)
class TestBuildComponentList:
def test_build_dict(self):
d = {"one": 1, "two": None, "three": 8, "four": 4}
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)
assert build_component_list(duplicate_bs, convert=lambda x: x.lower()) == [
"two",
"one",
]
duplicate_bs.set("one", duplicate_bs["one"], priority=20)
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(
ValueError, match="Some paths in .* convert to the same object"
):
build_component_list(duplicate_bs, convert=lambda x: x.lower())
def test_valid_numbers(self):
# work well with None and numeric values
d = {"a": 10, "b": None, "c": 15, "d": 5.0}
assert build_component_list(d, convert=lambda x: x) == ["d", "a", "c"]
d = {
"a": 33333333333333333333,
"b": 11111111111111111111,
"c": 22222222222222222222,
}
assert build_component_list(d, convert=lambda x: x) == ["b", "c", "a"]
class TestUtilsConf:
def test_arglist_to_dict(self):
assert arglist_to_dict(["arg1=val1", "arg2=val2"]) == {
"arg1": "val1",
"arg2": "val2",
}
class TestFeedExportConfig:
def test_feed_export_config_invalid_format(self):
settings = Settings()
with pytest.raises(UsageError):
feed_process_params_from_cli(settings, ["items.dat"])
def test_feed_export_config_mismatch(self):
settings = Settings()
with pytest.raises(UsageError):
feed_process_params_from_cli(settings, ["items1.dat", "items2.dat"])
def test_feed_export_config_explicit_formats(self):
settings = Settings()
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()
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()
assert {"stdout:": {"format": "pickle"}} == feed_process_params_from_cli(
settings, ["-:pickle"]
)
def test_feed_export_config_overwrite(self):
settings = Settings()
assert {
"output.json": {"format": "json", "overwrite": True}
} == feed_process_params_from_cli(
settings, [], overwrite_output=["output.json"]
)
def test_output_and_overwrite_output(self):
with pytest.raises(UsageError):
feed_process_params_from_cli(
Settings(), ["output1.json"], overwrite_output=["output2.json"]
)
def test_feed_complete_default_values_from_settings_empty(self):
feed = {}
settings = Settings(
{
"FEED_EXPORT_ENCODING": "custom encoding",
"FEED_EXPORT_FIELDS": ["f1", "f2", "f3"],
"FEED_EXPORT_INDENT": 42,
"FEED_STORE_EMPTY": True,
"FEED_URI_PARAMS": (1, 2, 3, 4),
"FEED_EXPORT_BATCH_ITEM_COUNT": 2,
}
)
new_feed = feed_complete_default_values_from_settings(feed, settings)
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 = {
"encoding": "other encoding",
"fields": None,
}
settings = Settings(
{
"FEED_EXPORT_ENCODING": "custom encoding",
"FEED_EXPORT_FIELDS": ["f1", "f2", "f3"],
"FEED_EXPORT_INDENT": 42,
"FEED_STORE_EMPTY": True,
"FEED_EXPORT_BATCH_ITEM_COUNT": 2,
}
)
new_feed = feed_complete_default_values_from_settings(feed, settings)
assert new_feed == {
"encoding": "other encoding",
"fields": None,
"indent": 42,
"store_empty": True,
"uri_params": None,
"batch_item_count": 2,
"item_export_kwargs": {},
}
|
