Improve unit tests for transactions.

refs #138

--HG--
rename : test/suite/test_txn01.py => test/suite/test_txn02.py

1 files changed, 34 insertions, 15 deletions

diff --git a/test/suite/test_txn01.py b/test/suite/test_txn01.py
index e6264082ea2..c0a4c6d78cd 100644
--- a/test/suite/test_txn01.py
+++ b/test/suite/test_txn01.py
@@ -25,8 +25,8 @@
 # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 # OTHER DEALINGS IN THE SOFTWARE.
 #
-# test_util12.py
-#	Utilities: wt write
+# test_txn01.py
+#	Transactions: basic functionality
 #
 
 import os, struct
@@ -34,8 +34,9 @@ import wiredtiger, wttest
 
 class test_txn01(wttest.WiredTigerTestCase):
     tablename = 'test_txn01'
+    uri = 'table:' + tablename
     nentries = 10000
-    session_params = 'key_format=r,value_format=S'
+    create_params = 'key_format=r,value_format=S'
 
     # Overrides WiredTigerTestCase
     def setUpConnectionOpen(self, dir):
@@ -45,36 +46,54 @@ class test_txn01(wttest.WiredTigerTestCase):
         self.pr(`conn`)
         return conn
 
-    def check_count(self, expected):
+    def check_checkpoint(self, expected):
         s = self.conn.open_session()
         s.checkpoint("snapshot=test")
         try:
-            cursor = s.open_cursor('table:' + self.tablename, None, "snapshot=test")
+            cursor = s.open_cursor(self.uri, None, "snapshot=test")
+            count = 0
+            for r in cursor:
+                count += 1
+        finally:
+            s.close()
+        self.assertEqual(count, expected)
+
+    def check_transaction(self, expected):
+        s = self.conn.open_session()
+        s.begin_transaction('isolation=snapshot')
+        try:
+            cursor = s.open_cursor(self.uri, None)
             count = 0
             for r in cursor:
                 count += 1
-            cursor.close()
         finally:
             s.close()
         self.assertEqual(count, expected)
 
-    def test_write(self):
-        self.session.create('table:' + self.tablename, self.session_params)
-        self.check_count(0)
+    def check_count(self, expected):
+        self.check_transaction(expected)
+        self.check_checkpoint(expected)
+
+    def test_visibilty(self):
+        self.session.create(self.uri, self.create_params)
+        committed_inserts = 0
+        self.check_count(committed_inserts)
         self.session.begin_transaction()
-        cursor = self.session.open_cursor('table:' + self.tablename, None, "append")
+        cursor = self.session.open_cursor(self.uri, None, "append")
         for i in xrange(self.nentries):
             if i > 0 and (i * 10) % self.nentries == 0:
-                self.check_count(i - self.nentries / 10)
+                self.check_count(committed_inserts)
                 self.session.commit_transaction()
+                committed_inserts = i
                 self.session.begin_transaction()
-                cursor = self.session.open_cursor('table:' + self.tablename, None, "append")
-            cursor.set_value(("value" + str(i)) * 100)
+                cursor = self.session.open_cursor(self.uri, None, "append")
+            cursor.set_value(("value%06d" % i) * 100)
             cursor.insert()
         cursor.close()
-        self.check_count(self.nentries - self.nentries / 10)
+        self.check_count(committed_inserts)
         self.session.commit_transaction()
-        self.check_count(self.nentries)
+        committed_inserts = self.nentries
+        self.check_count(committed_inserts)
 
 if __name__ == '__main__':
     wttest.run()