test_tx.py (6621B)
1 # standard imports 2 import os 3 import unittest 4 import logging 5 6 # local imports 7 from chainlib.eth.unittest.ethtester import EthTesterCase 8 from chainlib.eth.nonce import RPCNonceOracle 9 from chainlib.eth.gas import ( 10 RPCGasOracle, 11 Gas, 12 ) 13 from chainlib.eth.tx import ( 14 unpack, 15 pack, 16 raw, 17 transaction, 18 TxFormat, 19 TxFactory, 20 Tx, 21 ) 22 from chainlib.eth.contract import ( 23 ABIContractEncoder, 24 ABIContractType, 25 ) 26 from chainlib.eth.address import ( 27 to_checksum_address, 28 is_same_address, 29 ) 30 from hexathon import ( 31 strip_0x, 32 add_0x, 33 same as hex_same, 34 ) 35 from chainlib.eth.block import Block 36 37 logging.basicConfig(level=logging.DEBUG) 38 logg = logging.getLogger() 39 40 41 class TxTestCase(EthTesterCase): 42 43 def test_tx_basic(self): 44 tx_src = { 45 'hash': os.urandom(32).hex(), 46 'from': os.urandom(20).hex(), 47 'to': os.urandom(20).hex(), 48 'value': 13, 49 'data': '0xdeadbeef', 50 'nonce': 666, 51 'gasPrice': 100, 52 'gas': 21000, 53 } 54 55 tx = Tx(tx_src) 56 57 self.assertEqual(tx.hash, tx_src['hash']) 58 self.assertTrue(is_same_address(tx.outputs[0], tx_src['from'])) 59 self.assertTrue(is_same_address(tx.inputs[0], tx_src['to'])) 60 self.assertEqual(tx.value, tx_src['value']) 61 self.assertEqual(tx.nonce, tx_src['nonce']) 62 self.assertTrue(hex_same(tx.payload, tx_src['data'])) 63 64 65 def test_tx_reciprocal(self): 66 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 67 gas_oracle = RPCGasOracle(self.rpc) 68 c = Gas(signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_spec=self.chain_spec) 69 (tx_hash_hex, o) = c.create(self.accounts[0], self.accounts[1], 1024, tx_format=TxFormat.RLP_SIGNED) 70 tx = unpack(bytes.fromhex(strip_0x(o)), self.chain_spec) 71 self.assertTrue(is_same_address(tx['from'], self.accounts[0])) 72 self.assertTrue(is_same_address(tx['to'], self.accounts[1])) 73 74 75 def test_tx_repack(self): 76 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 77 gas_oracle = RPCGasOracle(self.rpc) 78 c = Gas(signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_spec=self.chain_spec) 79 (tx_hash_hex, o) = c.create(self.accounts[0], self.accounts[1], 1024) 80 self.rpc.do(o) 81 82 o = transaction(tx_hash_hex) 83 tx_src = self.rpc.do(o) 84 tx = Tx(tx_src) 85 tx_bin = pack(tx.src, self.chain_spec) 86 87 88 def test_tx_pack(self): 89 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 90 gas_oracle = RPCGasOracle(self.rpc) 91 92 mock_contract = to_checksum_address(add_0x(os.urandom(20).hex())) 93 94 f = TxFactory(self.chain_spec, signer=self.rpc) 95 enc = ABIContractEncoder() 96 enc.method('fooMethod') 97 enc.typ(ABIContractType.UINT256) 98 enc.uint256(13) 99 data = enc.get() 100 tx = f.template(self.accounts[0], mock_contract, use_nonce=True) 101 tx = f.set_code(tx, data) 102 (tx_hash, tx_signed_raw_hex) = f.finalize(tx, TxFormat.RLP_SIGNED) 103 logg.debug('tx result {}'.format(tx)) 104 o = raw(tx_signed_raw_hex) 105 r = self.rpc.do(o) 106 o = transaction(tx_hash) 107 tx_rpc_src = self.rpc.do(o) 108 logg.debug('rpc src {}'.format(tx_rpc_src)) 109 110 tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex)) 111 tx_src = unpack(tx_signed_raw_bytes, self.chain_spec) 112 txo = Tx(tx_src) 113 tx_signed_raw_bytes_recovered = pack(txo, self.chain_spec) 114 logg.debug('o {}'.format(tx_signed_raw_bytes.hex())) 115 logg.debug('r {}'.format(tx_signed_raw_bytes_recovered.hex())) 116 self.assertEqual(tx_signed_raw_bytes, tx_signed_raw_bytes_recovered) 117 118 119 def test_apply_block(self): 120 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 121 gas_oracle = RPCGasOracle(self.rpc) 122 c = Gas(signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_spec=self.chain_spec) 123 (tx_hash_hex, o) = c.create(self.accounts[0], self.accounts[1], 1024, tx_format=TxFormat.RLP_SIGNED) 124 tx_data = unpack(bytes.fromhex(strip_0x(o)), self.chain_spec) 125 126 block_hash = os.urandom(32).hex() 127 block = Block({ 128 'hash': block_hash, 129 'number': 42, 130 'timestamp': 13241324, 131 'transactions': [], 132 'author': os.urandom(20).hex(), 133 'gas_used': 21000, 134 'gas_limit': '0x2345', 135 'parent_hash': None, 136 }) 137 with self.assertRaises(AttributeError): 138 tx = Tx(tx_data, block=block) 139 140 tx_unknown_hash = os.urandom(32).hex() 141 block.txs = [add_0x(tx_unknown_hash)] 142 block.txs.append(add_0x(tx_data['hash'])) 143 tx = Tx(tx_data, block=block) 144 145 block.txs = [add_0x(tx_unknown_hash)] 146 block.txs.append(tx_data) 147 tx = Tx(tx_data, block=block) 148 149 150 def test_apply_receipt(self): 151 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 152 gas_oracle = RPCGasOracle(self.rpc) 153 c = Gas(signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_spec=self.chain_spec) 154 (tx_hash_hex, o) = c.create(self.accounts[0], self.accounts[1], 1024, tx_format=TxFormat.RLP_SIGNED) 155 tx_data = unpack(bytes.fromhex(strip_0x(o)), self.chain_spec) 156 157 rcpt = { 158 'transaction_hash': os.urandom(32).hex(), 159 'block_hash': os.urandom(32).hex(), 160 'status': 1, 161 'block_number': 42, 162 'transaction_index': 1, 163 'logs': [], 164 'gas_used': 21000, 165 } 166 with self.assertRaises(ValueError): 167 tx = Tx(tx_data, rcpt=rcpt) 168 169 rcpt['transaction_hash'] = tx_data['hash'] 170 tx = Tx(tx_data, rcpt=rcpt) 171 172 block_hash = os.urandom(32).hex() 173 block = Block({ 174 'hash': block_hash, 175 'number': 42, 176 'timestamp': 13241324, 177 'transactions': [], 178 'author': os.urandom(20).hex(), 179 'gas_used': 21000, 180 'gas_limit': '0x2345', 181 'parent_hash': None, 182 }) 183 184 block.txs = [add_0x(tx_data['hash'])] 185 with self.assertRaises(ValueError): 186 tx = Tx(tx_data, rcpt=rcpt, block=block) 187 188 rcpt['block_hash'] = block.hash 189 tx = Tx(tx_data, rcpt=rcpt, block=block) 190 191 192 if __name__ == '__main__': 193 unittest.main()