Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from scrapy.squeues import MarshalFifoDiskQueue, MarshalLifoDiskQueue, PickleFifoDiskQueue, PickleLifoDiskQueue
from scrapy.item import Item, Field
from scrapy.http import Request
from scrapy.loader import ItemLoader
class TestItem(Item):
name = Field()
def _test_procesor(x):
return x + x
class TestLoader(ItemLoader):
default_item_class = TestItem
name_out = staticmethod(_test_procesor)
class MarshalFifoDiskQueueTest(t.FifoDiskQueueTest):
chunksize = 100000
def queue(self):
return MarshalFifoDiskQueue(self.qpath, chunksize=self.chunksize)
def test_serialize(self):
q = self.queue()
q.push('a')
q.push(123)
q.push({'a': 'dict'})
self.assertEqual(q.pop(), 'a')
self.assertEqual(q.pop(), 123)
self.assertEqual(q.pop(), {'a': 'dict'})
def test_nonserializable_object(self):
except Exception:
# Trigger Twisted bug #7989
import twisted.persisted.styles # NOQA
self.assertRaises(ValueError, q.push, lambda x: x)
else:
# Use a different unpickleable object
class A(object): pass
a = A()
a.__reduce__ = a.__reduce_ex__ = None
self.assertRaises(ValueError, q.push, a)
# Selectors should fail (lxml.html.HtmlElement objects can't be pickled)
sel = Selector(text='<p>some text</p>')
self.assertRaises(ValueError, q.push, sel)
class MarshalFifoDiskQueueTest(t.FifoDiskQueueTest):
chunksize = 100000
def queue(self):
return MarshalFifoDiskQueue(self.qpath, chunksize=self.chunksize)
def test_serialize(self):
q = self.queue()
q.push('a')
q.push(123)
q.push({'a': 'dict'})
self.assertEqual(q.pop(), 'a')
self.assertEqual(q.pop(), 123)
self.assertEqual(q.pop(), {'a': 'dict'})
test_nonserializable_object = nonserializable_object_test