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
|
import unittest
import datetime
import json
from decimal import Decimal
from twisted.internet import defer
from scrapy.utils.serialize import SpiderReferencer, ScrapyJSONEncoder, ScrapyJSONDecoder
from scrapy.spider import Spider
from scrapy.http import Request, Response
class _EngineMock(object):
def __init__(self, open_spiders):
self.open_spiders = open_spiders
class CrawlerMock(object):
def __init__(self, open_spiders):
self.engine = _EngineMock(open_spiders)
class BaseTestCase(unittest.TestCase):
def setUp(self):
self.spider1 = Spider('name1')
self.spider2 = Spider('name2')
open_spiders = set([self.spider1, self.spider2])
crawler = CrawlerMock(open_spiders)
self.spref = SpiderReferencer(crawler)
self.encoder = ScrapyJSONEncoder(spref=self.spref)
self.decoder = ScrapyJSONDecoder(spref=self.spref)
class SpiderReferencerTestCase(BaseTestCase):
def test_spiders_and_references(self):
ref1 = self.spref.get_reference_from_spider(self.spider1)
assert isinstance(ref1, str)
assert self.spider1.name in ref1
ref2 = self.spref.get_reference_from_spider(self.spider2)
ref1_ = self.spref.get_reference_from_spider(self.spider1)
assert ref1 == ref1_
assert ref1 != ref2
sp1 = self.spref.get_spider_from_reference(ref1)
sp2 = self.spref.get_spider_from_reference(ref2)
sp1_ = self.spref.get_spider_from_reference(ref1)
assert isinstance(sp1, Spider)
assert sp1 is not sp2
assert sp1 is sp1_
# referring to spiders by name
assert sp1 is self.spref.get_spider_from_reference('spider::name1')
assert sp2 is self.spref.get_spider_from_reference('spider::name2')
# must return string as-is if spider id not found
assert 'lala' == self.spref.get_spider_from_reference('lala')
# must raise RuntimeError if spider id is not found and spider is not running
self.assertRaises(RuntimeError, self.spref.get_spider_from_reference, 'spider:fffffff')
def test_encode_decode(self):
sr = self.spref
sp1 = self.spider1
sp2 = self.spider2
ref1 = sr.get_reference_from_spider(sp1)
ref2 = sr.get_reference_from_spider(sp2)
examples = [
('lala', 'lala'),
(sp1, ref1),
(['lala', sp1], ['lala', ref1]),
({'lala': sp1}, {'lala': ref1}),
({sp1: sp2}, {ref1: ref2}),
({sp1: {sp2: ['lala', sp1]}}, {ref1: {ref2: ['lala', ref1]}})
]
for spiders, refs in examples:
self.assertEqual(sr.encode_references(spiders), refs)
self.assertEqual(sr.decode_references(refs), spiders)
class JsonEncoderTestCase(BaseTestCase):
def test_encode_decode(self):
sr = self.spref
sp1 = self.spider1
sp2 = self.spider2
ref1 = sr.get_reference_from_spider(sp1)
ref2 = sr.get_reference_from_spider(sp2)
dt = datetime.datetime(2010, 1, 2, 10, 11, 12)
dts = "2010-01-02 10:11:12"
d = datetime.date(2010, 1, 2)
ds = "2010-01-02"
t = datetime.time(10, 11, 12)
ts = "10:11:12"
dec = Decimal("1000.12")
decs = "1000.12"
examples_encode_decode = [
('lala', 'lala'),
(sp1, ref1),
(['lala', sp1], ['lala', ref1]),
({'lala': sp1}, {'lala': ref1}),
({sp1: sp2}, {ref1: ref2}),
({sp1: {sp2: ['lala', sp1]}}, {ref1: {ref2: ['lala', ref1]}})
]
for spiders, refs in examples_encode_decode:
self.assertEqual(self.encoder.encode(spiders), json.dumps(refs))
self.assertEqual(self.decoder.decode(json.dumps(refs)), spiders)
examples_encode_only = [
({sp1: dt}, {ref1: dts}),
({sp1: d}, {ref1: ds}),
({sp1: t}, {ref1: ts}),
({sp1: dec}, {ref1: decs}),
]
for spiders, refs in examples_encode_only:
self.assertEqual(self.encoder.encode(spiders), json.dumps(refs))
assert 'Deferred' in self.encoder.encode(defer.Deferred())
def test_encode_request(self):
r = Request("http://www.example.com/lala")
rs = self.encoder.encode(r)
assert r.method in rs
assert r.url in rs
def test_encode_response(self):
r = Response("http://www.example.com/lala")
rs = self.encoder.encode(r)
assert r.url in rs
assert str(r.status) in rs
if __name__ == "__main__":
unittest.main()
|