import sys
import uuid

import gdb.printing

try:
    import bson
    import bson.json_util
    import collections

            

Reported by Pylint.

                      self.val = val

        decl_vector = val["_decorations"]["_registry"]["_decorationInfo"]
        # TODO: abstract out navigating a std::vector
        self.start = decl_vector["_M_impl"]["_M_start"]
        finish = decl_vector["_M_impl"]["_M_finish"]
        decorable_t = val.type.template_argument(0)
        decinfo_t = gdb.lookup_type('mongo::DecorationRegistry<{}>::DecorationInfo'.format(
            str(decorable_t).replace("class", "").strip()))

            

Reported by Pylint.

              
            # In order to get the type stored in the decorable, we examine the type of its
            # constructor, and do some string manipulations.
            # TODO: abstract out navigating a std::function
            type_name = str(descriptor["constructor"])
            type_name = type_name[0:len(type_name) - 1]
            type_name = type_name[0:type_name.rindex(">")]
            type_name = type_name[type_name.index("constructAt<"):].replace("constructAt<", "")


            

Reported by Pylint.

              ###################################################################################################


class StatusPrinter(object):
    """Pretty-printer for mongo::Status."""

    def __init__(self, val):
        """Initialize StatusPrinter."""
        self.val = val

            

Reported by Pylint.

              ###################################################################################################


class StatusPrinter(object):
    """Pretty-printer for mongo::Status."""

    def __init__(self, val):
        """Initialize StatusPrinter."""
        self.val = val

            

Reported by Pylint.

                      return 'Status(%s, %s)' % (code, reason)


class StatusWithPrinter(object):
    """Pretty-printer for mongo::StatusWith<>."""

    def __init__(self, val):
        """Initialize StatusWithPrinter."""
        self.val = val

            

Reported by Pylint.

                      return 'Status(%s, %s)' % (code, reason)


class StatusWithPrinter(object):
    """Pretty-printer for mongo::StatusWith<>."""

    def __init__(self, val):
        """Initialize StatusWithPrinter."""
        self.val = val

            

Reported by Pylint.

                      return 'StatusWith(%s, %s)' % (code, reason)


class StringDataPrinter(object):
    """Pretty-printer for mongo::StringData."""

    def __init__(self, val):
        """Initialize StringDataPrinter."""
        self.val = val

            

Reported by Pylint.

                      return self.val['_data'].lazy_string(length=size)


class BSONObjPrinter(object):
    """Pretty-printer for mongo::BSONObj."""

    def __init__(self, val):
        """Initialize BSONObjPrinter."""
        self.val = val

            

Reported by Pylint.

                      return "%s BSONObj %s bytes @ %s%s" % (ownership, size, self.ptr, suffix)


class OplogEntryPrinter(object):
    """Pretty-printer for mongo::repl::OplogEntry."""

    def __init__(self, val):
        """Initialize OplogEntryPrinter."""
        self.val = val

            

Reported by Pylint.

              
MEM_STATIC void MEM_write16(void* memPtr, U16 value)
{
    memcpy(memPtr, &value, sizeof(value));
}

#endif /* MEM_FORCE_MEMORY_ACCESS */

MEM_STATIC U32 MEM_swap32(U32 in)

            

Reported by FlawFinder.

                                  if (normalizedCounter[s] >= largeLimit) DTableH.fastMode=0;
                    symbolNext[s] = normalizedCounter[s];
        }   }   }
        memcpy(dt, &DTableH, sizeof(DTableH));
    }

    /* Spread symbols */
    {   U32 const tableMask = tableSize-1;
        U32 const step = FSEv07_TABLESTEP(tableSize);

            

Reported by FlawFinder.

                      if (tableLog > (U32)(dtd.maxTableLog+1)) return ERROR(tableLog_tooLarge);   /* DTable too small, huffman tree cannot fit in */
        dtd.tableType = 0;
        dtd.tableLog = (BYTE)tableLog;
        memcpy(DTable, &dtd, sizeof(dtd));
    }

    /* Prepare ranks */
    {   U32 n, nextRankStart = 0;
        for (n=1; n<tableLog+1; n++) {

            

Reported by FlawFinder.

              
    dtd.tableLog = (BYTE)maxTableLog;
    dtd.tableType = 1;
    memcpy(DTable, &dtd, sizeof(dtd));
    return iSize;
}


static U32 HUFv07_decodeSymbolX4(void* op, BITv07_DStream_t* DStream, const HUFv07_DEltX4* dt, const U32 dtLog)

            

Reported by FlawFinder.

              static U32 HUFv07_decodeSymbolX4(void* op, BITv07_DStream_t* DStream, const HUFv07_DEltX4* dt, const U32 dtLog)
{
    const size_t val = BITv07_lookBitsFast(DStream, dtLog);   /* note : dtLog >= 1 */
    memcpy(op, dt+val, 2);
    BITv07_skipBits(DStream, dt[val].nbBits);
    return dt[val].length;
}

static U32 HUFv07_decodeLastSymbolX4(void* op, BITv07_DStream_t* DStream, const HUFv07_DEltX4* dt, const U32 dtLog)

            

Reported by FlawFinder.

              static U32 HUFv07_decodeLastSymbolX4(void* op, BITv07_DStream_t* DStream, const HUFv07_DEltX4* dt, const U32 dtLog)
{
    const size_t val = BITv07_lookBitsFast(DStream, dtLog);   /* note : dtLog >= 1 */
    memcpy(op, dt+val, 1);
    if (dt[val].length==1) BITv07_skipBits(DStream, dt[val].nbBits);
    else {
        if (DStream->bitsConsumed < (sizeof(DStream->bitContainer)*8)) {
            BITv07_skipBits(DStream, dt[val].nbBits);
            if (DStream->bitsConsumed > (sizeof(DStream->bitContainer)*8))

            

Reported by FlawFinder.

                  /* validation checks */
    if (dstSize == 0) return ERROR(dstSize_tooSmall);
    if (cSrcSize > dstSize) return ERROR(corruption_detected);   /* invalid */
    if (cSrcSize == dstSize) { memcpy(dst, cSrc, dstSize); return dstSize; }   /* not compressed */
    if (cSrcSize == 1) { memset(dst, *(const BYTE*)cSrc, dstSize); return dstSize; }   /* RLE */

    {   U32 const algoNb = HUFv07_selectDecoder(dstSize, cSrcSize);
        return decompress[algoNb](dst, dstSize, cSrc, cSrcSize);
    }

            

Reported by FlawFinder.

                  /* validation checks */
    if (dstSize == 0) return ERROR(dstSize_tooSmall);
    if (cSrcSize > dstSize) return ERROR(corruption_detected);   /* invalid */
    if (cSrcSize == dstSize) { memcpy(dst, cSrc, dstSize); return dstSize; }   /* not compressed */
    if (cSrcSize == 1) { memset(dst, *(const BYTE*)cSrc, dstSize); return dstSize; }   /* RLE */

    {   U32 const algoNb = HUFv07_selectDecoder(dstSize, cSrcSize);
        return algoNb ? HUFv07_decompress4X4_DCtx(dctx, dst, dstSize, cSrc, cSrcSize) :
                        HUFv07_decompress4X2_DCtx(dctx, dst, dstSize, cSrc, cSrcSize) ;

            

Reported by FlawFinder.

                  /* validation checks */
    if (dstSize == 0) return ERROR(dstSize_tooSmall);
    if (cSrcSize > dstSize) return ERROR(corruption_detected);   /* invalid */
    if (cSrcSize == dstSize) { memcpy(dst, cSrc, dstSize); return dstSize; }   /* not compressed */
    if (cSrcSize == 1) { memset(dst, *(const BYTE*)cSrc, dstSize); return dstSize; }   /* RLE */

    {   U32 const algoNb = HUFv07_selectDecoder(dstSize, cSrcSize);
        return algoNb ? HUFv07_decompress1X4_DCtx(dctx, dst, dstSize, cSrc, cSrcSize) :
                        HUFv07_decompress1X2_DCtx(dctx, dst, dstSize, cSrc, cSrcSize) ;

            

Reported by FlawFinder.

              /*-*******************************************
*  Shared functions to include for inlining
*********************************************/
static void ZSTDv07_copy8(void* dst, const void* src) { memcpy(dst, src, 8); }
#define COPY8(d,s) { ZSTDv07_copy8(d,s); d+=8; s+=8; }

/*! ZSTDv07_wildcopy() :
*   custom version of memcpy(), can copy up to 7 bytes too many (8 bytes if length==0) */
#define WILDCOPY_OVERLENGTH 8

            

Reported by FlawFinder.

              # OTHER DEALINGS IN THE SOFTWARE.

import os
import wiredtiger, wttest, run
from wtscenario import make_scenarios

# test_join09.py
#    Join bloom filters with false positives
class test_join09(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 os
import wiredtiger, wttest, run
from wtscenario import make_scenarios

# test_join09.py
#    Join bloom filters with false positives

            

Reported by Pylint.

              # OTHER DEALINGS IN THE SOFTWARE.

import os
import wiredtiger, wttest, run
from wtscenario import make_scenarios

# test_join09.py
#    Join bloom filters with false positives
class test_join09(wttest.WiredTigerTestCase):

            

Reported by Pylint.

              # OTHER DEALINGS IN THE SOFTWARE.

import os
import wiredtiger, wttest, run
from wtscenario import make_scenarios

# test_join09.py
#    Join bloom filters with false positives
class test_join09(wttest.WiredTigerTestCase):

            

Reported by Pylint.

                      jc = self.session.open_cursor('join:table:join09', None, None)
        c0 = self.session.open_cursor('index:join09:index0', None, None)
        c0.set_key('520')
        self.assertEquals(0, c0.search())
        self.session.join(jc, c0, 'compare=ge')

        joinconfig = 'compare=eq,' + self.config
        c1 = self.session.open_cursor('index:join09:index1', None, None)
        c1.set_key('555')

            

Reported by Pylint.

                      joinconfig = 'compare=eq,' + self.config
        c1 = self.session.open_cursor('index:join09:index1', None, None)
        c1.set_key('555')
        self.assertEquals(0, c1.search())
        self.session.join(jc, c1, joinconfig)

        mbr = set(range(520,600)) | set(range(53,60))

        fp_count = 0

            

Reported by Pylint.

                      while jc.next() == 0:
            [k] = jc.get_keys()
            [v0,v1] = jc.get_values()
            self.assertEquals(self.gen_values(k), [v0, v1])
            if not k in mbr:
                # With false positives, we can see extra values
                if self.false_positives:
                    fp_count += 1
                    continue

            

Reported by Pylint.

              
        if len(mbr) != 0:
            self.tty('**** ERROR: did not see these: ' + str(mbr))
        self.assertEquals(0, len(mbr))

        # Turning on false positives does not guarantee we'll see extra
        # values, but we've configured our test with a low count to
        # make sure it happens.
        if self.false_positives:

            

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.

              # OTHER DEALINGS IN THE SOFTWARE.

import os
import wiredtiger, wttest, run
from wtscenario import make_scenarios

# test_join09.py
#    Join bloom filters with false positives
class test_join09(wttest.WiredTigerTestCase):

            

Reported by Pylint.

              #
# We also have some special logic to handle function description comments even
# though those don't conform to our definition of a "block comment".
import re, sys

# List of words in the current block comment.
words = []

# Whether we're inside a potential block comment.

            

Reported by Pylint.

              # Fill out block comments to the full line length (currently 100).
#
# We're defining a "block comment" to be a multiline comment where each line
# begins with an alphabetic character.
#
# We also have some special logic to handle function description comments even
# though those don't conform to our definition of a "block comment".
import re, sys


            

Reported by Pylint.

              #
# We also have some special logic to handle function description comments even
# though those don't conform to our definition of a "block comment".
import re, sys

# List of words in the current block comment.
words = []

# Whether we're inside a potential block comment.

            

Reported by Pylint.

              words = []

# Whether we're inside a potential block comment.
multiline = False

# The maximum allowed line length.
line_length = 100

# How far to indent the current block comment.

            

Reported by Pylint.

              multiline = False

# The maximum allowed line length.
line_length = 100

# How far to indent the current block comment.
indentation = 0

# Whether we're inside a function description comment. This is not a block

            

Reported by Pylint.

              line_length = 100

# How far to indent the current block comment.
indentation = 0

# Whether we're inside a function description comment. This is not a block
# comment by our definition but we want to fill these too.
function_desc = False


            

Reported by Pylint.

              
# Whether we're inside a function description comment. This is not a block
# comment by our definition but we want to fill these too.
function_desc = False

# Whether we've seen a line in the multiline comment to indicate that it is NOT
# a block comment. In that case don't use the refilling logic and just print the
# contents verbatim.
block = False

            

Reported by Pylint.

              # Whether we've seen a line in the multiline comment to indicate that it is NOT
# a block comment. In that case don't use the refilling logic and just print the
# contents verbatim.
block = False

# The literal contents of the current block comment. If we realise halfway
# through the comment that it's not a block comment then we'll just print this
# out and pretend none of this ever happened.
comment = str()

            

Reported by Pylint.

              # The literal contents of the current block comment. If we realise halfway
# through the comment that it's not a block comment then we'll just print this
# out and pretend none of this ever happened.
comment = str()

for line in sys.stdin:
    sline = line.strip()
    # Beginning of a block comment.
    if sline == '/*':

            

Reported by Pylint.

                  # Beginning of a block comment.
    if sline == '/*':
        comment = line
        assert not multiline
        multiline = True
        block = True
        # Figure out how far we need to indent.
        indentation = 0
        for c in line:

            

Reported by Bandit.

              # rollback_to_stable:out_of_order_timestamps
# [END_TAGS]

from wiredtiger import stat, WT_NOTFOUND
from wtscenario import make_scenarios
from helper import simulate_crash_restart
from wtdataset import SimpleDataSet
from test_rollback_to_stable01 import test_rollback_to_stable_base


            

Reported by Pylint.

              
    scenarios = make_scenarios(key_format_values)

    def conn_config(self):
        config = 'cache_size=250MB,statistics=(all),statistics_log=(json,on_close,wait=1)'
        return config

    def test_rollback_to_stable(self):
        nrows = 1000

            

Reported by Pylint.

              
        for i in range(1, nrows + 1):
            evict_cursor.set_key(i)
            self.assertEquals(evict_cursor.search(), 0)
            self.assertEqual(evict_cursor.get_value(), valuea)
            evict_cursor.reset()

        s.rollback_transaction()
        self.conn.set_timestamp('stable_timestamp=' + self.timestamp_str(40))

            

Reported by Pylint.

              
        for i in range(1, nrows + 1):
            evict_cursor.set_key(i)
            self.assertEquals(evict_cursor.search(), 0)
            self.assertEqual(evict_cursor.get_value(), valuea)
            evict_cursor.reset()

        s.rollback_transaction()
        self.conn.set_timestamp('stable_timestamp=' + self.timestamp_str(40))

            

Reported by Pylint.

              
        for i in range(1, nrows + 1):
            evict_cursor.set_key(i)
            self.assertEquals(evict_cursor.search(), WT_NOTFOUND)
            evict_cursor.reset()

        s.rollback_transaction()
        self.conn.set_timestamp('stable_timestamp=' + self.timestamp_str(40))
        s.checkpoint()

            

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.

              
# test_rollback_to_stable21.py
# Test rollback to stable when an out of order prepared transaction is written to disk
class test_rollback_to_stable21(test_rollback_to_stable_base):
    key_format_values = [
        ('column', dict(key_format='r')),
        ('integer_row', dict(key_format='i')),
    ]


            

Reported by Pylint.

              
# test_rollback_to_stable21.py
# Test rollback to stable when an out of order prepared transaction is written to disk
class test_rollback_to_stable21(test_rollback_to_stable_base):
    key_format_values = [
        ('column', dict(key_format='r')),
        ('integer_row', dict(key_format='i')),
    ]


            

Reported by Pylint.

              
    scenarios = make_scenarios(key_format_values)

    def conn_config(self):
        config = 'cache_size=250MB,statistics=(all),statistics_log=(json,on_close,wait=1)'
        return config

    def test_rollback_to_stable(self):
        nrows = 1000

            

Reported by Pylint.

              
    scenarios = make_scenarios(key_format_values)

    def conn_config(self):
        config = 'cache_size=250MB,statistics=(all),statistics_log=(json,on_close,wait=1)'
        return config

    def test_rollback_to_stable(self):
        nrows = 1000

            

Reported by Pylint.

              # [END_TAGS]
#

import os, threading, time, wiredtiger, wttest
from wiredtiger import stat
from wtthread import checkpoint_thread, flush_tier_thread

# test_tiered08.py
#   Run background checkpoints and flush_tier operations while inserting

            

Reported by Pylint.

              #

import os, threading, time, wiredtiger, wttest
from wiredtiger import stat
from wtthread import checkpoint_thread, flush_tier_thread

# test_tiered08.py
#   Run background checkpoints and flush_tier operations while inserting
#   data into a table from another thread.

            

Reported by Pylint.

                  bucket_prefix = "pfx_"
    extension_name = "local_store"

    def conn_config(self):
        if not os.path.exists(self.bucket):
            os.mkdir(self.bucket)
        return \
          'statistics=(fast),' + \
          'tiered_storage=(auth_token=%s,' % self.auth_token + \

            

Reported by Pylint.

              # [END_TAGS]
#

import os, threading, time, wiredtiger, wttest
from wiredtiger import stat
from wtthread import checkpoint_thread, flush_tier_thread

# test_tiered08.py
#   Run background checkpoints and flush_tier operations while inserting

            

Reported by Pylint.

                          extlist.skip_if_missing = True
        extlist.extension('storage_sources', self.extension_name)

    def get_stat(self, stat):
        stat_cursor = self.session.open_cursor('statistics:')
        val = stat_cursor[stat][2]
        stat_cursor.close()
        return val


            

Reported by Pylint.

              
    def test_tiered08(self):

        # FIXME-WT-7833
        #     This test can trigger races in file handle access during flush_tier.
        #     We will re-enable it when that is fixed.
        self.skipTest('Concurrent flush_tier and insert operations not supported yet.')

        cfg = self.conn_config()

            

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.

              # [END_TAGS]
#

import os, threading, time, wiredtiger, wttest
from wiredtiger import stat
from wtthread import checkpoint_thread, flush_tier_thread

# test_tiered08.py
#   Run background checkpoints and flush_tier operations while inserting

            

Reported by Pylint.

              # [END_TAGS]
#

import os, threading, time, wiredtiger, wttest
from wiredtiger import stat
from wtthread import checkpoint_thread, flush_tier_thread

# test_tiered08.py
#   Run background checkpoints and flush_tier operations while inserting

            

Reported by Pylint.

              #

import os, threading, time, wiredtiger, wttest
from wiredtiger import stat
from wtthread import checkpoint_thread, flush_tier_thread

# test_tiered08.py
#   Run background checkpoints and flush_tier operations while inserting
#   data into a table from another thread.

            

Reported by Pylint.

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

import os, wiredtiger, wttest
from wiredtiger import stat
StorageSource = wiredtiger.StorageSource  # easy access to constants

# test_tiered04.py
#    Basic tiered storage API test.

            

Reported by Pylint.

              # OTHER DEALINGS IN THE SOFTWARE.

import os, wiredtiger, wttest
from wiredtiger import stat
StorageSource = wiredtiger.StorageSource  # easy access to constants

# test_tiered04.py
#    Basic tiered storage API test.
class test_tiered04(wttest.WiredTigerTestCase):

            

Reported by Pylint.

                  object_uri_val = 15 * 1024 * 1024
    retention = 600
    retention1 = 350
    def conn_config(self):
        os.mkdir(self.bucket)
        os.mkdir(self.bucket1)
        return \
          'statistics=(all),' + \
          'tiered_storage=(auth_token=%s,' % self.auth_token + \

            

Reported by Pylint.

                      c.close()
        self.assertTrue(val_str in val)

    def get_stat(self, stat, uri):
        if uri == None:
            stat_cursor = self.session.open_cursor('statistics:')
        else:
            stat_cursor = self.session.open_cursor('statistics:' + uri)
        val = stat_cursor[stat][2]

            

Reported by Pylint.

                      for i in range(0, n):
            self.assertEqual(tc[str(i)], str(i))
        tc.set_key(str(n))
        self.assertEquals(tc.search(), wiredtiger.WT_NOTFOUND)

    # Test calling the flush_tier API.
    def test_tiered(self):
        # Create three tables. One using the system tiered storage, one
        # specifying its own bucket and object size and one using no

            

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 os, wiredtiger, wttest
from wiredtiger import stat
StorageSource = wiredtiger.StorageSource  # easy access to constants

# test_tiered04.py
#    Basic tiered storage API test.

            

Reported by Pylint.

              # OTHER DEALINGS IN THE SOFTWARE.

import os, wiredtiger, wttest
from wiredtiger import stat
StorageSource = wiredtiger.StorageSource  # easy access to constants

# test_tiered04.py
#    Basic tiered storage API test.
class test_tiered04(wttest.WiredTigerTestCase):

            

Reported by Pylint.

              
# test_tiered04.py
#    Basic tiered storage API test.
class test_tiered04(wttest.WiredTigerTestCase):

    # If the 'uri' changes all the other names must change with it.
    fileuri_base = 'file:test_tiered04-000000000'
    objuri = 'object:test_tiered04-0000000001.wtobj'
    tiereduri = "tiered:test_tiered04"

            

Reported by Pylint.

              
# test_tiered04.py
#    Basic tiered storage API test.
class test_tiered04(wttest.WiredTigerTestCase):

    # If the 'uri' changes all the other names must change with it.
    fileuri_base = 'file:test_tiered04-000000000'
    objuri = 'object:test_tiered04-0000000001.wtobj'
    tiereduri = "tiered:test_tiered04"

            

Reported by Pylint.

              from helper import copy_wiredtiger_home
import random
from suite_subprocess import suite_subprocess
import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):
    table_ts_log     = 'table:ts06_ts_logged'
    table_ts_nolog   = 'table:ts06_ts_nologged'

            

Reported by Pylint.

              from helper import copy_wiredtiger_home
import random
from suite_subprocess import suite_subprocess
import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):
    table_ts_log     = 'table:ts06_ts_logged'
    table_ts_nolog   = 'table:ts06_ts_nologged'

            

Reported by Pylint.

                      # Count how many times the given value is present in the
        # logged timestamp table.
        actual_ts_log = 0
        for k, v in cur_ts_log:
            if check_value == v:
                actual_ts_log += 1
        cur_ts_log.close()
        # Count how many times the given value is present in the
        # not logged timestamp table

            

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 helper import copy_wiredtiger_home
import random
from suite_subprocess import suite_subprocess
import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):

            

Reported by Pylint.

              from helper import copy_wiredtiger_home
import random
from suite_subprocess import suite_subprocess
import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):
    table_ts_log     = 'table:ts06_ts_logged'
    table_ts_nolog   = 'table:ts06_ts_nologged'

            

Reported by Pylint.

              from helper import copy_wiredtiger_home
import random
from suite_subprocess import suite_subprocess
import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):
    table_ts_log     = 'table:ts06_ts_logged'
    table_ts_nolog   = 'table:ts06_ts_nologged'

            

Reported by Pylint.

              import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):
    table_ts_log     = 'table:ts06_ts_logged'
    table_ts_nolog   = 'table:ts06_ts_nologged'

    types = [
        # FLCS does not yet work in a timestamp world.

            

Reported by Pylint.

              import wiredtiger, wttest
from wtscenario import make_scenarios

class test_timestamp06(wttest.WiredTigerTestCase, suite_subprocess):
    table_ts_log     = 'table:ts06_ts_logged'
    table_ts_nolog   = 'table:ts06_ts_nologged'

    types = [
        # FLCS does not yet work in a timestamp world.

            

Reported by Pylint.

              
    conncfg = [
        ('nolog', dict(conn_config='create', using_log=False)),
        ('V1', dict(conn_config='create,log=(archive=false,enabled),compatibility=(release="2.9")', using_log=True)),
        ('V2', dict(conn_config='create,log=(archive=false,enabled)', using_log=True)),
    ]
    session_config = 'isolation=snapshot'

    scenarios = make_scenarios(conncfg, types, ckpt)

            

Reported by Pylint.

              
import unittest

from testtools.compat import _b, BytesIO

import subunit.chunked


class TestDecode(unittest.TestCase):

            

Reported by Pylint.

              #
#  subunit: extensions to python unittest to get test results from subprocesses.
#  Copyright (C) 2005  Robert Collins <robertc@robertcollins.net>
#  Copyright (C) 2011  Martin Pool <mbp@sourcefrog.net>
#
#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
#  license at the users choice. A copy of both licenses are available in the
#  project source as Apache-2.0 and BSD. You may not use this file except in
#  compliance with one of these two licences.

            

Reported by Pylint.

              #  license at the users choice. A copy of both licenses are available in the
#  project source as Apache-2.0 and BSD. You may not use this file except in
#  compliance with one of these two licences.
#  
#  Unless required by applicable law or agreed to in writing, software
#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
#  license you chose for the specific language governing permissions and
#  limitations under that license.

            

Reported by Pylint.

              import subunit.chunked


class TestDecode(unittest.TestCase):

    def setUp(self):
        unittest.TestCase.setUp(self)
        self.output = BytesIO()
        self.decoder = subunit.chunked.Decoder(self.output)

            

Reported by Pylint.

                      self.output = BytesIO()
        self.decoder = subunit.chunked.Decoder(self.output)

    def test_close_read_length_short_errors(self):
        self.assertRaises(ValueError, self.decoder.close)

    def test_close_body_short_errors(self):
        self.assertEqual(None, self.decoder.write(_b('2\r\na')))
        self.assertRaises(ValueError, self.decoder.close)

            

Reported by Pylint.

                  def test_close_read_length_short_errors(self):
        self.assertRaises(ValueError, self.decoder.close)

    def test_close_body_short_errors(self):
        self.assertEqual(None, self.decoder.write(_b('2\r\na')))
        self.assertRaises(ValueError, self.decoder.close)

    def test_close_body_buffered_data_errors(self):
        self.assertEqual(None, self.decoder.write(_b('2\r')))

            

Reported by Pylint.

                      self.assertEqual(None, self.decoder.write(_b('2\r\na')))
        self.assertRaises(ValueError, self.decoder.close)

    def test_close_body_buffered_data_errors(self):
        self.assertEqual(None, self.decoder.write(_b('2\r')))
        self.assertRaises(ValueError, self.decoder.close)

    def test_close_after_finished_stream_safe(self):
        self.assertEqual(None, self.decoder.write(_b('2\r\nab')))

            

Reported by Pylint.

                      self.assertEqual(None, self.decoder.write(_b('2\r')))
        self.assertRaises(ValueError, self.decoder.close)

    def test_close_after_finished_stream_safe(self):
        self.assertEqual(None, self.decoder.write(_b('2\r\nab')))
        self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
        self.decoder.close()

    def test_decode_nothing(self):

            

Reported by Pylint.

                      self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
        self.decoder.close()

    def test_decode_nothing(self):
        self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
        self.assertEqual(_b(''), self.output.getvalue())

    def test_decode_serialised_form(self):
        self.assertEqual(None, self.decoder.write(_b("F\r\n")))

            

Reported by Pylint.

                      self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
        self.assertEqual(_b(''), self.output.getvalue())

    def test_decode_serialised_form(self):
        self.assertEqual(None, self.decoder.write(_b("F\r\n")))
        self.assertEqual(None, self.decoder.write(_b("serialised\n")))
        self.assertEqual(_b(''), self.decoder.write(_b("form0\r\n")))

    def test_decode_short(self):

            

Reported by Pylint.

              #endif

#ifdef _MSC_VER
#  define snprintf _snprintf
#endif

#ifdef VMS
#  define unlink delete
#  define GZ_SUFFIX "-gz"

            

Reported by FlawFinder.

              #endif

#ifdef _MSC_VER
#  define snprintf _snprintf
#endif

#ifdef VMS
#  define unlink delete
#  define GZ_SUFFIX "-gz"

            

Reported by FlawFinder.

                      exit(1);
    }

    strcpy(outfile, file);
    strcat(outfile, GZ_SUFFIX);

    in = fopen(file, "rb");
    if (in == NULL) {
        perror(file);

            

Reported by FlawFinder.

                  }

    strcpy(outfile, file);
    strcat(outfile, GZ_SUFFIX);

    in = fopen(file, "rb");
    if (in == NULL) {
        perror(file);
        exit(1);

            

Reported by FlawFinder.

                      exit(1);
    }

    strcpy(buf, file);

    if (len > SUFFIX_LEN && strcmp(file+len-SUFFIX_LEN, GZ_SUFFIX) == 0) {
        infile = file;
        outfile = buf;
        outfile[len-3] = '\0';

            

Reported by FlawFinder.

                  } else {
        outfile = file;
        infile = buf;
        strcat(infile, GZ_SUFFIX);
    }
    in = gzopen(infile, "rb");
    if (in == NULL) {
        fprintf(stderr, "%s: can't gzopen %s\n", prog, infile);
        exit(1);

            

Reported by FlawFinder.

              static char *strwinerror (error)
     DWORD error;
{
    static char buf[1024];

    wchar_t *msgbuf;
    DWORD lasterr = GetLastError();
    DWORD chars = FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM
        | FORMAT_MESSAGE_ALLOCATE_BUFFER,

            

Reported by FlawFinder.

                      LocalFree(msgbuf);
    }
    else {
        sprintf(buf, "unknown win32 error (%ld)", error);
    }

    SetLastError(lasterr);
    return buf;
}

            

Reported by FlawFinder.

                      return NULL;
    }
    gz->file = path == NULL ? fdopen(fd, gz->write ? "wb" : "rb") :
                              fopen(path, gz->write ? "wb" : "rb");
    if (gz->file == NULL) {
        gz->write ? deflateEnd(&(gz->strm)) : inflateEnd(&(gz->strm));
        free(gz);
        return NULL;
    }

            

Reported by FlawFinder.

                  unsigned len;
{
    z_stream *strm;
    unsigned char out[BUFLEN];

    if (gz == NULL || !gz->write)
        return 0;
    strm = &(gz->strm);
    strm->next_in = (void *)buf;

            

Reported by FlawFinder.