from unittest import TestCase , skip
from mythril . ipc . client import EthIpc
from mythril . ipc . utils import hex_to_dec
class IpcTest ( TestCase ) :
client = None
def _new_filter ( self ) :
return self . client . eth_newFilter ( from_block = " 0x1 " , to_block = " 0x2 " , address = " 0x8888f1f195afa192cfee860698584c030f4c9db1 " )
def setUp ( self ) :
self . client = EthIpc ( )
def test_web3_clientVersion ( self ) :
version = self . client . web3_clientVersion ( )
self . assertTrue ( " Geth/ " in version , " We should use real `geth` to test " )
def test_web3_sha3 ( self ) :
sha3 = self . client . web3_sha3 ( " contract A {} " )
self . assertEqual ( sha3 , " 0x713073fa3404d26d3eb1ada031e6b51a3a0c135d0ee7c72ec87e0e77e004cf8c " )
def test_net_version ( self ) :
version = self . client . net_version ( )
self . assertEqual ( version , ' 1 ' , " We use mainnet for testing, whose version is ' 1 ' " )
def test_net_listening ( self ) :
listening = self . client . net_listening ( )
self . assertTrue ( listening , " should be listening " )
def test_net_peerCount ( self ) :
count = self . client . net_peerCount ( )
self . assertGreater ( count , 0 , " peerCount should >0 when syncing " )
def test_eth_protocolVersion ( self ) :
version = self . client . eth_protocolVersion ( )
self . assertTrue ( version . startswith ( " 0x " ) )
def test_eth_syncing ( self ) :
syncing = self . client . eth_syncing ( )
highest_block = hex_to_dec ( syncing [ " highestBlock " ] )
current_block = hex_to_dec ( syncing [ " currentBlock " ] )
self . assertTrue ( current_block > 0 , " current_block should > 0 since we make sure of it for testing " )
self . assertTrue ( highest_block > = current_block )
def test_eth_coinbase ( self ) :
coinbase = self . client . eth_coinbase ( )
self . assertTrue ( coinbase . startswith ( " 0x " ) , " coinbase should be a hex string " )
self . assertEqual ( len ( coinbase ) , 42 , " coinbase is a string with length of 42 " )
def test_eth_mining ( self ) :
mining = self . client . eth_mining ( )
self . assertEqual ( type ( mining ) , bool )
def test_eth_hashrate ( self ) :
hashrate = self . client . eth_hashrate ( )
self . assertEqual ( type ( hashrate ) , int )
def test_eth_gasPrice ( self ) :
price = self . client . eth_gasPrice ( )
self . assertGreater ( price , 0 , " gas price should be > 0 " )
def test_eth_accounts ( self ) :
accounts = self . client . eth_accounts ( )
self . assertGreater ( len ( accounts ) , 0 )
self . assertEqual ( len ( accounts [ 0 ] ) , 42 , " account is a string with length of 42 " )
self . assertTrue ( accounts [ 0 ] . startswith ( " 0x " ) , " account is hex string " )
def test_eth_blockNumber ( self ) :
block_number = self . client . eth_blockNumber ( )
self . assertGreater ( block_number , 0 , " we have made sure the blockNumber is > 0 for testing " )
def test_eth_getBalance ( self ) :
balance = self . client . eth_getBalance ( address = " 0x0000000000000000000000000000000000000000 " )
self . assertGreater ( balance , 10000000 , " specified address should have a lot of balance " )
def test_eth_getStorageAt ( self ) :
storage = self . client . eth_getStorageAt ( address = " 0x0000000000000000000000000000000000000000 " )
self . assertEqual ( storage , ' 0x0000000000000000000000000000000000000000000000000000000000000000 ' )
@skip ( " Not sure how to test this unless use some kind of mocks " )
def test_create_contract ( self ) :
self . client . create_contract ( from_ = " 0x0000000000000000000000000000000000000000 " , code = " 0x636f6e747261637420417b7d " , gas = 1 )
pass
def test_eth_call ( self ) :
# TODO need to give a better test case
result = self . client . eth_call ( to_address = " 0x0000000000000000000000000000000000000000 " )
self . assertEqual ( result , " 0x " )
@skip ( " { ' jsonrpc ' : ' 2.0 ' , ' id ' : 1, ' error ' : { ' code ' : -32602, ' message ' : ' too many arguments, want at most 1 ' }} " )
def test_eth_estimateGas ( self ) :
x = self . client . eth_estimateGas ( )
print ( x )
def test_eth_getBlockByHash ( self ) :
block = self . client . eth_getBlockByHash ( " 0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3 " )
self . assertEqual ( block [ " extraData " ] , " 0x11bbe8db4e347b4e8c937c1c8370e4b5ed33adb3db69cbdb7a38e1e50b1b82fa " , " the data of the first block should be right " )
def test_eth_getBlockByNumber ( self ) :
block = self . client . eth_getBlockByNumber ( 0 )
self . assertEqual ( block [ " extraData " ] , " 0x11bbe8db4e347b4e8c937c1c8370e4b5ed33adb3db69cbdb7a38e1e50b1b82fa " , " the data of the first block should be right " )
def test_eth_getBlockTransactionCountByHash ( self ) :
# TODO: can't find a block which contains transactions after checking thousands of blocks
count = self . client . eth_getBlockTransactionCountByHash ( " 0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3 " )
self . assertEqual ( count , 0 )
def test_eth_getBlockTransactionCountByNumber ( self ) :
# TODO: can't find a proper block
count = self . client . eth_getBlockTransactionCountByNumber ( 0 )
self . assertEqual ( count , 0 )
def test_eth_getCode ( self ) :
# TODO: can't find a proper address for getting code
code = self . client . eth_getCode ( address = " 0x0000000000000000000000000000000000000001 " )
self . assertEqual ( code , " 0x " )
def test_eth_getFilterChanges ( self ) :
filter_id = self . _new_filter ( )
changes = self . client . eth_getFilterChanges ( filter_id )
self . assertEqual ( type ( changes ) , list )
def test_eth_getFilterLogs ( self ) :
filter_id = self . _new_filter ( )
logs = self . client . eth_getFilterLogs ( filter_id )
self . assertEqual ( type ( logs ) , list )
def test_eth_getLogs ( self ) :
# TODO: can't find proper filter_object to return an non-empty response
logs = self . client . eth_getLogs ( filter_object = {
" topics " : [ " 0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b " ]
} )
self . assertEqual ( type ( logs ) , list )
def test_eth_getTransactionByBlockHashAndIndex ( self ) :
transaction = self . client . eth_getTransactionByBlockHashAndIndex ( block_hash = " 0x7e5a9336dd82efff0bfe8c25ccb0e8cf44b4c6f781b25b3fc3578f004f60b872 " )
self . assertEqual ( transaction [ " from " ] , " 0x22f2dcff5ad78c3eb6850b5cb951127b659522e6 " )
self . assertEqual ( transaction [ " to " ] , " 0x0000000000000000000000000000000000000000 " )
def test_eth_getTransactionByBlockNumberAndIndex ( self ) :
transaction = self . client . eth_getTransactionByBlockNumberAndIndex ( block = 46420 )
self . assertEqual ( transaction [ " blockHash " ] , " 0x7e5a9336dd82efff0bfe8c25ccb0e8cf44b4c6f781b25b3fc3578f004f60b872 " )
def test_eth_getTransactionByHash ( self ) :
transaction = self . client . eth_getTransactionByHash ( tx_hash = " 0xe363505adc6b2996111f8bd774f8653a61d244cc6567b5372c2e781c6c63b737 " )
self . assertEqual ( transaction [ " from " ] , " 0x22f2dcff5ad78c3eb6850b5cb951127b659522e6 " )
self . assertEqual ( transaction [ " to " ] , " 0x0000000000000000000000000000000000000000 " )
def test_eth_getTransactionCount ( self ) :
count = self . client . eth_getTransactionCount ( address = " 0x22f2dcff5ad78c3eb6850b5cb951127b659522e6 " )
self . assertGreater ( count , 17 , " The address should at least have 18 transactions " )
def test_eth_getTransactionReceipt ( self ) :
transaction = self . client . eth_getTransactionReceipt ( tx_hash = " 0xe363505adc6b2996111f8bd774f8653a61d244cc6567b5372c2e781c6c63b737 " )
self . assertEqual ( transaction [ " from " ] , " 0x22f2dcff5ad78c3eb6850b5cb951127b659522e6 " )
self . assertEqual ( transaction [ " to " ] , " 0x0000000000000000000000000000000000000000 " )
def test_eth_getUncleByBlockHashAndIndex ( self ) :
uncle = self . client . eth_getUncleByBlockHashAndIndex ( block_hash = " 0x07efc5598ed7e5e6e707eb087110f2dfb2772d6355508b75f4696d96d0f747ec " )
self . assertEqual ( uncle [ " hash " ] , " 0x78b57d7dd0b797cdf91f88807d761a7a0099f417be98b08ec600a2142e66af9e " )
def test_eth_getUncleByBlockNumberAndIndex ( self ) :
uncle = self . client . eth_getUncleByBlockNumberAndIndex ( block = 105 )
self . assertEqual ( uncle [ " hash " ] , " 0x78b57d7dd0b797cdf91f88807d761a7a0099f417be98b08ec600a2142e66af9e " )
def test_eth_getUncleCountByBlockHash ( self ) :
count = self . client . eth_getUncleCountByBlockHash ( block_hash = " 0x07efc5598ed7e5e6e707eb087110f2dfb2772d6355508b75f4696d96d0f747ec " )
self . assertEqual ( count , 1 , " there should be 1 uncle at block 105 " )
def test_eth_getUncleCountByBlockNumber ( self ) :
count = self . client . eth_getUncleCountByBlockNumber ( 105 )
self . assertEqual ( count , 1 , " there should be 1 uncle at block 105 " )
@skip ( """ { ' jsonrpc ' : ' 2.0 ' , ' id ' : 1, ' error ' : { ' code ' : -32000, ' message ' : " mining not ready: No work available yet, don ' t panic. " }} """ )
def test_eth_getWork ( self ) :
work = self . client . eth_getWork ( )
self . assertEqual ( len ( work ) , 3 )
self . assertTrue ( work [ 0 ] . startswith ( " 0x " ) , " should be a hex " )
self . assertTrue ( work [ 1 ] . startswith ( " 0x " ) , " should be a hex " )
self . assertTrue ( work [ 2 ] . startswith ( " 0x " ) , " should be a hex " )
def test_eth_newBlockFilter ( self ) :
filter_id = self . client . eth_newBlockFilter ( )
self . assertTrue ( filter_id . startswith ( " 0x " ) , " filter_id is a hex string " )
def test_eth_newFilter ( self ) :
filter_id = self . client . eth_newFilter ( from_block = " 0x1 " , to_block = " 0x2 " , address = " 0x8888f1f195afa192cfee860698584c030f4c9db1 " ,
topics = [ " 0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b " ] )
self . assertTrue ( filter_id . startswith ( " 0x " ) )
def test_eth_newPendingTransactionFilter ( self ) :
filter_id = self . client . eth_newPendingTransactionFilter ( )
self . assertTrue ( filter_id . startswith ( " 0x " ) , " filter_id should be a hex " )
@skip ( " { ' jsonrpc ' : ' 2.0 ' , ' id ' : 1, ' error ' : { ' code ' : -32000, ' message ' : ' rlp: element is larger than containing list ' }} " )
def test_eth_sendRawTransaction ( self ) :
x = self . client . eth_sendRawTransaction ( data = " 0xd46e8dd67c5d32be8d46e8dd67c5d32be8058bb8eb970870f072445675058bb8eb970870f072445675 " )
print ( x )
@skip ( " { ' jsonrpc ' : ' 2.0 ' , ' id ' : 1, ' error ' : { ' code ' : -32000, ' message ' : ' authentication needed: password or unlock ' }} " )
def test_eth_sendTransaction ( self ) :
x = self . client . eth_sendTransaction ( to_address = " 0x0000000000000000000000000000000000000000 " ,
gas = 30400 , gas_price = 10000000000000 , value = 2441406250 ,
data = " 0xd46e8dd67c5d32be8d46e8dd67c5d32be8058bb8eb970870f072445675058bb8eb970870f072445675 " )
print ( x )
@skip ( " server returns { ' jsonrpc ' : ' 2.0 ' , ' id ' : 1, ' error ' : { ' code ' : -32000, ' message ' : ' unknown account ' }}, not sure how to fix it " )
def test_eth_sign ( self ) :
x = self . client . eth_sign ( address = " 0x9b2055d370f73ec7d8a03e965129118dc8f5bf83 " , data = " 0xdeadbeaf " )
print ( x )
def test_eth_submitHashrate ( self ) :
submitted = self . client . eth_submitHashrate ( hash_rate = 5242880 ,
client_id = " 0x59daa26581d0acd1fce254fb7e85952f4c09d0915afd33d3886cd914bc7d283c " )
self . assertTrue ( submitted , " Should submit hashrate successfully " )
def test_eth_submitWork ( self ) :
work_accepted = self . client . eth_submitWork ( nonce = " 0x0000000000000001 " ,
header = " 0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef " ,
mix_digest = " 0xD1FE5700000000000000000000000000D1FE5700000000000000000000000000 " )
self . assertFalse ( work_accepted , " Work should be rejected " )
def test_eth_uninstallFilter ( self ) :
filter_id = self . _new_filter ( )
uninstalled = self . client . eth_uninstallFilter ( filter_id )
self . assertTrue ( uninstalled , " filter_id should be uninstalled " )
def test_get_contract_address ( self ) :
# TODO need to find a tx which has non-none contract address
address = self . client . get_contract_address ( tx = " 0xe363505adc6b2996111f8bd774f8653a61d244cc6567b5372c2e781c6c63b737 " )
self . assertEqual ( address , None )
def test_shh_version ( self ) :
version = self . client . shh_version ( )
self . assertEqual ( version , " 5.0 " , " shh_version should be 5.0 for current ethereum release " )
@skip ( " { ' jsonrpc ' : ' 2.0 ' , ' id ' : 1, ' error ' : { ' code ' : -32000, ' message ' : ' unknown account ' }} " )
def test_transfer ( self ) :
x = self . client . transfer ( from_ = " 0x22f2dcff5ad78c3eb6850b5cb951127b659522e6 " , to = " 0x0000000000000000000000000000000000000000 " , amount = 1 )
print ( x )
def test_getBlockRlp ( self ) :
block = self . client . getBlockRlp ( number = 1 )
self . assertGreater ( len ( block ) , 1000 , " The RLP encoded block is very long " )