from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_raises_rpc_error

from decimal import Decimal
import shutil
import os

def descriptors(out):
    return sorted(u['desc'] for u in out['unspents'])

            

Reported by Pylint.

              from test_framework.util import assert_equal, assert_raises_rpc_error

from decimal import Decimal
import shutil
import os

def descriptors(out):
    return sorted(u['desc'] for u in out['unspents'])


            

Reported by Pylint.

              
from decimal import Decimal
import shutil
import os

def descriptors(out):
    return sorted(u['desc'] for u in out['unspents'])

class ScantxoutsetTest(BitcoinTestFramework):

            

Reported by Pylint.

              import shutil
import os

def descriptors(out):
    return sorted(u['desc'] for u in out['unspents'])

class ScantxoutsetTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1

            

Reported by Pylint.

              def descriptors(out):
    return sorted(u['desc'] for u in out['unspents'])

class ScantxoutsetTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.setup_clean_chain = True

    def skip_test_if_missing_module(self):

            

Reported by Pylint.

                  def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()

    def run_test(self):
        self.log.info("Mining blocks...")
        self.nodes[0].generate(110)

        addr_P2SH_SEGWIT = self.nodes[0].getnewaddress("", "p2sh-segwit")
        pubk1 = self.nodes[0].getaddressinfo(addr_P2SH_SEGWIT)['pubkey']

            

Reported by Pylint.

                      self.log.info("Mining blocks...")
        self.nodes[0].generate(110)

        addr_P2SH_SEGWIT = self.nodes[0].getnewaddress("", "p2sh-segwit")
        pubk1 = self.nodes[0].getaddressinfo(addr_P2SH_SEGWIT)['pubkey']
        addr_LEGACY = self.nodes[0].getnewaddress("", "legacy")
        pubk2 = self.nodes[0].getaddressinfo(addr_LEGACY)['pubkey']
        addr_BECH32 = self.nodes[0].getnewaddress("", "bech32")
        pubk3 = self.nodes[0].getaddressinfo(addr_BECH32)['pubkey']

            

Reported by Pylint.

              
        addr_P2SH_SEGWIT = self.nodes[0].getnewaddress("", "p2sh-segwit")
        pubk1 = self.nodes[0].getaddressinfo(addr_P2SH_SEGWIT)['pubkey']
        addr_LEGACY = self.nodes[0].getnewaddress("", "legacy")
        pubk2 = self.nodes[0].getaddressinfo(addr_LEGACY)['pubkey']
        addr_BECH32 = self.nodes[0].getnewaddress("", "bech32")
        pubk3 = self.nodes[0].getaddressinfo(addr_BECH32)['pubkey']
        self.nodes[0].sendtoaddress(addr_P2SH_SEGWIT, 0.001)
        self.nodes[0].sendtoaddress(addr_LEGACY, 0.002)

            

Reported by Pylint.

                      pubk1 = self.nodes[0].getaddressinfo(addr_P2SH_SEGWIT)['pubkey']
        addr_LEGACY = self.nodes[0].getnewaddress("", "legacy")
        pubk2 = self.nodes[0].getaddressinfo(addr_LEGACY)['pubkey']
        addr_BECH32 = self.nodes[0].getnewaddress("", "bech32")
        pubk3 = self.nodes[0].getaddressinfo(addr_BECH32)['pubkey']
        self.nodes[0].sendtoaddress(addr_P2SH_SEGWIT, 0.001)
        self.nodes[0].sendtoaddress(addr_LEGACY, 0.002)
        self.nodes[0].sendtoaddress(addr_BECH32, 0.004)


            

Reported by Pylint.

                      self.nodes[0].sendtoaddress(addr_LEGACY, 0.002)
        self.nodes[0].sendtoaddress(addr_BECH32, 0.004)

        #send to child keys of tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK
        self.nodes[0].sendtoaddress("mkHV1C6JLheLoUSSZYk7x3FH5tnx9bu7yc", 0.008) # (m/0'/0'/0')
        self.nodes[0].sendtoaddress("mipUSRmJAj2KrjSvsPQtnP8ynUon7FhpCR", 0.016) # (m/0'/0'/1')
        self.nodes[0].sendtoaddress("n37dAGe6Mq1HGM9t4b6rFEEsDGq7Fcgfqg", 0.032) # (m/0'/0'/1500')
        self.nodes[0].sendtoaddress("mqS9Rpg8nNLAzxFExsgFLCnzHBsoQ3PRM6", 0.064) # (m/0'/0'/0)
        self.nodes[0].sendtoaddress("mnTg5gVWr3rbhHaKjJv7EEEc76ZqHgSj4S", 0.128) # (m/0'/0'/1)

            

Reported by Pylint.

              class P2PVersionStore(P2PInterface):
    version_received = None

    def on_version(self, msg):
        # Responds with an appropriate verack
        super().on_version(msg)
        self.version_received = msg



            

Reported by Pylint.

                      assert_greater_than_or_equal(ver.nTime, time.time() - 3600)
        assert_greater_than_or_equal(time.time() + 3600, ver.nTime)
        assert_equal(ver.addrFrom.port, 0)
        assert_equal(ver.addrFrom.ip, '0.0.0.0')
        assert_equal(ver.nStartingHeight, 201)
        assert_equal(ver.relay, 1)

        self.log.info('Check that old peers are disconnected')
        p2p_old_peer = self.nodes[0].add_p2p_connection(P2PInterface(), send_version=False, wait_for_verack=False)

            

Reported by Bandit.

              PEER_TIMEOUT = 3


class LazyPeer(P2PInterface):
    def __init__(self):
        super().__init__()
        self.unexpected_msg = False
        self.ever_connected = False
        self.got_wtxidrelay = False

            

Reported by Pylint.

              PEER_TIMEOUT = 3


class LazyPeer(P2PInterface):
    def __init__(self):
        super().__init__()
        self.unexpected_msg = False
        self.ever_connected = False
        self.got_wtxidrelay = False

            

Reported by Pylint.

                      self.got_wtxidrelay = False
        self.got_sendaddrv2 = False

    def bad_message(self, message):
        self.unexpected_msg = True
        print("should not have received message: %s" % message.msgtype)

    def on_open(self):
        self.ever_connected = True

            

Reported by Pylint.

                      self.ever_connected = True

    # Does not respond to "version" with "verack"
    def on_version(self, message): self.bad_message(message)
    def on_verack(self, message): self.bad_message(message)
    def on_inv(self, message): self.bad_message(message)
    def on_addr(self, message): self.bad_message(message)
    def on_getdata(self, message): self.bad_message(message)
    def on_getblocks(self, message): self.bad_message(message)

            

Reported by Pylint.

              
    # Does not respond to "version" with "verack"
    def on_version(self, message): self.bad_message(message)
    def on_verack(self, message): self.bad_message(message)
    def on_inv(self, message): self.bad_message(message)
    def on_addr(self, message): self.bad_message(message)
    def on_getdata(self, message): self.bad_message(message)
    def on_getblocks(self, message): self.bad_message(message)
    def on_tx(self, message): self.bad_message(message)

            

Reported by Pylint.

                  # Does not respond to "version" with "verack"
    def on_version(self, message): self.bad_message(message)
    def on_verack(self, message): self.bad_message(message)
    def on_inv(self, message): self.bad_message(message)
    def on_addr(self, message): self.bad_message(message)
    def on_getdata(self, message): self.bad_message(message)
    def on_getblocks(self, message): self.bad_message(message)
    def on_tx(self, message): self.bad_message(message)
    def on_block(self, message): self.bad_message(message)

            

Reported by Pylint.

                  def on_version(self, message): self.bad_message(message)
    def on_verack(self, message): self.bad_message(message)
    def on_inv(self, message): self.bad_message(message)
    def on_addr(self, message): self.bad_message(message)
    def on_getdata(self, message): self.bad_message(message)
    def on_getblocks(self, message): self.bad_message(message)
    def on_tx(self, message): self.bad_message(message)
    def on_block(self, message): self.bad_message(message)
    def on_getaddr(self, message): self.bad_message(message)

            

Reported by Pylint.

                  def on_verack(self, message): self.bad_message(message)
    def on_inv(self, message): self.bad_message(message)
    def on_addr(self, message): self.bad_message(message)
    def on_getdata(self, message): self.bad_message(message)
    def on_getblocks(self, message): self.bad_message(message)
    def on_tx(self, message): self.bad_message(message)
    def on_block(self, message): self.bad_message(message)
    def on_getaddr(self, message): self.bad_message(message)
    def on_headers(self, message): self.bad_message(message)

            

Reported by Pylint.

              
    def _load_dyn_tags(self) -> None:
        self.dyn_tags = []
        for idx, section in enumerate(self.sections):
            if section.sh_type == SHT_DYNAMIC: # find dynamic tag tables
                strtab = self.sections[section.sh_link].contents() # associated string table
                self.dyn_tags += _parse_dyn_tags(section, strtab, self.hdr)

    def _section_to_segment_mapping(self) -> None:

            

Reported by Pylint.

              SHT_STRTAB = 3
SHT_DYNAMIC = 6
SHT_DYNSYM = 11
SHT_GNU_verneed = 0x6ffffffe
SHT_GNU_versym = 0x6fffffff

# relevant values for p_type
PT_LOAD = 1
PT_GNU_STACK = 0x6474e551

            

Reported by Pylint.

              SHT_DYNAMIC = 6
SHT_DYNSYM = 11
SHT_GNU_verneed = 0x6ffffffe
SHT_GNU_versym = 0x6fffffff

# relevant values for p_type
PT_LOAD = 1
PT_GNU_STACK = 0x6474e551
PT_GNU_RELRO = 0x6474e552

            

Reported by Pylint.

              # rrlevant values for ST_BIND subfield of st_info (symbol binding)
STB_LOCAL = 0

class ELFRecord(types.SimpleNamespace):
    '''Unified parsing for ELF records.'''
    def __init__(self, data: bytes, offset: int, eh: 'ELFHeader', total_size: Optional[int]) -> None:
        hdr_struct = self.STRUCT[eh.ei_class][0][eh.ei_data]
        if total_size is not None and hdr_struct.size > total_size:
            raise ValueError(f'{self.__class__.__name__} header size too small ({total_size} < {hdr_struct.size})')

            

Reported by Pylint.

              
class ELFRecord(types.SimpleNamespace):
    '''Unified parsing for ELF records.'''
    def __init__(self, data: bytes, offset: int, eh: 'ELFHeader', total_size: Optional[int]) -> None:
        hdr_struct = self.STRUCT[eh.ei_class][0][eh.ei_data]
        if total_size is not None and hdr_struct.size > total_size:
            raise ValueError(f'{self.__class__.__name__} header size too small ({total_size} < {hdr_struct.size})')
        for field, value in zip(self.STRUCT[eh.ei_class][1], hdr_struct.unpack(data[offset:offset + hdr_struct.size])):
            setattr(self, field, value)

            

Reported by Pylint.

              
class ELFRecord(types.SimpleNamespace):
    '''Unified parsing for ELF records.'''
    def __init__(self, data: bytes, offset: int, eh: 'ELFHeader', total_size: Optional[int]) -> None:
        hdr_struct = self.STRUCT[eh.ei_class][0][eh.ei_data]
        if total_size is not None and hdr_struct.size > total_size:
            raise ValueError(f'{self.__class__.__name__} header size too small ({total_size} < {hdr_struct.size})')
        for field, value in zip(self.STRUCT[eh.ei_class][1], hdr_struct.unpack(data[offset:offset + hdr_struct.size])):
            setattr(self, field, value)

            

Reported by Pylint.

                  def __init__(self, data: bytes, offset: int, eh: 'ELFHeader', total_size: Optional[int]) -> None:
        hdr_struct = self.STRUCT[eh.ei_class][0][eh.ei_data]
        if total_size is not None and hdr_struct.size > total_size:
            raise ValueError(f'{self.__class__.__name__} header size too small ({total_size} < {hdr_struct.size})')
        for field, value in zip(self.STRUCT[eh.ei_class][1], hdr_struct.unpack(data[offset:offset + hdr_struct.size])):
            setattr(self, field, value)

def BiStruct(chars: str) -> Dict[int, struct.Struct]:
    '''Compile a struct parser for both endians.'''

            

Reported by Pylint.

                      hdr_struct = self.STRUCT[eh.ei_class][0][eh.ei_data]
        if total_size is not None and hdr_struct.size > total_size:
            raise ValueError(f'{self.__class__.__name__} header size too small ({total_size} < {hdr_struct.size})')
        for field, value in zip(self.STRUCT[eh.ei_class][1], hdr_struct.unpack(data[offset:offset + hdr_struct.size])):
            setattr(self, field, value)

def BiStruct(chars: str) -> Dict[int, struct.Struct]:
    '''Compile a struct parser for both endians.'''
    return {

            

Reported by Pylint.

                      for field, value in zip(self.STRUCT[eh.ei_class][1], hdr_struct.unpack(data[offset:offset + hdr_struct.size])):
            setattr(self, field, value)

def BiStruct(chars: str) -> Dict[int, struct.Struct]:
    '''Compile a struct parser for both endians.'''
    return {
        ELFDATA2LSB: struct.Struct('<' + chars),
        ELFDATA2MSB: struct.Struct('>' + chars),
    }

            

Reported by Pylint.

                      ELFDATA2MSB: struct.Struct('>' + chars),
    }

class ELFHeader(ELFRecord):
    FIELDS = ['e_type', 'e_machine', 'e_version', 'e_entry', 'e_phoff', 'e_shoff', 'e_flags', 'e_ehsize', 'e_phentsize', 'e_phnum', 'e_shentsize', 'e_shnum', 'e_shstrndx']
    STRUCT = {
        ELFCLASS32: (BiStruct('HHIIIIIHHHHHH'), FIELDS),
        ELFCLASS64: (BiStruct('HHIQQQIHHHHHH'), FIELDS),
    }

            

Reported by Pylint.

                      self.skip_if_no_wallet()

    def get_keys(self):
        self.pub = []
        self.priv = []
        node0, node1, node2 = self.nodes
        for _ in range(self.nkeys):
            k = ECKey()
            k.generate()

            

Reported by Pylint.

              
    def get_keys(self):
        self.pub = []
        self.priv = []
        node0, node1, node2 = self.nodes
        for _ in range(self.nkeys):
            k = ECKey()
            k.generate()
            self.pub.append(k.get_pubkey().get_bytes().hex())

            

Reported by Pylint.

                  def get_keys(self):
        self.pub = []
        self.priv = []
        node0, node1, node2 = self.nodes
        for _ in range(self.nkeys):
            k = ECKey()
            k.generate()
            self.pub.append(k.get_pubkey().get_bytes().hex())
            self.priv.append(bytes_to_wif(k.get_bytes(), k.is_compressed))

            

Reported by Pylint.

                  def get_keys(self):
        self.pub = []
        self.priv = []
        node0, node1, node2 = self.nodes
        for _ in range(self.nkeys):
            k = ECKey()
            k.generate()
            self.pub.append(k.get_pubkey().get_bytes().hex())
            self.priv.append(bytes_to_wif(k.get_bytes(), k.is_compressed))

            

Reported by Pylint.

                  def get_keys(self):
        self.pub = []
        self.priv = []
        node0, node1, node2 = self.nodes
        for _ in range(self.nkeys):
            k = ECKey()
            k.generate()
            self.pub.append(k.get_pubkey().get_bytes().hex())
            self.priv.append(bytes_to_wif(k.get_bytes(), k.is_compressed))

            

Reported by Pylint.

                          k.generate()
            self.pub.append(k.get_pubkey().get_bytes().hex())
            self.priv.append(bytes_to_wif(k.get_bytes(), k.is_compressed))
        self.final = node2.getnewaddress()

    def run_test(self):
        node0, node1, node2 = self.nodes

        self.check_addmultisigaddress_errors()

            

Reported by Pylint.

                      self.final = node2.getnewaddress()

    def run_test(self):
        node0, node1, node2 = self.nodes

        self.check_addmultisigaddress_errors()

        self.log.info('Generating blocks ...')
        node0.generate(149)

            

Reported by Pylint.

                      node0.generate(149)
        self.sync_all()

        self.moved = 0
        for self.nkeys in [3, 5]:
            for self.nsigs in [2, 3]:
                for self.output_type in ["bech32", "p2sh-segwit", "legacy"]:
                    self.get_keys()
                    self.do_multisig()

            

Reported by Pylint.

                      self.sync_all()

        self.moved = 0
        for self.nkeys in [3, 5]:
            for self.nsigs in [2, 3]:
                for self.output_type in ["bech32", "p2sh-segwit", "legacy"]:
                    self.get_keys()
                    self.do_multisig()


            

Reported by Pylint.

              
        self.moved = 0
        for self.nkeys in [3, 5]:
            for self.nsigs in [2, 3]:
                for self.output_type in ["bech32", "p2sh-segwit", "legacy"]:
                    self.get_keys()
                    self.do_multisig()

        self.checkbalances()

            

Reported by Pylint.

              ]


def DirectoryOfThisScript():
  """Returns the absolute path to the directory containing this script."""
  return os.path.dirname(os.path.abspath(__file__))


def MakeRelativePathsInFlagsAbsolute(flags, build_root):

            

Reported by Pylint.

              

def DirectoryOfThisScript():
  """Returns the absolute path to the directory containing this script."""
  return os.path.dirname(os.path.abspath(__file__))


def MakeRelativePathsInFlagsAbsolute(flags, build_root):
  """Expands relative paths in a list of Clang command-line flags.

            

Reported by Pylint.

              
def DirectoryOfThisScript():
  """Returns the absolute path to the directory containing this script."""
  return os.path.dirname(os.path.abspath(__file__))


def MakeRelativePathsInFlagsAbsolute(flags, build_root):
  """Expands relative paths in a list of Clang command-line flags.


            

Reported by Pylint.

                return os.path.dirname(os.path.abspath(__file__))


def MakeRelativePathsInFlagsAbsolute(flags, build_root):
  """Expands relative paths in a list of Clang command-line flags.

  Args:
    flags: The list of flags passed to Clang.
    build_root: The current directory when running the Clang compiler. Should be

            

Reported by Pylint.

              

def MakeRelativePathsInFlagsAbsolute(flags, build_root):
  """Expands relative paths in a list of Clang command-line flags.

  Args:
    flags: The list of flags passed to Clang.
    build_root: The current directory when running the Clang compiler. Should be
        an absolute path.

            

Reported by Pylint.

                Returns:
    A list of flags with relative paths replaced by absolute paths.
  """
  new_flags = []
  make_next_absolute = False
  for flag in flags:
    new_flag = flag

    if make_next_absolute:

            

Reported by Pylint.

                  A list of flags with relative paths replaced by absolute paths.
  """
  new_flags = []
  make_next_absolute = False
  for flag in flags:
    new_flag = flag

    if make_next_absolute:
      make_next_absolute = False

            

Reported by Pylint.

                """
  new_flags = []
  make_next_absolute = False
  for flag in flags:
    new_flag = flag

    if make_next_absolute:
      make_next_absolute = False
      if not flag.startswith('/'):

            

Reported by Pylint.

                new_flags = []
  make_next_absolute = False
  for flag in flags:
    new_flag = flag

    if make_next_absolute:
      make_next_absolute = False
      if not flag.startswith('/'):
        new_flag = os.path.join(build_root, flag)

            

Reported by Pylint.

                for flag in flags:
    new_flag = flag

    if make_next_absolute:
      make_next_absolute = False
      if not flag.startswith('/'):
        new_flag = os.path.join(build_root, flag)

    for path_flag in PATH_FLAGS:

            

Reported by Pylint.

                  format_string = format_string.replace('%%', 'X')
    n = 0
    in_specifier = False
    for i, char in enumerate(format_string):
        if char == "%":
            in_specifier = True
            n += 1
        elif char in "aAcdeEfFgGinopsuxX":
            in_specifier = False

            

Reported by Pylint.

              #!/usr/bin/env python3
#
# Copyright (c) 2018-2019 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
#
# Lint format strings: This program checks that the number of arguments passed
# to a variadic format string function matches the number of format specifiers
# in the format string.

            

Reported by Pylint.

              #!/usr/bin/env python3
#
# Copyright (c) 2018-2019 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
#
# Lint format strings: This program checks that the number of arguments passed
# to a variadic format string function matches the number of format specifiers
# in the format string.

            

Reported by Pylint.

              FALSE_POSITIVES = [
    ("src/dbwrapper.cpp", "vsnprintf(p, limit - p, format, backup_ap)"),
    ("src/index/base.cpp", "FatalError(const char* fmt, const Args&... args)"),
    ("src/netbase.cpp", "LogConnectFailure(bool manual_connection, const char* fmt, const Args&... args)"),
    ("src/util/system.cpp", "strprintf(_(COPYRIGHT_HOLDERS).translated, COPYRIGHT_HOLDERS_SUBSTITUTION)"),
    ("src/validationinterface.cpp", "LogPrint(BCLog::VALIDATION, fmt \"\\n\", __VA_ARGS__)"),
    ("src/wallet/wallet.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),
    ("src/wallet/wallet.h", "LogPrintf((\"%s \" + fmt).c_str(), GetDisplayName(), parameters...)"),
    ("src/wallet/scriptpubkeyman.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),

            

Reported by Pylint.

                  ("src/dbwrapper.cpp", "vsnprintf(p, limit - p, format, backup_ap)"),
    ("src/index/base.cpp", "FatalError(const char* fmt, const Args&... args)"),
    ("src/netbase.cpp", "LogConnectFailure(bool manual_connection, const char* fmt, const Args&... args)"),
    ("src/util/system.cpp", "strprintf(_(COPYRIGHT_HOLDERS).translated, COPYRIGHT_HOLDERS_SUBSTITUTION)"),
    ("src/validationinterface.cpp", "LogPrint(BCLog::VALIDATION, fmt \"\\n\", __VA_ARGS__)"),
    ("src/wallet/wallet.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),
    ("src/wallet/wallet.h", "LogPrintf((\"%s \" + fmt).c_str(), GetDisplayName(), parameters...)"),
    ("src/wallet/scriptpubkeyman.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),
    ("src/wallet/scriptpubkeyman.h", "LogPrintf((\"%s \" + fmt).c_str(), m_storage.GetDisplayName(), parameters...)"),

            

Reported by Pylint.

                  ("src/wallet/wallet.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),
    ("src/wallet/wallet.h", "LogPrintf((\"%s \" + fmt).c_str(), GetDisplayName(), parameters...)"),
    ("src/wallet/scriptpubkeyman.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),
    ("src/wallet/scriptpubkeyman.h", "LogPrintf((\"%s \" + fmt).c_str(), m_storage.GetDisplayName(), parameters...)"),
    ("src/logging.h", "LogPrintf(const char* fmt, const Args&... args)"),
    ("src/wallet/scriptpubkeyman.h", "WalletLogPrintf(const std::string& fmt, const Params&... parameters)"),
]



            

Reported by Pylint.

                  ("src/wallet/scriptpubkeyman.h",  "WalletLogPrintf(std::string fmt, Params... parameters)"),
    ("src/wallet/scriptpubkeyman.h", "LogPrintf((\"%s \" + fmt).c_str(), m_storage.GetDisplayName(), parameters...)"),
    ("src/logging.h", "LogPrintf(const char* fmt, const Args&... args)"),
    ("src/wallet/scriptpubkeyman.h", "WalletLogPrintf(const std::string& fmt, const Params&... parameters)"),
]


def parse_function_calls(function_name, source_code):
    """Return an array with all calls to function function_name in string source_code.

            

Reported by Pylint.

                  >>> len(parse_function_calls("foo", "#define FOO foo();"))
    0
    """
    assert type(function_name) is str and type(source_code) is str and function_name
    lines = [re.sub("// .*", " ", line).strip()
             for line in source_code.split("\n")
             if not line.strip().startswith("#")]
    return re.findall(r"[^a-zA-Z_](?=({}\(.*).*)".format(function_name), " " + " ".join(lines))


            

Reported by Bandit.

                  >>> len(parse_function_calls("foo", "#define FOO foo();"))
    0
    """
    assert type(function_name) is str and type(source_code) is str and function_name
    lines = [re.sub("// .*", " ", line).strip()
             for line in source_code.split("\n")
             if not line.strip().startswith("#")]
    return re.findall(r"[^a-zA-Z_](?=({}\(.*).*)".format(function_name), " " + " ".join(lines))


            

Reported by Pylint.

                  >>> len(parse_function_calls("foo", "#define FOO foo();"))
    0
    """
    assert type(function_name) is str and type(source_code) is str and function_name
    lines = [re.sub("// .*", " ", line).strip()
             for line in source_code.split("\n")
             if not line.strip().startswith("#")]
    return re.findall(r"[^a-zA-Z_](?=({}\(.*).*)".format(function_name), " " + " ".join(lines))


            

Reported by Pylint.

                      self.skip_if_no_wallet()

    def run_test(self):
        self.relayfee = self.nodes[0].getnetworkinfo()["relayfee"]

        # Generate some coins
        self.nodes[0].generate(110)

        self.log.info("Running test disable flag")

            

Reported by Pylint.

              # RPC error for non-BIP68 final transactions
NOT_FINAL_ERROR = "non-BIP68-final"

class BIP68Test(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 2
        self.extra_args = [
            [
                "-acceptnonstdtxn=1",

            

Reported by Pylint.

                      self.activateCSV()

        self.log.info("Verifying nVersion=2 transactions are standard.")
        self.log.info("Note that nVersion=2 transactions are always standard (independent of BIP68 activation status).")
        self.test_version2_relay()

        self.log.info("Passed")

    # Test that BIP68 is not in effect if tx version is 1, or if

            

Reported by Pylint.

              
    # Test that BIP68 is not in effect if tx version is 1, or if
    # the first sequence bit is set.
    def test_disable_flag(self):
        # Create some unconfirmed inputs
        new_addr = self.nodes[0].getnewaddress()
        self.nodes[0].sendtoaddress(new_addr, 2) # send 2 BTC

        utxos = self.nodes[0].listunspent(0, 0)

            

Reported by Pylint.

                      self.nodes[0].sendtoaddress(new_addr, 2) # send 2 BTC

        utxos = self.nodes[0].listunspent(0, 0)
        assert len(utxos) > 0

        utxo = utxos[0]

        tx1 = CTransaction()
        value = int(satoshi_round(utxo["amount"] - self.relayfee)*COIN)

            

Reported by Bandit.

                      tx2.vout = [CTxOut(int(value - self.relayfee * COIN), DUMMY_P2WPKH_SCRIPT)]
        tx2.rehash()

        assert_raises_rpc_error(-26, NOT_FINAL_ERROR, self.nodes[0].sendrawtransaction, tx2.serialize().hex())

        # Setting the version back down to 1 should disable the sequence lock,
        # so this should be accepted.
        tx2.nVersion = 1


            

Reported by Pylint.

              
    # Calculate the median time past of a prior block ("confirmations" before
    # the current tip).
    def get_median_time_past(self, confirmations):
        block_hash = self.nodes[0].getblockhash(self.nodes[0].getblockcount()-confirmations)
        return self.nodes[0].getblockheader(block_hash)["mediantime"]

    # Test that sequence locks are respected for transactions spending confirmed inputs.
    def test_sequence_lock_confirmed_inputs(self):

            

Reported by Pylint.

                      return self.nodes[0].getblockheader(block_hash)["mediantime"]

    # Test that sequence locks are respected for transactions spending confirmed inputs.
    def test_sequence_lock_confirmed_inputs(self):
        # Create lots of confirmed utxos, and use them to generate lots of random
        # transactions.
        max_outputs = 50
        addresses = []
        while len(addresses) < max_outputs:

            

Reported by Pylint.

                      return self.nodes[0].getblockheader(block_hash)["mediantime"]

    # Test that sequence locks are respected for transactions spending confirmed inputs.
    def test_sequence_lock_confirmed_inputs(self):
        # Create lots of confirmed utxos, and use them to generate lots of random
        # transactions.
        max_outputs = 50
        addresses = []
        while len(addresses) < max_outputs:

            

Reported by Pylint.

                      return self.nodes[0].getblockheader(block_hash)["mediantime"]

    # Test that sequence locks are respected for transactions spending confirmed inputs.
    def test_sequence_lock_confirmed_inputs(self):
        # Create lots of confirmed utxos, and use them to generate lots of random
        # transactions.
        max_outputs = 50
        addresses = []
        while len(addresses) < max_outputs:

            

Reported by Pylint.

              def get_count_of_copyrights_of_any_style_any_holder(contents):
    return len(ANY_COPYRIGHT_COMPILED.findall(contents))

def file_has_dominant_style_copyright_for_holder(contents, holder_name):
    match = DOMINANT_STYLE_COMPILED[holder_name].search(contents)
    return match is not None

def file_has_year_list_style_copyright_for_holder(contents, holder_name):
    match = YEAR_LIST_STYLE_COMPILED[holder_name].search(contents)

            

Reported by Pylint.

                  match = DOMINANT_STYLE_COMPILED[holder_name].search(contents)
    return match is not None

def file_has_year_list_style_copyright_for_holder(contents, holder_name):
    match = YEAR_LIST_STYLE_COMPILED[holder_name].search(contents)
    return match is not None

def file_has_without_c_style_copyright_for_holder(contents, holder_name):
    match = WITHOUT_C_STYLE_COMPILED[holder_name].search(contents)

            

Reported by Pylint.

                  match = YEAR_LIST_STYLE_COMPILED[holder_name].search(contents)
    return match is not None

def file_has_without_c_style_copyright_for_holder(contents, holder_name):
    match = WITHOUT_C_STYLE_COMPILED[holder_name].search(contents)
    return match is not None

################################################################################
# get file info

            

Reported by Pylint.

                  info['dominant_style'] = {}
    info['year_list_style'] = {}
    info['without_c_style'] = {}
    for holder_name in EXPECTED_HOLDER_NAMES:
        has_dominant_style = (
            file_has_dominant_style_copyright_for_holder(c, holder_name))
        has_year_list_style = (
            file_has_year_list_style_copyright_for_holder(c, holder_name))
        has_without_c_style = (

            

Reported by Pylint.

                  print(SEPARATOR)
    print('Copyrights with dominant style:\ne.g. "Copyright (c)" and '
          '"<year>" or "<startYear>-<endYear>":\n')
    for holder_name in EXPECTED_HOLDER_NAMES:
        dominant_style = [i['filename'] for i in file_infos if
                          i['dominant_style'][holder_name]]
        if len(dominant_style) > 0:
            print("%4d with '%s'" % (len(dominant_style),
                                     holder_name.replace('\n', '\\n')))

            

Reported by Pylint.

              #!/usr/bin/env python3
# Copyright (c) 2016-2020 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.

import re
import fnmatch
import sys
import subprocess

            

Reported by Pylint.

              import re
import fnmatch
import sys
import subprocess
import datetime
import os

################################################################################
# file filtering

            

Reported by Bandit.

              INCLUDE = ['*.h', '*.cpp', '*.cc', '*.c', '*.mm', '*.py', '*.sh', '*.bash-completion']
INCLUDE_COMPILED = re.compile('|'.join([fnmatch.translate(m) for m in INCLUDE]))

def applies_to_file(filename):
    for excluded_dir in EXCLUDE_DIRS:
        if filename.startswith(excluded_dir):
            return False
    return ((EXCLUDE_COMPILED.match(filename) is None) and
            (INCLUDE_COMPILED.match(filename) is not None))

            

Reported by Pylint.

              GIT_LS_CMD = 'git ls-files --full-name'.split(' ')
GIT_TOPLEVEL_CMD = 'git rev-parse --show-toplevel'.split(' ')

def call_git_ls(base_directory):
    out = subprocess.check_output([*GIT_LS_CMD, base_directory])
    return [f for f in out.decode("utf-8").split('\n') if f != '']

def call_git_toplevel():
    "Returns the absolute path to the project root"

            

Reported by Pylint.

              GIT_TOPLEVEL_CMD = 'git rev-parse --show-toplevel'.split(' ')

def call_git_ls(base_directory):
    out = subprocess.check_output([*GIT_LS_CMD, base_directory])
    return [f for f in out.decode("utf-8").split('\n') if f != '']

def call_git_toplevel():
    "Returns the absolute path to the project root"
    return subprocess.check_output(GIT_TOPLEVEL_CMD).strip().decode("utf-8")

            

Reported by Bandit.

              )


class DecodeScriptTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1

    def decodescript_script_sig(self):

            

Reported by Pylint.

                      self.setup_clean_chain = True
        self.num_nodes = 1

    def decodescript_script_sig(self):
        signature = '304502207fa7a6d1e0ee81132a269ad84e68d695483745cde8b541e3bf630749894e342a022100c1f7ab20e13e22fb95281a870f3dcf38d782e53023ee313d741ad0cfbc0c509001'
        push_signature = '48' + signature
        public_key = '03b0da749730dc9b4b1f4a14d6902877a92541f5368778853d9c4a0cb7802dcfb2'
        push_public_key = '21' + public_key


            

Reported by Pylint.

                      self.num_nodes = 1

    def decodescript_script_sig(self):
        signature = '304502207fa7a6d1e0ee81132a269ad84e68d695483745cde8b541e3bf630749894e342a022100c1f7ab20e13e22fb95281a870f3dcf38d782e53023ee313d741ad0cfbc0c509001'
        push_signature = '48' + signature
        public_key = '03b0da749730dc9b4b1f4a14d6902877a92541f5368778853d9c4a0cb7802dcfb2'
        push_public_key = '21' + public_key

        # below are test cases for all of the standard transaction types

            

Reported by Pylint.

                      # 5) null data scriptSig - no such thing because null data scripts can not be spent.
        # thus, no test case for that standard transaction type is here.

    def decodescript_script_pub_key(self):
        public_key = '03b0da749730dc9b4b1f4a14d6902877a92541f5368778853d9c4a0cb7802dcfb2'
        push_public_key = '21' + public_key
        public_key_hash = '5dd1d3a048119c27b28293056724d9522f26d945'
        push_public_key_hash = '14' + public_key_hash
        uncompressed_public_key = '04b0da749730dc9b4b1f4a14d6902877a92541f5368778853d9c4a0cb7802dcfb25e01fc8fde47c96c98a4f3a8123e33a38a50cf9025cc8c4494a518f991792bb7'

            

Reported by Pylint.

                      push_public_key = '21' + public_key
        public_key_hash = '5dd1d3a048119c27b28293056724d9522f26d945'
        push_public_key_hash = '14' + public_key_hash
        uncompressed_public_key = '04b0da749730dc9b4b1f4a14d6902877a92541f5368778853d9c4a0cb7802dcfb25e01fc8fde47c96c98a4f3a8123e33a38a50cf9025cc8c4494a518f991792bb7'
        push_uncompressed_public_key = '41' + uncompressed_public_key
        p2wsh_p2pk_script_hash = 'd8590cf8ea0674cf3d49fd7ca249b85ef7485dea62c138468bddeb20cd6519f7'

        # below are test cases for all of the standard transaction types


            

Reported by Pylint.

                      # 2) P2PKH scriptPubKey
        # OP_DUP OP_HASH160 <PubKeyHash> OP_EQUALVERIFY OP_CHECKSIG
        rpc_result = self.nodes[0].decodescript('76a9' + push_public_key_hash + '88ac')
        assert_equal('OP_DUP OP_HASH160 ' + public_key_hash + ' OP_EQUALVERIFY OP_CHECKSIG', rpc_result['asm'])
        # P2PKH is translated to P2WPKH
        assert_equal('0 ' + public_key_hash, rpc_result['segwit']['asm'])

        # 3) multisig scriptPubKey
        # <m> <A pubkey> <B pubkey> <C pubkey> <n> OP_CHECKMULTISIG

            

Reported by Pylint.

                      # 3) multisig scriptPubKey
        # <m> <A pubkey> <B pubkey> <C pubkey> <n> OP_CHECKMULTISIG
        # just imagine that the pub keys used below are different.
        # for our purposes here it does not matter that they are the same even though it is unrealistic.
        multisig_script = '52' + push_public_key + push_public_key + push_public_key + '53ae'
        rpc_result = self.nodes[0].decodescript(multisig_script)
        assert_equal('2 ' + public_key + ' ' + public_key + ' ' + public_key +  ' 3 OP_CHECKMULTISIG', rpc_result['asm'])
        # multisig in P2WSH
        multisig_script_hash = sha256(bytes.fromhex(multisig_script)).hex()

            

Reported by Pylint.

                      # for our purposes here it does not matter that they are the same even though it is unrealistic.
        multisig_script = '52' + push_public_key + push_public_key + push_public_key + '53ae'
        rpc_result = self.nodes[0].decodescript(multisig_script)
        assert_equal('2 ' + public_key + ' ' + public_key + ' ' + public_key +  ' 3 OP_CHECKMULTISIG', rpc_result['asm'])
        # multisig in P2WSH
        multisig_script_hash = sha256(bytes.fromhex(multisig_script)).hex()
        assert_equal('0 ' + multisig_script_hash, rpc_result['segwit']['asm'])

        # 4) P2SH scriptPubKey

            

Reported by Pylint.

                      rpc_result = self.nodes[0].decodescript('a9' + push_public_key_hash + '87')
        assert_equal('OP_HASH160 ' + public_key_hash + ' OP_EQUAL', rpc_result['asm'])
        # P2SH does not work in segwit secripts. decodescript should not return a result for it.
        assert 'segwit' not in rpc_result

        # 5) null data scriptPubKey
        # use a signature look-alike here to make sure that we do not decode random data as a signature.
        # this matters if/when signature sighash decoding comes along.
        # would want to make sure that no such decoding takes place in this case.

            

Reported by Bandit.

                      assert 'segwit' not in rpc_result

        # 5) null data scriptPubKey
        # use a signature look-alike here to make sure that we do not decode random data as a signature.
        # this matters if/when signature sighash decoding comes along.
        # would want to make sure that no such decoding takes place in this case.
        signature_imposter = '48304502207fa7a6d1e0ee81132a269ad84e68d695483745cde8b541e3bf630749894e342a022100c1f7ab20e13e22fb95281a870f3dcf38d782e53023ee313d741ad0cfbc0c509001'
        # OP_RETURN <data>
        rpc_result = self.nodes[0].decodescript('6a' + signature_imposter)

            

Reported by Pylint.

                      logging.error("No fuzz targets found")
        sys.exit(1)

    logging.debug("{} fuzz target(s) found: {}".format(len(test_list_all), " ".join(sorted(test_list_all))))

    args.target = args.target or test_list_all  # By default run all
    test_list_error = list(set(args.target).difference(set(test_list_all)))
    if test_list_error:
        logging.error("Unknown fuzz targets selected: {}".format(test_list_error))

            

Reported by Pylint.

                  args.target = args.target or test_list_all  # By default run all
    test_list_error = list(set(args.target).difference(set(test_list_all)))
    if test_list_error:
        logging.error("Unknown fuzz targets selected: {}".format(test_list_error))
    test_list_selection = list(set(test_list_all).intersection(set(args.target)))
    if not test_list_selection:
        logging.error("No fuzz targets selected")
    if args.exclude:
        for excluded_target in args.exclude.split(","):

            

Reported by Pylint.

                  if args.exclude:
        for excluded_target in args.exclude.split(","):
            if excluded_target not in test_list_selection:
                logging.error("Target \"{}\" not found in current target list.".format(excluded_target))
                continue
            test_list_selection.remove(excluded_target)
    test_list_selection.sort()

    logging.info("{} of {} detected fuzz target(s) selected: {}".format(len(test_list_selection), len(test_list_all), " ".join(test_list_selection)))

            

Reported by Pylint.

                          test_list_selection.remove(excluded_target)
    test_list_selection.sort()

    logging.info("{} of {} detected fuzz target(s) selected: {}".format(len(test_list_selection), len(test_list_all), " ".join(test_list_selection)))

    if not args.generate:
        test_list_missing_corpus = []
        for t in test_list_selection:
            corpus_path = os.path.join(args.corpus_dir, t)

            

Reported by Pylint.

                      test_list_missing_corpus.sort()
        if test_list_missing_corpus:
            logging.info(
                "Fuzzing harnesses lacking a corpus: {}".format(
                    " ".join(test_list_missing_corpus)
                )
            )
            logging.info("Please consider adding a fuzz corpus at https://github.com/bitcoin-core/qa-assets")


            

Reported by Pylint.

                  Run {targets} without input, and outputs the generated corpus to
    {corpus_dir}.
    """
    logging.info("Generating corpus to {}".format(corpus_dir))

    def job(command, t):
        logging.debug("Running '{}'\n".format(" ".join(command)))
        logging.debug("Command '{}' output:\n'{}'\n".format(
            ' '.join(command),

            

Reported by Pylint.

                  logging.info("Generating corpus to {}".format(corpus_dir))

    def job(command, t):
        logging.debug("Running '{}'\n".format(" ".join(command)))
        logging.debug("Command '{}' output:\n'{}'\n".format(
            ' '.join(command),
            subprocess.run(
                command,
                env=get_fuzz_env(target=t, source_dir=src_dir),

            

Reported by Pylint.

              
    def job(command, t):
        logging.debug("Running '{}'\n".format(" ".join(command)))
        logging.debug("Command '{}' output:\n'{}'\n".format(
            ' '.join(command),
            subprocess.run(
                command,
                env=get_fuzz_env(target=t, source_dir=src_dir),
                check=True,

            

Reported by Pylint.

              

def merge_inputs(*, fuzz_pool, corpus, test_list, src_dir, build_dir, merge_dir):
    logging.info("Merge the inputs from the passed dir into the corpus_dir. Passed dir {}".format(merge_dir))
    jobs = []
    for t in test_list:
        args = [
            os.path.join(build_dir, 'src', 'test', 'fuzz', 'fuzz'),
            '-merge=1',

            

Reported by Pylint.

              
        def job(t, args):
            output = 'Run {} with args {}'.format(t, args)
            result = subprocess.run(
                args,
                env=get_fuzz_env(target=t, source_dir=src_dir),
                stderr=subprocess.PIPE,
                universal_newlines=True,
            )

            

Reported by Pylint.