Python code coverage for Lib/bsddb/test/test_dbshelve.py

#	count	content
1	n/a	"""
2	n/a	TestCases for checking dbShelve objects.
3	n/a	"""
4	n/a
5	n/a	import os, string, sys
6	n/a	import random
7	n/a	import unittest
8	n/a
9	n/a
10	n/a	from test_all import db, dbshelve, test_support, verbose, \
11	n/a	get_new_environment_path, get_new_database_path
12	n/a
13	n/a
14	n/a
15	n/a	if sys.version_info < (2, 4) :
16	n/a	from sets import Set as set
17	n/a
18	n/a
19	n/a
20	n/a	#----------------------------------------------------------------------
21	n/a
22	n/a	# We want the objects to be comparable so we can test dbshelve.values
23	n/a	# later on.
24	n/a	class DataClass:
25	n/a	def __init__(self):
26	n/a	self.value = random.random()
27	n/a
28	n/a	def __repr__(self) : # For Python 3.0 comparison
29	n/a	return "DataClass %f" %self.value
30	n/a
31	n/a	def __cmp__(self, other): # For Python 2.x comparison
32	n/a	return cmp(self.value, other)
33	n/a
34	n/a
35	n/a	class DBShelveTestCase(unittest.TestCase):
36	n/a	if sys.version_info < (2, 4):
37	n/a	def assertTrue(self, expr, msg=None):
38	n/a	return self.failUnless(expr,msg=msg)
39	n/a
40	n/a	if (sys.version_info < (2, 7)) or ((sys.version_info >= (3, 0)) and
41	n/a	(sys.version_info < (3, 2))) :
42	n/a	def assertIn(self, a, b, msg=None) :
43	n/a	return self.assertTrue(a in b, msg=msg)
44	n/a
45	n/a
46	n/a	def setUp(self):
47	n/a	if sys.version_info[0] >= 3 :
48	n/a	from test_all import do_proxy_db_py3k
49	n/a	self._flag_proxy_db_py3k = do_proxy_db_py3k(False)
50	n/a	self.filename = get_new_database_path()
51	n/a	self.do_open()
52	n/a
53	n/a	def tearDown(self):
54	n/a	if sys.version_info[0] >= 3 :
55	n/a	from test_all import do_proxy_db_py3k
56	n/a	do_proxy_db_py3k(self._flag_proxy_db_py3k)
57	n/a	self.do_close()
58	n/a	test_support.unlink(self.filename)
59	n/a
60	n/a	def mk(self, key):
61	n/a	"""Turn key into an appropriate key type for this db"""
62	n/a	# override in child class for RECNO
63	n/a	if sys.version_info[0] < 3 :
64	n/a	return key
65	n/a	else :
66	n/a	return bytes(key, "iso8859-1") # 8 bits
67	n/a
68	n/a	def populateDB(self, d):
69	n/a	for x in string.letters:
70	n/a	d[self.mk('S' + x)] = 10 * x # add a string
71	n/a	d[self.mk('I' + x)] = ord(x) # add an integer
72	n/a	d[self.mk('L' + x)] = [x] * 10 # add a list
73	n/a
74	n/a	inst = DataClass() # add an instance
75	n/a	inst.S = 10 * x
76	n/a	inst.I = ord(x)
77	n/a	inst.L = [x] * 10
78	n/a	d[self.mk('O' + x)] = inst
79	n/a
80	n/a
81	n/a	# overridable in derived classes to affect how the shelf is created/opened
82	n/a	def do_open(self):
83	n/a	self.d = dbshelve.open(self.filename)
84	n/a
85	n/a	# and closed...
86	n/a	def do_close(self):
87	n/a	self.d.close()
88	n/a
89	n/a
90	n/a
91	n/a	def test01_basics(self):
92	n/a	if verbose:
93	n/a	print '\n', '-=' * 30
94	n/a	print "Running %s.test01_basics..." % self.__class__.__name__
95	n/a
96	n/a	self.populateDB(self.d)
97	n/a	self.d.sync()
98	n/a	self.do_close()
99	n/a	self.do_open()
100	n/a	d = self.d
101	n/a
102	n/a	l = len(d)
103	n/a	k = d.keys()
104	n/a	s = d.stat()
105	n/a	f = d.fd()
106	n/a
107	n/a	if verbose:
108	n/a	print "length:", l
109	n/a	print "keys:", k
110	n/a	print "stats:", s
111	n/a
112	n/a	self.assertEqual(0, d.has_key(self.mk('bad key')))
113	n/a	self.assertEqual(1, d.has_key(self.mk('IA')))
114	n/a	self.assertEqual(1, d.has_key(self.mk('OA')))
115	n/a
116	n/a	d.delete(self.mk('IA'))
117	n/a	del d[self.mk('OA')]
118	n/a	self.assertEqual(0, d.has_key(self.mk('IA')))
119	n/a	self.assertEqual(0, d.has_key(self.mk('OA')))
120	n/a	self.assertEqual(len(d), l-2)
121	n/a
122	n/a	values = []
123	n/a	for key in d.keys():
124	n/a	value = d[key]
125	n/a	values.append(value)
126	n/a	if verbose:
127	n/a	print "%s: %s" % (key, value)
128	n/a	self.checkrec(key, value)
129	n/a
130	n/a	dbvalues = d.values()
131	n/a	self.assertEqual(len(dbvalues), len(d.keys()))
132	n/a	if sys.version_info < (2, 6) :
133	n/a	values.sort()
134	n/a	dbvalues.sort()
135	n/a	self.assertEqual(values, dbvalues)
136	n/a	else : # XXX: Convert all to strings. Please, improve
137	n/a	values.sort(key=lambda x : str(x))
138	n/a	dbvalues.sort(key=lambda x : str(x))
139	n/a	self.assertEqual(repr(values), repr(dbvalues))
140	n/a
141	n/a	items = d.items()
142	n/a	self.assertEqual(len(items), len(values))
143	n/a
144	n/a	for key, value in items:
145	n/a	self.checkrec(key, value)
146	n/a
147	n/a	self.assertEqual(d.get(self.mk('bad key')), None)
148	n/a	self.assertEqual(d.get(self.mk('bad key'), None), None)
149	n/a	self.assertEqual(d.get(self.mk('bad key'), 'a string'), 'a string')
150	n/a	self.assertEqual(d.get(self.mk('bad key'), [1, 2, 3]), [1, 2, 3])
151	n/a
152	n/a	d.set_get_returns_none(0)
153	n/a	self.assertRaises(db.DBNotFoundError, d.get, self.mk('bad key'))
154	n/a	d.set_get_returns_none(1)
155	n/a
156	n/a	d.put(self.mk('new key'), 'new data')
157	n/a	self.assertEqual(d.get(self.mk('new key')), 'new data')
158	n/a	self.assertEqual(d[self.mk('new key')], 'new data')
159	n/a
160	n/a
161	n/a
162	n/a	def test02_cursors(self):
163	n/a	if verbose:
164	n/a	print '\n', '-=' * 30
165	n/a	print "Running %s.test02_cursors..." % self.__class__.__name__
166	n/a
167	n/a	self.populateDB(self.d)
168	n/a	d = self.d
169	n/a
170	n/a	count = 0
171	n/a	c = d.cursor()
172	n/a	rec = c.first()
173	n/a	while rec is not None:
174	n/a	count = count + 1
175	n/a	if verbose:
176	n/a	print rec
177	n/a	key, value = rec
178	n/a	self.checkrec(key, value)
179	n/a	# Hack to avoid conversion by 2to3 tool
180	n/a	rec = getattr(c, "next")()
181	n/a	del c
182	n/a
183	n/a	self.assertEqual(count, len(d))
184	n/a
185	n/a	count = 0
186	n/a	c = d.cursor()
187	n/a	rec = c.last()
188	n/a	while rec is not None:
189	n/a	count = count + 1
190	n/a	if verbose:
191	n/a	print rec
192	n/a	key, value = rec
193	n/a	self.checkrec(key, value)
194	n/a	rec = c.prev()
195	n/a
196	n/a	self.assertEqual(count, len(d))
197	n/a
198	n/a	c.set(self.mk('SS'))
199	n/a	key, value = c.current()
200	n/a	self.checkrec(key, value)
201	n/a	del c
202	n/a
203	n/a
204	n/a	def test03_append(self):
205	n/a	# NOTE: this is overridden in RECNO subclass, don't change its name.
206	n/a	if verbose:
207	n/a	print '\n', '-=' * 30
208	n/a	print "Running %s.test03_append..." % self.__class__.__name__
209	n/a
210	n/a	self.assertRaises(dbshelve.DBShelveError,
211	n/a	self.d.append, 'unit test was here')
212	n/a
213	n/a
214	n/a	def test04_iterable(self) :
215	n/a	self.populateDB(self.d)
216	n/a	d = self.d
217	n/a	keys = d.keys()
218	n/a	keyset = set(keys)
219	n/a	self.assertEqual(len(keyset), len(keys))
220	n/a
221	n/a	for key in d :
222	n/a	self.assertIn(key, keyset)
223	n/a	keyset.remove(key)
224	n/a	self.assertEqual(len(keyset), 0)
225	n/a
226	n/a	def checkrec(self, key, value):
227	n/a	# override this in a subclass if the key type is different
228	n/a
229	n/a	if sys.version_info[0] >= 3 :
230	n/a	if isinstance(key, bytes) :
231	n/a	key = key.decode("iso8859-1") # 8 bits
232	n/a
233	n/a	x = key[1]
234	n/a	if key[0] == 'S':
235	n/a	self.assertEqual(type(value), str)
236	n/a	self.assertEqual(value, 10 * x)
237	n/a
238	n/a	elif key[0] == 'I':
239	n/a	self.assertEqual(type(value), int)
240	n/a	self.assertEqual(value, ord(x))
241	n/a
242	n/a	elif key[0] == 'L':
243	n/a	self.assertEqual(type(value), list)
244	n/a	self.assertEqual(value, [x] * 10)
245	n/a
246	n/a	elif key[0] == 'O':
247	n/a	if sys.version_info[0] < 3 :
248	n/a	from types import InstanceType
249	n/a	self.assertEqual(type(value), InstanceType)
250	n/a	else :
251	n/a	self.assertEqual(type(value), DataClass)
252	n/a
253	n/a	self.assertEqual(value.S, 10 * x)
254	n/a	self.assertEqual(value.I, ord(x))
255	n/a	self.assertEqual(value.L, [x] * 10)
256	n/a
257	n/a	else:
258	n/a	self.assert_(0, 'Unknown key type, fix the test')
259	n/a
260	n/a	#----------------------------------------------------------------------
261	n/a
262	n/a	class BasicShelveTestCase(DBShelveTestCase):
263	n/a	def do_open(self):
264	n/a	self.d = dbshelve.DBShelf()
265	n/a	self.d.open(self.filename, self.dbtype, self.dbflags)
266	n/a
267	n/a	def do_close(self):
268	n/a	self.d.close()
269	n/a
270	n/a
271	n/a	class BTreeShelveTestCase(BasicShelveTestCase):
272	n/a	dbtype = db.DB_BTREE
273	n/a	dbflags = db.DB_CREATE
274	n/a
275	n/a
276	n/a	class HashShelveTestCase(BasicShelveTestCase):
277	n/a	dbtype = db.DB_HASH
278	n/a	dbflags = db.DB_CREATE
279	n/a
280	n/a
281	n/a	class ThreadBTreeShelveTestCase(BasicShelveTestCase):
282	n/a	dbtype = db.DB_BTREE
283	n/a	dbflags = db.DB_CREATE \| db.DB_THREAD
284	n/a
285	n/a
286	n/a	class ThreadHashShelveTestCase(BasicShelveTestCase):
287	n/a	dbtype = db.DB_HASH
288	n/a	dbflags = db.DB_CREATE \| db.DB_THREAD
289	n/a
290	n/a
291	n/a	#----------------------------------------------------------------------
292	n/a
293	n/a	class BasicEnvShelveTestCase(DBShelveTestCase):
294	n/a	def do_open(self):
295	n/a	self.env = db.DBEnv()
296	n/a	self.env.open(self.homeDir,
297	n/a	self.envflags \| db.DB_INIT_MPOOL \| db.DB_CREATE)
298	n/a
299	n/a	self.filename = os.path.split(self.filename)[1]
300	n/a	self.d = dbshelve.DBShelf(self.env)
301	n/a	self.d.open(self.filename, self.dbtype, self.dbflags)
302	n/a
303	n/a
304	n/a	def do_close(self):
305	n/a	self.d.close()
306	n/a	self.env.close()
307	n/a
308	n/a
309	n/a	def setUp(self) :
310	n/a	self.homeDir = get_new_environment_path()
311	n/a	DBShelveTestCase.setUp(self)
312	n/a
313	n/a	def tearDown(self):
314	n/a	if sys.version_info[0] >= 3 :
315	n/a	from test_all import do_proxy_db_py3k
316	n/a	do_proxy_db_py3k(self._flag_proxy_db_py3k)
317	n/a	self.do_close()
318	n/a	test_support.rmtree(self.homeDir)
319	n/a
320	n/a
321	n/a	class EnvBTreeShelveTestCase(BasicEnvShelveTestCase):
322	n/a	envflags = 0
323	n/a	dbtype = db.DB_BTREE
324	n/a	dbflags = db.DB_CREATE
325	n/a
326	n/a
327	n/a	class EnvHashShelveTestCase(BasicEnvShelveTestCase):
328	n/a	envflags = 0
329	n/a	dbtype = db.DB_HASH
330	n/a	dbflags = db.DB_CREATE
331	n/a
332	n/a
333	n/a	class EnvThreadBTreeShelveTestCase(BasicEnvShelveTestCase):
334	n/a	envflags = db.DB_THREAD
335	n/a	dbtype = db.DB_BTREE
336	n/a	dbflags = db.DB_CREATE \| db.DB_THREAD
337	n/a
338	n/a
339	n/a	class EnvThreadHashShelveTestCase(BasicEnvShelveTestCase):
340	n/a	envflags = db.DB_THREAD
341	n/a	dbtype = db.DB_HASH
342	n/a	dbflags = db.DB_CREATE \| db.DB_THREAD
343	n/a
344	n/a
345	n/a	#----------------------------------------------------------------------
346	n/a	# test cases for a DBShelf in a RECNO DB.
347	n/a
348	n/a	class RecNoShelveTestCase(BasicShelveTestCase):
349	n/a	dbtype = db.DB_RECNO
350	n/a	dbflags = db.DB_CREATE
351	n/a
352	n/a	def setUp(self):
353	n/a	BasicShelveTestCase.setUp(self)
354	n/a
355	n/a	# pool to assign integer key values out of
356	n/a	self.key_pool = list(range(1, 5000))
357	n/a	self.key_map = {} # map string keys to the number we gave them
358	n/a	self.intkey_map = {} # reverse map of above
359	n/a
360	n/a	def mk(self, key):
361	n/a	if key not in self.key_map:
362	n/a	self.key_map[key] = self.key_pool.pop(0)
363	n/a	self.intkey_map[self.key_map[key]] = key
364	n/a	return self.key_map[key]
365	n/a
366	n/a	def checkrec(self, intkey, value):
367	n/a	key = self.intkey_map[intkey]
368	n/a	BasicShelveTestCase.checkrec(self, key, value)
369	n/a
370	n/a	def test03_append(self):
371	n/a	if verbose:
372	n/a	print '\n', '-=' * 30
373	n/a	print "Running %s.test03_append..." % self.__class__.__name__
374	n/a
375	n/a	self.d[1] = 'spam'
376	n/a	self.d[5] = 'eggs'
377	n/a	self.assertEqual(6, self.d.append('spam'))
378	n/a	self.assertEqual(7, self.d.append('baked beans'))
379	n/a	self.assertEqual('spam', self.d.get(6))
380	n/a	self.assertEqual('spam', self.d.get(1))
381	n/a	self.assertEqual('baked beans', self.d.get(7))
382	n/a	self.assertEqual('eggs', self.d.get(5))
383	n/a
384	n/a
385	n/a	#----------------------------------------------------------------------
386	n/a
387	n/a	def test_suite():
388	n/a	suite = unittest.TestSuite()
389	n/a
390	n/a	suite.addTest(unittest.makeSuite(DBShelveTestCase))
391	n/a	suite.addTest(unittest.makeSuite(BTreeShelveTestCase))
392	n/a	suite.addTest(unittest.makeSuite(HashShelveTestCase))
393	n/a	suite.addTest(unittest.makeSuite(ThreadBTreeShelveTestCase))
394	n/a	suite.addTest(unittest.makeSuite(ThreadHashShelveTestCase))
395	n/a	suite.addTest(unittest.makeSuite(EnvBTreeShelveTestCase))
396	n/a	suite.addTest(unittest.makeSuite(EnvHashShelveTestCase))
397	n/a	suite.addTest(unittest.makeSuite(EnvThreadBTreeShelveTestCase))
398	n/a	suite.addTest(unittest.makeSuite(EnvThreadHashShelveTestCase))
399	n/a	suite.addTest(unittest.makeSuite(RecNoShelveTestCase))
400	n/a
401	n/a	return suite
402	n/a
403	n/a
404	n/a	if __name__ == '__main__':
405	n/a	unittest.main(defaultTest='test_suite')