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
|
import os
import unittest
import six
import bmemcached
from bmemcached.exceptions import MemcachedException
if six.PY3:
from unittest import mock
else:
import mock
class TestMemcachedErrors(unittest.TestCase):
def testGet(self):
"""
Raise MemcachedException if request wasn't successful and
wasn't a 'key not found' error.
"""
client = bmemcached.Client('{}:11211'.format(os.environ['MEMCACHED_HOST']), 'user', 'password')
with mock.patch.object(bmemcached.protocol.Protocol, '_get_response') as mocked_response:
mocked_response.return_value = (0, 0, 0, 0, 0, 0x81, 0, 0, 0, 0)
self.assertRaises(MemcachedException, client.get, 'foo')
def testSet(self):
"""
Raise MemcachedException if request wasn't successful and
wasn't a 'key not found' or 'key exists' error.
"""
client = bmemcached.Client('{}:11211'.format(os.environ['MEMCACHED_HOST']), 'user', 'password')
with mock.patch.object(bmemcached.protocol.Protocol, '_get_response') as mocked_response:
mocked_response.return_value = (0, 0, 0, 0, 0, 0x81, 0, 0, 0, 0)
self.assertRaises(MemcachedException, client.set, 'foo', 'bar', 300)
def testIncrDecr(self):
"""
Incr/Decr raise MemcachedException unless the request wasn't
successful.
"""
client = bmemcached.Client('{}:11211'.format(os.environ['MEMCACHED_HOST']), 'user', 'password')
client.set('foo', 1)
with mock.patch.object(bmemcached.protocol.Protocol, '_get_response') as mocked_response:
mocked_response.return_value = (0, 0, 0, 0, 0, 0x81, 0, 0, 0, 2)
self.assertRaises(MemcachedException, client.incr, 'foo', 1)
self.assertRaises(MemcachedException, client.decr, 'foo', 1)
def testDelete(self):
"""
Raise MemcachedException if the delete request isn't successful.
"""
client = bmemcached.Client('{}:11211'.format(os.environ['MEMCACHED_HOST']), 'user', 'password')
client.flush_all()
with mock.patch.object(bmemcached.protocol.Protocol, '_get_response') as mocked_response:
mocked_response.return_value = (0, 0, 0, 0, 0, 0x81, 0, 0, 0, 0)
self.assertRaises(MemcachedException, client.delete, 'foo')
def testFlushAll(self):
"""
Raise MemcachedException if the flush wasn't successful.
"""
client = bmemcached.Client('{}:11211'.format(os.environ['MEMCACHED_HOST']), 'user', 'password')
with mock.patch.object(bmemcached.protocol.Protocol, '_get_response') as mocked_response:
mocked_response.return_value = (0, 0, 0, 0, 0, 0x81, 0, 0, 0, 0)
self.assertRaises(MemcachedException, client.flush_all)
|
