diff --git a/pymempool/api.py b/pymempool/api.py index 4bf9dc3..cbfa4db 100644 --- a/pymempool/api.py +++ b/pymempool/api.py @@ -1,35 +1,68 @@ import json +import logging import warnings import requests from requests.adapters import HTTPAdapter from requests.packages import urllib3 +logger = logging.getLogger(__name__) + class MempoolAPI: - __API_URL_BASE = "https://mempool.space/api/" + __API_URL_BASE = [ + "https://mempool.space/api/", + "https://mempool.emzy.de/api/", + "https://mempool.bitcoin-21.org/api/", + ] def __init__( - self, api_base_url=__API_URL_BASE, retries=5, request_verify=True, proxies=None + self, api_base_url=__API_URL_BASE, retries=3, request_verify=True, proxies=None ): - self.api_base_url = api_base_url + self.set_api_base_url(api_base_url) self.proxies = proxies - self.request_timeout = 120 + self.connect_timeout = 1 + self.reading_timeout = 120 + self.sending_timeout = 120 self.request_verify = request_verify if not request_verify: warnings.filterwarnings('ignore', message='Unverified HTTPS request') self.session = requests.Session() retries = urllib3.util.retry.Retry( - total=retries, backoff_factor=0.5, status_forcelist=[502, 503, 504, 429] + total=retries, backoff_factor=0.1, status_forcelist=[502, 503, 504, 429] ) self.session.mount('https://', HTTPAdapter(max_retries=retries)) + def set_api_base_url(self, api_base_url): + if isinstance(api_base_url, list): + self.api_base_url = api_base_url + else: + self.api_base_url = api_base_url.split(",") + + def get_api_base_url(self, index=0): + if len(self.api_base_url) > index: + return self.api_base_url[index] + else: + return self.api_base_url[0] + + def _request(self, url): + index = 0 + while True: + complete_url = f"{self.get_api_base_url(index)}{url}" + try: + return self.__request(complete_url) + except Exception as e: + logger.info(f"Timeout on {complete_url} - {e}") + index += 1 + if index >= len(self.api_base_url): + raise + def __request(self, url): - # print(url) + logger.info(url) try: response = self.session.get( url, - timeout=self.request_timeout, + timeout=(self.connect_timeout, self.reading_timeout), verify=self.request_verify, proxies=self.proxies, ) @@ -57,13 +90,27 @@ def __request(self, url): return response.content.decode('utf-8') raise + def _send(self, url, data): + index = 0 + while True: + complete_url = f"{self.get_api_base_url(index)}{url}" + try: + return self.__send(complete_url, data) + except Exception as e: + logger.info(f"Timeout on {complete_url} - {e}") + index += 1 + if index >= len(self.api_base_url): + raise + def __send(self, url, data): - # print(url) + logger.info(url) try: req = requests.Request('POST', url, data=data) prepped = req.prepare() response = self.session.send( - prepped, timeout=self.request_timeout, verify=self.request_verify + prepped, + timeout=(self.connect_timeout, self.sending_timeout), + verify=self.request_verify, ) except requests.exceptions.RequestException: raise @@ -85,99 +132,97 @@ def __send(self, url, data): def get_difficulty_adjustment(self): """Returns details about difficulty adjustment.""" - api_url = f'{self.api_base_url}v1/difficulty-adjustment' - return self.__request(api_url) + api_url = 'v1/difficulty-adjustment' + return self._request(api_url) def get_address(self, address): """Returns details about an address.""" address = address.replace(' ', '') - api_url = f'{self.api_base_url}address/{address}' - return self.__request(api_url) + api_url = f'address/{address}' + return self._request(api_url) def get_address_transactions(self, address): """Get transaction history for the specified address/scripthash, sorted with newest first.""" address = address.replace(' ', '') - api_url = f'{self.api_base_url}address/{address}/txs' - return self.__request(api_url) + api_url = f'address/{address}/txs' + return self._request(api_url) def get_address_transactions_chain(self, address, last_seen_txid=None): """Get confirmed transaction history for the specified address/scripthash, sorted with newest first.""" address = address.replace(' ', '') if last_seen_txid is None: - api_url = f'{self.api_base_url}address/{address}/txs/chain' + api_url = f'address/{address}/txs/chain' else: - api_url = '{}address/{}/txs/chain/{}'.format( - self.api_base_url, address, last_seen_txid - ) - return self.__request(api_url) + api_url = f'address/{address}/txs/chain/{last_seen_txid}' + return self._request(api_url) def get_address_transactions_mempool(self, address): """Get unconfirmed transaction history for the specified address/scripthash.""" address = address.replace(' ', '') - api_url = f'{self.api_base_url}address/{address}/txs/mempool' - return self.__request(api_url) + api_url = f'address/{address}/txs/mempool' + return self._request(api_url) def get_address_utxo(self, address): """Get the list of unspent transaction outputs associated with the address/scripthash.""" address = address.replace(' ', '') - api_url = f'{self.api_base_url}address/{address}/utxo' - return self.__request(api_url) + api_url = f'address/{address}/utxo' + return self._request(api_url) def get_block(self, hash_value): """Returns details about a block.""" hash_value = hash_value.replace(' ', '') - api_url = f'{self.api_base_url}block/{hash_value}' - return self.__request(api_url) + api_url = f'block/{hash_value}' + return self._request(api_url) def get_block_header(self, hash_value): """Returns the hex-encoded block header.""" hash_value = hash_value.replace(' ', '') - api_url = f'{self.api_base_url}block/{hash_value}/header' - return self.__request(api_url) + api_url = f'block/{hash_value}/header' + return self._request(api_url) def get_block_height(self, height): """Returns the hash of the block currently at height.""" height = int(height) - api_url = f'{self.api_base_url}block-height/{height}' - return self.__request(api_url) + api_url = f'block-height/{height}' + return self._request(api_url) def get_block_raw(self, hash_value): """Returns the raw block representation in binary.""" hash_value = hash_value.replace(' ', '') - api_url = f'{self.api_base_url}block/{hash_value}/raw' - return self.__request(api_url) + api_url = f'block/{hash_value}/raw' + return self._request(api_url) def get_block_status(self, hash_value): """Returns the confirmation status of a block.""" hash_value = hash_value.replace(' ', '') - api_url = f'{self.api_base_url}block/{hash_value}/status' - return self.__request(api_url) + api_url = f'block/{hash_value}/status' + return self._request(api_url) def get_block_tip_height(self): """Returns the height of the last block.""" - api_url = f'{self.api_base_url}blocks/tip/height' - return self.__request(api_url) + api_url = 'blocks/tip/height' + return self._request(api_url) def get_block_tip_hash(self): """Returns the hash of the last block.""" - api_url = f'{self.api_base_url}blocks/tip/hash' - return self.__request(api_url) + api_url = 'blocks/tip/hash' + return self._request(api_url) def get_block_transaction_id(self, hash_value, index): """Returns the transaction at index index within the specified block.""" hash_value = hash_value.replace(' ', '') index = int(index) - api_url = f'{self.api_base_url}block/{hash_value}/txid/{index}' - return self.__request(api_url) + api_url = f'block/{hash_value}/txid/{index}' + return self._request(api_url) def get_block_transaction_ids(self, hash_value): """Returns a list of all txids in the block.""" hash_value = hash_value.replace(' ', '') - api_url = f'{self.api_base_url}block/{hash_value}/txids' - return self.__request(api_url) + api_url = f'block/{hash_value}/txids' + return self._request(api_url) def get_block_transactions(self, hash_value, start_index=None): """Returns a list of transactions in the block (up to 25 transactions beginning @@ -186,21 +231,21 @@ def get_block_transactions(self, hash_value, start_index=None): if start_index is not None: start_index = int(start_index) api_url = '{}block/{}/txs/{}'.format( - self.api_base_url, hash_value, start_index + self.get_api_base_url(), hash_value, start_index ) else: - api_url = f'{self.api_base_url}block/{hash_value}/txs' - return self.__request(api_url) + api_url = f'block/{hash_value}/txs' + return self._request(api_url) def get_blocks(self, start_height=None): """Returns the 10 newest blocks starting at the tip or at :start_height if specified.""" if start_height is None: - api_url = f'{self.api_base_url}v1/blocks' + api_url = 'v1/blocks' else: start_height = int(start_height) - api_url = f'{self.api_base_url}v1/blocks/{start_height}' - return self.__request(api_url) + api_url = f'v1/blocks/{start_height}' + return self._request(api_url) def get_blocks_bulk(self, min_height, max_height=None): """Returns details on the range of blocks between :min_height and :max_height, @@ -210,38 +255,38 @@ def get_blocks_bulk(self, min_height, max_height=None): """ if max_height is None: min_height = int(min_height) - api_url = f'{self.api_base_url}v1/blocks-bulk/{min_height}' + api_url = f'v1/blocks-bulk/{min_height}' else: min_height = int(min_height) max_height = int(max_height) - api_url = f'{self.api_base_url}v1/blocks-bulk/{min_height}/{max_height}' - return self.__request(api_url) + api_url = f'v1/blocks-bulk/{min_height}/{max_height}' + return self._request(api_url) def get_mining_pools(self, time_period): """Returns a list of all known mining pools ordered by blocks found over the specified trailing time_period.""" - api_url = f'{self.api_base_url}v1/mining/pools/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/pools/{time_period}' + return self._request(api_url) def get_mining_pool(self, slug): """Returns details about the mining pool specified by slug.""" - api_url = f'{self.api_base_url}v1/mining/pool/{slug}' - return self.__request(api_url) + api_url = f'v1/mining/pool/{slug}' + return self._request(api_url) def get_mining_pool_hashrates(self, time_period): """Returns average hashrates (and share of total hashrate) of mining pools active in the specified trailing time_period, in descending order of hashrate.""" - api_url = f'{self.api_base_url}v1/mining/hashrate/pools/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/hashrate/pools/{time_period}' + return self._request(api_url) def get_mining_pool_hashrate(self, slug): """Returns all known hashrate data for the mining pool specified by slug. Hashrate values are weekly averages. """ - api_url = f'{self.api_base_url}v1/mining/pool/{slug}/hashrate' - return self.__request(api_url) + api_url = f'v1/mining/pool/{slug}/hashrate' + return self._request(api_url) def get_mining_pool_block(self, slug, block_height=None): """Returns past 10 blocks mined by the specified mining pool (slug) before the @@ -251,27 +296,27 @@ def get_mining_pool_block(self, slug, block_height=None): recent blocks are returned. """ if block_height is None: - api_url = f'{self.api_base_url}v1/mining/pool/{slug}/blocks' + api_url = f'v1/mining/pool/{slug}/blocks' else: - api_url = f'{self.api_base_url}v1/mining/pool/{slug}/blocks/{block_height}' - return self.__request(api_url) + api_url = f'v1/mining/pool/{slug}/blocks/{block_height}' + return self._request(api_url) def get_hashrate(self, time_period=None): """Returns network-wide hashrate and difficulty figures over the specified trailing :timePeriod:""" if time_period is None: - api_url = f'{self.api_base_url}v1/mining/hashrate' + api_url = 'v1/mining/hashrate' else: - api_url = f'{self.api_base_url}v1/mining/hashrate/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/hashrate/{time_period}' + return self._request(api_url) def get_reward_stats(self, block_count): """Returns block reward and total transactions confirmed for the past. :blockCount blocks. """ - api_url = f'{self.api_base_url}v1/mining/reward-stats/{block_count}' - return self.__request(api_url) + api_url = f'v1/mining/reward-stats/{block_count}' + return self._request(api_url) def get_block_fees(self, time_period): """Returns average total fees for blocks in the specified :timePeriod, ordered @@ -280,8 +325,8 @@ def get_block_fees(self, time_period): :timePeriod can be any of the following: 24h, 3d, 1w, 1m, 3m, 6m, 1y, 2y, 3y. """ - api_url = f'{self.api_base_url}v1/mining/blocks/fees/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/blocks/fees/{time_period}' + return self._request(api_url) def get_block_rewards(self, time_period): """Returns average block rewards for blocks in the specified :timePeriod, @@ -290,8 +335,8 @@ def get_block_rewards(self, time_period): :timePeriod can be any of the following: 24h, 3d, 1w, 1m, 3m, 6m, 1y, 2y, 3y. """ - api_url = f'{self.api_base_url}v1/mining/blocks/rewards/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/blocks/rewards/{time_period}' + return self._request(api_url) def get_block_feerates(self, time_period): """Returns average feerate percentiles for blocks in the specified. @@ -299,8 +344,8 @@ def get_block_feerates(self, time_period): :timePeriod, ordered oldest to newest. :timePeriod can be any of the following: 24h, 3d, 1w, 1m, 3m, 6m, 1y, 2y, 3y. """ - api_url = f'{self.api_base_url}v1/mining/blocks/fee-rates/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/blocks/fee-rates/{time_period}' + return self._request(api_url) def get_block_sizes_and_weights(self, time_period): """Returns average size (bytes) and average weight (weight units) for blocks in @@ -309,96 +354,96 @@ def get_block_sizes_and_weights(self, time_period): :timePeriod can be any of the following: 24h, 3d, 1w, 1m, 3m, 6m, 1y, 2y, 3y. """ - api_url = f'{self.api_base_url}v1/mining/blocks/sizes-weights/{time_period}' - return self.__request(api_url) + api_url = f'v1/mining/blocks/sizes-weights/{time_period}' + return self._request(api_url) def get_mempool_blocks_fee(self): """Returns current mempool as projected blocks.""" - api_url = f'{self.api_base_url}v1/fees/mempool-blocks' - return self.__request(api_url) + api_url = 'v1/fees/mempool-blocks' + return self._request(api_url) def get_recommended_fees(self): """Returns our currently suggested fees for new transactions.""" - api_url = f'{self.api_base_url}v1/fees/recommended' - return self.__request(api_url) + api_url = 'v1/fees/recommended' + return self._request(api_url) def get_mempool(self): """Returns current mempool backlog statistics.""" - api_url = f'{self.api_base_url}mempool' - return self.__request(api_url) + api_url = 'mempool' + return self._request(api_url) def get_mempool_transactions_ids(self): """Get the full list of txids in the mempool as an array.""" - api_url = f'{self.api_base_url}mempool/txids' - return self.__request(api_url) + api_url = 'mempool/txids' + return self._request(api_url) def get_mempool_recent(self): """Get a list of the last 10 transactions to enter the mempool.""" - api_url = f'{self.api_base_url}mempool/recent' - return self.__request(api_url) + api_url = 'mempool/recent' + return self._request(api_url) def get_children_pay_for_parents(self, txid): """Returns the ancestors and the best descendant fees for a transaction.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}v1/cpfp/{txid}' - return self.__request(api_url) + api_url = f'v1/cpfp/{txid}' + return self._request(api_url) def get_transaction(self, txid): """Returns details about a transaction.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}' - return self.__request(api_url) + api_url = f'tx/{txid}' + return self._request(api_url) def get_transaction_hex(self, txid): """Returns a transaction serialized as hex.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/hex' - return self.__request(api_url) + api_url = f'tx/{txid}/hex' + return self._request(api_url) def get_transaction_merkleblock_proof(self, txid): """Returns a merkle inclusion proof for the transaction using bitcoind's merkleblock format.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/merkleblock-proof' - return self.__request(api_url) + api_url = f'tx/{txid}/merkleblock-proof' + return self._request(api_url) def get_transaction_merkle_proof(self, txid): """Returns a merkle inclusion proof for the transaction using Electrum's blockchain.transaction.get_merkle format.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/merkle-proof' - return self.__request(api_url) + api_url = f'tx/{txid}/merkle-proof' + return self._request(api_url) def get_transaction_outspend(self, txid, vout): """Returns the spending status of a transaction output.""" txid = txid.replace(' ', '') vout = vout.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/outspend/{vout}' - return self.__request(api_url) + api_url = f'tx/{txid}/outspend/{vout}' + return self._request(api_url) def get_transaction_outspends(self, txid): """Returns the spending status of all transaction outputs.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/outspends' - return self.__request(api_url) + api_url = f'tx/{txid}/outspends' + return self._request(api_url) def get_transaction_raw(self, txid): """Returns a transaction as binary data.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/raw' - return self.__request(api_url) + api_url = f'tx/{txid}/raw' + return self._request(api_url) def get_transaction_status(self, txid): """Returns the confirmation status of a transaction.""" txid = txid.replace(' ', '') - api_url = f'{self.api_base_url}tx/{txid}/status' - return self.__request(api_url) + api_url = f'tx/{txid}/status' + return self._request(api_url) def post_transaction(self, txHex): """Broadcast a raw transaction to the network.""" txHex = txHex.replace(' ', '') - api_url = f'{self.api_base_url}tx' - return self.__send(api_url, txHex) + api_url = 'tx' + return self._send(api_url, txHex) def get_network_stats(self, interval): """Returns network-wide stats such as total number of channels and nodes, total @@ -407,86 +452,86 @@ def get_network_stats(self, interval): Pass one of the following for interval: latest, 24h, 3d, 1w, 1m, 3m, 6m, 1y, 2y, 3y. """ - api_url = f'{self.api_base_url}v1/lightning/statistics/{interval}' - return self.__request(api_url) + api_url = f'v1/lightning/statistics/{interval}' + return self._request(api_url) def get_nodes_channels(self, query): """Returns Lightning nodes and channels that match a full-text, case-insensitive search :query across node aliases, node pubkeys, channel IDs, and short channel IDs.""" - api_url = f'{self.api_base_url}v1/lightning/search?searchText=:{query}' - return self.__request(api_url) + api_url = f'v1/lightning/search?searchText=:{query}' + return self._request(api_url) def get_nodes_in_country(self, country): """Returns a list of Lightning nodes running on clearnet in the requested country, where :country is an ISO Alpha-2 country code.""" - api_url = f'{self.api_base_url}v1/lightning/nodes/country/{country}' - return self.__request(api_url) + api_url = f'v1/lightning/nodes/country/{country}' + return self._request(api_url) def get_node_stat_per_country(self): """Returns aggregate capacity and number of clearnet nodes per country. Capacity figures are in satoshis. """ - api_url = f'{self.api_base_url}v1/lightning/nodes/countries' - return self.__request(api_url) + api_url = 'v1/lightning/nodes/countries' + return self._request(api_url) def get_isp_nodes(self, isp): """Returns a list of nodes hosted by a specified isp, where isp is an ISP's ASN.""" - api_url = f'{self.api_base_url}v1/lightning/nodes/nodes/isp/{isp}' - return self.__request(api_url) + api_url = f'v1/lightning/nodes/nodes/isp/{isp}' + return self._request(api_url) def get_node_stat_per_isp(self): """Returns aggregate capacity, number of nodes, and number of channels per ISP. Capacity figures are in satoshis. """ - api_url = f'{self.api_base_url}v1/lightning/nodes/isp-ranking' - return self.__request(api_url) + api_url = 'v1/lightning/nodes/isp-ranking' + return self._request(api_url) def get_top_100_nodes(self): """Returns two lists of the top 100 nodes: one ordered by liquidity (aggregate channel capacity) and the other ordered by connectivity (number of open channels).""" - api_url = f'{self.api_base_url}v1/lightning/nodes/rankings' - return self.__request(api_url) + api_url = 'v1/lightning/nodes/rankings' + return self._request(api_url) def get_top_100_nodes_by_liquidity(self): """Returns a list of the top 100 nodes by liquidity (aggregate channel capacity).""" - api_url = f'{self.api_base_url}v1/lightning/nodes/rankings/liquidity' - return self.__request(api_url) + api_url = 'v1/lightning/nodes/rankings/liquidity' + return self._request(api_url) def get_top_100_nodes_by_connectivity(self): """Returns a list of the top 100 nodes by connectivity (number of open channels).""" - api_url = f'{self.api_base_url}v1/lightning/nodes/rankings/connectivity' - return self.__request(api_url) + api_url = 'v1/lightning/nodes/rankings/connectivity' + return self._request(api_url) def get_top_100_oldest_nodes(self): """Returns a list of the top 100 oldest nodes.""" - api_url = f'{self.api_base_url}v1/lightning/nodes/rankings/age' - return self.__request(api_url) + api_url = 'v1/lightning/nodes/rankings/age' + return self._request(api_url) def get_node_stats(self, pubkey): """Returns details about a node with the given pubKey.""" - api_url = f'{self.api_base_url}v1/lightning/nodes/{pubkey}' - return self.__request(api_url) + api_url = f'v1/lightning/nodes/{pubkey}' + return self._request(api_url) def get_historical_node_stats(self, pubkey): """Returns details about a node with the given pubKey.""" - api_url = f'{self.api_base_url}v1/lightning/nodes/{pubkey}/statistics' - return self.__request(api_url) + api_url = f'v1/lightning/nodes/{pubkey}/statistics' + return self._request(api_url) def get_channel(self, channelid): """Returns info about a Lightning channel with the given :channelId.""" - api_url = f'{self.api_base_url}v1/lightning/channels/{channelid}' - return self.__request(api_url) + api_url = f'v1/lightning/channels/{channelid}' + return self._request(api_url) def get_channel_from_txid(self, txids): """Returns info about a Lightning channel with the given :channelId.""" - api_url = f'{self.api_base_url}v1/lightning/channels/txids' + api_url = 'v1/lightning/channels/txids' if isinstance(txids, "str"): first = True for txid in txids.split(","): @@ -503,7 +548,7 @@ def get_channel_from_txid(self, txids): first = True else: api_url += f'&txId[]={txid}' - return self.__request(api_url) + return self._request(api_url) def get_channels_from_node_pubkey(self, pubkey, channel_status, index=None): """Returns a list of a node's channels given its :pubKey. @@ -511,19 +556,19 @@ def get_channels_from_node_pubkey(self, pubkey, channel_status, index=None): Ten channels are returned at a time. Use :index for paging. :channelStatus can be open, active, or closed. """ - api_url = f'{self.api_base_url}v1/lightning/channels' + api_url = 'v1/lightning/channels' api_url += f'?pub_key={pubkey}&status={channel_status}' if index is not None: api_url += f'&index={index}' - return self.__request(api_url) + return self._request(api_url) def get_channel_geodata(self): """Returns a list of channels with corresponding node geodata.""" - api_url = f'{self.api_base_url}v1/lightning/channels-geo' - return self.__request(api_url) + api_url = 'v1/lightning/channels-geo' + return self._request(api_url) def get_channel_geodata_for_node(self, pubkey): """Returns a list of channels with corresponding geodata for a node with the given :pubKey.""" - api_url = f'{self.api_base_url}v1/lightning/channels-geo/{pubkey}' - return self.__request(api_url) + api_url = f'v1/lightning/channels-geo/{pubkey}' + return self._request(api_url) diff --git a/pymempool/cli.py b/pymempool/cli.py index 6a84d04..797f397 100644 --- a/pymempool/cli.py +++ b/pymempool/cli.py @@ -12,7 +12,7 @@ app = typer.Typer() console = Console() -state = {"verbose": 3, "api": "https://mempool.space/api/"} +state = {} @app.command() @@ -99,7 +99,11 @@ def block(hash: str): @app.callback() -def main(verbose: int = 3, api: str = "https://mempool.space/api/"): +def main( + verbose: int = 3, + api: str = "https://mempool.space/api/,https://mempool.emzy.de/api/," + "https://mempool.bitcoin-21.org/api/", +): """Python CLI for mempool.space, enjoy.""" # Logging state["verbose"] = verbose diff --git a/tests/test_api.py b/tests/test_api.py index cb19230..2f59a23 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -12,7 +12,7 @@ class TestWrapper(unittest.TestCase): @responses.activate def test_connection_error(self): with pytest.raises(requests.exceptions.ConnectionError): - MempoolAPI().get_block_tip_height() + MempoolAPI(api_base_url="https://mempool.space/api/").get_block_tip_height() @responses.activate def test_failed_height(self): @@ -24,7 +24,7 @@ def test_failed_height(self): # Act Assert with pytest.raises(HTTPError): - MempoolAPI().get_block_tip_height() + MempoolAPI(api_base_url="https://mempool.space/api/").get_block_tip_height() @responses.activate def test_get_adress(self): @@ -38,7 +38,9 @@ def test_get_adress(self): ) # Act - response = MempoolAPI().get_address("1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv") + response = MempoolAPI(api_base_url="https://mempool.space/api/").get_address( + "1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv" + ) self.assertEqual(response, ping_json) @@ -50,7 +52,7 @@ def test_post(self): '{"code":-25,"message":"bad-txns-inputs-missingorspent"}' ) with pytest.raises(ValueError): - MempoolAPI().post_transaction( + MempoolAPI(api_base_url="https://mempool.space/api/").post_transaction( "0200000001fd5b5fcd1cb066c27cfc9fda5428b9be850b81ac440ea51f1ddba2f9871" "89ac1010000008a4730440220686a40e9d2dbffeab4ca1ff66341d06a17806767f12a" "1fc4f55740a7af24c6b5022049dd3c9a85ac6c51fecd5f4baff7782a518781bbdd944" @@ -60,3 +62,110 @@ def test_post(self): "864a9297b20d74c61f4787d71d0000000000001976a9140a59837ccd4df25adc31cd" "ad39be6a8d97557ed688ac00000000" ) + + @responses.activate + def test_difficulty_adjustment(self): + base_api_url = "https://mempool.space/api/" + # Arrange + res_json = {} + responses.add( + responses.GET, + f'{base_api_url}v1/difficulty-adjustment', + json=res_json, + status=200, + ) + + # Act + response = MempoolAPI(api_base_url=base_api_url).get_difficulty_adjustment() + self.assertEqual(response, res_json) + + @responses.activate + def test_address_transactions(self): + base_api_url = "https://mempool.space/api/" + address = "1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv" + # Arrange + res_json = {} + responses.add( + responses.GET, + f'{base_api_url}address/{address}/txs', + json=res_json, + status=200, + ) + + # Act + response = MempoolAPI(api_base_url=base_api_url).get_address_transactions( + address + ) + self.assertEqual(response, res_json) + + @responses.activate + def test_address_transactions_chain(self): + base_api_url = "https://mempool.space/api/" + address = "1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv" + # Arrange + res_json = {} + responses.add( + responses.GET, + f'{base_api_url}address/{address}/txs/chain', + json=res_json, + status=200, + ) + + # Act + response = MempoolAPI(api_base_url=base_api_url).get_address_transactions_chain( + address + ) + self.assertEqual(response, res_json) + + last_seen_txid = ( + "4654a83d953c68ba2c50473a80921bb4e1f01d428b18c65ff0128920865cc314" + ) + res_json2 = {} + responses.add( + responses.GET, + f'{base_api_url}address/{address}/txs/chain/{last_seen_txid}', + json=res_json2, + status=200, + ) + + # Act + response = MempoolAPI(api_base_url=base_api_url).get_address_transactions_chain( + address, last_seen_txid + ) + self.assertEqual(response, res_json2) + + @responses.activate + def test_address_transactions_mempool(self): + base_api_url = "https://mempool.space/api/" + address = "1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv" + # Arrange + res_json = {} + responses.add( + responses.GET, + f'{base_api_url}address/{address}/txs/mempool', + json=res_json, + status=200, + ) + + # Act + response = MempoolAPI( + api_base_url=base_api_url + ).get_address_transactions_mempool(address) + self.assertEqual(response, res_json) + + @responses.activate + def test_address_utxo(self): + base_api_url = "https://mempool.space/api/" + address = "1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv" + # Arrange + res_json = {} + responses.add( + responses.GET, + f'{base_api_url}address/{address}/utxo', + json=res_json, + status=200, + ) + + # Act + response = MempoolAPI(api_base_url=base_api_url).get_address_utxo(address) + self.assertEqual(response, res_json)