import unittest import warnings from scrapy.exceptions import UsageError, ScrapyDeprecationWarning 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 BuildComponentListTest(unittest.TestCase): 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']) def test_backward_compatible_build_dict(self): base = {'one': 1, 'two': 2, 'three': 3, 'five': 5, 'six': None} custom = {'two': None, 'three': 8, 'four': 4} self.assertEqual(build_component_list(base, custom, convert=lambda x: x), ['one', 'four', 'five', 'three']) def test_return_list(self): custom = ['a', 'b', 'c'] self.assertEqual(build_component_list(None, custom, convert=lambda x: x), custom) def test_map_dict(self): custom = {'one': 1, 'two': 2, 'three': 3} self.assertEqual(build_component_list({}, custom, convert=lambda x: x.upper()), ['ONE', 'TWO', 'THREE']) def test_map_list(self): custom = ['a', 'b', 'c'] self.assertEqual(build_component_list(None, custom, lambda x: x.upper()), ['A', 'B', 'C']) def test_duplicate_components_in_dict(self): duplicate_dict = {'one': 1, 'two': 2, 'ONE': 4} self.assertRaises(ValueError, build_component_list, {}, duplicate_dict, convert=lambda x: x.lower()) def test_duplicate_components_in_list(self): duplicate_list = ['a', 'b', 'a'] with self.assertRaises(ValueError) as cm: build_component_list(None, duplicate_list, convert=lambda x: x) self.assertIn(str(duplicate_list), str(cm.exception)) 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']) duplicate_bs.set('one', duplicate_bs['one'], priority=20) self.assertEqual(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) self.assertRaises(ValueError, 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} self.assertEqual(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']) # raise exception for invalid values d = {'one': '5'} self.assertRaises(ValueError, build_component_list, {}, d, convert=lambda x: x) d = {'one': '1.0'} self.assertRaises(ValueError, build_component_list, {}, d, convert=lambda x: x) d = {'one': [1, 2, 3]} self.assertRaises(ValueError, build_component_list, {}, d, convert=lambda x: x) d = {'one': {'a': 'a', 'b': 2}} self.assertRaises(ValueError, build_component_list, {}, d, convert=lambda x: x) d = {'one': 'lorem ipsum'} self.assertRaises(ValueError, build_component_list, {}, d, convert=lambda x: x) class UtilsConfTestCase(unittest.TestCase): def test_arglist_to_dict(self): self.assertEqual( arglist_to_dict(['arg1=val1', 'arg2=val2']), {'arg1': 'val1', 'arg2': 'val2'}) class FeedExportConfigTestCase(unittest.TestCase): def test_feed_export_config_invalid_format(self): settings = Settings() self.assertRaises(UsageError, feed_process_params_from_cli, settings, ['items.dat'], 'noformat') def test_feed_export_config_mismatch(self): settings = Settings() self.assertRaises( UsageError, feed_process_params_from_cli, settings, ['items1.dat', 'items2.dat'], 'noformat' ) def test_feed_export_config_backward_compatible(self): with warnings.catch_warnings(record=True) as cw: settings = Settings() self.assertEqual( {'items.dat': {'format': 'csv'}}, feed_process_params_from_cli(settings, ['items.dat'], 'csv') ) self.assertEqual(cw[0].category, ScrapyDeprecationWarning) 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']) ) 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']) ) def test_feed_export_config_stdout(self): settings = Settings() self.assertEqual( {'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, [], None, ['output.json']) ) def test_output_and_overwrite_output(self): with self.assertRaises(UsageError): feed_process_params_from_cli( Settings(), ['output1.json'], None, ['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) 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": dict(), }) 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) self.assertEqual(new_feed, { "encoding": "other encoding", "fields": None, "indent": 42, "store_empty": True, "uri_params": None, "batch_item_count": 2, "item_export_kwargs": dict(), }) if __name__ == "__main__": unittest.main()