File: test_utils_serialize.py

package info (click to toggle)
python-scrapy 0.24.2-1
  • links: PTS, VCS
  • area: main
  • in suites: jessie, jessie-kfreebsd
  • size: 3,240 kB
  • ctags: 4,259
  • sloc: python: 21,170; xml: 199; makefile: 67; sh: 44
file content (133 lines) | stat: -rw-r--r-- 4,623 bytes parent folder | download
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()