1
1
import logging
2
+ import time
3
+ from decimal import Decimal
4
+ from json import JSONDecodeError
5
+ from os import environ
2
6
from typing import List
3
7
4
8
import pytest
5
9
import requests
6
- import time
7
- from json import JSONDecodeError
8
- from decimal import Decimal
9
- from os import environ
10
+ from selenium .common import TimeoutException
11
+
10
12
import tests
11
13
12
14
13
- class NetworkApi ( object ) :
15
+ class NetworkApi :
14
16
def __init__ (self ):
15
- self .network_url = 'http://api-goerli.etherscan.io/api?'
16
- self .headers = {
17
- 'User-Agent' :"Mozilla\\ 5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit\\ 537.36 (KHTML, like Gecko) Chrome\\ 7"
18
- "7.0.3865.90 Safari\\ 537.36" , }
19
- self .chat_bot_url = 'http://offsite.chat:8099'
17
+ self .network_url = 'http://api-sepolia.etherscan.io/api'
20
18
self .api_key = environ .get ('ETHERSCAN_API_KEY' )
21
19
22
20
def log (self , text : str ):
23
21
tests .test_suite_data .current_test .testruns [- 1 ].steps .append (text )
24
22
logging .info (text )
25
23
26
- def send_etherscan_request (self , method , extracted_param : str ):
27
- for attempt in range (3 ):
28
- try :
29
- response = requests .request ('GET' , url = method , headers = self .headers ).json ()
30
- if response :
31
- return response [extracted_param ]
32
- except TypeError as e :
33
- self .log ("Check response from etherscan API. Returned values do not match expected. %s" % str (e ))
34
- except JSONDecodeError as e :
35
- self .log ("No valid JSON response from Etherscan: %s " % str (e ))
36
- pass
37
- time .sleep (30 )
24
+ def send_etherscan_request (self , params ):
25
+ params ['apikey' ] = self .api_key
26
+ try :
27
+ response = requests .get (url = self .network_url , params = params ).json ()
28
+ if response :
29
+ return response ['result' ]
30
+ except TypeError as e :
31
+ self .log ("Check response from etherscan API. Returned values do not match expected. %s" % str (e ))
32
+ except JSONDecodeError as e :
33
+ self .log ("No valid JSON response from Etherscan: %s " % str (e ))
34
+ pass
38
35
39
36
def get_token_transactions (self , address : str ) -> List [dict ]:
40
- method = self .network_url + 'module=account&action=tokentx&address=0x%s&sort=desc&apikey=%s' % (
41
- address , self .api_key )
42
- return self .send_etherscan_request (method , 'result' )
37
+ params = {'module' : 'account' , 'action' : 'tokentx' , 'address' : address , 'sort' : 'desc' }
38
+ return self .send_etherscan_request (params )
43
39
44
40
def get_transactions (self , address : str ) -> List [dict ]:
45
- method = self . network_url + 'module= account& action= txlist& address=0x%s&sort=desc&apikey=%s' % ( address , self . api_key )
46
- return self .send_etherscan_request (method , 'result' )
41
+ params = { 'module' : ' account' , ' action' : ' txlist' , ' address' : address , 'sort' : 'desc' }
42
+ return self .send_etherscan_request (params )
47
43
48
44
def is_transaction_successful (self , transaction_hash : str ) -> int :
49
- method = self .network_url + 'module=transaction&action=getstatus&txhash=%s' % transaction_hash
50
- return not int (requests .request ('GET' , url = method , headers = self .headers ).json ()['result' ]['isError' ])
51
-
52
- def get_balance (self , address ):
53
- address = '0x' + address
54
- method = self .network_url + 'module=account&action=balance&address=%s&tag=latest&apikey=%s' % (
55
- address , self .api_key )
56
- balance = self .send_etherscan_request (method , 'result' )
45
+ params = {'module' : 'transaction' , 'action' : 'getstatus' , 'txhash' : transaction_hash }
46
+ return not int (self .send_etherscan_request (params )['isError' ])
47
+
48
+ def get_balance (self , address : str ):
49
+ params = {'module' : 'account' , 'action' : 'balance' , 'address' : address , 'tag' : 'latest' }
50
+ balance = self .send_etherscan_request (params )
57
51
if balance :
58
52
self .log ('Balance is %s Gwei' % balance )
59
53
return int (balance )
60
54
else :
61
55
self .log ('Cannot extract balance!' )
62
56
63
57
def get_latest_block_number (self ) -> int :
64
- method = self . network_url + 'module= proxy& action= eth_blockNumber'
65
- return int (requests . request ( 'GET' , url = method ). json ()[ 'result' ] , 0 )
58
+ params = { 'module' : ' proxy' , ' action' : ' eth_blockNumber'}
59
+ return int (self . send_etherscan_request ( params ) , 0 )
66
60
67
61
def find_transaction_by_hash (self , transaction_hash : str ):
68
- method = self .network_url + 'module=transaction&action=gettxreceiptstatus&txhash=%s&apikey=%s' % (
69
- transaction_hash , self .api_key )
70
- result = self .send_etherscan_request (method , 'result' )
62
+ params = {'module' : 'transaction' , 'action' : 'gettxreceiptstatus' , 'txhash' : transaction_hash }
63
+ result = self .send_etherscan_request (params )
71
64
72
65
if result :
73
66
final_status = True
@@ -86,13 +79,14 @@ def find_transaction_by_unique_amount(self, address, amount, token=False, decima
86
79
while True :
87
80
if counter >= wait_time :
88
81
for entry in range (0 , 5 ):
89
- self .log ('Transaction #%s, amount is %s' % (entry + 1 , float (int (transactions [entry ]['value' ]) / 10 ** decimals )))
82
+ self .log ('Transaction #%s, amount is %s' % (
83
+ entry + 1 , float (int (transactions [entry ]['value' ]) / 10 ** decimals )))
90
84
self .log (str (transactions [entry ]))
91
85
pytest .fail (
92
86
'Transaction with amount %s is not found in list of %s, address is %s during %ss' %
93
87
(amount , additional_info , address , wait_time ))
94
88
else :
95
- self .log ("Finding tx in %s, attempt #%s" % (additional_info , str (int (counter / 30 )+ 1 )))
89
+ self .log ("Finding tx in %s, attempt #%s" % (additional_info , str (int (counter / 30 ) + 1 )))
96
90
try :
97
91
if token :
98
92
transactions = self .get_token_transactions (address )
@@ -126,7 +120,8 @@ def wait_for_confirmation_of_transaction(self, address, amount, confirmations=6,
126
120
if int (transaction ['confirmations' ]) >= confirmations :
127
121
return
128
122
time .sleep (20 )
129
- pytest .fail ('Transaction with amount %s was not confirmed, address is %s, still has %s confirmations' % (amount , address , int (transaction ['confirmations' ])))
123
+ pytest .fail ('Transaction with amount %s was not confirmed, address is %s, still has %s confirmations' % (
124
+ amount , address , int (transaction ['confirmations' ])))
130
125
131
126
def verify_balance_is_updated (self , initial_balance , recipient_address , wait_time = 360 ):
132
127
counter = 0
@@ -141,10 +136,13 @@ def verify_balance_is_updated(self, initial_balance, recipient_address, wait_tim
141
136
self .log ('Balance is updated!' )
142
137
return
143
138
144
- def verify_balance_is (self , expected_balance : int , recipient_address : str , errors : list ):
145
- balance = self .get_balance (recipient_address )
146
- if balance / 1000000000000000000 != expected_balance :
147
- errors .append ('Recipients balance is not updated on etherscan' )
139
+ def wait_for_balance_to_be (self , address : str , expected_balance : int , less : bool = True ):
140
+ for _ in range (5 ):
141
+ balance = self .get_balance (address ) / 1000000000000000000
142
+ if (less and balance < expected_balance ) or (not less and balance > expected_balance ):
143
+ return
144
+ time .sleep (10 )
145
+ raise TimeoutException ('Balance is not updated on Etherscan' )
148
146
149
147
# Do not use until web3 update
150
148
# def faucet(self, address):
@@ -160,7 +158,6 @@ def verify_balance_is(self, expected_balance: int, recipient_address: str, error
160
158
# address = "0x" + address
161
159
# w3.donate_testnet_eth(address=address, amount=0.01, inscrease_default_gas_price=10)
162
160
163
-
164
161
# def get_donate(self, address, external_faucet=False, wait_time=300):
165
162
# initial_balance = self.get_balance(address)
166
163
# counter = 0
@@ -180,27 +177,21 @@ def verify_balance_is(self, expected_balance: int, recipient_address: str, error
180
177
# self.log('Got %s Gwei for %s' % (self.get_balance(address), address))
181
178
# return
182
179
183
- def start_chat_bot (self , chat_name : str , messages_number : int , interval : int = 1 ) -> list :
184
- url = '%s/ping/%s?count=%s&interval=%s' % (self .chat_bot_url , chat_name , messages_number , interval )
185
- text = requests .request ('GET' , url ).text
186
- return [i .split (maxsplit = 5 )[- 1 ].strip ('*' ) for i in text .splitlines ()]
187
-
188
180
def get_rounded_balance (self , fetched_balance , actual_balance ):
189
181
fetched_balance , actual_balance = str (fetched_balance ), str (actual_balance )
190
182
# get actual number of decimals on account balance
191
183
decimals = abs (Decimal (fetched_balance ).as_tuple ().exponent )
192
184
rounded_balance = round (float (actual_balance ), decimals )
193
185
return rounded_balance
194
186
195
- def get_tx_param_by_hash (self , hash : str , param : str ):
196
- method = self .network_url + 'module=proxy&action=eth_getTransactionByHash&txhash=%s&apikey=%s' % (
197
- hash , self .api_key )
198
- res = self .send_etherscan_request (method , 'result' )
187
+ def get_tx_param_by_hash (self , transaction_hash : str , param : str ):
188
+ params = {'module' : 'proxy' , 'action' : 'eth_getTransactionByHash' , 'txhash' : transaction_hash }
189
+ res = self .send_etherscan_request (params )
199
190
return int (res [param ], 16 )
200
191
201
192
def get_custom_fee_tx_params (self , hash : str ):
202
193
return {
203
- 'fee_cap' : str (self .get_tx_param_by_hash (hash , 'maxFeePerGas' )/ 1000000000 ),
204
- 'tip_cap' : str (self .get_tx_param_by_hash (hash , 'maxPriorityFeePerGas' )/ 1000000000 ),
194
+ 'fee_cap' : str (self .get_tx_param_by_hash (hash , 'maxFeePerGas' ) / 1000000000 ),
195
+ 'tip_cap' : str (self .get_tx_param_by_hash (hash , 'maxPriorityFeePerGas' ) / 1000000000 ),
205
196
'gas_limit' : str (self .get_tx_param_by_hash (hash , 'gas' ))
206
- }
197
+ }
0 commit comments