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

#	count	content
1	n/a	"""TestCases for distributed transactions.
2	n/a	"""
3	n/a
4	n/a	import os
5	n/a	import unittest
6	n/a
7	n/a	from test_all import db, test_support, get_new_environment_path, \
8	n/a	get_new_database_path
9	n/a
10	n/a	try :
11	n/a	a=set()
12	n/a	except : # Python 2.3
13	n/a	from sets import Set as set
14	n/a	else :
15	n/a	del a
16	n/a
17	n/a	from test_all import verbose
18	n/a
19	n/a	#----------------------------------------------------------------------
20	n/a
21	n/a	class DBTxn_distributed(unittest.TestCase):
22	n/a	num_txns=1234
23	n/a	nosync=True
24	n/a	must_open_db=False
25	n/a	def _create_env(self, must_open_db) :
26	n/a	self.dbenv = db.DBEnv()
27	n/a	self.dbenv.set_tx_max(self.num_txns)
28	n/a	self.dbenv.set_lk_max_lockers(self.num_txns*2)
29	n/a	self.dbenv.set_lk_max_locks(self.num_txns*2)
30	n/a	self.dbenv.set_lk_max_objects(self.num_txns*2)
31	n/a	if self.nosync :
32	n/a	self.dbenv.set_flags(db.DB_TXN_NOSYNC,True)
33	n/a	self.dbenv.open(self.homeDir, db.DB_CREATE \| db.DB_THREAD \|
34	n/a	db.DB_RECOVER \|
35	n/a	db.DB_INIT_TXN \| db.DB_INIT_LOG \| db.DB_INIT_MPOOL \|
36	n/a	db.DB_INIT_LOCK, 0666)
37	n/a	self.db = db.DB(self.dbenv)
38	n/a	self.db.set_re_len(db.DB_GID_SIZE)
39	n/a	if must_open_db :
40	n/a	if db.version() >= (4,2) :
41	n/a	txn=self.dbenv.txn_begin()
42	n/a	self.db.open(self.filename,
43	n/a	db.DB_QUEUE, db.DB_CREATE \| db.DB_THREAD, 0666,
44	n/a	txn=txn)
45	n/a	txn.commit()
46	n/a	else :
47	n/a	self.db.open(self.filename,
48	n/a	db.DB_QUEUE, db.DB_CREATE \| db.DB_THREAD, 0666)
49	n/a
50	n/a	def setUp(self) :
51	n/a	self.homeDir = get_new_environment_path()
52	n/a	self.filename = "test"
53	n/a	return self._create_env(must_open_db=True)
54	n/a
55	n/a	def _destroy_env(self):
56	n/a	if self.nosync or (db.version()[:2] == (4,6)): # Known bug
57	n/a	self.dbenv.log_flush()
58	n/a	self.db.close()
59	n/a	self.dbenv.close()
60	n/a
61	n/a	def tearDown(self):
62	n/a	self._destroy_env()
63	n/a	test_support.rmtree(self.homeDir)
64	n/a
65	n/a	def _recreate_env(self,must_open_db) :
66	n/a	self._destroy_env()
67	n/a	self._create_env(must_open_db)
68	n/a
69	n/a	def test01_distributed_transactions(self) :
70	n/a	txns=set()
71	n/a	adapt = lambda x : x
72	n/a	import sys
73	n/a	if sys.version_info[0] >= 3 :
74	n/a	adapt = lambda x : bytes(x, "ascii")
75	n/a	# Create transactions, "prepare" them, and
76	n/a	# let them be garbage collected.
77	n/a	for i in xrange(self.num_txns) :
78	n/a	txn = self.dbenv.txn_begin()
79	n/a	gid = "%%%dd" %db.DB_GID_SIZE
80	n/a	gid = adapt(gid %i)
81	n/a	self.db.put(i, gid, txn=txn, flags=db.DB_APPEND)
82	n/a	txns.add(gid)
83	n/a	txn.prepare(gid)
84	n/a	del txn
85	n/a
86	n/a	self._recreate_env(self.must_open_db)
87	n/a
88	n/a	# Get "to be recovered" transactions but
89	n/a	# let them be garbage collected.
90	n/a	recovered_txns=self.dbenv.txn_recover()
91	n/a	self.assertEquals(self.num_txns,len(recovered_txns))
92	n/a	for gid,txn in recovered_txns :
93	n/a	self.assert_(gid in txns)
94	n/a	del txn
95	n/a	del recovered_txns
96	n/a
97	n/a	self._recreate_env(self.must_open_db)
98	n/a
99	n/a	# Get "to be recovered" transactions. Commit, abort and
100	n/a	# discard them.
101	n/a	recovered_txns=self.dbenv.txn_recover()
102	n/a	self.assertEquals(self.num_txns,len(recovered_txns))
103	n/a	discard_txns=set()
104	n/a	committed_txns=set()
105	n/a	state=0
106	n/a	for gid,txn in recovered_txns :
107	n/a	if state==0 or state==1:
108	n/a	committed_txns.add(gid)
109	n/a	txn.commit()
110	n/a	elif state==2 :
111	n/a	txn.abort()
112	n/a	elif state==3 :
113	n/a	txn.discard()
114	n/a	discard_txns.add(gid)
115	n/a	state=-1
116	n/a	state+=1
117	n/a	del txn
118	n/a	del recovered_txns
119	n/a
120	n/a	self._recreate_env(self.must_open_db)
121	n/a
122	n/a	# Verify the discarded transactions are still
123	n/a	# around, and dispose them.
124	n/a	recovered_txns=self.dbenv.txn_recover()
125	n/a	self.assertEquals(len(discard_txns),len(recovered_txns))
126	n/a	for gid,txn in recovered_txns :
127	n/a	txn.abort()
128	n/a	del txn
129	n/a	del recovered_txns
130	n/a
131	n/a	self._recreate_env(must_open_db=True)
132	n/a
133	n/a	# Be sure there are not pending transactions.
134	n/a	# Check also database size.
135	n/a	recovered_txns=self.dbenv.txn_recover()
136	n/a	self.assert_(len(recovered_txns)==0)
137	n/a	self.assertEquals(len(committed_txns),self.db.stat()["nkeys"])
138	n/a
139	n/a	class DBTxn_distributedSYNC(DBTxn_distributed):
140	n/a	nosync=False
141	n/a
142	n/a	class DBTxn_distributed_must_open_db(DBTxn_distributed):
143	n/a	must_open_db=True
144	n/a
145	n/a	class DBTxn_distributedSYNC_must_open_db(DBTxn_distributed):
146	n/a	nosync=False
147	n/a	must_open_db=True
148	n/a
149	n/a	#----------------------------------------------------------------------
150	n/a
151	n/a	def test_suite():
152	n/a	suite = unittest.TestSuite()
153	n/a	if db.version() >= (4,5) :
154	n/a	suite.addTest(unittest.makeSuite(DBTxn_distributed))
155	n/a	suite.addTest(unittest.makeSuite(DBTxn_distributedSYNC))
156	n/a	if db.version() >= (4,6) :
157	n/a	suite.addTest(unittest.makeSuite(DBTxn_distributed_must_open_db))
158	n/a	suite.addTest(unittest.makeSuite(DBTxn_distributedSYNC_must_open_db))
159	n/a	return suite
160	n/a
161	n/a
162	n/a	if __name__ == '__main__':
163	n/a	unittest.main(defaultTest='test_suite')