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

#	count	content
1	n/a	"""TestCases for checking that it does not segfault when a DBEnv object
2	n/a	is closed before its DB objects.
3	n/a	"""
4	n/a
5	n/a	import os, sys
6	n/a	import unittest
7	n/a
8	n/a	from test_all import db, test_support, verbose, get_new_environment_path, get_new_database_path
9	n/a
10	n/a	# We're going to get warnings in this module about trying to close the db when
11	n/a	# its env is already closed. Let's just ignore those.
12	n/a	try:
13	n/a	import warnings
14	n/a	except ImportError:
15	n/a	pass
16	n/a	else:
17	n/a	warnings.filterwarnings('ignore',
18	n/a	message='DB could not be closed in',
19	n/a	category=RuntimeWarning)
20	n/a
21	n/a
22	n/a	#----------------------------------------------------------------------
23	n/a
24	n/a	class DBEnvClosedEarlyCrash(unittest.TestCase):
25	n/a	def setUp(self):
26	n/a	self.homeDir = get_new_environment_path()
27	n/a	self.filename = "test"
28	n/a
29	n/a	def tearDown(self):
30	n/a	test_support.rmtree(self.homeDir)
31	n/a
32	n/a	def test01_close_dbenv_before_db(self):
33	n/a	dbenv = db.DBEnv()
34	n/a	dbenv.open(self.homeDir,
35	n/a	db.DB_INIT_CDB\| db.DB_CREATE \|db.DB_THREAD\|db.DB_INIT_MPOOL,
36	n/a	0666)
37	n/a
38	n/a	d = db.DB(dbenv)
39	n/a	d2 = db.DB(dbenv)
40	n/a	d.open(self.filename, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
41	n/a
42	n/a	self.assertRaises(db.DBNoSuchFileError, d2.open,
43	n/a	self.filename+"2", db.DB_BTREE, db.DB_THREAD, 0666)
44	n/a
45	n/a	d.put("test","this is a test")
46	n/a	self.assertEqual(d.get("test"), "this is a test", "put!=get")
47	n/a	dbenv.close() # This "close" should close the child db handle also
48	n/a	self.assertRaises(db.DBError, d.get, "test")
49	n/a
50	n/a	def test02_close_dbenv_before_dbcursor(self):
51	n/a	dbenv = db.DBEnv()
52	n/a	dbenv.open(self.homeDir,
53	n/a	db.DB_INIT_CDB\| db.DB_CREATE \|db.DB_THREAD\|db.DB_INIT_MPOOL,
54	n/a	0666)
55	n/a
56	n/a	d = db.DB(dbenv)
57	n/a	d.open(self.filename, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
58	n/a
59	n/a	d.put("test","this is a test")
60	n/a	d.put("test2","another test")
61	n/a	d.put("test3","another one")
62	n/a	self.assertEqual(d.get("test"), "this is a test", "put!=get")
63	n/a	c=d.cursor()
64	n/a	c.first()
65	n/a	c.next()
66	n/a	d.close() # This "close" should close the child db handle also
67	n/a	# db.close should close the child cursor
68	n/a	self.assertRaises(db.DBError,c.next)
69	n/a
70	n/a	d = db.DB(dbenv)
71	n/a	d.open(self.filename, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
72	n/a	c=d.cursor()
73	n/a	c.first()
74	n/a	c.next()
75	n/a	dbenv.close()
76	n/a	# The "close" should close the child db handle also, with cursors
77	n/a	self.assertRaises(db.DBError, c.next)
78	n/a
79	n/a	def test03_close_db_before_dbcursor_without_env(self):
80	n/a	import os.path
81	n/a	path=os.path.join(self.homeDir,self.filename)
82	n/a	d = db.DB()
83	n/a	d.open(path, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
84	n/a
85	n/a	d.put("test","this is a test")
86	n/a	d.put("test2","another test")
87	n/a	d.put("test3","another one")
88	n/a	self.assertEqual(d.get("test"), "this is a test", "put!=get")
89	n/a	c=d.cursor()
90	n/a	c.first()
91	n/a	c.next()
92	n/a	d.close()
93	n/a	# The "close" should close the child db handle also
94	n/a	self.assertRaises(db.DBError, c.next)
95	n/a
96	n/a	def test04_close_massive(self):
97	n/a	dbenv = db.DBEnv()
98	n/a	dbenv.open(self.homeDir,
99	n/a	db.DB_INIT_CDB\| db.DB_CREATE \|db.DB_THREAD\|db.DB_INIT_MPOOL,
100	n/a	0666)
101	n/a
102	n/a	dbs=[db.DB(dbenv) for i in xrange(16)]
103	n/a	cursors=[]
104	n/a	for i in dbs :
105	n/a	i.open(self.filename, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
106	n/a
107	n/a	dbs[10].put("test","this is a test")
108	n/a	dbs[10].put("test2","another test")
109	n/a	dbs[10].put("test3","another one")
110	n/a	self.assertEqual(dbs[4].get("test"), "this is a test", "put!=get")
111	n/a
112	n/a	for i in dbs :
113	n/a	cursors.extend([i.cursor() for j in xrange(32)])
114	n/a
115	n/a	for i in dbs[::3] :
116	n/a	i.close()
117	n/a	for i in cursors[::3] :
118	n/a	i.close()
119	n/a
120	n/a	# Check for missing exception in DB! (after DB close)
121	n/a	self.assertRaises(db.DBError, dbs[9].get, "test")
122	n/a
123	n/a	# Check for missing exception in DBCursor! (after DB close)
124	n/a	self.assertRaises(db.DBError, cursors[101].first)
125	n/a
126	n/a	cursors[80].first()
127	n/a	cursors[80].next()
128	n/a	dbenv.close() # This "close" should close the child db handle also
129	n/a	# Check for missing exception! (after DBEnv close)
130	n/a	self.assertRaises(db.DBError, cursors[80].next)
131	n/a
132	n/a	def test05_close_dbenv_delete_db_success(self):
133	n/a	dbenv = db.DBEnv()
134	n/a	dbenv.open(self.homeDir,
135	n/a	db.DB_INIT_CDB\| db.DB_CREATE \|db.DB_THREAD\|db.DB_INIT_MPOOL,
136	n/a	0666)
137	n/a
138	n/a	d = db.DB(dbenv)
139	n/a	d.open(self.filename, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
140	n/a
141	n/a	dbenv.close() # This "close" should close the child db handle also
142	n/a
143	n/a	del d
144	n/a	try:
145	n/a	import gc
146	n/a	except ImportError:
147	n/a	gc = None
148	n/a	if gc:
149	n/a	# force d.__del__ [DB_dealloc] to be called
150	n/a	gc.collect()
151	n/a
152	n/a	def test06_close_txn_before_dup_cursor(self) :
153	n/a	dbenv = db.DBEnv()
154	n/a	dbenv.open(self.homeDir,db.DB_INIT_TXN \| db.DB_INIT_MPOOL \|
155	n/a	db.DB_INIT_LOG \| db.DB_CREATE)
156	n/a	d = db.DB(dbenv)
157	n/a	txn = dbenv.txn_begin()
158	n/a	d.open(self.filename, dbtype = db.DB_HASH, flags = db.DB_CREATE,
159	n/a	txn=txn)
160	n/a	d.put("XXX", "yyy", txn=txn)
161	n/a	txn.commit()
162	n/a	txn = dbenv.txn_begin()
163	n/a	c1 = d.cursor(txn)
164	n/a	c2 = c1.dup()
165	n/a	self.assertEquals(("XXX", "yyy"), c1.first())
166	n/a
167	n/a	# Not interested in warnings about implicit close.
168	n/a	import warnings
169	n/a	if sys.version_info < (2, 6) :
170	n/a	# Completely resetting the warning state is
171	n/a	# problematic with python >=2.6 with -3 (py3k warning),
172	n/a	# because some stdlib modules selectively ignores warnings.
173	n/a	warnings.simplefilter("ignore")
174	n/a	txn.commit()
175	n/a	warnings.resetwarnings()
176	n/a	else :
177	n/a	# When we drop support for python 2.3 and 2.4
178	n/a	# we could use: (in 2.5 we need a __future__ statement)
179	n/a	#
180	n/a	# with warnings.catch_warnings():
181	n/a	# warnings.simplefilter("ignore")
182	n/a	# txn.commit()
183	n/a	#
184	n/a	# We can not use "with" as is, because it would be invalid syntax
185	n/a	# in python 2.3, 2.4 and (with no __future__) 2.5.
186	n/a	# Here we simulate "with" following PEP 343 :
187	n/a	w = warnings.catch_warnings()
188	n/a	w.__enter__()
189	n/a	try :
190	n/a	warnings.simplefilter("ignore")
191	n/a	txn.commit()
192	n/a	finally :
193	n/a	w.__exit__()
194	n/a
195	n/a	self.assertRaises(db.DBCursorClosedError, c2.first)
196	n/a
197	n/a	if db.version() > (4,3,0) :
198	n/a	def test07_close_db_before_sequence(self):
199	n/a	import os.path
200	n/a	path=os.path.join(self.homeDir,self.filename)
201	n/a	d = db.DB()
202	n/a	d.open(path, db.DB_BTREE, db.DB_CREATE \| db.DB_THREAD, 0666)
203	n/a	dbs=db.DBSequence(d)
204	n/a	d.close() # This "close" should close the child DBSequence also
205	n/a	dbs.close() # If not closed, core dump (in Berkeley DB 4.6.*)
206	n/a
207	n/a	#----------------------------------------------------------------------
208	n/a
209	n/a	def test_suite():
210	n/a	suite = unittest.TestSuite()
211	n/a	suite.addTest(unittest.makeSuite(DBEnvClosedEarlyCrash))
212	n/a	return suite
213	n/a
214	n/a
215	n/a	if __name__ == '__main__':
216	n/a	unittest.main(defaultTest='test_suite')