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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
import datetime
import pylibmc
import _pylibmc
from pylibmc.test import make_test_client
from tests import PylibmcTestCase
from tests import get_refcounts
class RefcountTests(PylibmcTestCase):
"""Test that refcounts are invariant under pylibmc operations.
In general, pylibmc should not retain any references to keys or values,
so the refcounts of operands should be the same before and after operations.
Assert that this is so, using sys.getrefcount.
Caveats:
1. In general, calling sys.getrefcount will change the refcount, because
passing the object as an argument creates an additional reference.
Therefore it is necessary to always measure the refcounts in exactly the
same way.
2. The test code itself must not retain any references to the refcounted
objects (basically, don't define locals).
"""
def _test_get(self, key, val):
bc = make_test_client(binary=True)
refcountables = [key, val]
initial_refcounts = get_refcounts(refcountables)
bc.set(key, val)
assert get_refcounts(refcountables) == initial_refcounts
assert bc.get(key) == val
assert get_refcounts(refcountables) == initial_refcounts
def test_get_complex_type(self):
self._test_get("refcountest", datetime.datetime.fromtimestamp(0))
def test_get_simple(self):
self._test_get(b"refcountest2", 485295)
def test_get_singleton(self):
self._test_get(b"refcountest3", False)
def test_get_with_default(self):
bc = make_test_client(binary=True)
key = b'refcountest4'
val = 'some_value'
default = object()
refcountables = [key, val, default]
initial_refcounts = get_refcounts(refcountables)
bc.set(key, val)
assert get_refcounts(refcountables) == initial_refcounts
assert bc.get(key) == val
assert get_refcounts(refcountables) == initial_refcounts
assert bc.get(key, default) == val
assert get_refcounts(refcountables) == initial_refcounts
bc.delete(key)
assert bc.get(key) is None
assert get_refcounts(refcountables) == initial_refcounts
assert bc.get(key, default) is default
assert get_refcounts(refcountables) == initial_refcounts
def test_get_multi(self):
bc = make_test_client(binary=True)
keys = ["first", "second", "", b""]
value = "first_value"
refcountables = keys + [value]
initial_refcounts = get_refcounts(refcountables)
bc.set(keys[0], value)
assert get_refcounts(refcountables) == initial_refcounts
assert bc.get_multi(keys) == {keys[0]: value}
assert get_refcounts(refcountables) == initial_refcounts
def test_get_multi_bytes_and_unicode(self):
bc = make_test_client(binary=True)
keys = ["third", b"fourth"]
value = "another_value"
kv = {k: value for k in keys}
refcountables = [keys] + [value]
initial_refcounts = get_refcounts(refcountables)
bc.set_multi(kv)
assert get_refcounts(refcountables) == initial_refcounts
assert bc.get_multi(keys)[keys[0]] == value
assert get_refcounts(refcountables) == initial_refcounts
def test_delete(self):
bc = make_test_client(binary=True)
keys = ["delone", b"deltwo"]
values = [b"valone", "valtwo"]
refcountables = keys + values
initial_refcounts = get_refcounts(refcountables)
bc.set(keys[0], values[0])
assert get_refcounts(refcountables) == initial_refcounts
bc.set(keys[1], values[1])
assert get_refcounts(refcountables) == initial_refcounts
bc.delete(keys[0])
assert get_refcounts(refcountables) == initial_refcounts
bc.delete(keys[1])
assert get_refcounts(refcountables) == initial_refcounts
bc.delete(keys[0])
assert get_refcounts(refcountables) == initial_refcounts
bc.delete(keys[1])
assert get_refcounts(refcountables) == initial_refcounts
def test_incr(self):
bc = make_test_client(binary=True)
keys = [b"increment_key", "increment_key_again"]
refcountables = keys
initial_refcounts = get_refcounts(refcountables)
bc.set(keys[0], 1)
assert get_refcounts(refcountables) == initial_refcounts
bc.incr(keys[0])
assert get_refcounts(refcountables) == initial_refcounts
bc.set(keys[1], 5)
assert get_refcounts(refcountables) == initial_refcounts
bc.incr(keys[1])
assert get_refcounts(refcountables) == initial_refcounts
def test_set_and_delete_multi(self):
bc = make_test_client(binary=True)
keys = ["delone", b"deltwo", "delthree", "delfour"]
values = [b"valone", "valtwo", object(), 2]
refcountables = keys + values
initial_refcounts = get_refcounts(refcountables)
bc.set_multi(dict(zip(keys, values)))
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi([keys[0]])
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi([keys[1]])
assert get_refcounts(refcountables) == initial_refcounts
bc.set_multi(dict(zip(keys, values)))
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi(keys)
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi(keys)
assert get_refcounts(refcountables) == initial_refcounts
def test_prefixes(self):
bc = make_test_client(binary=True)
keys = ["prefixone", b"prefixtwo"]
prefix = "testprefix-"
values = [b"valone", "valtwo"]
refcountables = keys + values + [prefix]
initial_refcounts = get_refcounts(refcountables)
bc.set_multi(dict(zip(keys, values)), key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
bc.get_multi(keys, key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi([keys[0]], key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi([keys[1]], key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
bc.set_multi(dict(zip(keys, values)), key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi(keys, key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
bc.delete_multi(keys, key_prefix=prefix)
assert get_refcounts(refcountables) == initial_refcounts
def test_get_invalid_key(self):
bc = make_test_client(binary=True)
key = object()
initial_refcount = get_refcounts([key])
raised = False
try:
bc.get(key)
except TypeError:
raised = True
assert raised
assert get_refcounts([key]) == initial_refcount
def test_cas(self):
k = "testkey"
val = 1138478589238
mc = make_test_client(binary=False, behaviors={'cas': True})
refcountables = [k, val]
initial_refcounts = get_refcounts(refcountables)
assert mc.set(k, 0)
assert get_refcounts(refcountables) == initial_refcounts
while True:
rv, cas = mc.gets(k)
assert get_refcounts(refcountables) == initial_refcounts
assert mc.cas(k, rv + 1, cas)
assert get_refcounts(refcountables) == initial_refcounts
if rv == 10:
break
|
