def sha256(bytestr):
def double_sha256(bytestr):
def double_sha256_checksum(bytestr):
def ripemd160_sha256(bytestr): FIELD_SIZE = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F
GROUP_ORDER = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
TONELLI_SHANKS_CONSTANT = (FIELD_SIZE + 1) // 4
def parity(num):
def x_to_y(x, y_parity): def verify_sig(signature, data, public_key):
    """Verifies some data was signed by the owner of a public key.

    :param signature: The signature to verify.
    :type signature: ``bytes``
    :param data: The data that was supposedly signed.
    :type data: ``bytes``
    :param public_key: The public key.
    :type public_key: ``bytes``
    :returns: ``True`` if all checks pass, ``False`` otherwise.
    """
    return _vs(signature, data, public_key) class NetworkAPI:
    IGNORED_ERRORS = (
        ConnectionError,
        requests.exceptions.ConnectionError,
        requests.exceptions.Timeout,
        requests.exceptions.ReadTimeout,
        ExcessiveAddress,
    )

    GET_BALANCE_MAIN = [
        BlockchairAPI.get_balance,
        BlockstreamAPI.get_balance,
        BitcoreAPI.get_balance,
        SmartbitAPI.get_balance,
        BlockchainAPI.get_balance,
    ]
    GET_TRANSACTIONS_MAIN = [
        BlockchairAPI.get_transactions,  # Limit 1000
        BlockstreamAPI.get_transactions,  # Limit 1000
        SmartbitAPI.get_transactions,  # Limit 1000
        BlockchainAPI.get_transactions,  # No limit, requires multiple requests
    ]
    GET_TRANSACTION_BY_ID_MAIN = [
        BlockchairAPI.get_transaction_by_id,
        BlockstreamAPI.get_transaction_by_id,
        SmartbitAPI.get_transaction_by_id,
        BlockchainAPI.get_transaction_by_id,
    ]
    GET_UNSPENT_MAIN = [
        BlockstreamAPI.get_unspent,
        BlockchairAPI.get_unspent,
        SmartbitAPI.get_unspent,  # Limit 1000
        BlockchainAPI.get_unspent,
        BitcoreAPI.get_unspent,  # No limit
    ]
    BROADCAST_TX_MAIN = [
        BlockchairAPI.broadcast_tx,
        BlockstreamAPI.broadcast_tx,
        BitcoreAPI.broadcast_tx,
        SmartbitAPI.broadcast_tx,  # Limit 5/minute
        BlockchainAPI.broadcast_tx,
    ]

    GET_BALANCE_TEST = [
        BlockchairAPI.get_balance_testnet,
        BlockstreamAPI.get_balance_testnet,
        BitcoreAPI.get_balance_testnet,
        SmartbitAPI.get_balance_testnet,
    ]
    GET_TRANSACTIONS_TEST = [
        BlockchairAPI.get_transactions_testnet,  # Limit 1000
        BlockstreamAPI.get_transactions_testnet,
        SmartbitAPI.get_transactions_testnet,  # Limit 1000
    ]
    GET_TRANSACTION_BY_ID_TEST = [
        BlockchairAPI.get_transaction_by_id_testnet,
        BlockstreamAPI.get_transaction_by_id_testnet,
        SmartbitAPI.get_transaction_by_id_testnet,
    ]
    GET_UNSPENT_TEST = [
        BlockstreamAPI.get_unspent_testnet,
        BlockchairAPI.get_unspent_testnet,
        SmartbitAPI.get_unspent_testnet,  # Limit 1000
        BitcoreAPI.get_unspent_testnet,  # No limit
    ]
    BROADCAST_TX_TEST = [
        BlockchairAPI.broadcast_tx_testnet,
        BlockstreamAPI.broadcast_tx_testnet,
        BitcoreAPI.broadcast_tx_testnet,
        SmartbitAPI.broadcast_tx_testnet,  # Limit 5/minute
    ]

    @classmethod
    def connect_to_node(cls, user, password, host='localhost', port=8332, use_https=False, testnet=False, path=""):
        """Connect to a remote Bitcoin node instead of using web APIs.
        Allows to connect to a testnet and mainnet Bitcoin node simultaneously.

        :param user: The RPC user to a Bitcoin node
        :type user: ``str``
        :param password: The RPC password to a Bitcoin node
        :type password: ``str``
        :param host: The host to a Bitcoin node
        :type host: ``str``
        :param port: The port to a Bitcoin node
        :type port: ``int``
        :param use_https: Connect to the Bitcoin node via HTTPS. Either a
            boolean, in which case it controls whether we connect to the node
            via HTTP or HTTPS, or a string, in which case we connect via HTTPS
            and it must be a path to the CA bundle to use. Defaults to False.
        :type use_https: ``bool`` or ``string``
        :param testnet: Defines if the node should be used for testnet
        :type testnet: ``bool``
        :returns: The node exposing its RPCs for direct interaction.
        :rtype: ``RPCHost``
        """
        node = RPCHost(user=user, password=password, host=host, port=port, use_https=use_https, path=path)

        # Inject remote node into NetworkAPI
        if testnet is False:
            cls.GET_BALANCE_MAIN = [node.get_balance]
            cls.GET_TRANSACTIONS_MAIN = [node.get_transactions]
            cls.GET_TRANSACTION_BY_ID_MAIN = [node.get_transaction_by_id]
            cls.GET_UNSPENT_MAIN = [node.get_unspent]
            cls.BROADCAST_TX_MAIN = [node.broadcast_tx]
        else:
            cls.GET_BALANCE_TEST = [node.get_balance_testnet]
            cls.GET_TRANSACTIONS_TEST = [node.get_transactions_testnet]
            cls.GET_TRANSACTION_BY_ID_TEST = [node.get_transaction_by_id_testnet]
            cls.GET_UNSPENT_TEST = [node.get_unspent_testnet]
            cls.BROADCAST_TX_TEST = [node.broadcast_tx_testnet]

        return node

    @classmethod
    def get_balance(cls, address):
        """Gets the balance of an address in satoshi.

        :param address: The address in question.
        :type address: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``int``
        """

        for api_call in cls.GET_BALANCE_MAIN:
            try:
                return api_call(address)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_balance_testnet(cls, address):
        """Gets the balance of an address on the test network in satoshi.

        :param address: The address in question.
        :type address: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``int``
        """

        for api_call in cls.GET_BALANCE_TEST:
            try:
                return api_call(address)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_transactions(cls, address):
        """Gets the ID of all transactions related to an address.

        :param address: The address in question.
        :type address: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``list`` of ``str``
        """

        for api_call in cls.GET_TRANSACTIONS_MAIN:
            try:
                return api_call(address)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_transactions_testnet(cls, address):
        """Gets the ID of all transactions related to an address on the test
        network.

        :param address: The address in question.
        :type address: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``list`` of ``str``
        """

        for api_call in cls.GET_TRANSACTIONS_TEST:
            try:
                return api_call(address)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_transaction_by_id(cls, txid):
        """Gets a raw transaction hex by its transaction id (txid).

        :param txid: The id of the transaction
        :type txid: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``string``
        """

        for api_call in cls.GET_TRANSACTION_BY_ID_MAIN:
            try:
                return api_call(txid)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_transaction_by_id_testnet(cls, txid):
        """Gets a raw transaction hex by its transaction id (txid) on the test.

        :param txid: The id of the transaction
        :type txid: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``string``
        """

        for api_call in cls.GET_TRANSACTION_BY_ID_TEST:
            try:
                return api_call(txid)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_unspent(cls, address):
        """Gets all unspent transaction outputs belonging to an address.

        :param address: The address in question.
        :type address: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``list`` of :class:`~bit.network.meta.Unspent`
        """

        for api_call in cls.GET_UNSPENT_MAIN:
            try:
                return api_call(address)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def get_unspent_testnet(cls, address):
        """Gets all unspent transaction outputs belonging to an address on the
        test network.

        :param address: The address in question.
        :type address: ``str``
        :raises ConnectionError: If all API services fail.
        :rtype: ``list`` of :class:`~bit.network.meta.Unspent`
        """

        for api_call in cls.GET_UNSPENT_TEST:
            try:
                return api_call(address)
            except cls.IGNORED_ERRORS:
                pass

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def broadcast_tx(cls, tx_hex):  # pragma: no cover
        """Broadcasts a transaction to the blockchain.

        :param tx_hex: A signed transaction in hex form.
        :type tx_hex: ``str``
        :raises ConnectionError: If all API services fail.
        """
        success = None

        for api_call in cls.BROADCAST_TX_MAIN:
            try:
                success = api_call(tx_hex)
                if not success:
                    continue
                return
            except cls.IGNORED_ERRORS:
                pass

        if success is False:
            raise ConnectionError('Transaction broadcast failed, or Unspents were already used.')

        raise ConnectionError('All APIs are unreachable.')

    @classmethod
    def broadcast_tx_testnet(cls, tx_hex):  # pragma: no cover
        """Broadcasts a transaction to the test network's blockchain.

        :param tx_hex: A signed transaction in hex form.
        :type tx_hex: ``str``
        :raises ConnectionError: If all API services fail.
        """
        success = None

        for api_call in cls.BROADCAST_TX_TEST:
            try:
                success = api_call(tx_hex)
                if not success:
                    continue
                return
            except cls.IGNORED_ERRORS:
                pass

        if success is False:
            raise ConnectionError('Transaction broadcast failed, or Unspents were already used.')

        raise ConnectionError('All APIs are unreachable.') class Unspent:
    """Represents an unspent transaction output (UTXO)."""

    __slots__ = ('amount', 'confirmations', 'script', 'txid', 'txindex', 'type', 'vsize', 'segwit', 'sequence')

    def __init__(self, amount, confirmations, script, txid, txindex, type='p2pkh', vsize=None, segwit=None,
                 sequence=int.from_bytes(SEQUENCE, byteorder='little')):
        self.amount = amount
        self.confirmations = confirmations
        self.script = script
        self.txid = txid
        self.txindex = txindex
        self.type = type if type in UNSPENT_TYPES else 'unknown'
        self.vsize = vsize if vsize else UNSPENT_TYPES[self.type]['vsize']
        self.segwit = UNSPENT_TYPES[self.type]['segwit']
        self.sequence = sequence

    def to_dict(self):
        return {attr: getattr(self, attr) for attr in Unspent.__slots__}

    @classmethod
    def from_dict(cls, d):
        return Unspent(**{attr: d[attr] for attr in Unspent.__slots__})

    def __eq__(self, other):
        return (
            self.amount == other.amount
            and self.script == other.script
            and self.txid == other.txid
            and self.txindex == other.txindex
            and self.segwit == other.segwit
            and self.sequence == other.sequence
        )

    def __repr__(self):
        return 'Unspent(amount={}, confirmations={}, script={}, txid={}, txindex={}, segwit={}, sequence={})'.format(
            repr(self.amount),
            repr(self.confirmations),
            repr(self.script),
            repr(self.txid),
            repr(self.txindex),
            repr(self.segwit),
            repr(self.sequence)
        )

    def set_type(self, type, vsize=0):
        self.type = type if type in UNSPENT_TYPES else 'unknown'
        self.vsize = vsize if vsize else UNSPENT_TYPES[self.type]['vsize']
        self.segwit = UNSPENT_TYPES[self.type]['segwit']
        return self

    def opt_in_for_RBF(self):
        if self.sequence > 4294967293:
            self.sequence = 4294967293 def wif_to_key(wif):
    def __init__(self, wif=None):
    def public_key(self):
    def public_point(self):
    def sign(self, data):
    def verify(self, signature, data):
    def pub_to_hex(self):
    def to_hex(self):
    def to_bytes(self):
    def to_der(self):
    def to_pem(self):
    def to_int(self):
    def is_compressed(self):
    def __eq__(self, other):
    def __init__(self, wif=None):
    def address(self):
    def segwit_address(self):
    def scriptcode(self):
    def segwit_scriptcode(self):
    def can_sign_unspent(self, unspent):
    def to_wif(self):
    def balance_as(self, currency):
    def get_balance(self, currency='satoshi'):
    def get_unspents(self):
    def get_transactions(self):
    def create_transaction(
        self,
        outputs,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def send(
        self,
        outputs,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def prepare_transaction(
        cls,
        address,
        outputs,
        compressed=True,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def sign_transaction(self, tx_data, unspents=None):  # pragma: no cover
    def from_hex(cls, hexed):
    def from_bytes(cls, bytestr):
    def from_der(cls, der):
    def from_pem(cls, pem):
    def from_int(cls, num):
    def __repr__(self):
    def __init__(self, wif=None):
    def address(self):
    def segwit_address(self):
    def scriptcode(self):
    def segwit_scriptcode(self):
    def can_sign_unspent(self, unspent):
    def to_wif(self):
    def balance_as(self, currency):
    def get_balance(self, currency='satoshi'):
    def get_unspents(self):
    def get_transactions(self):
    def create_transaction(
        self,
        outputs,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def send(
        self,
        outputs,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def prepare_transaction(
        cls,
        address,
        outputs,
        compressed=True,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def sign_transaction(self, tx_data, unspents=None):  # pragma: no cover
    def from_hex(cls, hexed):
    def from_bytes(cls, bytestr):
    def from_der(cls, der):
    def from_pem(cls, pem):
    def from_int(cls, num):
    def __repr__(self):
    def __init__(self, private_key, public_keys, m):
    def address(self):
    def segwit_address(self):
    def scriptcode(self):
    def segwit_scriptcode(self):
    def can_sign_unspent(self, unspent):
    def sign(self, data):  # pragma: no cover
    def balance_as(self, currency):
    def get_balance(self, currency='satoshi'):
    def get_unspents(self):
    def get_transactions(self):
    def create_transaction(
        self,
        outputs,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def prepare_transaction(
        cls,
        address,
        outputs,
        compressed=True,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def sign_transaction(self, tx_data, unspents=None):  # pragma: no cover
    def __repr__(self):
    def __init__(self, private_key, public_keys, m):
    def address(self):
    def segwit_address(self):
    def scriptcode(self):
    def segwit_scriptcode(self):
    def can_sign_unspent(self, unspent):
    def sign(self, data):  # pragma: no cover
    def balance_as(self, currency):
    def get_balance(self, currency='satoshi'):
    def get_unspents(self):
    def get_transactions(self):
    def create_transaction(
        self,
        outputs,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def prepare_transaction(
        cls,
        address,
        outputs,
        compressed=True,
        fee=None,
        absolute_fee=False,
        leftover=None,
        combine=True,
        message=None,
        unspents=None,
        message_is_hex=False,
        replace_by_fee=False
    ):  # pragma: no cover
    def sign_transaction(self, tx_data, unspents=None):  # pragma: no cover
    def __repr__(self):
class BaseKey:
class PrivateKey(BaseKey):
class PrivateKeyTestnet(BaseKey):
class MultiSig:
class MultiSigTestnet: def bytes_to_hex(bytestr, upper=False):
    hexed = hexlify(bytestr).decode()
    return hexed.upper() if upper else hexed BITCOIN_ADDRESS = '1ELReFsTCUY2mfaDTy32qxYiT49z786eFg' BITCOIN_ADDRESS_TEST = 'mtrNwJxS1VyHYn3qBY1Qfsm3K3kh1mGRMS' BITCOIN_ADDRESS_P2SH_MULTISIG = '3LtE4jSa7MDD1GEJGEBYjntYQB9KcqzRCG' BITCOIN_ADDRESS_NP2SH_MULTISIG = '32syayPneML4mTuMfimt6ZSLPz87Fya1r6' BITCOIN_ADDRESS_TEST_P2SH_MULTISIG = '2NFEYyojDi7eFSZPwBRxSQPn1NPnS9p5upP' BITCOIN_ADDRESS_TEST_NP2SH_MULTISIG = '2NCWeVbWmaUp92dSFP3RddPk6r3GTd6cDd6' PRIVATE_KEY_BYTES = b'\xc2\x8a\x9f\x80s\x8fw\rRx\x03\xa5f\xcfo\xc3\xed\xf6\xce\xa5\x86\xc4\xfcJR#\xa5\xady~\x1a\xc3' PRIVATE_KEY_DER = (
    b"0\x81\x84\x02\x01\x000\x10\x06\x07*\x86H\xce=\x02\x01\x06"
    b"\x05+\x81\x04\x00\n\x04m0k\x02\x01\x01\x04 \xc2\x8a\x9f"
    b"\x80s\x8fw\rRx\x03\xa5f\xcfo\xc3\xed\xf6\xce\xa5\x86\xc4"
    b"\xfcJR#\xa5\xady~\x1a\xc3\xa1D\x03B\x00\x04=\\(u\xc9\xbd"
    b"\x11hu\xa7\x1a]\xb6L\xff\xcb\x139k\x16=\x03\x9b\x1d\x93'"
    b"\x82H\x91\x80C4v\xa45**\xdd\x00\xeb\xb0\xd5\xc9LQ[r\xeb"
    b"\x10\xf1\xfd\x8f?\x03\xb4/J+%[\xfc\x9a\xa9\xe3"
) PRIVATE_KEY_HEX = 'c28a9f80738f770d527803a566cf6fc3edf6cea586c4fc4a5223a5ad797e1ac3' PRIVATE_KEY_NUM = 87993618360805341115891506172036624893404292644470266399436498750715784469187 PRIVATE_KEY_PEM = (
    b'-----BEGIN PRIVATE KEY-----\n'
    b'MIGEAgEAMBAGByqGSM49AgEGBSuBBAAKBG0wawIBAQQgwoqfgHOPdw1SeAOlZs9v\n'
    b'w+32zqWGxPxKUiOlrXl+GsOhRANCAAQ9XCh1yb0RaHWnGl22TP/LEzlrFj0Dmx2T\n'
    b'J4JIkYBDNHakNSoq3QDrsNXJTFFbcusQ8f2PPwO0L0orJVv8mqnj\n'
    b'-----END PRIVATE KEY-----\n'
) PUBKEY_HASH_HEX = '043d5c2875c9bd116875a71a5db64cffcb13396b163d039b1d932782489180433476a4352a2add00ebb0d5c94c515b72eb10f1fd8f3f03b42f4a2b255bfc9aa9e3' PUBLIC_KEY_COMPRESSED = b"\x03=\\(u\xc9\xbd\x11hu\xa7\x1a]\xb6L\xff\xcb\x139k\x16=\x03\x9b\x1d\x93'\x82H\x91\x80C4" PUBLIC_KEY_UNCOMPRESSED = (
    b"\x04=\\(u\xc9\xbd\x11hu\xa7\x1a]\xb6L\xff\xcb\x139k\x16=\x03"
    b"\x9b\x1d\x93'\x82H\x91\x80C4v\xa45**\xdd\x00\xeb\xb0\xd5\xc9"
    b"LQ[r\xeb\x10\xf1\xfd\x8f?\x03\xb4/J+%[\xfc\x9a\xa9\xe3"
) PUBLIC_KEY_X = 27753912938952041417634381842191885283234814940840273460372041880794577257268 PUBLIC_KEY_Y = 53663045980837260634637807506183816949039230809110041985901491152185762425315 WALLET_FORMAT_COMPRESSED_MAIN = 'L3jsepcttyuJK3HKezD4qqRKGtwc8d2d1Nw6vsoPDX9cMcUxqqMv' WALLET_FORMAT_COMPRESSED_TEST = 'cU6s7jckL3bZUUkb3Q2CD9vNu8F1o58K5R5a3JFtidoccMbhEGKZ' WALLET_FORMAT_COMPRESSED_SEND_TEST = os.environ.get('WALLET_FORMAT_COMPRESSED_SEND_TEST', '') WALLET_FORMAT_MAIN = '5KHxtARu5yr1JECrYGEA2YpCPdh1i9ciEgQayAF8kcqApkGzT9s' WALLET_FORMAT_MAIN_1 = 'L3MWPaUtPpnBx7QZtMTrcKz437JFCLzz8GQ916UAZtk3P51w7cpo' WALLET_FORMAT_MAIN_2 = 'Kxat4TMUAv3f7H4g52NKbAyEncj72h3wujGVfoGhU52WdQxrZuFd' WALLET_FORMAT_TEST = '934bTuFSgCv9GHi9Ac84u9NA3J3isK9uadGY3nbe6MaDbnQdcbn' WALLET_FORMAT_SEND_TEST = os.environ.get('WALLET_FORMAT_SEND_TEST', '') WALLET_FORMAT_TEST_1 = 'KzDFE5K1Mb6cP2SAUySLUMx8F8KWn5RYiUFTFXcDMY5X22Jp1MvH' WALLET_FORMAT_TEST_2 = 'KxhDjpU1TWDbD2ukWpTcj9T55bkKVJsyn2hcDucCecpjApetCRG9' BITCOIN_ADDRESS_NP2WKH = '3291hXxutb58vbDVVumaJpopanmfxjVpgJ' BITCOIN_ADDRESS_TEST_NP2WKH = '2MshDmGtwW3aV8Nr3B3PSvmo5o8yqhwJFKZ'
import os
import sys
import time
import pytest
from unittest import mock
from bit.crypto import ECPrivateKey, sha256
from bit.curve import Point
from bit.format import verify_sig
from bit.network import NetworkAPI
from bit.network.meta import Unspent
from bit.wallet import BaseKey, Key, PrivateKey, PrivateKeyTestnet, MultiSig, MultiSigTestnet, wif_to_key
from bit.utils import bytes_to_hex
from .samples import (
    BITCOIN_ADDRESS,
    BITCOIN_ADDRESS_TEST,
    BITCOIN_ADDRESS_P2SH_MULTISIG,
    BITCOIN_ADDRESS_NP2SH_MULTISIG,
    BITCOIN_ADDRESS_TEST_P2SH_MULTISIG,
    BITCOIN_ADDRESS_TEST_NP2SH_MULTISIG,
    PRIVATE_KEY_BYTES,
    PRIVATE_KEY_DER,
    PRIVATE_KEY_HEX,
    PRIVATE_KEY_NUM,
    PRIVATE_KEY_PEM,
    PUBKEY_HASH_HEX,
    PUBLIC_KEY_COMPRESSED,
    PUBLIC_KEY_UNCOMPRESSED,
    PUBLIC_KEY_X,
    PUBLIC_KEY_Y,
    WALLET_FORMAT_COMPRESSED_MAIN,
    WALLET_FORMAT_COMPRESSED_TEST,
    WALLET_FORMAT_COMPRESSED_SEND_TEST,
    WALLET_FORMAT_MAIN,
    WALLET_FORMAT_MAIN_1,
    WALLET_FORMAT_MAIN_2,
    WALLET_FORMAT_TEST,
    WALLET_FORMAT_SEND_TEST,
    WALLET_FORMAT_TEST_1,
    WALLET_FORMAT_TEST_2,
    BITCOIN_ADDRESS_NP2WKH,
    BITCOIN_ADDRESS_TEST_NP2WKH,
)
    def test_send(self):
        if not TRAVIS or sys.version_info[:2] != (3, 7):
            return

        private_key = PrivateKeyTestnet(WALLET_FORMAT_COMPRESSED_SEND_TEST)

        # Monkey patching:
        # Ensures get_unspents() retrieves only non-segwit unspents.
        # Note: Necessary since Insight (Bitpay) testnet API does not support
        # pushing segwit transactions
        class _MonkeyPrivateKeyTestnet(PrivateKeyTestnet):
            segwit_address = None

        private_key.__class__ = _MonkeyPrivateKeyTestnet

        private_key.get_unspents()

        initial = len(private_key.get_transactions())
        current = initial
        tries = 0

        private_key.send([], leftover=private_key.address, fee=10)

        while tries < 180:  # pragma: no cover
            current = len(private_key.get_transactions())
            if current > initial:
                break
            time.sleep(6)
            tries += 1

        assert current > initial

    def test_cold_storage(self):
        if not TRAVIS or sys.version_info[:2] != (3, 7):
            return

        private_key = PrivateKeyTestnet(WALLET_FORMAT_SEND_TEST)
        address = private_key.address

        prepared = PrivateKeyTestnet.prepare_transaction(address, [], leftover=address, fee=10)
        tx_hex = private_key.sign_transaction(prepared)

        initial = len(private_key.get_transactions())
        current = initial
        tries = 0

        NetworkAPI.broadcast_tx_testnet(tx_hex)

        while tries < 180:  # pragma: no cover
            current = len(private_key.get_transactions())
            if current > initial:
                break
            time.sleep(6)
            tries += 1

        assert current > initial

    def test_from_hex(self):
        key = PrivateKeyTestnet.from_hex(PRIVATE_KEY_HEX)
        assert isinstance(key, PrivateKeyTestnet)
        assert key.to_hex() == PRIVATE_KEY_HEX

    def test_from_der(self):
        key = PrivateKeyTestnet.from_der(PRIVATE_KEY_DER)
        assert isinstance(key, PrivateKeyTestnet)
        assert key.to_der() == PRIVATE_KEY_DER

    def test_from_pem(self):
        key = PrivateKeyTestnet.from_pem(PRIVATE_KEY_PEM)
        assert isinstance(key, PrivateKeyTestnet)
        assert key.to_pem() == PRIVATE_KEY_PEM

    def test_from_int(self):
        key = PrivateKeyTestnet.from_int(PRIVATE_KEY_NUM)
        assert isinstance(key, PrivateKeyTestnet)
        assert key.to_int() == PRIVATE_KEY_NUM

    def test_repr(self):
        assert repr(PrivateKeyTestnet(WALLET_FORMAT_MAIN)) == '<PrivateKeyTestnet: mtrNwJxS1VyHYn3qBY1Qfsm3K3kh1mGRMS>'


class TestMultiSig:
    def test_init_default(self):
        key1 = PrivateKey()
        key2 = PrivateKey()
        multisig = MultiSig(key1, [key1.public_key, key2.public_key], 2)
        assert multisig._address is None
        assert multisig.balance == 0
        assert multisig.unspents == []
        assert multisig.transactions == []
        assert multisig.m == 2

        multisig2 = MultiSig(key2, [bytes_to_hex(key1.public_key), bytes_to_hex(key2.public_key)], 2)
        assert multisig2._address is None
        assert multisig2.balance == 0
        assert multisig2.unspents == []
        assert multisig2.transactions == []
        assert multisig2.m == 2

    def test_invalid_init(self):
        key1 = PrivateKey()
        key2 = PrivateKey()

        with pytest.raises(TypeError):
            MultiSig(PrivateKeyTestnet(), [key1.public_key, key2.public_key], 2)

        with pytest.raises(ValueError):
            MultiSig(PrivateKey(), [key1.public_key, key2.public_key], 2)

        with pytest.raises(TypeError):
            # Public keys must be provided in list or set
            MultiSig(key1, (key1.public_key, key2.public_key), 1)

    def test_lexicographical_ordering_pubkeys(self):
        key1 = PrivateKey(WALLET_FORMAT_MAIN_1)
        key2 = PrivateKey(WALLET_FORMAT_MAIN_2)

        multisig = MultiSig(key1, [key2.public_key, key1.public_key], 2)
        assert multisig.public_keys == [key2.public_key, key1.public_key]


        multisig = MultiSig(key1, [key1.public_key, key2.public_key], 2)