2018-08-15 12:51:52 +00:00
import logging
2018-10-11 20:28:52 +00:00
from typing import List
2018-05-25 17:29:07 +00:00
import pytest
import requests
import time
2018-09-03 12:52:31 +00:00
from json import JSONDecodeError
2019-10-04 13:50:44 +00:00
from decimal import Decimal
2020-02-24 11:30:45 +00:00
from os import environ
2019-11-29 11:48:07 +00:00
import tests
2020-12-04 17:27:21 +00:00
import support . api . web3_api as w3
2018-05-25 17:29:07 +00:00
2018-10-11 20:28:52 +00:00
class NetworkApi ( object ) :
2018-05-25 17:29:07 +00:00
def __init__ ( self ) :
2019-11-29 11:48:07 +00:00
self . network_url = ' http://api- %s .etherscan.io/api? ' % tests . pytest_config_global [ ' network ' ]
2019-03-22 11:01:22 +00:00
self . faucet_url = ' https://faucet-ropsten.status.im/donate '
2020-12-04 17:27:21 +00:00
self . faucet_backup_address = w3 . account_address
2019-10-02 07:20:53 +00:00
self . headers = {
' User-Agent ' : " Mozilla \ /5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit \
/ 537.36 ( KHTML , like Gecko ) Chrome \/ 77.0 .3865 .90 Safari \/ 537.36 " , }
2018-06-07 15:13:37 +00:00
self . chat_bot_url = ' http://offsite.chat:8099 '
2020-02-24 12:53:16 +00:00
self . api_key = environ . get ( ' ETHERSCAN_API_KEY ' )
2018-05-25 17:29:07 +00:00
2020-04-13 19:03:00 +00:00
def log ( self , text : str ) :
tests . test_suite_data . current_test . testruns [ - 1 ] . steps . append ( text )
logging . info ( text )
2018-10-11 20:28:52 +00:00
def get_transactions ( self , address : str ) - > List [ dict ] :
2020-02-24 12:53:16 +00:00
method = self . network_url + ' module=account&action=txlist&address=0x %s &sort=desc&apikey= %s ' % ( address , self . api_key )
2020-02-24 11:30:45 +00:00
try :
2020-10-20 15:34:52 +00:00
transactions_response = requests . request ( ' GET ' , url = method , headers = self . headers ) . json ( )
if transactions_response :
return transactions_response [ ' result ' ]
2020-04-13 19:03:00 +00:00
except TypeError as e :
self . log ( " Check response from etherscan API. Returned values do not match expected. %s " % e )
2020-10-06 14:48:08 +00:00
except JSONDecodeError as e :
self . log ( str ( e ) )
pass
2018-05-25 17:29:07 +00:00
2018-10-11 20:28:52 +00:00
def get_token_transactions ( self , address : str ) - > List [ dict ] :
2020-02-24 12:53:16 +00:00
method = self . network_url + ' module=account&action=tokentx&address=0x %s &sort=desc&apikey= %s ' % ( address , self . api_key )
2020-02-24 11:30:45 +00:00
try :
2020-10-20 15:34:52 +00:00
transactions_response = requests . request ( ' GET ' , url = method , headers = self . headers ) . json ( )
if transactions_response :
return transactions_response [ ' result ' ]
2020-04-13 19:03:00 +00:00
except TypeError as e :
2020-10-06 14:48:08 +00:00
self . log ( " Check response from etherscan API. Returned values do not match expected. %s " % str ( e ) )
except JSONDecodeError as e :
self . log ( str ( e ) )
pass
2018-07-03 18:50:18 +00:00
2018-05-25 17:29:07 +00:00
def is_transaction_successful ( self , transaction_hash : str ) - > int :
2018-08-07 09:08:54 +00:00
method = self . network_url + ' module=transaction&action=getstatus&txhash= %s ' % transaction_hash
2019-10-02 07:20:53 +00:00
return not int ( requests . request ( ' GET ' , url = method , headers = self . headers ) . json ( ) [ ' result ' ] [ ' isError ' ] )
2018-05-25 17:29:07 +00:00
def get_balance ( self , address ) :
2020-12-08 15:56:57 +00:00
address = ' 0x ' + address
balance = w3 . balance_of_address ( address )
self . log ( ' Balance is %s Gwei ' % balance )
return int ( balance )
2018-05-25 17:29:07 +00:00
2018-06-30 12:17:38 +00:00
def get_latest_block_number ( self ) - > int :
2018-08-07 09:08:54 +00:00
method = self . network_url + ' module=proxy&action=eth_blockNumber '
2018-06-30 12:17:38 +00:00
return int ( requests . request ( ' GET ' , url = method ) . json ( ) [ ' result ' ] , 0 )
2018-05-25 17:29:07 +00:00
def find_transaction_by_hash ( self , address : str , transaction_hash : str ) :
transactions = self . get_transactions ( address = address )
for transaction in transactions :
if transaction [ ' hash ' ] == transaction_hash :
2018-08-15 12:51:52 +00:00
logging . info ( ' Transaction is found in Ropsten network ' )
2018-05-25 17:29:07 +00:00
return
pytest . fail ( ' Transaction is not found in Ropsten network ' )
2018-07-13 10:56:36 +00:00
def find_transaction_by_unique_amount ( self , address , amount , token = False , decimals = 18 , wait_time = 600 ) :
2021-01-25 16:35:40 +00:00
additional_info = ' token transactions ' if token else ' ETH transactions '
2018-05-25 17:29:07 +00:00
counter = 0
while True :
if counter > = wait_time :
2020-06-17 14:45:40 +00:00
for entry in range ( 0 , 5 ) :
self . log ( ' Transaction # %s , amount is %s ' % ( entry + 1 , float ( int ( transactions [ entry ] [ ' value ' ] ) / 10 * * decimals ) ) )
self . log ( str ( transactions [ entry ] ) )
2018-05-25 17:29:07 +00:00
pytest . fail (
2021-01-25 16:35:40 +00:00
' Transaction with amount %s is not found in list of %s , address is %s during %s s ' %
( amount , additional_info , address , wait_time ) )
2018-05-25 17:29:07 +00:00
else :
counter + = 10
time . sleep ( 10 )
2018-09-03 12:52:31 +00:00
try :
if token :
transactions = self . get_token_transactions ( address )
else :
transactions = self . get_transactions ( address )
2020-04-13 19:03:00 +00:00
except JSONDecodeError as e :
self . log ( str ( e ) )
2018-09-03 12:52:31 +00:00
continue
2020-04-13 19:03:00 +00:00
try :
for transaction in transactions :
if float ( int ( transaction [ ' value ' ] ) / 10 * * decimals ) == float ( amount ) :
return transaction
except TypeError as e :
2020-10-20 15:34:52 +00:00
self . log ( " Failed iterate transactions: " + str ( e ) )
2020-04-13 19:03:00 +00:00
continue
2018-07-13 10:56:36 +00:00
2020-01-03 15:27:50 +00:00
def wait_for_confirmation_of_transaction ( self , address , amount , confirmations = 12 , token = False ) :
2018-07-13 10:56:36 +00:00
start_time = time . time ( )
2021-01-25 16:35:40 +00:00
if token :
token_info = " token transaction "
else :
token_info = " ETH transaction "
self . log ( ' Waiting %s %s for %s to have %s confirmations ' % ( amount , token_info , address , confirmations ) )
2018-08-02 10:11:45 +00:00
while round ( time . time ( ) - start_time , ndigits = 2 ) < 900 : # should be < idleTimeout capability
2019-11-07 13:38:14 +00:00
transaction = self . find_transaction_by_unique_amount ( address , amount , token )
2020-10-28 19:39:02 +00:00
self . log (
' Expected amount of confirmations is %s , in fact %s ' % ( confirmations , transaction [ ' confirmations ' ] ) )
2020-01-03 15:27:50 +00:00
if int ( transaction [ ' confirmations ' ] ) > = confirmations :
2018-07-13 10:56:36 +00:00
return
2020-10-28 19:39:02 +00:00
time . sleep ( 20 )
2020-07-22 13:30:07 +00:00
pytest . fail ( ' Transaction with amount %s was not confirmed, address is %s , still has %s confirmations ' % ( amount , address , int ( transaction [ ' confirmations ' ] ) ) )
2018-05-25 17:29:07 +00:00
def verify_balance_is_updated ( self , initial_balance , recipient_address , wait_time = 360 ) :
counter = 0
while True :
if counter > = wait_time :
2020-10-20 15:34:52 +00:00
pytest . fail ( ' Balance is not changed during %s seconds ' % wait_time )
2018-05-25 17:29:07 +00:00
elif initial_balance == self . get_balance ( recipient_address ) :
counter + = 10
time . sleep ( 10 )
2020-10-20 15:34:52 +00:00
self . log ( ' Waiting %s seconds for for changing account balance from %s ' % ( counter , initial_balance ) )
2018-05-25 17:29:07 +00:00
else :
2020-10-20 15:34:52 +00:00
self . log ( ' Balance is updated! ' )
2018-05-25 17:29:07 +00:00
return
2018-06-28 18:46:51 +00:00
def verify_balance_is ( self , expected_balance : int , recipient_address : str , errors : list ) :
balance = self . get_balance ( recipient_address )
if balance / 1000000000000000000 != expected_balance :
errors . append ( ' Recipients balance is not updated on etherscan ' )
2018-05-25 17:29:07 +00:00
def faucet ( self , address ) :
2020-10-07 16:07:42 +00:00
try :
self . log ( " Trying to get funds from %s " % self . faucet_url )
return requests . request ( ' GET ' , ' %s /0x %s ' % ( self . faucet_url , address ) ) . json ( )
except JSONDecodeError as e :
self . log ( str ( e ) )
pass
2018-05-25 17:29:07 +00:00
2019-05-08 15:40:08 +00:00
def faucet_backup ( self , address ) :
2020-12-04 17:27:21 +00:00
self . log ( " Trying to get funds from %s " % self . faucet_backup_address )
address = " 0x " + address
w3 . donate_testnet_eth ( address = address , amount = 0.005 , inscrease_default_gas_price = 10 )
2019-05-08 15:40:08 +00:00
2020-12-04 17:27:21 +00:00
def get_donate ( self , address , external_faucet = False , wait_time = 300 ) :
2018-05-25 17:29:07 +00:00
initial_balance = self . get_balance ( address )
counter = 0
if initial_balance < 1000000000000000000 :
2020-02-27 16:52:31 +00:00
if external_faucet :
self . faucet_backup ( address )
2020-12-04 17:27:21 +00:00
self . faucet ( address )
2018-05-25 17:29:07 +00:00
while True :
if counter > = wait_time :
pytest . fail ( " Donation was not received during %s seconds! " % wait_time )
elif self . get_balance ( address ) == initial_balance :
counter + = 10
time . sleep ( 10 )
2020-04-13 19:03:00 +00:00
self . log ( ' Waiting %s seconds for donation ' % counter )
2018-05-25 17:29:07 +00:00
else :
2020-12-04 17:27:21 +00:00
self . log ( ' Got %s Gwei for %s ' % ( self . get_balance ( address ) , address ) )
2018-05-25 17:29:07 +00:00
return
2018-03-31 20:05:11 +00:00
def start_chat_bot ( self , chat_name : str , messages_number : int , interval : int = 1 ) - > list :
2018-06-07 15:13:37 +00:00
url = ' %s /ping/ %s ?count= %s &interval= %s ' % ( self . chat_bot_url , chat_name , messages_number , interval )
2018-03-31 20:05:11 +00:00
text = requests . request ( ' GET ' , url ) . text
return [ i . split ( maxsplit = 5 ) [ - 1 ] . strip ( ' * ' ) for i in text . splitlines ( ) ]
2019-10-04 13:50:44 +00:00
def get_rounded_balance ( self , fetched_balance , actual_balance ) :
fetched_balance , actual_balance = str ( fetched_balance ) , str ( actual_balance )
# get actual number of decimals on account balance
decimals = abs ( Decimal ( fetched_balance ) . as_tuple ( ) . exponent )
rounded_balance = round ( float ( actual_balance ) , decimals )
return rounded_balance