"""Call one key import RPC."""
        rescan = self.rescan == Rescan.yes

        assert_equal(self.address["solvable"], True)
        assert_equal(self.address["isscript"], self.address_type == AddressType.p2sh_segwit)
        assert_equal(self.address["iswitness"], self.address_type == AddressType.bech32)
        if self.address["isscript"]:
            assert_equal(self.address["embedded"]["isscript"], False)
            assert_equal(self.address["embedded"]["iswitness"], True)

            

Reported by Pylint.

                      rescan = self.rescan == Rescan.yes

        assert_equal(self.address["solvable"], True)
        assert_equal(self.address["isscript"], self.address_type == AddressType.p2sh_segwit)
        assert_equal(self.address["iswitness"], self.address_type == AddressType.bech32)
        if self.address["isscript"]:
            assert_equal(self.address["embedded"]["isscript"], False)
            assert_equal(self.address["embedded"]["iswitness"], True)


            

Reported by Pylint.

              
        assert_equal(self.address["solvable"], True)
        assert_equal(self.address["isscript"], self.address_type == AddressType.p2sh_segwit)
        assert_equal(self.address["iswitness"], self.address_type == AddressType.bech32)
        if self.address["isscript"]:
            assert_equal(self.address["embedded"]["isscript"], False)
            assert_equal(self.address["embedded"]["iswitness"], True)

        if self.call == Call.single:

            

Reported by Pylint.

                      assert_equal(self.address["solvable"], True)
        assert_equal(self.address["isscript"], self.address_type == AddressType.p2sh_segwit)
        assert_equal(self.address["iswitness"], self.address_type == AddressType.bech32)
        if self.address["isscript"]:
            assert_equal(self.address["embedded"]["isscript"], False)
            assert_equal(self.address["embedded"]["iswitness"], True)

        if self.call == Call.single:
            if self.data == Data.address:

            

Reported by Pylint.

                      assert_equal(self.address["isscript"], self.address_type == AddressType.p2sh_segwit)
        assert_equal(self.address["iswitness"], self.address_type == AddressType.bech32)
        if self.address["isscript"]:
            assert_equal(self.address["embedded"]["isscript"], False)
            assert_equal(self.address["embedded"]["iswitness"], True)

        if self.call == Call.single:
            if self.data == Data.address:
                response = self.node.importaddress(address=self.address["address"], label=self.label, rescan=rescan)

            

Reported by Pylint.

                      assert_equal(self.address["iswitness"], self.address_type == AddressType.bech32)
        if self.address["isscript"]:
            assert_equal(self.address["embedded"]["isscript"], False)
            assert_equal(self.address["embedded"]["iswitness"], True)

        if self.call == Call.single:
            if self.data == Data.address:
                response = self.node.importaddress(address=self.address["address"], label=self.label, rescan=rescan)
            elif self.data == Data.pub:

            

Reported by Pylint.

              
        if self.call == Call.single:
            if self.data == Data.address:
                response = self.node.importaddress(address=self.address["address"], label=self.label, rescan=rescan)
            elif self.data == Data.pub:
                response = self.node.importpubkey(pubkey=self.address["pubkey"], label=self.label, rescan=rescan)
            elif self.data == Data.priv:
                response = self.node.importprivkey(privkey=self.key, label=self.label, rescan=rescan)
            assert_equal(response, None)

            

Reported by Pylint.

              
        if self.call == Call.single:
            if self.data == Data.address:
                response = self.node.importaddress(address=self.address["address"], label=self.label, rescan=rescan)
            elif self.data == Data.pub:
                response = self.node.importpubkey(pubkey=self.address["pubkey"], label=self.label, rescan=rescan)
            elif self.data == Data.priv:
                response = self.node.importprivkey(privkey=self.key, label=self.label, rescan=rescan)
            assert_equal(response, None)

            

Reported by Pylint.

              
        if self.call == Call.single:
            if self.data == Data.address:
                response = self.node.importaddress(address=self.address["address"], label=self.label, rescan=rescan)
            elif self.data == Data.pub:
                response = self.node.importpubkey(pubkey=self.address["pubkey"], label=self.label, rescan=rescan)
            elif self.data == Data.priv:
                response = self.node.importprivkey(privkey=self.key, label=self.label, rescan=rescan)
            assert_equal(response, None)

            

Reported by Pylint.

                          if self.data == Data.address:
                response = self.node.importaddress(address=self.address["address"], label=self.label, rescan=rescan)
            elif self.data == Data.pub:
                response = self.node.importpubkey(pubkey=self.address["pubkey"], label=self.label, rescan=rescan)
            elif self.data == Data.priv:
                response = self.node.importprivkey(privkey=self.key, label=self.label, rescan=rescan)
            assert_equal(response, None)

        elif self.call in (Call.multiaddress, Call.multiscript):

            

Reported by Pylint.

                  def run_test(self):
        self.log.info("Generate blocks to create UTXOs")
        node = self.nodes[0]
        self.privkeys = [node.get_deterministic_priv_key().key]
        self.address = node.get_deterministic_priv_key().address
        self.coins = []
        # The last 100 coinbase transactions are premature
        for b in node.generatetoaddress(200, self.address)[:100]:
            coinbase = node.getblock(blockhash=b, verbosity=2)["tx"][0]

            

Reported by Pylint.

                      self.log.info("Generate blocks to create UTXOs")
        node = self.nodes[0]
        self.privkeys = [node.get_deterministic_priv_key().key]
        self.address = node.get_deterministic_priv_key().address
        self.coins = []
        # The last 100 coinbase transactions are premature
        for b in node.generatetoaddress(200, self.address)[:100]:
            coinbase = node.getblock(blockhash=b, verbosity=2)["tx"][0]
            self.coins.append({

            

Reported by Pylint.

                      node = self.nodes[0]
        self.privkeys = [node.get_deterministic_priv_key().key]
        self.address = node.get_deterministic_priv_key().address
        self.coins = []
        # The last 100 coinbase transactions are premature
        for b in node.generatetoaddress(200, self.address)[:100]:
            coinbase = node.getblock(blockhash=b, verbosity=2)["tx"][0]
            self.coins.append({
                "txid": coinbase["txid"],

            

Reported by Pylint.

                  make_chain,
)

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

    def run_test(self):

            

Reported by Pylint.

                      self.address = node.get_deterministic_priv_key().address
        self.coins = []
        # The last 100 coinbase transactions are premature
        for b in node.generatetoaddress(200, self.address)[:100]:
            coinbase = node.getblock(blockhash=b, verbosity=2)["tx"][0]
            self.coins.append({
                "txid": coinbase["txid"],
                "amount": coinbase["vout"][0]["value"],
                "scriptPubKey": coinbase["vout"][0]["scriptPubKey"],

            

Reported by Pylint.

                      self.test_anc_size_limits()
        self.test_desc_size_limits()

    def test_chain_limits_helper(self, mempool_count, package_count):
        node = self.nodes[0]
        assert_equal(0, node.getmempoolinfo()["size"])
        first_coin = self.coins.pop()
        spk = None
        txid = first_coin["txid"]

            

Reported by Pylint.

                      value = first_coin["amount"]

        for i in range(mempool_count + package_count):
            (tx, txhex, value, spk) = make_chain(node, self.address, self.privkeys, txid, value, 0, spk)
            txid = tx.rehash()
            if i < mempool_count:
                node.sendrawtransaction(txhex)
                assert_equal(node.getrawmempool(verbose=True)[txid]["ancestorcount"], i + 1)
            else:

            

Reported by Pylint.

                      value = first_coin["amount"]

        for i in range(mempool_count + package_count):
            (tx, txhex, value, spk) = make_chain(node, self.address, self.privkeys, txid, value, 0, spk)
            txid = tx.rehash()
            if i < mempool_count:
                node.sendrawtransaction(txhex)
                assert_equal(node.getrawmempool(verbose=True)[txid]["ancestorcount"], i + 1)
            else:

            

Reported by Pylint.

              
        # Clear mempool and check that the package passes now
        node.generate(1)
        assert all([res["allowed"] for res in node.testmempoolaccept(rawtxs=chain_hex)])

    def test_chain_limits(self):
        """Create chains from mempool and package transactions that are longer than 25,
        but only if both in-mempool and in-package transactions are considered together.
        This checks that both mempool and in-package transactions are taken into account when

            

Reported by Bandit.

              
        # Clear mempool and check that the package passes now
        node.generate(1)
        assert all([res["allowed"] for res in node.testmempoolaccept(rawtxs=chain_hex)])

    def test_chain_limits(self):
        """Create chains from mempool and package transactions that are longer than 25,
        but only if both in-mempool and in-package transactions are considered together.
        This checks that both mempool and in-package transactions are taken into account when

            

Reported by Pylint.

              import sys
from typing import List, Optional

import lief
import pixie

from utils import determine_wellknown_cmd

# Debian 8 (Jessie) EOL: 2020. https://wiki.debian.org/DebianReleases#Production_Releases

            

Reported by Pylint.

                  else:
        return ver <= max_versions[lib][arch]

def check_imported_symbols(filename) -> bool:
    elf = pixie.load(filename)
    cppfilt = CPPFilt()
    ok: bool = True

    for symbol in elf.dyn_symbols:

            

Reported by Pylint.

                          ok = False
    return ok

def check_exported_symbols(filename) -> bool:
    elf = pixie.load(filename)
    cppfilt = CPPFilt()
    ok: bool = True
    for symbol in elf.dyn_symbols:
        if not symbol.is_export:

            

Reported by Pylint.

                      ok = False
    return ok

def check_ELF_libraries(filename) -> bool:
    ok: bool = True
    elf = pixie.load(filename)
    for library_name in elf.query_dyn_tags(pixie.DT_NEEDED):
        assert(isinstance(library_name, bytes))
        if library_name.decode() not in ELF_ALLOWED_LIBRARIES:

            

Reported by Pylint.

                          ok = False
    return ok

def check_MACHO_libraries(filename) -> bool:
    ok: bool = True
    binary = lief.parse(filename)
    for dylib in binary.libraries:
        split = dylib.name.split('/')
        if split[-1] not in MACHO_ALLOWED_LIBRARIES:

            

Reported by Pylint.

                          ok = False
    return ok

def check_MACHO_min_os(filename) -> bool:
    binary = lief.parse(filename)
    if binary.build_version.minos == [10,14,0]:
        return True
    return False


            

Reported by Pylint.

                      return True
    return False

def check_MACHO_sdk(filename) -> bool:
    binary = lief.parse(filename)
    if binary.build_version.sdk == [10, 15, 6]:
        return True
    return False


            

Reported by Pylint.

                      return True
    return False

def check_PE_libraries(filename) -> bool:
    ok: bool = True
    binary = lief.parse(filename)
    for dylib in binary.libraries:
        if dylib not in PE_ALLOWED_LIBRARIES:
            print(f'{dylib} is not in ALLOWED_LIBRARIES!')

            

Reported by Pylint.

                          ok = False
    return ok

def check_PE_subsystem_version(filename) -> bool:
    binary = lief.parse(filename)
    major: int = binary.optional_header.major_subsystem_version
    minor: int = binary.optional_header.minor_subsystem_version
    if major == 6 and minor == 1:
        return True

            

Reported by Pylint.

              ]
}

def identify_executable(executable) -> Optional[str]:
    with open(filename, 'rb') as f:
        magic = f.read(4)
    if magic.startswith(b'MZ'):
        return 'PE'
    elif magic.startswith(b'\x7fELF'):

            

Reported by Pylint.

                      assert self.nodes[0].verifychain(4, 0)

    def mine_chain(self):
        self.log.info(f"Generate {HEIGHT} blocks after the genesis block in ten-minute steps")
        for t in range(TIME_GENESIS_BLOCK, TIME_RANGE_END, TIME_RANGE_STEP):
            self.nodes[0].setmocktime(t)
            self.nodes[0].generatetoaddress(1, ADDRESS_BCRT1_P2WSH_OP_TRUE)
        assert_equal(self.nodes[0].getblockchaininfo()['blocks'], HEIGHT)


            

Reported by Pylint.

                      assert_equal(res['transactions'], HEIGHT)
        assert_equal(res['height'], HEIGHT)
        assert_equal(res['txouts'], HEIGHT)
        assert_equal(res['bogosize'], 16800),
        assert_equal(res['bestblock'], node.getblockhash(HEIGHT))
        size = res['disk_size']
        assert size > 6400
        assert size < 64000
        assert_equal(len(res['bestblock']), 64)

            

Reported by Pylint.

                      assert_equal(res2['total_amount'], Decimal('0'))
        assert_equal(res2['height'], 0)
        assert_equal(res2['txouts'], 0)
        assert_equal(res2['bogosize'], 0),
        assert_equal(res2['bestblock'], node.getblockhash(0))
        assert_equal(len(res2['hash_serialized_2']), 64)

        self.log.info("Test gettxoutsetinfo returns the same result after invalidate/reconsider block")
        node.reconsiderblock(b1hash)

            

Reported by Pylint.

              from decimal import Decimal
import http.client
import os
import subprocess

from test_framework.address import ADDRESS_BCRT1_P2WSH_OP_TRUE
from test_framework.blocktools import (
    DERSIG_HEIGHT,
    create_block,

            

Reported by Bandit.

              TIME_RANGE_END = TIME_GENESIS_BLOCK + HEIGHT * TIME_RANGE_STEP


class BlockchainTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.supports_cli = False


            

Reported by Pylint.

              
    def run_test(self):
        self.mine_chain()
        self.restart_node(0, extra_args=['-stopatheight=207', '-prune=1'])  # Set extra args with pruning after rescan is complete

        self._test_getblockchaininfo()
        self._test_getchaintxstats()
        self._test_gettxoutsetinfo()
        self._test_getblockheader()

            

Reported by Pylint.

                      self._test_stopatheight()
        self._test_waitforblockheight()
        self._test_getblock()
        assert self.nodes[0].verifychain(4, 0)

    def mine_chain(self):
        self.log.info(f"Generate {HEIGHT} blocks after the genesis block in ten-minute steps")
        for t in range(TIME_GENESIS_BLOCK, TIME_RANGE_END, TIME_RANGE_STEP):
            self.nodes[0].setmocktime(t)

            

Reported by Bandit.

                      self._test_getblock()
        assert self.nodes[0].verifychain(4, 0)

    def mine_chain(self):
        self.log.info(f"Generate {HEIGHT} blocks after the genesis block in ten-minute steps")
        for t in range(TIME_GENESIS_BLOCK, TIME_RANGE_END, TIME_RANGE_STEP):
            self.nodes[0].setmocktime(t)
            self.nodes[0].generatetoaddress(1, ADDRESS_BCRT1_P2WSH_OP_TRUE)
        assert_equal(self.nodes[0].getblockchaininfo()['blocks'], HEIGHT)

            

Reported by Pylint.

              
    def mine_chain(self):
        self.log.info(f"Generate {HEIGHT} blocks after the genesis block in ten-minute steps")
        for t in range(TIME_GENESIS_BLOCK, TIME_RANGE_END, TIME_RANGE_STEP):
            self.nodes[0].setmocktime(t)
            self.nodes[0].generatetoaddress(1, ADDRESS_BCRT1_P2WSH_OP_TRUE)
        assert_equal(self.nodes[0].getblockchaininfo()['blocks'], HEIGHT)

    def _test_getblockchaininfo(self):

            

Reported by Pylint.

                      assert_greater_than_or_equal(res['pruneheight'], 0)

        # check other pruning fields given that prune=1
        assert res['pruned']
        assert not res['automatic_pruning']

        self.restart_node(0, ['-stopatheight=207'])
        res = self.nodes[0].getblockchaininfo()
        # should have exact keys

            

Reported by Bandit.

                  while True:

        # Log a message to prevent Travis from timing out
        logging.debug("verify-commits: [in-progress] processing commit {}".format(current_commit[:8]))

        if current_commit == verified_root:
            print('There is a valid path from "{}" to {} where all commits are signed!'.format(initial_commit, verified_root))
            sys.exit(0)
        if current_commit == verified_sha512_root:

            

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.
"""Verify commits against a trusted keys list."""
import argparse
import hashlib
import logging
import os

            

Reported by Pylint.

              import hashlib
import logging
import os
import subprocess
import sys
import time

GIT = os.getenv('GIT', 'git')


            

Reported by Bandit.

              
GIT = os.getenv('GIT', 'git')

def tree_sha512sum(commit='HEAD'):
    """Calculate the Tree-sha512 for the commit.

    This is copied from github-merge.py. See https://github.com/bitcoin-core/bitcoin-maintainer-tools."""

    # request metadata for entire tree, recursively

            

Reported by Pylint.

              def tree_sha512sum(commit='HEAD'):
    """Calculate the Tree-sha512 for the commit.

    This is copied from github-merge.py. See https://github.com/bitcoin-core/bitcoin-maintainer-tools."""

    # request metadata for entire tree, recursively
    files = []
    blob_by_name = {}
    for line in subprocess.check_output([GIT, 'ls-tree', '--full-tree', '-r', commit]).splitlines():

            

Reported by Pylint.

              def tree_sha512sum(commit='HEAD'):
    """Calculate the Tree-sha512 for the commit.

    This is copied from github-merge.py. See https://github.com/bitcoin-core/bitcoin-maintainer-tools."""

    # request metadata for entire tree, recursively
    files = []
    blob_by_name = {}
    for line in subprocess.check_output([GIT, 'ls-tree', '--full-tree', '-r', commit]).splitlines():

            

Reported by Pylint.

                  # request metadata for entire tree, recursively
    files = []
    blob_by_name = {}
    for line in subprocess.check_output([GIT, 'ls-tree', '--full-tree', '-r', commit]).splitlines():
        name_sep = line.index(b'\t')
        metadata = line[:name_sep].split()  # perms, 'blob', blobid
        assert metadata[1] == b'blob'
        name = line[name_sep + 1:]
        files.append(name)

            

Reported by Bandit.

                  for line in subprocess.check_output([GIT, 'ls-tree', '--full-tree', '-r', commit]).splitlines():
        name_sep = line.index(b'\t')
        metadata = line[:name_sep].split()  # perms, 'blob', blobid
        assert metadata[1] == b'blob'
        name = line[name_sep + 1:]
        files.append(name)
        blob_by_name[name] = metadata[2]

    files.sort()

            

Reported by Bandit.

                  files.sort()
    # open connection to git-cat-file in batch mode to request data for all blobs
    # this is much faster than launching it per file
    p = subprocess.Popen([GIT, 'cat-file', '--batch'], stdout=subprocess.PIPE, stdin=subprocess.PIPE)
    overall = hashlib.sha512()
    for f in files:
        blob = blob_by_name[f]
        # request blob
        p.stdin.write(blob + b'\n')

            

Reported by Pylint.

                  files.sort()
    # open connection to git-cat-file in batch mode to request data for all blobs
    # this is much faster than launching it per file
    p = subprocess.Popen([GIT, 'cat-file', '--batch'], stdout=subprocess.PIPE, stdin=subprocess.PIPE)
    overall = hashlib.sha512()
    for f in files:
        blob = blob_by_name[f]
        # request blob
        p.stdin.write(blob + b'\n')

            

Reported by Pylint.

                      peerinfo = self.nodes[0].getpeerinfo()[0]
        addrs_processed = peerinfo['addr_processed']
        addrs_rate_limited = peerinfo['addr_rate_limited']
        self.log.debug(f"addrs_processed = {addrs_processed}, addrs_rate_limited = {addrs_rate_limited}")

        if no_relay:
            assert_equal(addrs_processed, 0)
            assert_equal(addrs_rate_limited, 0)
        else:

            

Reported by Pylint.

                      self.nodes[0].setmocktime(self.mocktime)

        for contype, no_relay in [("outbound-full-relay", False), ("block-relay-only", True), ("inbound", False)]:
            self.log.info(f'Test rate limiting of addr processing for {contype} peers')
            if contype == "inbound":
                peer = self.nodes[0].add_p2p_connection(AddrReceiver())
            else:
                peer = self.nodes[0].add_outbound_p2p_connection(AddrReceiver(), p2p_idx=0, connection_type=contype)


            

Reported by Pylint.

              )
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_greater_than
import random
import time


class AddrReceiver(P2PInterface):
    num_ipv4_received = 0

            

Reported by Pylint.

              from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_greater_than
import random
import time


class AddrReceiver(P2PInterface):
    num_ipv4_received = 0
    test_addr_contents = False

            

Reported by Pylint.

              import time


class AddrReceiver(P2PInterface):
    num_ipv4_received = 0
    test_addr_contents = False
    _tokens = 1
    send_getaddr = True


            

Reported by Pylint.

                  def on_addr(self, message):
        for addr in message.addrs:
            self.num_ipv4_received += 1
            if(self.test_addr_contents):
                # relay_tests checks the content of the addr messages match
                # expectations based on the message creation in setup_addr_msg
                assert_equal(addr.nServices, 9)
                if not 8333 <= addr.port < 8343:
                    raise AssertionError("Invalid addr.port of {} (8333-8342 expected)".format(addr.port))

            

Reported by Pylint.

                              # expectations based on the message creation in setup_addr_msg
                assert_equal(addr.nServices, 9)
                if not 8333 <= addr.port < 8343:
                    raise AssertionError("Invalid addr.port of {} (8333-8342 expected)".format(addr.port))
                assert addr.ip.startswith('123.123.123.')

    def on_getaddr(self, message):
        # When the node sends us a getaddr, it increments the addr relay tokens for the connection by 1000
        self._tokens += 1000

            

Reported by Pylint.

                              assert_equal(addr.nServices, 9)
                if not 8333 <= addr.port < 8343:
                    raise AssertionError("Invalid addr.port of {} (8333-8342 expected)".format(addr.port))
                assert addr.ip.startswith('123.123.123.')

    def on_getaddr(self, message):
        # When the node sends us a getaddr, it increments the addr relay tokens for the connection by 1000
        self._tokens += 1000


            

Reported by Bandit.

                              assert addr.ip.startswith('123.123.123.')

    def on_getaddr(self, message):
        # When the node sends us a getaddr, it increments the addr relay tokens for the connection by 1000
        self._tokens += 1000

    @property
    def tokens(self):
        with p2p_lock:

            

Reported by Pylint.

                      self._tokens += 1000

    @property
    def tokens(self):
        with p2p_lock:
            return self._tokens

    def increment_tokens(self, n):
        # When we move mocktime forward, the node increments the addr relay tokens for its peers

            

Reported by Pylint.

                      txid = int(ctx.rehash(), 16)

        self.log.info(
            "Announce the transaction to all nodes from all {} incoming peers, but never send it".format(NUM_INBOUND))
        msg = msg_inv([CInv(t=MSG_TX, h=txid)])
        for p in self.peers:
            p.send_and_ping(msg)

        self.log.info("Put the tx in node 0's mempool")

            

Reported by Pylint.

                      # * 2 seconds to avoid races
        assert self.nodes[1].getpeerinfo()[0]['inbound'] == False
        timeout = 2 + INBOUND_PEER_TX_DELAY + GETDATA_TX_INTERVAL
        self.log.info("Tx should be received at node 1 after {} seconds".format(timeout))
        self.sync_mempools(timeout=timeout)

    def test_in_flight_max(self):
        self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format(MAX_GETDATA_IN_FLIGHT))
        txids = [i for i in range(MAX_GETDATA_IN_FLIGHT + 2)]

            

Reported by Pylint.

                      self.sync_mempools(timeout=timeout)

    def test_in_flight_max(self):
        self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format(MAX_GETDATA_IN_FLIGHT))
        txids = [i for i in range(MAX_GETDATA_IN_FLIGHT + 2)]

        p = self.nodes[0].p2ps[0]

        with p2p_lock:

            

Reported by Pylint.

                      for i in range(MAX_GETDATA_IN_FLIGHT, len(txids)):
            p.send_message(msg_inv([CInv(t=MSG_WTX, h=txids[i])]))
        p.sync_with_ping()
        self.log.info("No more than {} requests should be seen within {} seconds after announcement".format(MAX_GETDATA_IN_FLIGHT, INBOUND_PEER_TX_DELAY + OVERLOADED_PEER_DELAY - 1))
        self.nodes[0].setmocktime(mock_time + INBOUND_PEER_TX_DELAY + OVERLOADED_PEER_DELAY - 1)
        p.sync_with_ping()
        with p2p_lock:
            assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT)
        self.log.info("If we wait {} seconds after announcement, we should eventually get more requests".format(INBOUND_PEER_TX_DELAY + OVERLOADED_PEER_DELAY))

            

Reported by Pylint.

                      p.sync_with_ping()
        with p2p_lock:
            assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT)
        self.log.info("If we wait {} seconds after announcement, we should eventually get more requests".format(INBOUND_PEER_TX_DELAY + OVERLOADED_PEER_DELAY))
        self.nodes[0].setmocktime(mock_time + INBOUND_PEER_TX_DELAY + OVERLOADED_PEER_DELAY)
        p.wait_until(lambda: p.tx_getdata_count == len(txids))

    def test_expiry_fallback(self):
        self.log.info('Check that expiry will select another peer for download')

            

Reported by Pylint.

                      # One of the peers is asked for the tx
        peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
        with p2p_lock:
            peer_expiry, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
            assert_equal(peer_fallback.tx_getdata_count, 0)
        self.nodes[0].setmocktime(int(time.time()) + GETDATA_TX_INTERVAL + 1)  # Wait for request to peer_expiry to expire
        peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
        self.restart_node(0)  # reset mocktime


            

Reported by Pylint.

                          self.log.info('Check invs from preferred peers are downloaded immediately')
            self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1'])
        else:
            self.log.info('Check invs from non-preferred peers are downloaded after {} s'.format(NONPREF_PEER_TX_DELAY))
        mock_time = int(time.time() + 1)
        self.nodes[0].setmocktime(mock_time)
        peer = self.nodes[0].add_p2p_connection(TestP2PConn())
        peer.send_message(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
        peer.sync_with_ping()

            

Reported by Pylint.

                          peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)

    def test_txid_inv_delay(self, glob_wtxid=False):
        self.log.info('Check that inv from a txid-relay peers are delayed by {} s, with a wtxid peer {}'.format(TXID_RELAY_DELAY, glob_wtxid))
        self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1'])
        mock_time = int(time.time() + 1)
        self.nodes[0].setmocktime(mock_time)
        peer = self.nodes[0].add_p2p_connection(TestP2PConn(wtxidrelay=False))
        if glob_wtxid:

            

Reported by Pylint.

                          self.start_nodes()
            self.connect_nodes(1, 0)
            # Setup the p2p connections
            self.peers = []
            for node in self.nodes:
                for _ in range(NUM_INBOUND):
                    self.peers.append(node.add_p2p_connection(TestP2PConn()))
            self.log.info("Nodes are setup with {} incoming connections each".format(NUM_INBOUND))
            test()

            

Reported by Pylint.

                          for node in self.nodes:
                for _ in range(NUM_INBOUND):
                    self.peers.append(node.add_p2p_connection(TestP2PConn()))
            self.log.info("Nodes are setup with {} incoming connections each".format(NUM_INBOUND))
            test()


if __name__ == '__main__':
    TxDownloadTest().main()

            

Reported by Pylint.

              import enum
import unittest

from .script import hash256, hash160, sha256, CScript, OP_0
from .segwit_addr import encode_segwit_address
from .util import assert_equal

ADDRESS_BCRT1_UNSPENDABLE = 'bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj'
ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR = 'addr(bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj)#juyq9d97'

            

Reported by Pylint.

              import unittest

from .script import hash256, hash160, sha256, CScript, OP_0
from .segwit_addr import encode_segwit_address
from .util import assert_equal

ADDRESS_BCRT1_UNSPENDABLE = 'bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj'
ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR = 'addr(bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj)#juyq9d97'
# Coins sent to this address can be spent with a witness stack of just OP_TRUE

            

Reported by Pylint.

              
from .script import hash256, hash160, sha256, CScript, OP_0
from .segwit_addr import encode_segwit_address
from .util import assert_equal

ADDRESS_BCRT1_UNSPENDABLE = 'bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj'
ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR = 'addr(bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj)#juyq9d97'
# Coins sent to this address can be spent with a witness stack of just OP_TRUE
ADDRESS_BCRT1_P2WSH_OP_TRUE = 'bcrt1qft5p2uhsdcdc3l2ua4ap5qqfg4pjaqlp250x7us7a8qqhrxrxfsqseac85'

            

Reported by Pylint.

              
def byte_to_base58(b, version):
    result = ''
    str = b.hex()
    str = chr(version).encode('latin-1').hex() + str
    checksum = hash256(bytes.fromhex(str)).hex()
    str += checksum[:8]
    value = int('0x' + str, 0)
    while value > 0:

            

Reported by Pylint.

                  return res[1:-4], int(res[0])


def keyhash_to_p2pkh(hash, main=False):
    assert len(hash) == 20
    version = 0 if main else 111
    return byte_to_base58(hash, version)

def scripthash_to_p2sh(hash, main=False):

            

Reported by Pylint.

                  version = 0 if main else 111
    return byte_to_base58(hash, version)

def scripthash_to_p2sh(hash, main=False):
    assert len(hash) == 20
    version = 5 if main else 196
    return byte_to_base58(hash, version)

def key_to_p2pkh(key, main=False):

            

Reported by Pylint.

              from .util import assert_equal

ADDRESS_BCRT1_UNSPENDABLE = 'bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj'
ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR = 'addr(bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq3xueyj)#juyq9d97'
# Coins sent to this address can be spent with a witness stack of just OP_TRUE
ADDRESS_BCRT1_P2WSH_OP_TRUE = 'bcrt1qft5p2uhsdcdc3l2ua4ap5qqfg4pjaqlp250x7us7a8qqhrxrxfsqseac85'


class AddressType(enum.Enum):

            

Reported by Pylint.

              ADDRESS_BCRT1_P2WSH_OP_TRUE = 'bcrt1qft5p2uhsdcdc3l2ua4ap5qqfg4pjaqlp250x7us7a8qqhrxrxfsqseac85'


class AddressType(enum.Enum):
    bech32 = 'bech32'
    p2sh_segwit = 'p2sh-segwit'
    legacy = 'legacy'  # P2PKH



            

Reported by Pylint.

              

class AddressType(enum.Enum):
    bech32 = 'bech32'
    p2sh_segwit = 'p2sh-segwit'
    legacy = 'legacy'  # P2PKH


chars = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'

            

Reported by Pylint.

              
class AddressType(enum.Enum):
    bech32 = 'bech32'
    p2sh_segwit = 'p2sh-segwit'
    legacy = 'legacy'  # P2PKH


chars = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'


            

Reported by Pylint.

              import abc

from typing import Optional
from test_framework.messages import (
    COutPoint,
    CTransaction,
    CTxIn,
    CTxOut,
    MAX_MONEY,

            

Reported by Pylint.

                  CTxOut,
    MAX_MONEY,
)
from test_framework.blocktools import create_tx_with_script, MAX_BLOCK_SIGOPS
from test_framework.script import (
    CScript,
    OP_0,
    OP_2DIV,
    OP_2MUL,

            

Reported by Pylint.

                  MAX_MONEY,
)
from test_framework.blocktools import create_tx_with_script, MAX_BLOCK_SIGOPS
from test_framework.script import (
    CScript,
    OP_0,
    OP_2DIV,
    OP_2MUL,
    OP_AND,

            

Reported by Pylint.

                  OP_TRUE,
    OP_XOR,
)
from test_framework.script_util import (
    script_to_p2sh_script,
)
basic_p2sh = script_to_p2sh_script(CScript([OP_0]))



            

Reported by Pylint.

                  @abc.abstractmethod
    def get_tx(self, *args, **kwargs):
        """Return a CTransaction that is invalid per the subclass."""
        pass


class OutputMissing(BadTxTemplate):
    reject_reason = "bad-txns-vout-empty"
    expect_disconnect = True

            

Reported by Pylint.

                  reject_reason = "bad-txns-vout-empty"
    expect_disconnect = True

    def get_tx(self):
        tx = CTransaction()
        tx.vin.append(self.valid_txin)
        tx.calc_sha256()
        return tx


            

Reported by Pylint.

                  # Otherwise the transaction will fail the
    # "surpufluous witness" check during deserialization
    # rather than the input count check.
    def get_tx(self):
        tx = CTransaction()
        tx.calc_sha256()
        return tx



            

Reported by Pylint.

                  expect_disconnect = False
    valid_in_block = True

    def get_tx(self):
        tx = CTransaction()
        tx.vin.append(self.valid_txin)
        tx.vout.append(CTxOut(0, CScript([OP_TRUE])))
        tx.calc_sha256()
        return tx

            

Reported by Pylint.

                  reject_reason = None
    expect_disconnect = False

    def get_tx(self):
        num_indices = len(self.spend_tx.vin)
        bad_idx = num_indices + 100

        tx = CTransaction()
        tx.vin.append(CTxIn(COutPoint(self.spend_tx.sha256, bad_idx), b"", 0xffffffff))

            

Reported by Pylint.

                  reject_reason = 'bad-txns-inputs-duplicate'
    expect_disconnect = True

    def get_tx(self):
        tx = CTransaction()
        tx.vin.append(self.valid_txin)
        tx.vin.append(self.valid_txin)
        tx.vout.append(CTxOut(1, basic_p2sh))
        tx.calc_sha256()

            

Reported by Pylint.

              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.