9380708004

Committed 05 Jun 2024 08:00AM UTC coverage: 85.118% (-0.2%) from 85.291%

Build Type

push

github

Committed by Cryp Toon

Commit Message

Revert "Merge pull request #395 from HoverHell/deferred-config-dir"

This reverts commit 2c7d9e626, reversing
changes made to 60804c647.

Run Details

64 of 74 new or added lines in 9 files covered. (86.49%)

252 existing lines in 14 files now uncovered.

7893 of 9273 relevant lines covered (85.12%)

1.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.32

/bitcoinlib/services/mempool.py

# -*- coding: utf-8 -*-
#
#    BitcoinLib - Python Cryptocurrency Library
#    mempool.space client
#    © 2021-2023 May - 1200 Web Development <http://1200wd.com/>
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import logging
from datetime import datetime, timezone
from bitcoinlib.main import MAX_TRANSACTIONS
from bitcoinlib.services.baseclient import BaseClient, ClientError
from bitcoinlib.transactions import Transaction
from bitcoinlib.encoding import varstr

PROVIDERNAME = 'mempool'
# Please note: In the Blockstream API, the first couple of Bitcoin blocks are not correctly indexed,
# so transactions from these blocks are missing.

_logger = logging.getLogger(__name__)


class MempoolClient(BaseClient):

    def __init__(self, network, base_url, denominator, *args):
        super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)

    def compose_request(self, function, data='', parameter='', parameter2='', variables=None, post_data='',
                        method='get'):
        url_path = function
        if data:
            url_path += '/' + data
        if parameter:
            url_path += '/' + parameter
        if parameter2:
            url_path += '/' + parameter2
        if variables is None:
            variables = {}
        if self.api_key:
            variables.update({'token': self.api_key})
        return self.request(url_path, variables, method, post_data=post_data)

    def getbalance(self, addresslist):
        balance = 0
        for address in addresslist:
            res = self.compose_request('address', address)
            balance += res['chain_stats']['funded_txo_sum'] - res['chain_stats']['spent_txo_sum']
        return balance

    def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
        res = self.compose_request('address', address, 'utxo')
        utxos = []
        # # key=lambda k: (k[2], pow(10, 20)-k[0].transaction_id, k[3]), reverse=True
        res = sorted(res, key=lambda k: 0 if 'block_height' not in k['status'] else k['status']['block_height'])
        for a in res:
            confirmations = 0
            block_height = None
            if 'block_height' in a['status']:
                block_height = a['status']['block_height']
                confirmations = self.latest_block - block_height
            utxos.append({
                'address': address,
                'txid': a['txid'],
                'confirmations': confirmations,
                'output_n': a['vout'],
                'input_n': 0,
                'block_height': block_height,
                'fee': None,
                'size': 0,
                'value': a['value'],
                'script': '',
                'date': None if 'block_time' not in a['status'] else
                datetime.fromtimestamp(a['status']['block_time'], timezone.utc)
            })
            if a['txid'] == after_txid:
                utxos = []
        return utxos[:limit]

    def _parse_transaction(self, tx):
        block_height = None if 'block_height' not in tx['status'] else tx['status']['block_height']
        confirmations = 0
        tx_date = None
        status = 'unconfirmed'
        if tx['status']['confirmed']:
            if block_height:
                self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
                confirmations = self.latest_block - block_height + 1
            tx_date = datetime.fromtimestamp(tx['status']['block_time'], timezone.utc)
            status = 'confirmed'

        t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network, block_height=block_height,
                        fee=tx['fee'], size=tx['size'], txid=tx['txid'], date=tx_date, confirmations=confirmations,
                        status=status, coinbase=tx['vin'][0]['is_coinbase'])
        for ti in tx['vin']:
            if ti['is_coinbase']:
                t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], unlocking_script=ti['scriptsig'], value=0,
                            sequence=ti['sequence'], strict=self.strict)
            else:
                t.add_input(prev_txid=ti['txid'], output_n=ti['vout'],
                            unlocking_script=ti['scriptsig'], value=ti['prevout']['value'],
                            address=ti['prevout'].get('scriptpubkey_address', ''),
                            locking_script=ti['prevout']['scriptpubkey'], sequence=ti['sequence'],
                            witnesses=None if 'witness' not in ti else [bytes.fromhex(w) for w in ti['witness']],
                            strict=self.strict)
        for to in tx['vout']:
            t.add_output(value=to['value'], address=to.get('scriptpubkey_address', ''), spent=None,
                         lock_script=to['scriptpubkey'], strict=self.strict)
        if 'segwit' in [i.witness_type for i in t.inputs] or 'p2sh-segwit' in [i.witness_type for i in t.inputs]:
            t.witness_type = 'segwit'
        t.update_totals()
        return t

    def gettransaction(self, txid):
        tx = self.compose_request('tx', txid)
        return self._parse_transaction(tx)

    def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        prtxs = []
        before_txid = ''
        while True:
            txs = self.compose_request('address', address, 'txs', before_txid)
            prtxs += txs
            if len(txs) == 25:
                before_txid = txs[-1:][0]['txid']
            else:
                break
            if len(prtxs) > 100:
                break
        txs = []
        for tx in prtxs[::-1]:
            t = self._parse_transaction(tx)
            if t:
                txs.append(t)
            if t.txid == after_txid:
                txs = []
            if len(txs) > limit:
                break
        return txs[:limit]

    def getrawtransaction(self, txid):
        return self.compose_request('tx', txid, 'hex')

    def sendrawtransaction(self, rawtx):
        res = self.compose_request('tx', post_data=rawtx, method='post')
        _logger.debug('mempool response: %s', res)
        return {
            'txid': res,
            'response_dict': {}
        }

    def estimatefee(self, blocks):
        estimates = self.compose_request('v1/fees', 'recommended')
        if blocks < 2:
            return estimates['fastestFee'] * 1000
        elif blocks < 4:
            return estimates['halfHourFee'] * 1000
        if blocks < 7:
            return estimates['hourFee'] * 1000
        else:
            return estimates['minimumFee'] * 1000

    def blockcount(self):
        res = self.compose_request('blocks', 'tip', 'height')
        return res

    def mempool(self, txid=''):
        txids = self.compose_request('mempool', 'txids')
        if not txid:
            return txids
        if txid in txids:
            return [txid]
        return []

    def getblock(self, blockid, parse_transactions, page, limit):
        if isinstance(blockid, int):
            blockid = self.compose_request('block-height', str(blockid))
        if (page == 1 and limit == 10) or limit > 25:
            limit = 25
        bd = self.compose_request('block', blockid)
        btxs = self.compose_request('block', blockid, 'txs', str((page-1)*limit))
        if parse_transactions:
            txs = []
            for tx in btxs[:limit]:
                txs.append(self._parse_transaction(tx))
        else:
            txs = [tx['txid'] for tx in btxs]

        block = {
            'bits': bd['bits'],
            'depth': None,
            'block_hash': bd['id'],
            'height': bd['height'],
            'merkle_root': bd['merkle_root'],
            'nonce': bd['nonce'],
            'prev_block': bd['previousblockhash'],
            'time': bd['timestamp'],
            'tx_count': bd['tx_count'],
            'txs': txs,
            'version': bd['version'],
            'page': page,
            'pages': None if not limit else int(bd['tx_count'] // limit) + (bd['tx_count'] % limit > 0),
            'limit': limit
        }
        return block

    def getrawblock(self, blockid):
        if isinstance(blockid, int):
            blockid = self.compose_request('block-height', str(blockid))
        return self.compose_request('block', blockid, 'raw').hex()

    def isspent(self, txid, output_n):
        res = self.compose_request('tx', txid, 'outspend', str(output_n))
        return 1 if res['spent'] else 0

    # def getinfo(self):

1	# -- coding: utf-8 --
2	#
3	# BitcoinLib - Python Cryptocurrency Library
4	# mempool.space client
5	# © 2021-2023 May - 1200 Web Development <http://1200wd.com/>
6	#
7	# This program is free software: you can redistribute it and/or modify
8	# it under the terms of the GNU Affero General Public License as
9	# published by the Free Software Foundation, either version 3 of the
10	# License, or (at your option) any later version.
11	#
12	# This program is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU Affero General Public License for more details.
16	#
17	# You should have received a copy of the GNU Affero General Public License
18	# along with this program. If not, see <http://www.gnu.org/licenses/>.
19	#
20
21	import logging	2✔
22	from datetime import datetime, timezone	2✔
23	from bitcoinlib.main import MAX_TRANSACTIONS	2✔
24	from bitcoinlib.services.baseclient import BaseClient, ClientError	2✔
25	from bitcoinlib.transactions import Transaction	2✔
26	from bitcoinlib.encoding import varstr	2✔
27
28	PROVIDERNAME = 'mempool'	2✔
29	# Please note: In the Blockstream API, the first couple of Bitcoin blocks are not correctly indexed,
30	# so transactions from these blocks are missing.
31
32	_logger = logging.getLogger(__name__)	2✔
33
34
35	class MempoolClient(BaseClient):	2✔
36
37	def __init__(self, network, base_url, denominator, *args):	2✔
38	super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)	2✔
39
40	def compose_request(self, function, data='', parameter='', parameter2='', variables=None, post_data='',	2✔
41	method='get'):
42	url_path = function	2✔
43	if data:	2✔
44	url_path += '/' + data	2✔
45	if parameter:	2✔
46	url_path += '/' + parameter	2✔
47	if parameter2:	2✔
48	url_path += '/' + parameter2	2✔
49	if variables is None:	2✔
50	variables = {}	2✔
51	if self.api_key:	2✔
52	variables.update({'token': self.api_key})	×
53	return self.request(url_path, variables, method, post_data=post_data)	2✔
54
55	def getbalance(self, addresslist):	2✔
56	balance = 0	1✔
57	for address in addresslist:	1✔
58	res = self.compose_request('address', address)	1✔
59	balance += res['chain_stats']['funded_txo_sum'] - res['chain_stats']['spent_txo_sum']	1✔
60	return balance	1✔
61
62	def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
63	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	1✔
64	res = self.compose_request('address', address, 'utxo')	1✔
65	utxos = []	1✔
66	# # key=lambda k: (k[2], pow(10, 20)-k[0].transaction_id, k[3]), reverse=True
67	res = sorted(res, key=lambda k: 0 if 'block_height' not in k['status'] else k['status']['block_height'])	1✔
68	for a in res:	1✔
69	confirmations = 0	1✔
70	block_height = None	1✔
71	if 'block_height' in a['status']:	1✔
72	block_height = a['status']['block_height']	1✔
73	confirmations = self.latest_block - block_height	1✔
74	utxos.append({	1✔
75	'address': address,
76	'txid': a['txid'],
77	'confirmations': confirmations,
78	'output_n': a['vout'],
79	'input_n': 0,
80	'block_height': block_height,
81	'fee': None,
82	'size': 0,
83	'value': a['value'],
84	'script': '',
85	'date': None if 'block_time' not in a['status'] else
86	datetime.fromtimestamp(a['status']['block_time'], timezone.utc)
87	})
88	if a['txid'] == after_txid:	1✔
UNCOV 89	utxos = []	×
90	return utxos[:limit]	1✔
91
92	def _parse_transaction(self, tx):	2✔
93	block_height = None if 'block_height' not in tx['status'] else tx['status']['block_height']	2✔
94	confirmations = 0	2✔
95	tx_date = None	2✔
96	status = 'unconfirmed'	2✔
97	if tx['status']['confirmed']:	2✔
98	if block_height:	2✔
99	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	2✔
100	confirmations = self.latest_block - block_height + 1	2✔
101	tx_date = datetime.fromtimestamp(tx['status']['block_time'], timezone.utc)	2✔
102	status = 'confirmed'	2✔
103
104	t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network, block_height=block_height,	2✔
105	fee=tx['fee'], size=tx['size'], txid=tx['txid'], date=tx_date, confirmations=confirmations,
106	status=status, coinbase=tx['vin'][0]['is_coinbase'])
107	for ti in tx['vin']:	2✔
108	if ti['is_coinbase']:	2✔
109	t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], unlocking_script=ti['scriptsig'], value=0,	2✔
110	sequence=ti['sequence'], strict=self.strict)
111	else:
112	t.add_input(prev_txid=ti['txid'], output_n=ti['vout'],	2✔
113	unlocking_script=ti['scriptsig'], value=ti['prevout']['value'],
114	address=ti['prevout'].get('scriptpubkey_address', ''),
115	locking_script=ti['prevout']['scriptpubkey'], sequence=ti['sequence'],
116	witnesses=None if 'witness' not in ti else [bytes.fromhex(w) for w in ti['witness']],
117	strict=self.strict)
118	for to in tx['vout']:	2✔
119	t.add_output(value=to['value'], address=to.get('scriptpubkey_address', ''), spent=None,	2✔
120	lock_script=to['scriptpubkey'], strict=self.strict)
121	if 'segwit' in [i.witness_type for i in t.inputs] or 'p2sh-segwit' in [i.witness_type for i in t.inputs]:	2✔
122	t.witness_type = 'segwit'	2✔
123	t.update_totals()	2✔
124	return t	2✔
125
126	def gettransaction(self, txid):	2✔
127	tx = self.compose_request('tx', txid)	2✔
128	return self._parse_transaction(tx)	2✔
129
130	def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
131	prtxs = []	2✔
132	before_txid = ''	2✔
133	while True:	1✔
134	txs = self.compose_request('address', address, 'txs', before_txid)	2✔
135	prtxs += txs	2✔
136	if len(txs) == 25:	2✔
137	before_txid = txs[-1:][0]['txid']	×
138	else:
139	break	1✔
140	if len(prtxs) > 100:	×
141	break	×
142	txs = []	2✔
143	for tx in prtxs[::-1]:	2✔
144	t = self._parse_transaction(tx)	2✔
145	if t:	2✔
146	txs.append(t)	2✔
147	if t.txid == after_txid:	2✔
148	txs = []	1✔
149	if len(txs) > limit:	2✔
UNCOV 150	break	×
151	return txs[:limit]	2✔
152
153	def getrawtransaction(self, txid):	2✔
UNCOV 154	return self.compose_request('tx', txid, 'hex')	×
155
156	def sendrawtransaction(self, rawtx):	2✔
157	res = self.compose_request('tx', post_data=rawtx, method='post')	1✔
158	_logger.debug('mempool response: %s', res)	×
159	return {	×
160	'txid': res,
161	'response_dict': {}
162	}
163
164	def estimatefee(self, blocks):	2✔
165	estimates = self.compose_request('v1/fees', 'recommended')	2✔
166	if blocks < 2:	2✔
167	return estimates['fastestFee'] * 1000	×
168	elif blocks < 4:	2✔
169	return estimates['halfHourFee'] * 1000	2✔
170	if blocks < 7:	1✔
171	return estimates['hourFee'] * 1000	1✔
172	else:
173	return estimates['minimumFee'] * 1000	1✔
174
175	def blockcount(self):	2✔
176	res = self.compose_request('blocks', 'tip', 'height')	2✔
177	return res	2✔
178
179	def mempool(self, txid=''):	2✔
180	txids = self.compose_request('mempool', 'txids')	2✔
181	if not txid:	2✔
182	return txids	1✔
183	if txid in txids:	2✔
184	return [txid]	×
185	return []	2✔
186
187	def getblock(self, blockid, parse_transactions, page, limit):	2✔
188	if isinstance(blockid, int):	2✔
189	blockid = self.compose_request('block-height', str(blockid))	1✔
190	if (page == 1 and limit == 10) or limit > 25:	2✔
191	limit = 25	1✔
192	bd = self.compose_request('block', blockid)	2✔
193	btxs = self.compose_request('block', blockid, 'txs', str((page-1)*limit))	2✔
194	if parse_transactions:	2✔
195	txs = []	2✔
196	for tx in btxs[:limit]:	2✔
197	txs.append(self._parse_transaction(tx))	2✔
198	else:
199	txs = [tx['txid'] for tx in btxs]	1✔
200
201	block = {	2✔
202	'bits': bd['bits'],
203	'depth': None,
204	'block_hash': bd['id'],
205	'height': bd['height'],
206	'merkle_root': bd['merkle_root'],
207	'nonce': bd['nonce'],
208	'prev_block': bd['previousblockhash'],
209	'time': bd['timestamp'],
210	'tx_count': bd['tx_count'],
211	'txs': txs,
212	'version': bd['version'],
213	'page': page,
214	'pages': None if not limit else int(bd['tx_count'] // limit) + (bd['tx_count'] % limit > 0),
215	'limit': limit
216	}
217	return block	2✔
218
219	def getrawblock(self, blockid):	2✔
UNCOV 220	if isinstance(blockid, int):	×
UNCOV 221	blockid = self.compose_request('block-height', str(blockid))	×
UNCOV 222	return self.compose_request('block', blockid, 'raw').hex()	×
223
224	def isspent(self, txid, output_n):	2✔
UNCOV 225	res = self.compose_request('tx', txid, 'outspend', str(output_n))	×
UNCOV 226	return 1 if res['spent'] else 0	×
227
228	# def getinfo(self):

1200wd / bitcoinlib / 9380708004

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous