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

#	count	content
1	n/a	#!/usr/bin/env python
2	n/a	#
3	n/a	#-----------------------------------------------------------------------
4	n/a	# A test suite for the table interface built on bsddb.db
5	n/a	#-----------------------------------------------------------------------
6	n/a	#
7	n/a	# Copyright (C) 2000, 2001 by Autonomous Zone Industries
8	n/a	# Copyright (C) 2002 Gregory P. Smith
9	n/a	#
10	n/a	# March 20, 2000
11	n/a	#
12	n/a	# License: This is free software. You may use this software for any
13	n/a	# purpose including modification/redistribution, so long as
14	n/a	# this header remains intact and that you do not claim any
15	n/a	# rights of ownership or authorship of this software. This
16	n/a	# software has been tested, but no warranty is expressed or
17	n/a	# implied.
18	n/a	#
19	n/a	# -- Gregory P. Smith <greg@krypto.org>
20	n/a	#
21	n/a	# $Id: test_dbtables.py 79285 2010-03-22 14:22:26Z jesus.cea $
22	n/a
23	n/a	import os, re, sys
24	n/a
25	n/a	if sys.version_info[0] < 3 :
26	n/a	try:
27	n/a	import cPickle
28	n/a	pickle = cPickle
29	n/a	except ImportError:
30	n/a	import pickle
31	n/a	else :
32	n/a	import pickle
33	n/a
34	n/a	import unittest
35	n/a	from test_all import db, dbtables, test_support, verbose, \
36	n/a	get_new_environment_path, get_new_database_path
37	n/a
38	n/a	#----------------------------------------------------------------------
39	n/a
40	n/a	class TableDBTestCase(unittest.TestCase):
41	n/a	db_name = 'test-table.db'
42	n/a
43	n/a	def setUp(self):
44	n/a	import sys
45	n/a	if sys.version_info[0] >= 3 :
46	n/a	from test_all import do_proxy_db_py3k
47	n/a	self._flag_proxy_db_py3k = do_proxy_db_py3k(False)
48	n/a
49	n/a	self.testHomeDir = get_new_environment_path()
50	n/a	self.tdb = dbtables.bsdTableDB(
51	n/a	filename='tabletest.db', dbhome=self.testHomeDir, create=1)
52	n/a
53	n/a	def tearDown(self):
54	n/a	self.tdb.close()
55	n/a	import sys
56	n/a	if sys.version_info[0] >= 3 :
57	n/a	from test_all import do_proxy_db_py3k
58	n/a	do_proxy_db_py3k(self._flag_proxy_db_py3k)
59	n/a	test_support.rmtree(self.testHomeDir)
60	n/a
61	n/a	def test01(self):
62	n/a	tabname = "test01"
63	n/a	colname = 'cool numbers'
64	n/a	try:
65	n/a	self.tdb.Drop(tabname)
66	n/a	except dbtables.TableDBError:
67	n/a	pass
68	n/a	self.tdb.CreateTable(tabname, [colname])
69	n/a	import sys
70	n/a	if sys.version_info[0] < 3 :
71	n/a	self.tdb.Insert(tabname, {colname: pickle.dumps(3.14159, 1)})
72	n/a	else :
73	n/a	self.tdb.Insert(tabname, {colname: pickle.dumps(3.14159,
74	n/a	1).decode("iso8859-1")}) # 8 bits
75	n/a
76	n/a	if verbose:
77	n/a	self.tdb._db_print()
78	n/a
79	n/a	values = self.tdb.Select(
80	n/a	tabname, [colname], conditions={colname: None})
81	n/a
82	n/a	import sys
83	n/a	if sys.version_info[0] < 3 :
84	n/a	colval = pickle.loads(values[0][colname])
85	n/a	else :
86	n/a	colval = pickle.loads(bytes(values[0][colname], "iso8859-1"))
87	n/a	self.assert_(colval > 3.141)
88	n/a	self.assert_(colval < 3.142)
89	n/a
90	n/a
91	n/a	def test02(self):
92	n/a	tabname = "test02"
93	n/a	col0 = 'coolness factor'
94	n/a	col1 = 'but can it fly?'
95	n/a	col2 = 'Species'
96	n/a
97	n/a	import sys
98	n/a	if sys.version_info[0] < 3 :
99	n/a	testinfo = [
100	n/a	{col0: pickle.dumps(8, 1), col1: 'no', col2: 'Penguin'},
101	n/a	{col0: pickle.dumps(-1, 1), col1: 'no', col2: 'Turkey'},
102	n/a	{col0: pickle.dumps(9, 1), col1: 'yes', col2: 'SR-71A Blackbird'}
103	n/a	]
104	n/a	else :
105	n/a	testinfo = [
106	n/a	{col0: pickle.dumps(8, 1).decode("iso8859-1"),
107	n/a	col1: 'no', col2: 'Penguin'},
108	n/a	{col0: pickle.dumps(-1, 1).decode("iso8859-1"),
109	n/a	col1: 'no', col2: 'Turkey'},
110	n/a	{col0: pickle.dumps(9, 1).decode("iso8859-1"),
111	n/a	col1: 'yes', col2: 'SR-71A Blackbird'}
112	n/a	]
113	n/a
114	n/a	try:
115	n/a	self.tdb.Drop(tabname)
116	n/a	except dbtables.TableDBError:
117	n/a	pass
118	n/a	self.tdb.CreateTable(tabname, [col0, col1, col2])
119	n/a	for row in testinfo :
120	n/a	self.tdb.Insert(tabname, row)
121	n/a
122	n/a	import sys
123	n/a	if sys.version_info[0] < 3 :
124	n/a	values = self.tdb.Select(tabname, [col2],
125	n/a	conditions={col0: lambda x: pickle.loads(x) >= 8})
126	n/a	else :
127	n/a	values = self.tdb.Select(tabname, [col2],
128	n/a	conditions={col0: lambda x:
129	n/a	pickle.loads(bytes(x, "iso8859-1")) >= 8})
130	n/a
131	n/a	self.assertEqual(len(values), 2)
132	n/a	if values[0]['Species'] == 'Penguin' :
133	n/a	self.assertEqual(values[1]['Species'], 'SR-71A Blackbird')
134	n/a	elif values[0]['Species'] == 'SR-71A Blackbird' :
135	n/a	self.assertEqual(values[1]['Species'], 'Penguin')
136	n/a	else :
137	n/a	if verbose:
138	n/a	print "values= %r" % (values,)
139	n/a	raise RuntimeError("Wrong values returned!")
140	n/a
141	n/a	def test03(self):
142	n/a	tabname = "test03"
143	n/a	try:
144	n/a	self.tdb.Drop(tabname)
145	n/a	except dbtables.TableDBError:
146	n/a	pass
147	n/a	if verbose:
148	n/a	print '...before CreateTable...'
149	n/a	self.tdb._db_print()
150	n/a	self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e'])
151	n/a	if verbose:
152	n/a	print '...after CreateTable...'
153	n/a	self.tdb._db_print()
154	n/a	self.tdb.Drop(tabname)
155	n/a	if verbose:
156	n/a	print '...after Drop...'
157	n/a	self.tdb._db_print()
158	n/a	self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e'])
159	n/a
160	n/a	try:
161	n/a	self.tdb.Insert(tabname,
162	n/a	{'a': "",
163	n/a	'e': pickle.dumps([{4:5, 6:7}, 'foo'], 1),
164	n/a	'f': "Zero"})
165	n/a	self.fail('Expected an exception')
166	n/a	except dbtables.TableDBError:
167	n/a	pass
168	n/a
169	n/a	try:
170	n/a	self.tdb.Select(tabname, [], conditions={'foo': '123'})
171	n/a	self.fail('Expected an exception')
172	n/a	except dbtables.TableDBError:
173	n/a	pass
174	n/a
175	n/a	self.tdb.Insert(tabname,
176	n/a	{'a': '42',
177	n/a	'b': "bad",
178	n/a	'c': "meep",
179	n/a	'e': 'Fuzzy wuzzy was a bear'})
180	n/a	self.tdb.Insert(tabname,
181	n/a	{'a': '581750',
182	n/a	'b': "good",
183	n/a	'd': "bla",
184	n/a	'c': "black",
185	n/a	'e': 'fuzzy was here'})
186	n/a	self.tdb.Insert(tabname,
187	n/a	{'a': '800000',
188	n/a	'b': "good",
189	n/a	'd': "bla",
190	n/a	'c': "black",
191	n/a	'e': 'Fuzzy wuzzy is a bear'})
192	n/a
193	n/a	if verbose:
194	n/a	self.tdb._db_print()
195	n/a
196	n/a	# this should return two rows
197	n/a	values = self.tdb.Select(tabname, ['b', 'a', 'd'],
198	n/a	conditions={'e': re.compile('wuzzy').search,
199	n/a	'a': re.compile('^[0-9]+$').match})
200	n/a	self.assertEqual(len(values), 2)
201	n/a
202	n/a	# now lets delete one of them and try again
203	n/a	self.tdb.Delete(tabname, conditions={'b': dbtables.ExactCond('good')})
204	n/a	values = self.tdb.Select(
205	n/a	tabname, ['a', 'd', 'b'],
206	n/a	conditions={'e': dbtables.PrefixCond('Fuzzy')})
207	n/a	self.assertEqual(len(values), 1)
208	n/a	self.assertEqual(values[0]['d'], None)
209	n/a
210	n/a	values = self.tdb.Select(tabname, ['b'],
211	n/a	conditions={'c': lambda c: c == 'meep'})
212	n/a	self.assertEqual(len(values), 1)
213	n/a	self.assertEqual(values[0]['b'], "bad")
214	n/a
215	n/a
216	n/a	def test04_MultiCondSelect(self):
217	n/a	tabname = "test04_MultiCondSelect"
218	n/a	try:
219	n/a	self.tdb.Drop(tabname)
220	n/a	except dbtables.TableDBError:
221	n/a	pass
222	n/a	self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e'])
223	n/a
224	n/a	try:
225	n/a	self.tdb.Insert(tabname,
226	n/a	{'a': "",
227	n/a	'e': pickle.dumps([{4:5, 6:7}, 'foo'], 1),
228	n/a	'f': "Zero"})
229	n/a	self.fail('Expected an exception')
230	n/a	except dbtables.TableDBError:
231	n/a	pass
232	n/a
233	n/a	self.tdb.Insert(tabname, {'a': "A", 'b': "B", 'c': "C", 'd': "D",
234	n/a	'e': "E"})
235	n/a	self.tdb.Insert(tabname, {'a': "-A", 'b': "-B", 'c': "-C", 'd': "-D",
236	n/a	'e': "-E"})
237	n/a	self.tdb.Insert(tabname, {'a': "A-", 'b': "B-", 'c': "C-", 'd': "D-",
238	n/a	'e': "E-"})
239	n/a
240	n/a	if verbose:
241	n/a	self.tdb._db_print()
242	n/a
243	n/a	# This select should return 0 rows. it is designed to test
244	n/a	# the bug identified and fixed in sourceforge bug # 590449
245	n/a	# (Big Thanks to "Rob Tillotson (n9mtb)" for tracking this down
246	n/a	# and supplying a fix!! This one caused many headaches to say
247	n/a	# the least...)
248	n/a	values = self.tdb.Select(tabname, ['b', 'a', 'd'],
249	n/a	conditions={'e': dbtables.ExactCond('E'),
250	n/a	'a': dbtables.ExactCond('A'),
251	n/a	'd': dbtables.PrefixCond('-')
252	n/a	} )
253	n/a	self.assertEqual(len(values), 0, values)
254	n/a
255	n/a
256	n/a	def test_CreateOrExtend(self):
257	n/a	tabname = "test_CreateOrExtend"
258	n/a
259	n/a	self.tdb.CreateOrExtendTable(
260	n/a	tabname, ['name', 'taste', 'filling', 'alcohol content', 'price'])
261	n/a	try:
262	n/a	self.tdb.Insert(tabname,
263	n/a	{'taste': 'crap',
264	n/a	'filling': 'no',
265	n/a	'is it Guinness?': 'no'})
266	n/a	self.fail("Insert should've failed due to bad column name")
267	n/a	except:
268	n/a	pass
269	n/a	self.tdb.CreateOrExtendTable(tabname,
270	n/a	['name', 'taste', 'is it Guinness?'])
271	n/a
272	n/a	# these should both succeed as the table should contain the union of both sets of columns.
273	n/a	self.tdb.Insert(tabname, {'taste': 'crap', 'filling': 'no',
274	n/a	'is it Guinness?': 'no'})
275	n/a	self.tdb.Insert(tabname, {'taste': 'great', 'filling': 'yes',
276	n/a	'is it Guinness?': 'yes',
277	n/a	'name': 'Guinness'})
278	n/a
279	n/a
280	n/a	def test_CondObjs(self):
281	n/a	tabname = "test_CondObjs"
282	n/a
283	n/a	self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e', 'p'])
284	n/a
285	n/a	self.tdb.Insert(tabname, {'a': "the letter A",
286	n/a	'b': "the letter B",
287	n/a	'c': "is for cookie"})
288	n/a	self.tdb.Insert(tabname, {'a': "is for aardvark",
289	n/a	'e': "the letter E",
290	n/a	'c': "is for cookie",
291	n/a	'd': "is for dog"})
292	n/a	self.tdb.Insert(tabname, {'a': "the letter A",
293	n/a	'e': "the letter E",
294	n/a	'c': "is for cookie",
295	n/a	'p': "is for Python"})
296	n/a
297	n/a	values = self.tdb.Select(
298	n/a	tabname, ['p', 'e'],
299	n/a	conditions={'e': dbtables.PrefixCond('the l')})
300	n/a	self.assertEqual(len(values), 2, values)
301	n/a	self.assertEqual(values[0]['e'], values[1]['e'], values)
302	n/a	self.assertNotEqual(values[0]['p'], values[1]['p'], values)
303	n/a
304	n/a	values = self.tdb.Select(
305	n/a	tabname, ['d', 'a'],
306	n/a	conditions={'a': dbtables.LikeCond('%aardvark%')})
307	n/a	self.assertEqual(len(values), 1, values)
308	n/a	self.assertEqual(values[0]['d'], "is for dog", values)
309	n/a	self.assertEqual(values[0]['a'], "is for aardvark", values)
310	n/a
311	n/a	values = self.tdb.Select(tabname, None,
312	n/a	{'b': dbtables.Cond(),
313	n/a	'e':dbtables.LikeCond('%letter%'),
314	n/a	'a':dbtables.PrefixCond('is'),
315	n/a	'd':dbtables.ExactCond('is for dog'),
316	n/a	'c':dbtables.PrefixCond('is for'),
317	n/a	'p':lambda s: not s})
318	n/a	self.assertEqual(len(values), 1, values)
319	n/a	self.assertEqual(values[0]['d'], "is for dog", values)
320	n/a	self.assertEqual(values[0]['a'], "is for aardvark", values)
321	n/a
322	n/a	def test_Delete(self):
323	n/a	tabname = "test_Delete"
324	n/a	self.tdb.CreateTable(tabname, ['x', 'y', 'z'])
325	n/a
326	n/a	# prior to 2001-05-09 there was a bug where Delete() would
327	n/a	# fail if it encountered any rows that did not have values in
328	n/a	# every column.
329	n/a	# Hunted and Squashed by <Donwulff> (Jukka Santala - donwulff@nic.fi)
330	n/a	self.tdb.Insert(tabname, {'x': 'X1', 'y':'Y1'})
331	n/a	self.tdb.Insert(tabname, {'x': 'X2', 'y':'Y2', 'z': 'Z2'})
332	n/a
333	n/a	self.tdb.Delete(tabname, conditions={'x': dbtables.PrefixCond('X')})
334	n/a	values = self.tdb.Select(tabname, ['y'],
335	n/a	conditions={'x': dbtables.PrefixCond('X')})
336	n/a	self.assertEqual(len(values), 0)
337	n/a
338	n/a	def test_Modify(self):
339	n/a	tabname = "test_Modify"
340	n/a	self.tdb.CreateTable(tabname, ['Name', 'Type', 'Access'])
341	n/a
342	n/a	self.tdb.Insert(tabname, {'Name': 'Index to MP3 files.doc',
343	n/a	'Type': 'Word', 'Access': '8'})
344	n/a	self.tdb.Insert(tabname, {'Name': 'Nifty.MP3', 'Access': '1'})
345	n/a	self.tdb.Insert(tabname, {'Type': 'Unknown', 'Access': '0'})
346	n/a
347	n/a	def set_type(type):
348	n/a	if type is None:
349	n/a	return 'MP3'
350	n/a	return type
351	n/a
352	n/a	def increment_access(count):
353	n/a	return str(int(count)+1)
354	n/a
355	n/a	def remove_value(value):
356	n/a	return None
357	n/a
358	n/a	self.tdb.Modify(tabname,
359	n/a	conditions={'Access': dbtables.ExactCond('0')},
360	n/a	mappings={'Access': remove_value})
361	n/a	self.tdb.Modify(tabname,
362	n/a	conditions={'Name': dbtables.LikeCond('%MP3%')},
363	n/a	mappings={'Type': set_type})
364	n/a	self.tdb.Modify(tabname,
365	n/a	conditions={'Name': dbtables.LikeCond('%')},
366	n/a	mappings={'Access': increment_access})
367	n/a
368	n/a	try:
369	n/a	self.tdb.Modify(tabname,
370	n/a	conditions={'Name': dbtables.LikeCond('%')},
371	n/a	mappings={'Access': 'What is your quest?'})
372	n/a	except TypeError:
373	n/a	# success, the string value in mappings isn't callable
374	n/a	pass
375	n/a	else:
376	n/a	raise RuntimeError, "why was TypeError not raised for bad callable?"
377	n/a
378	n/a	# Delete key in select conditions
379	n/a	values = self.tdb.Select(
380	n/a	tabname, None,
381	n/a	conditions={'Type': dbtables.ExactCond('Unknown')})
382	n/a	self.assertEqual(len(values), 1, values)
383	n/a	self.assertEqual(values[0]['Name'], None, values)
384	n/a	self.assertEqual(values[0]['Access'], None, values)
385	n/a
386	n/a	# Modify value by select conditions
387	n/a	values = self.tdb.Select(
388	n/a	tabname, None,
389	n/a	conditions={'Name': dbtables.ExactCond('Nifty.MP3')})
390	n/a	self.assertEqual(len(values), 1, values)
391	n/a	self.assertEqual(values[0]['Type'], "MP3", values)
392	n/a	self.assertEqual(values[0]['Access'], "2", values)
393	n/a
394	n/a	# Make sure change applied only to select conditions
395	n/a	values = self.tdb.Select(
396	n/a	tabname, None, conditions={'Name': dbtables.LikeCond('%doc%')})
397	n/a	self.assertEqual(len(values), 1, values)
398	n/a	self.assertEqual(values[0]['Type'], "Word", values)
399	n/a	self.assertEqual(values[0]['Access'], "9", values)
400	n/a
401	n/a
402	n/a	def test_suite():
403	n/a	suite = unittest.TestSuite()
404	n/a	suite.addTest(unittest.makeSuite(TableDBTestCase))
405	n/a	return suite
406	n/a
407	n/a
408	n/a	if __name__ == '__main__':
409	n/a	unittest.main(defaultTest='test_suite')