Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_failover_good(caplog):
daemon = Daemon(coin, ','.join(urls))
with caplog.at_level(logging.INFO):
result = daemon.failover()
assert result is True
assert daemon.current_url() == urls[1]
logged_url = daemon.logged_url()
assert in_caplog(caplog, f'failing over to {logged_url}')
# And again
result = daemon.failover()
assert result is True
assert daemon.current_url() == urls[0]
async def test_set_urls_bad():
with pytest.raises(CoinError):
Daemon(coin, '')
with pytest.raises(CoinError):
Daemon(coin, 'a')
async def test_set_urls_two(caplog):
with caplog.at_level(logging.INFO):
daemon = Daemon(coin, ','.join(urls))
assert daemon.current_url() == urls[0]
assert len(daemon.urls) == 2
logged_url = daemon.logged_url()
assert logged_url == '127.0.0.1:8332/'
assert in_caplog(caplog, f'daemon #1 at {logged_url} (current)')
assert in_caplog(caplog, 'daemon #2 at 192.168.0.1:8332')
return aiohttp.TCPConnector(verify_ssl=False)
class PreLegacyRPCDaemon(LegacyRPCDaemon):
'''Handles connections to a daemon at the given URL.
This class is useful for daemons that don't have the new 'getblock'
RPC call that returns the block in hex, and need the False parameter
for the getblock'''
async def deserialised_block(self, hex_hash):
'''Return the deserialised block with the given hex hash.'''
return await self._send_single('getblock', (hex_hash, False))
class SmartCashDaemon(Daemon):
async def masternode_broadcast(self, params):
'''Broadcast a smartnode to the network.'''
return await self._send_single('smartnodebroadcast', params)
async def masternode_list(self, params):
'''Return the smartnode status.'''
return await self._send_single('smartnodelist', params)
async def smartrewards(self, params):
'''Return smartrewards data.'''
return await self._send_single('smartrewards', params)
class ZcoinMtpDaemon(Daemon):
else:
raw_block += b'\x00'
return raw_block
def timestamp_safe(self, t):
if isinstance(t, int):
return t
return timegm(time.strptime(t, "%Y-%m-%d %H:%M:%S %Z"))
class FakeEstimateLegacyRPCDaemon(LegacyRPCDaemon, FakeEstimateFeeDaemon):
pass
class DecredDaemon(Daemon):
async def raw_blocks(self, hex_hashes):
'''Return the raw binary blocks with the given hex hashes.'''
params_iterable = ((h, False) for h in hex_hashes)
blocks = await self._send_vector('getblock', params_iterable)
raw_blocks = []
valid_tx_tree = {}
for block in blocks:
# Convert to bytes from hex
raw_block = hex_to_bytes(block)
raw_blocks.append(raw_block)
# Check if previous block is valid
prev = self.prev_hex_hash(raw_block)
votebits = unpack_le_uint16_from(raw_block[100:102])[0]
valid_tx_tree[prev] = self.is_valid_tx_tree(votebits)
'''Broadcast a transaction to the network.'''
return await self._send_single('sendrawtransaction', (raw_tx, ))
async def height(self):
'''Query the daemon for its current height.'''
self._height = await self._send_single('getblockcount')
return self._height
def cached_height(self):
'''Return the cached daemon height.
If the daemon has not been queried yet this returns None.'''
return self._height
class DashDaemon(Daemon):
async def masternode_broadcast(self, params):
'''Broadcast a transaction to the network.'''
return await self._send_single('masternodebroadcast', params)
async def masternode_list(self, params):
'''Return the masternode status.'''
return await self._send_single('masternodelist', params)
async def protx(self, params):
'''Set of commands to execute ProTx related actions.'''
return await self._send_single('protx', params)
class FakeEstimateFeeDaemon(Daemon):
'''Daemon that simulates estimatefee and relayfee RPC calls. Coin that
class SmartCashDaemon(Daemon):
async def masternode_broadcast(self, params):
'''Broadcast a smartnode to the network.'''
return await self._send_single('smartnodebroadcast', params)
async def masternode_list(self, params):
'''Return the smartnode status.'''
return await self._send_single('smartnodelist', params)
async def smartrewards(self, params):
'''Return smartrewards data.'''
return await self._send_single('smartrewards', params)
class ZcoinMtpDaemon(Daemon):
def strip_mtp_data(self, raw_block):
if self.coin.is_mtp(raw_block):
return \
raw_block[:self.coin.MTP_HEADER_DATA_START*2] + \
raw_block[self.coin.MTP_HEADER_DATA_END*2:]
return raw_block
async def raw_blocks(self, hex_hashes):
'''Return the raw binary blocks with the given hex hashes.'''
params_iterable = ((h, False) for h in hex_hashes)
blocks = await self._send_vector('getblock', params_iterable)
# Convert hex string to bytes
return [hex_to_bytes(self.strip_mtp_data(block)) for block in blocks]
class FakeEstimateFeeDaemon(Daemon):
'''Daemon that simulates estimatefee and relayfee RPC calls. Coin that
wants to use this daemon must define ESTIMATE_FEE & RELAY_FEE'''
async def estimatefee(self, block_count):
'''Return the fee estimate for the given parameters.'''
return self.coin.ESTIMATE_FEE
async def relayfee(self):
'''The minimum fee a low-priority tx must pay in order to be accepted
to the daemon's memory pool.'''
return self.coin.RELAY_FEE
class LegacyRPCDaemon(Daemon):
'''Handles connections to a daemon at the given URL.
This class is useful for daemons that don't have the new 'getblock'
RPC call that returns the block in hex, the workaround is to manually
recreate the block bytes. The recreated block bytes may not be the exact
as in the underlying blockchain but it is good enough for our indexing
purposes.'''
async def raw_blocks(self, hex_hashes):
'''Return the raw binary blocks with the given hex hashes.'''
params_iterable = ((h, ) for h in hex_hashes)
block_info = await self._send_vector('getblock', params_iterable)
blocks = []
for i in block_info:
raw_block = await self.make_raw_block(i)
class FakeEstimateFeeDaemon(Daemon):
'''Daemon that simulates estimatefee and relayfee RPC calls. Coin that
wants to use this daemon must define ESTIMATE_FEE & RELAY_FEE'''
async def estimatefee(self, block_count):
'''Return the fee estimate for the given parameters.'''
return self.coin.ESTIMATE_FEE
async def relayfee(self):
'''The minimum fee a low-priority tx must pay in order to be accepted
to the daemon's memory pool.'''
return self.coin.RELAY_FEE
class LegacyRPCDaemon(Daemon):
'''Handles connections to a daemon at the given URL.
This class is useful for daemons that don't have the new 'getblock'
RPC call that returns the block in hex, the workaround is to manually
recreate the block bytes. The recreated block bytes may not be the exact
as in the underlying blockchain but it is good enough for our indexing
purposes.'''
async def raw_blocks(self, hex_hashes):
'''Return the raw binary blocks with the given hex hashes.'''
params_iterable = ((h, ) for h in hex_hashes)
block_info = await self._send_vector('getblock', params_iterable)
blocks = []
for i in block_info:
raw_block = await self.make_raw_block(i)
num_txs = len(transactions)
if num_txs > 0:
raw_block += pack_varint(num_txs)
raw_block += b''.join(transactions)
else:
raw_block += b'\x00'
return raw_block
def timestamp_safe(self, t):
if isinstance(t, int):
return t
return timegm(strptime(t, "%Y-%m-%d %H:%M:%S %Z"))
class DecredDaemon(Daemon):
async def raw_blocks(self, hex_hashes):
'''Return the raw binary blocks with the given hex hashes.'''
params_iterable = ((h, False) for h in hex_hashes)
blocks = await self._send_vector('getblock', params_iterable)
raw_blocks = []
valid_tx_tree = {}
for block in blocks:
# Convert to bytes from hex
raw_block = hex_to_bytes(block)
raw_blocks.append(raw_block)
# Check if previous block is valid
prev = self.prev_hex_hash(raw_block)
votebits = unpack_le_uint16_from(raw_block[100:102])[0]
valid_tx_tree[prev] = self.is_valid_tx_tree(votebits)