-
Type: Task
-
Resolution: Done
-
Affects Version/s: None
-
Component/s: None
-
None
Here is a test case:
import wiredtiger, wttest from helper import key_populate, value_populate class test_f(wttest.WiredTigerTestCase): uri = 'table:x' def test_f(self): self.session.create(self.uri, 'key_format=S,value_format=S') self.session.checkpoint('') curs1 = self.session.open_cursor(self.uri, None) for i in xrange(10): curs1.set_key(str(i) + "TEST_KEY") curs1.set_value("TEST_VAL") curs1.insert() curs1.close() curs1 = self.session.open_cursor(self.uri, None, 'checkpoint=WiredTigerCheckpoint') for i in xrange(10): curs1.set_key(str(i) + "TEST_KEY") self.assertEqual(curs1.search(), wiredtiger.WT_NOTFOUND) curs1.close() # Pin the original checkpoint sess2 = self.conn.open_session() curs2 = sess2.open_cursor(self.uri, None, 'checkpoint=WiredTigerCheckpoint') # Take a new checkpoint, make sure the data is there. self.session.checkpoint('') curs1 = self.session.open_cursor(self.uri, None, 'checkpoint=WiredTigerCheckpoint') for i in xrange(10): curs1.set_key(str(i) + "TEST_KEY") self.assertEqual(curs1.search(), 0) curs1.close() # Release the pinned cursor / session curs2.close() sess2.close() # Add more data curs3 = self.session.open_cursor(self.uri, None) for i in xrange(10, 20): curs3.set_key(str(i) + "TEST_KEY") curs3.set_value("TEST_VAL") curs3.insert() curs3.close() self.session.checkpoint('') if __name__ == '__main__': wttest.run()