# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest
from wtdataset import ComplexDataSet

# test_bug012.py
class test_bug012(wttest.WiredTigerTestCase):


            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest
from wtdataset import ComplexDataSet

# test_bug012.py
class test_bug012(wttest.WiredTigerTestCase):


            

Reported by Pylint.

              from wtdataset import ComplexDataSet

# test_bug012.py
class test_bug012(wttest.WiredTigerTestCase):

    # Test that we detect illegal collators.
    def test_illegal_collator(self):
        msg = '/unknown collator/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError, lambda:

            

Reported by Pylint.

              from wtdataset import ComplexDataSet

# test_bug012.py
class test_bug012(wttest.WiredTigerTestCase):

    # Test that we detect illegal collators.
    def test_illegal_collator(self):
        msg = '/unknown collator/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError, lambda:

            

Reported by Pylint.

              class test_bug012(wttest.WiredTigerTestCase):

    # Test that we detect illegal collators.
    def test_illegal_collator(self):
        msg = '/unknown collator/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError, lambda:
            self.session.create('table:A', 'collator="xyzzy"'), msg)

    # Test we detect illegal key and value formats. Key and value formats are

            

Reported by Pylint.

                  # created.)
    #
    # Test that we detect illegal key formats.
    def test_illegal_key_format(self):
        msg = '/Invalid type/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.create('table:A', 'key_format="xyzzy"'), msg)
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.create('table:A',

            

Reported by Pylint.

                              'type=lsm,lsm=(bloom_config=(key_format="xyzzy"))'), msg)

    # Test that we detect illegal value formats.
    def test_illegal_value_format(self):
        msg = '/Invalid type/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.create('table:A', 'value_format="xyzzy"'), msg)
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.create('table:A',

            

Reported by Pylint.

                              'type=lsm,lsm=(bloom_config=(value_format="xyzzy"))'), msg)

    # Test that we detect illegal compressors.
    def test_illegal_compressor(self):
        msg = '/unknown compressor/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError, lambda:
            self.session.create('table:A', 'block_compressor="xyzzy"'), msg)

    # Test that we detect illegal extractors.

            

Reported by Pylint.

                  #
    # This test is a little fragile, we're depending on ComplexDataSet to do
    # the heavy-lifting, so if that function changes, this could break.
    def test_illegal_extractor(self):
        ds = ComplexDataSet(self, 'table:A', 10)
        ds.populate()
        msg = '/unknown extractor/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError, lambda:
            self.session.create('index:A:xyzzy',

            

Reported by Pylint.

              #      This is an example demonstrating how to query database statistics.

import os
from wiredtiger import wiredtiger_open,WIREDTIGER_VERSION_STRING,stat

def main():
    # Create a clean test directory for this run of the test program
    os.system('rm -rf WT_HOME')
    os.makedirs('WT_HOME')

            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              import os
from wiredtiger import wiredtiger_open,WIREDTIGER_VERSION_STRING,stat

def main():
    # Create a clean test directory for this run of the test program
    os.system('rm -rf WT_HOME')
    os.makedirs('WT_HOME')
    # Connect to the database and open a session
    conn = wiredtiger_open('WT_HOME', 'create,statistics=(all)')

            

Reported by Pylint.

              
def main():
    # Create a clean test directory for this run of the test program
    os.system('rm -rf WT_HOME')
    os.makedirs('WT_HOME')
    # Connect to the database and open a session
    conn = wiredtiger_open('WT_HOME', 'create,statistics=(all)')
    session = conn.open_session()


            

Reported by Bandit.

              
def main():
    # Create a clean test directory for this run of the test program
    os.system('rm -rf WT_HOME')
    os.makedirs('WT_HOME')
    # Connect to the database and open a session
    conn = wiredtiger_open('WT_HOME', 'create,statistics=(all)')
    session = conn.open_session()


            

Reported by Bandit.

                  print_derived_stats(session)
    conn.close()

def print_database_stats(session):
    statcursor = session.open_cursor("statistics:")
    print_cursor(statcursor)
    statcursor.close()

def print_file_stats(session):

            

Reported by Pylint.

                  print_cursor(statcursor)
    statcursor.close()

def print_file_stats(session):
    fstatcursor = session.open_cursor("statistics:table:access")
    print_cursor(fstatcursor)
    fstatcursor.close()

def print_overflow_pages(session):

            

Reported by Pylint.

                  print_cursor(fstatcursor)
    fstatcursor.close()

def print_overflow_pages(session):
    ostatcursor = session.open_cursor("statistics:table:access")
    val = ostatcursor[stat.dsrc.btree_overflow]
    if val != 0:
        print('%s=%s' % (str(val[0]), str(val[1])))
    ostatcursor.close()

            

Reported by Pylint.

                      print('%s=%s' % (str(val[0]), str(val[1])))
    ostatcursor.close()

def print_derived_stats(session):
    dstatcursor = session.open_cursor("statistics:table:access")
    ckpt_size = dstatcursor[stat.dsrc.block_checkpoint_size][1]
    file_size = dstatcursor[stat.dsrc.block_size][1]
    percent = 0
    if file_size != 0:

            

Reported by Pylint.

                  fs_writes = int(dstatcursor[stat.dsrc.cache_bytes_write][1])

    if app_insert + app_remove + app_update != 0:
        print("Write amplification is " + '{:.2f}'.format(fs_writes / (app_insert + app_remove + app_update)))
    dstatcursor.close()

def print_cursor(mycursor):
    while mycursor.next() == 0:
        val = mycursor.get_value()

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wtscenario, wttest
from wtdataset import SimpleDataSet

# test_lsm01.py
#    Test LSM tree configuration options.
class test_lsm01(wttest.WiredTigerTestCase):

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wtscenario, wttest
from wtdataset import SimpleDataSet

# test_lsm01.py
#    Test LSM tree configuration options.
class test_lsm01(wttest.WiredTigerTestCase):

            

Reported by Pylint.

                          'Test LSM with config: ' + args + ' count: ' + str(self.nrecs))
        SimpleDataSet(self, self.uri, self.nrecs, config=args).populate()

        # TODO: Adding an explicit drop here can cause deadlocks, if a merge
        # is still happening. See issue #349.
        # self.session.drop(self.uri)

if __name__ == '__main__':
    wttest.run()

            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wtscenario, wttest
from wtdataset import SimpleDataSet

# test_lsm01.py
#    Test LSM tree configuration options.
class test_lsm01(wttest.WiredTigerTestCase):

            

Reported by Pylint.

              
# test_lsm01.py
#    Test LSM tree configuration options.
class test_lsm01(wttest.WiredTigerTestCase):
    K = 1024
    M = 1024 * K
    G = 1024 * M
    uri = "lsm:test_lsm01"


            

Reported by Pylint.

              
# test_lsm01.py
#    Test LSM tree configuration options.
class test_lsm01(wttest.WiredTigerTestCase):
    K = 1024
    M = 1024 * K
    G = 1024 * M
    uri = "lsm:test_lsm01"


            

Reported by Pylint.

                      prune=100, prunelong=500)

    # Test drop of an object.
    def test_lsm(self):
        args = 'key_format=S'
        args += ',lsm=(' # Start the LSM configuration options.
        # add names to args, e.g. args += ',session_max=30'
        for var in self.config_vars:
            value = getattr(self, 's_' + var)

            

Reported by Pylint.

                      # add names to args, e.g. args += ',session_max=30'
        for var in self.config_vars:
            value = getattr(self, 's_' + var)
            if value != None:
                if var == 'verbose':
                    value = '[' + str(value) + ']'
                if value == True:
                    value = 'true'
                if value == False:

            

Reported by Pylint.

                          if value != None:
                if var == 'verbose':
                    value = '[' + str(value) + ']'
                if value == True:
                    value = 'true'
                if value == False:
                    value = 'false'
                args += ',' + var + '=' + str(value)
        args += ')' # Close the LSM configuration option group

            

Reported by Pylint.

              from testtools import TestCase
from testtools.matchers import (
    Equals,
    NotEquals,
    Not,
    )
from testtools.matchers._dict import (
    ContainedByDict,
    ContainsDict,

            

Reported by Pylint.

              from testtools.tests.matchers.helpers import TestMatchersInterface


class TestMatchesAllDictInterface(TestCase, TestMatchersInterface):

    matches_matcher = MatchesAllDict({'a': NotEquals(1), 'b': NotEquals(2)})
    matches_matches = [3, 4]
    matches_mismatches = [1, 2]


            

Reported by Pylint.

                      ]


class TestKeysEqualWithList(TestCase, TestMatchersInterface):

    matches_matcher = KeysEqual('foo', 'bar')
    matches_matches = [
        {'foo': 0, 'bar': 1},
        ]

            

Reported by Pylint.

              
    describe_examples = []

    def test_description(self):
        matchee = {'foo': 0, 'bar': 1, 'baz': 2}
        mismatch = KeysEqual('foo', 'bar').match(matchee)
        description = mismatch.describe()
        self.assertThat(
            description, Equals(

            

Reported by Pylint.

                              % (matchee,)))


class TestKeysEqualWithDict(TestKeysEqualWithList):

    matches_matcher = KeysEqual({'foo': 3, 'bar': 4})


class TestSubDictOf(TestCase, TestMatchersInterface):

            

Reported by Pylint.

                  matches_matcher = KeysEqual({'foo': 3, 'bar': 4})


class TestSubDictOf(TestCase, TestMatchersInterface):

    matches_matcher = _SubDictOf({'foo': 'bar', 'baz': 'qux'})

    matches_matches = [
        {'foo': 'bar', 'baz': 'qux'},

            

Reported by Pylint.

                  describe_examples = []


class TestMatchesDict(TestCase, TestMatchersInterface):

    matches_matcher = MatchesDict(
        {'foo': Equals('bar'), 'baz': Not(Equals('qux'))})

    matches_matches = [

            

Reported by Pylint.

                      ]


class TestContainsDict(TestCase, TestMatchersInterface):

    matches_matcher = ContainsDict(
        {'foo': Equals('bar'), 'baz': Not(Equals('qux'))})

    matches_matches = [

            

Reported by Pylint.

                      ]


class TestContainedByDict(TestCase, TestMatchersInterface):

    matches_matcher = ContainedByDict(
        {'foo': Equals('bar'), 'baz': Not(Equals('qux'))})

    matches_matches = [

            

Reported by Pylint.

                      ]


def test_suite():
    from unittest import TestLoader
    return TestLoader().loadTestsFromName(__name__)

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest

# test_bug013.py
#    Test data consistency in LSM with updates. Ensure that overwrite
#    cursors see all entries in the tree (i.e: they open cursors on all
#    chunks in the LSM tree).

            

Reported by Pylint.

                      # Test by iterating.
        cursor = self.session.open_cursor(self.uri, None, None)
        i = 0
        for i1, i2, i3, v1 in cursor:
            self.assertEqual( keys[i], [i1, i2, i3])
            i += 1
        cursor.close()
        self.assertEqual(i, len(keys))


            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest

# test_bug013.py
#    Test data consistency in LSM with updates. Ensure that overwrite
#    cursors see all entries in the tree (i.e: they open cursors on all
#    chunks in the LSM tree).

            

Reported by Pylint.

              #    cursors see all entries in the tree (i.e: they open cursors on all
#    chunks in the LSM tree).
#    See JIRA BF-829
class test_bug013(wttest.WiredTigerTestCase):
    """
    Test LSM data consistency.
    """
    uri = 'table:test_bug013'


            

Reported by Pylint.

                  """
    uri = 'table:test_bug013'

    def check_entries(self, keys):
        # Test by iterating.
        cursor = self.session.open_cursor(self.uri, None, None)
        i = 0
        for i1, i2, i3, v1 in cursor:
            self.assertEqual( keys[i], [i1, i2, i3])

            

Reported by Pylint.

                      # Test by iterating.
        cursor = self.session.open_cursor(self.uri, None, None)
        i = 0
        for i1, i2, i3, v1 in cursor:
            self.assertEqual( keys[i], [i1, i2, i3])
            i += 1
        cursor.close()
        self.assertEqual(i, len(keys))


            

Reported by Pylint.

                      # Test by iterating.
        cursor = self.session.open_cursor(self.uri, None, None)
        i = 0
        for i1, i2, i3, v1 in cursor:
            self.assertEqual( keys[i], [i1, i2, i3])
            i += 1
        cursor.close()
        self.assertEqual(i, len(keys))


            

Reported by Pylint.

                      # Test by iterating.
        cursor = self.session.open_cursor(self.uri, None, None)
        i = 0
        for i1, i2, i3, v1 in cursor:
            self.assertEqual( keys[i], [i1, i2, i3])
            i += 1
        cursor.close()
        self.assertEqual(i, len(keys))


            

Reported by Pylint.

                      # Test by iterating.
        cursor = self.session.open_cursor(self.uri, None, None)
        i = 0
        for i1, i2, i3, v1 in cursor:
            self.assertEqual( keys[i], [i1, i2, i3])
            i += 1
        cursor.close()
        self.assertEqual(i, len(keys))


            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest

# test_debug_mode01.py
#    Test the debug mode settings. Test rollback_error in this one.
class test_debug_mode01(wttest.WiredTigerTestCase):
    conn_config = 'log=(enabled=true),debug_mode=(rollback_error=5)'

            

Reported by Pylint.

                          # conflict exception is raised, false if no exception occurred.
            if insert:
                conflict = self.assertRaisesException(wiredtiger.WiredTigerError, \
                    lambda:c.insert(), msg, True)
            else:
                conflict = self.assertRaisesException(wiredtiger.WiredTigerError, \
                    lambda:c.update(), msg, True)

            if conflict:

            

Reported by Pylint.

                                  lambda:c.insert(), msg, True)
            else:
                conflict = self.assertRaisesException(wiredtiger.WiredTigerError, \
                    lambda:c.update(), msg, True)

            if conflict:
                rollback += 1
                self.pr("Key: " + str(k) + " Rolled back")
                self.session.rollback_transaction()

            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest

# test_debug_mode01.py
#    Test the debug mode settings. Test rollback_error in this one.
class test_debug_mode01(wttest.WiredTigerTestCase):
    conn_config = 'log=(enabled=true),debug_mode=(rollback_error=5)'

            

Reported by Pylint.

              
# test_debug_mode01.py
#    Test the debug mode settings. Test rollback_error in this one.
class test_debug_mode01(wttest.WiredTigerTestCase):
    conn_config = 'log=(enabled=true),debug_mode=(rollback_error=5)'
    session_config = 'isolation=snapshot'
    uri = 'file:test_debug'

    entries = 22

            

Reported by Pylint.

              
# test_debug_mode01.py
#    Test the debug mode settings. Test rollback_error in this one.
class test_debug_mode01(wttest.WiredTigerTestCase):
    conn_config = 'log=(enabled=true),debug_mode=(rollback_error=5)'
    session_config = 'isolation=snapshot'
    uri = 'file:test_debug'

    entries = 22

            

Reported by Pylint.

                  entries = 22
    min_error = entries // 5

    def rollback_error(self, val, insert=True):
        keys = range(1, self.entries)
        c = self.session.open_cursor(self.uri, None)
        # We expect some operations to return an exception so we cannot
        # use the simple 'c[k] = 1'. But we must explicitly set the key
        # and value and then use the insert or update primitives.

            

Reported by Pylint.

              
    def rollback_error(self, val, insert=True):
        keys = range(1, self.entries)
        c = self.session.open_cursor(self.uri, None)
        # We expect some operations to return an exception so we cannot
        # use the simple 'c[k] = 1'. But we must explicitly set the key
        # and value and then use the insert or update primitives.
        #
        # Look for a generic 'WT_ROLLBACK' string not the specific

            

Reported by Pylint.

                      c.close()
        return rollback

    def test_rollback_error(self):
        self.session.create(self.uri, 'key_format=i,value_format=i')
        rollback = self.rollback_error(1)
        rollback += self.rollback_error(2, False)
        self.pr("Rollback: " + str(rollback))
        self.pr("Minimum: " + str(self.min_error))

            

Reported by Pylint.

              #   continue to generate more log files.
#

import fnmatch, os, shutil, time
from suite_subprocess import suite_subprocess
import wttest

class test_txn16(wttest.WiredTigerTestCase, suite_subprocess):
    t1 = 'table:test_txn16_1'

            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              #   continue to generate more log files.
#

import fnmatch, os, shutil, time
from suite_subprocess import suite_subprocess
import wttest

class test_txn16(wttest.WiredTigerTestCase, suite_subprocess):
    t1 = 'table:test_txn16_1'

            

Reported by Pylint.

              from suite_subprocess import suite_subprocess
import wttest

class test_txn16(wttest.WiredTigerTestCase, suite_subprocess):
    t1 = 'table:test_txn16_1'
    t2 = 'table:test_txn16_2'
    t3 = 'table:test_txn16_3'
    nentries = 1000
    create_params = 'key_format=i,value_format=i'

            

Reported by Pylint.

              from suite_subprocess import suite_subprocess
import wttest

class test_txn16(wttest.WiredTigerTestCase, suite_subprocess):
    t1 = 'table:test_txn16_1'
    t2 = 'table:test_txn16_2'
    t3 = 'table:test_txn16_3'
    nentries = 1000
    create_params = 'key_format=i,value_format=i'

            

Reported by Pylint.

                      'transaction_sync=(method=dsync,enabled)'
    conn_off = 'config_base=false,log=(enabled=false)'

    def populate_table(self, uri):
        self.session.create(uri, self.create_params)
        c = self.session.open_cursor(uri, None, None)
        # Populate with an occasional checkpoint to generate
        # some varying LSNs.
        for i in range(self.nentries):

            

Reported by Pylint.

              
    def populate_table(self, uri):
        self.session.create(uri, self.create_params)
        c = self.session.open_cursor(uri, None, None)
        # Populate with an occasional checkpoint to generate
        # some varying LSNs.
        for i in range(self.nentries):
            c[i] = i + 1
            if i % 900 == 0:

            

Reported by Pylint.

                      # close the original connection.
        self.close_conn()

    def run_toggle(self, homedir):
        loop = 0
        # Record original log files.  There should never be overlap
        # with these even after they're removed.
        orig_logs = fnmatch.filter(os.listdir(homedir), "*gerLog*")
        while loop < 3:

            

Reported by Pylint.

                              self.assertEqual(scur.isdisjoint(sorig), True)
                if loop > 1:
                    # We should be creating the same log files each time.
                    for l in cur_logs:
                        self.assertEqual(l in last_logs, True)
                    for l in last_logs:
                        self.assertEqual(l in cur_logs, True)
                last_logs = cur_logs
            loop += 1

            

Reported by Pylint.

                                  # We should be creating the same log files each time.
                    for l in cur_logs:
                        self.assertEqual(l in last_logs, True)
                    for l in last_logs:
                        self.assertEqual(l in cur_logs, True)
                last_logs = cur_logs
            loop += 1
            # Remove all log files before opening without logging.
            cur_logs = fnmatch.filter(os.listdir(homedir), "*gerLog*")

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest

# test_base04.py
#    Test that tables are reconciled correctly when they are empty.
class test_base04(wttest.WiredTigerTestCase):
    '''Test various tree types becoming empty'''

            

Reported by Pylint.

              
    def test_insert_delete(self):
        '''Create a table, add a key, get it back'''
        for reconcile in (False, True):
            self.create_table()
            self.insert('key1', 'value1')
            self.check_exists('key1', 0)
            self.remove('key1')
            self.check_exists('key1', wiredtiger.WT_NOTFOUND)

            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import wiredtiger, wttest

# test_base04.py
#    Test that tables are reconciled correctly when they are empty.
class test_base04(wttest.WiredTigerTestCase):
    '''Test various tree types becoming empty'''

            

Reported by Pylint.

              
# test_base04.py
#    Test that tables are reconciled correctly when they are empty.
class test_base04(wttest.WiredTigerTestCase):
    '''Test various tree types becoming empty'''

    tablename = 'table:test_base04'

    def __init__(self, *args, **kwargs):

            

Reported by Pylint.

                      wttest.WiredTigerTestCase.__init__(self, *args, **kwargs)
        self.reconcile = False

    def create_table(self):
        self.pr('create table')
        self.session.create(
            self.tablename, 'key_format=S,value_format=S')

    def drop_table(self):

            

Reported by Pylint.

                      self.session.create(
            self.tablename, 'key_format=S,value_format=S')

    def drop_table(self):
        self.pr('drop table')
        self.session.drop(self.tablename, None)

    def cursor(self):
        self.pr('open cursor')

            

Reported by Pylint.

                      self.pr('drop table')
        self.session.drop(self.tablename, None)

    def cursor(self):
        self.pr('open cursor')
        return self.session.open_cursor(self.tablename, None, None)

    def check_exists(self, key, expected):
        cursor = self.cursor()

            

Reported by Pylint.

                      self.pr('open cursor')
        return self.session.open_cursor(self.tablename, None, None)

    def check_exists(self, key, expected):
        cursor = self.cursor()
        cursor.set_key(key)
        self.pr('search')
        self.assertEqual(cursor.search(), expected)
        self.pr('closing cursor')

            

Reported by Pylint.

                      self.pr('closing cursor')
        cursor.close()

    def insert(self, key, value):
        self.pr('insert')
        cursor = self.cursor()
        cursor[key] = value
        cursor.close()
        if self.reconcile:

            

Reported by Pylint.

                  std::streampos seek( stream_offset off, BOOST_IOS::seekdir way, 
                         BOOST_IOS::openmode which = 
                             BOOST_IOS::in | BOOST_IOS::out );
    void open( const std::string& path,
               BOOST_IOS::openmode mode =
                   BOOST_IOS::in | BOOST_IOS::out,
               BOOST_IOS::openmode base_mode =
                   BOOST_IOS::in | BOOST_IOS::out );
    bool is_open() const;

            

Reported by FlawFinder.

              private:
    struct impl {
        impl(const std::string& path, BOOST_IOS::openmode mode)
            { file_.open(path.c_str(), mode); }
        ~impl() { if (file_.is_open()) file_.close(); }
        BOOST_IOSTREAMS_BASIC_FILEBUF(Ch) file_;
    };
    shared_ptr<impl> pimpl_;
};

            

Reported by FlawFinder.

                                         BOOST_IOS::in )
        : basic_file<Ch>(path, mode & ~BOOST_IOS::out, BOOST_IOS::in)
        { }
    void open( const std::string& path,
               BOOST_IOS::openmode mode = BOOST_IOS::in )
    {
        basic_file<Ch>::open(path, mode & ~BOOST_IOS::out, BOOST_IOS::in);
    }
};

            

Reported by FlawFinder.

                  void open( const std::string& path,
               BOOST_IOS::openmode mode = BOOST_IOS::in )
    {
        basic_file<Ch>::open(path, mode & ~BOOST_IOS::out, BOOST_IOS::in);
    }
};

typedef basic_file_source<char>     file_source;
typedef basic_file_source<wchar_t>  wfile_source;

            

Reported by FlawFinder.

                                   BOOST_IOS::openmode mode = BOOST_IOS::out )
        : basic_file<Ch>(path, mode & ~BOOST_IOS::in, BOOST_IOS::out)
        { }
    void open( const std::string& path,
               BOOST_IOS::openmode mode = BOOST_IOS::out )
    {
        basic_file<Ch>::open(path, mode & ~BOOST_IOS::in, BOOST_IOS::out);
    }
};

            

Reported by FlawFinder.

                  void open( const std::string& path,
               BOOST_IOS::openmode mode = BOOST_IOS::out )
    {
        basic_file<Ch>::open(path, mode & ~BOOST_IOS::in, BOOST_IOS::out);
    }
};

typedef basic_file_sink<char>     file_sink;
typedef basic_file_sink<wchar_t>  wfile_sink;

            

Reported by FlawFinder.

                  ( const std::string& path, BOOST_IOS::openmode mode, 
      BOOST_IOS::openmode base_mode )
{ 
    open(path, mode, base_mode);
}

template<typename Ch>
inline std::streamsize basic_file<Ch>::read
    (char_type* s, std::streamsize n)

            

Reported by FlawFinder.

              { return iostreams::seek(pimpl_->file_, off, way); }

template<typename Ch>
void basic_file<Ch>::open
    ( const std::string& path, BOOST_IOS::openmode mode, 
      BOOST_IOS::openmode base_mode )
{ 
    pimpl_.reset(new impl(path, mode | base_mode));
}

            

Reported by FlawFinder.

                                  BOOST_IOS::in | BOOST_IOS::out,
                BOOST_IOS::openmode base_mode =
                    BOOST_IOS::in | BOOST_IOS::out );
    std::streamsize read(char_type* s, std::streamsize n);
    bool putback(char_type c);
    std::streamsize write(const char_type* s, std::streamsize n);
    std::streampos seek( stream_offset off, BOOST_IOS::seekdir way, 
                         BOOST_IOS::openmode which = 
                             BOOST_IOS::in | BOOST_IOS::out );

            

Reported by FlawFinder.

                        device_tag,
          closable_tag
        { };
    using basic_file<Ch>::read;
    using basic_file<Ch>::putback;
    using basic_file<Ch>::seek;
    using basic_file<Ch>::is_open;
    using basic_file<Ch>::close;
    basic_file_source( const std::string& path,

            

Reported by FlawFinder.

              #

from suite_subprocess import suite_subprocess
import wiredtiger, wttest

class test_txn17(wttest.WiredTigerTestCase, suite_subprocess):
    def test_txn_api(self):
        # Test API functionality tagged as requires_transaction.
        # Cannot set a timestamp on a non-running transaction.

            

Reported by Pylint.

              
        # Cannot call commit on a non-running transaction.
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.commit_transaction(),
                '/only permitted in a running transaction/')

        # Cannot call rollback on a non-running transaction.
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.rollback_transaction(),

            

Reported by Pylint.

              
        # Cannot call rollback on a non-running transaction.
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.rollback_transaction(),
                '/only permitted in a running transaction/')

        # Cannot begin a transaction while a transaction is already running.
        self.session.begin_transaction()
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,

            

Reported by Pylint.

                      # Cannot begin a transaction while a transaction is already running.
        self.session.begin_transaction()
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.begin_transaction(),
                '/not permitted in a running transaction/')
        self.session.rollback_transaction()

        # Cannot take a checkpoint while a transaction is running.
        self.session.begin_transaction()

            

Reported by Pylint.

                      # Cannot take a checkpoint while a transaction is running.
        self.session.begin_transaction()
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.checkpoint(),
                '/not permitted in a running transaction/')
        self.session.rollback_transaction()

if __name__ == '__main__':
    wttest.run()

            

Reported by Pylint.

              #!/usr/bin/env python
#
# Public Domain 2014-present MongoDB, Inc.
# Public Domain 2008-2014 WiredTiger, Inc.
#
# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled

            

Reported by Pylint.

              #

from suite_subprocess import suite_subprocess
import wiredtiger, wttest

class test_txn17(wttest.WiredTigerTestCase, suite_subprocess):
    def test_txn_api(self):
        # Test API functionality tagged as requires_transaction.
        # Cannot set a timestamp on a non-running transaction.

            

Reported by Pylint.

              #

from suite_subprocess import suite_subprocess
import wiredtiger, wttest

class test_txn17(wttest.WiredTigerTestCase, suite_subprocess):
    def test_txn_api(self):
        # Test API functionality tagged as requires_transaction.
        # Cannot set a timestamp on a non-running transaction.

            

Reported by Pylint.

              from suite_subprocess import suite_subprocess
import wiredtiger, wttest

class test_txn17(wttest.WiredTigerTestCase, suite_subprocess):
    def test_txn_api(self):
        # Test API functionality tagged as requires_transaction.
        # Cannot set a timestamp on a non-running transaction.
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.timestamp_transaction(

            

Reported by Pylint.

              from suite_subprocess import suite_subprocess
import wiredtiger, wttest

class test_txn17(wttest.WiredTigerTestCase, suite_subprocess):
    def test_txn_api(self):
        # Test API functionality tagged as requires_transaction.
        # Cannot set a timestamp on a non-running transaction.
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.session.timestamp_transaction(

            

Reported by Pylint.