Ticket #2506: test_memcache.py

File test_memcache.py, 6.2 KB (added by therve, 3 years ago)

memcache tests

Line 
1import memcache
2
3from twisted.trial import unittest
4from twisted.test.proto_helpers import StringTransportWithDisconnection
5
6class Bean(object):
7    def __init__(self, **kwargs):
8        for k, v in kwargs.items():
9            setattr(self, k, v)
10
11class MemCacheTestCase(unittest.TestCase):
12    def setUp(self):
13        self.proto = memcache.MemCacheProtocol()
14        self.transport = StringTransportWithDisconnection()
15        self.transport.protocol = self.proto
16        self.proto.makeConnection(self.transport)
17
18    def test_get(self):
19        def cb(res):
20            self.assertEquals(res, "bar")
21        d = self.proto.get("foo")
22        d.addCallback(cb)
23        self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n")
24        return d
25
26    def test_emptyGet(self):
27        def cb(res):
28            self.assertEquals(res, None)
29        d = self.proto.get("foo")
30        d.addCallback(cb)
31        self.proto.dataReceived("END\r\n")
32        return d
33
34    def test_set(self):
35        def cb(res):
36            self.assert_(res)
37        d = self.proto.set("foo", "bar")
38        d.addCallback(cb)
39        self.proto.dataReceived("STORED\r\n")
40        return d
41
42    def test_add(self):
43        def cb(res):
44            self.assert_(res)
45        d = self.proto.add("foo", "bar")
46        d.addCallback(cb)
47        self.proto.dataReceived("STORED\r\n")
48        return d
49
50    def test_replace(self):
51        def cb(res):
52            self.assert_(res)
53        d = self.proto.replace("foo", "bar")
54        d.addCallback(cb)
55        self.proto.dataReceived("STORED\r\n")
56        return d
57
58    def test_errorAdd(self):
59        def cb(res):
60            self.assertFalse(res)
61        d = self.proto.add("foo", "bar")
62        d.addCallback(cb)
63        self.proto.dataReceived("NOT STORED\r\n")
64        return d
65
66    def test_errorReplace(self):
67        def cb(res):
68            self.assertFalse(res)
69        d = self.proto.replace("foo", "bar")
70        d.addCallback(cb)
71        self.proto.dataReceived("NOT STORED\r\n")
72        return d
73
74    def test_delete(self):
75        def cb(res):
76            self.assert_(res)
77        d = self.proto.delete("bar")
78        d.addCallback(cb)
79        self.proto.dataReceived("DELETED\r\n")
80        return d
81
82    def test_errorDelete(self):
83        def cb(res):
84            self.assertFalse(res)
85        d = self.proto.delete("bar")
86        d.addCallback(cb)
87        self.proto.dataReceived("NOT FOUND\r\n")
88        return d
89
90    def test_increment(self):
91        def cb(res):
92            self.assertEquals(res, 4)
93        d = self.proto.incr("foo")
94        d.addCallback(cb)
95        self.proto.dataReceived("4\r\n")
96        return d
97
98    def test_decrement(self):
99        def cb(res):
100            self.assertEquals(res, 5)
101        d = self.proto.decr("foo")
102        d.addCallback(cb)
103        self.proto.dataReceived("5\r\n")
104        return d
105
106    def test_stats(self):
107        def cb(res):
108            self.assertEquals(res, {"foo": "bar"})
109        d = self.proto.stats()
110        d.addCallback(cb)
111        self.proto.dataReceived("STAT foo bar\r\nEND\r\n")
112        return d
113
114    def test_version(self):
115        def cb(res):
116            self.assertEquals(res, "1.1")
117        d = self.proto.version()
118        d.addCallback(cb)
119        self.proto.dataReceived("VERSION 1.1\r\n")
120        return d
121
122    def test_flushAll(self):
123        def cb(res):
124            self.assert_(res)
125        d = self.proto.flushAll()
126        d.addCallback(cb)
127        self.proto.dataReceived("OK\r\n")
128        return d
129
130    def test_strSet(self):
131        def cb(res):
132            self.assert_(res)
133        a = "eggspamm"
134        d = self.proto.set("foo", a)
135        d.addCallback(cb)
136        self.proto.dataReceived("STORED\r\n")
137        return d
138
139    def test_marshalSet(self):
140        def cb(res):
141            self.assert_(res)
142        a = [3, 'r', 5]
143        d = self.proto.set("foo", a)
144        d.addCallback(cb)
145        self.proto.dataReceived("STORED\r\n")
146        return d
147
148    def test_complexSet(self):
149        def cb(res):
150            self.assert_(res)
151        a = Bean(foo="bar")
152        d = self.proto.set("foo", a)
153        d.addCallback(cb)
154        self.proto.dataReceived("STORED\r\n")
155        return d
156
157    def test_strGet(self):
158        def cb(res):
159            self.assertEquals(res, "spamegg")
160        s = "spamegg"
161        d = self.proto.get("foo")
162        d.addCallback(cb)
163        self.proto.dataReceived("VALUE foo 32 %s\r\n%s\r\nEND\r\n" % (len(s), s))
164        return d
165
166    def test_marshalGet(self):
167        def cb(res):
168            self.assertEquals(res, [3, 'r', 5])
169        a = [3, 'r', 5]
170        d = self.proto.get("foo")
171        d.addCallback(cb)
172        s = memcache.marshal.dumps(a)
173        self.proto.dataReceived("VALUE foo 16 %s\r\n%s\r\nEND\r\n" % (len(s), s))
174        return d
175
176    def test_pickleGet(self):
177        def cb(res):
178            self.assertEquals(res.foo, "bar")
179        a = Bean(foo="bar")
180        d = self.proto.get("foo")
181        d.addCallback(cb)
182        s = memcache.pickle.dumps(a)
183        self.proto.dataReceived("VALUE foo 1 %s\r\n%s\r\nEND\r\n" % (len(s), s))
184        return d
185
186    def test_ntGet(self):
187        def cb(res):
188            self.assertEquals(res, 12)
189            self.assert_(isinstance(res, int))
190        d = self.proto.get("foo")
191        d.addCallback(cb)
192        self.proto.dataReceived("VALUE foo 2 2\r\n%d\r\nEND\r\n" % 12)
193        return d
194
195    def test_longGet(self):
196        def cb(res):
197            self.assertEquals(res, 12)
198            self.assert_(isinstance(res, long))
199        d = self.proto.get("foo")
200        d.addCallback(cb)
201        self.proto.dataReceived("VALUE foo 4 2\r\n%d\r\nEND\r\n" % 12)
202        return d
203
204    def test_bigSet(self):
205        def cb(res):
206            self.assert_(res)
207        b = "X" * 200000
208        a = Bean(foo=b)
209        d = self.proto.set("foo", a)
210        d.addCallback(cb)
211        self.proto.dataReceived("STORED\r\n")
212        return d
213
214    def test_bigGet(self):
215        b = "X" * 200000
216        def cb(res):
217            self.assertEquals(res.foo, b)
218        a = Bean(foo=b)
219        d = self.proto.get("foo")
220        d.addCallback(cb)
221        s = memcache.pickle.dumps(a)
222        self.proto.dataReceived("VALUE foo 1 %s\r\n%s\r\nEND\r\n" % (len(s), s))
223        return d
224