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
|
import json
import os
from io import BytesIO
try:
import cPickle as pickle
except ImportError:
import pickle
import unittest
import bmemcached
class PickleableThing(object):
pass
class JsonPickler(object):
def __init__(self, f, protocol=0):
self.f = f
def dump(self, obj):
if isinstance(self.f, BytesIO):
return self.f.write(json.dumps(obj).encode())
return json.dump(obj, self.f)
def load(self):
if isinstance(self.f, BytesIO):
return json.loads(self.f.read().decode())
return json.load(self.f)
class PicklerTests(unittest.TestCase):
def setUp(self):
self.server = '{}:11211'.format(os.environ['MEMCACHED_HOST'])
self.json_client = bmemcached.Client(self.server, 'user', 'password', pickler=JsonPickler,
unpickler=JsonPickler)
self.pickle_client = bmemcached.Client(self.server, 'user', 'password',
pickler=pickle.Pickler,
unpickler=pickle.Unpickler)
self.data = {'a': 'b'}
def tearDown(self):
self.json_client.delete('test_key')
self.json_client.disconnect_all()
self.pickle_client.disconnect_all()
def testPickleDict(self):
self.pickle_client.set('test_key', self.data)
self.assertEqual(self.data, self.pickle_client.get('test_key'))
def testPickleClassInstance(self):
to_pickle = PickleableThing()
self.pickle_client.set('test_key', to_pickle)
unpickled = self.pickle_client.get('test_key')
self.assertEqual(type(unpickled), PickleableThing)
self.assertFalse(unpickled is to_pickle)
def testPickleVsJson(self):
self.pickle_client.set('test_key', self.data)
self.assertRaises(ValueError, self.json_client.get, 'test_key')
def testJsonVsPickle(self):
self.json_client.set('test_key', self.data)
self.assertRaises(pickle.UnpicklingError, self.pickle_client.get, 'test_key')
|
