eth-accounts-index

Accounts index evm contract tooling with permissioned writes
Info | Log | Files | Refs

test_app.py (7828B)

      1 # standard imports
      2 import os
      3 import unittest
      4 import json
      5 import logging
      6 
      7 # external imports
      8 from chainlib.eth.unittest.ethtester import EthTesterCase
      9 from chainlib.connection import RPCConnection
     10 from chainlib.eth.nonce import RPCNonceOracle
     11 from chainlib.eth.address import to_checksum_address
     12 from chainlib.eth.tx import (
     13         receipt,
     14         transaction,
     15         TxFormat,
     16         )
     17 from chainlib.eth.contract import (
     18         abi_decode_single,
     19         ABIContractType,
     20         )
     21 from chainlib.eth.block import (
     22         block_latest,
     23         block_by_number,
     24         )
     25 
     26 # local imports
     27 from eth_accounts_index.registry import AccountRegistry
     28 from eth_accounts_index import AccountsIndex
     29 from eth_accounts_index.unittest import TestAccountsIndex
     30 
     31 logging.basicConfig(level=logging.DEBUG)
     32 logg = logging.getLogger()
     33 
     34 testdir = os.path.dirname(__file__)
     35 
     36 
     37 class Test(TestAccountsIndex):
     38 
     39     def setUp(self):
     40         super(Test, self).setUp()
     41 
     42 
     43     def test_1_count(self):
     44         #o = self.o.count(self.address, sender_address=self.accounts[0])
     45         c = AccountsIndex(self.chain_spec)
     46         o = c.entry_count(self.address, sender_address=self.accounts[0])
     47         r = self.conn.do(o)
     48         r = abi_decode_single(ABIContractType.UINT256, r)
     49         self.assertEqual(r, 0)
     50 
     51 
     52     def test_2_add(self):
     53         b = os.urandom(20)
     54         a = to_checksum_address(b.hex())
     55         b = os.urandom(20)
     56         b = to_checksum_address(b.hex())
     57 
     58         nonce_oracle = RPCNonceOracle(self.accounts[0], self.conn)
     59         c = AccountsIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
     60         (tx_hash, o) = c.add(self.address, self.accounts[0], a)
     61         self.conn.do(o)
     62         o = receipt(tx_hash)
     63         r = self.conn.do(o)
     64         self.assertEqual(r['status'], 1)
     65 
     66         self.helper.mine_block()
     67         o = c.have(self.address, a, sender_address=self.accounts[0])
     68         r = self.conn.do(o)
     69         self.assertEqual(int(r, 16), 1)
     70 
     71         o = c.have(self.address, b, sender_address=self.accounts[0])
     72         r = self.conn.do(o)
     73         self.assertEqual(int(r, 16), 0)
     74 
     75 
     76     def test_3_add_rlpsigned(self):
     77         b = os.urandom(20)
     78         a = to_checksum_address(b.hex())
     79 
     80         nonce_oracle = RPCNonceOracle(self.accounts[0], self.conn)
     81         c = AccountsIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
     82         (tx_hash, o) = c.add(self.address, self.accounts[0], a) #, tx_format=TxFormat.RLP_SIGNED)
     83         r = self.conn.do(o)
     84         self.assertEqual(tx_hash, r)
     85         logg.debug('o {}'.format(o))
     86 
     87 
     88     def test_4_indices(self):
     89         nonce_oracle = RPCNonceOracle(self.accounts[0], self.conn)
     90         c = AccountsIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
     91 
     92         o = c.entry_count(self.address, sender_address=self.accounts[0])
     93         r = self.conn.do(o)
     94         count = c.parse_entry_count(r)
     95         self.assertEqual(count, 0)
     96 
     97         b = os.urandom(20)
     98         a = to_checksum_address(b.hex())
     99         (tx_hash, o) = c.add(self.address, self.accounts[0], a)
    100         r = self.conn.do(o)
    101 
    102         b = os.urandom(20)
    103         aa = to_checksum_address(b.hex())
    104         (tx_hash, o) = c.add(self.address, self.accounts[0], aa)
    105         r = self.conn.do(o)
    106 
    107         o = c.entry_count(self.address, sender_address=self.accounts[0])
    108         r = self.conn.do(o)
    109         count = c.parse_entry_count(r)
    110         self.assertEqual(count, 2)
    111 
    112 
    113     def test_5_deactivate(self):
    114         nonce_oracle = RPCNonceOracle(self.accounts[0], self.conn)
    115         c = AccountsIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
    116 
    117         b = os.urandom(20)
    118         a = to_checksum_address(b.hex())
    119         (tx_hash, o) = c.add(self.address, self.accounts[0], a)
    120         r = self.conn.do(o)
    121 
    122         o = c.entry_count(self.address, sender_address=self.accounts[0])
    123         r = self.conn.do(o)
    124         count = c.parse_entry_count(r)
    125         self.assertEqual(count, 1)
    126 
    127         b = os.urandom(20)
    128         a = to_checksum_address(b.hex())
    129         (tx_hash, o) = c.add(self.address, self.accounts[0], a)
    130         r = self.conn.do(o)
    131 
    132         o = c.entry_count(self.address, sender_address=self.accounts[0])
    133         r = self.conn.do(o)
    134         count = c.parse_entry_count(r)
    135         self.assertEqual(count, 2)
    136 
    137         (tx_hash, o) = c.deactivate(self.address, self.accounts[0], a)
    138         r = self.conn.do(o)
    139         o = receipt(r)
    140         r = self.conn.do(o)
    141         self.assertEqual(r['status'], 1)
    142 
    143         o = c.entry_count(self.address, sender_address=self.accounts[0])
    144         r = self.conn.do(o)
    145         count = c.parse_entry_count(r)
    146         self.assertEqual(count, 2)
    147 
    148         o = c.have(self.address, a, sender_address=self.accounts[0])
    149         r = self.conn.do(o)
    150         self.assertEqual(int(r, 16), 1)
    151 
    152         o = c.is_active(self.address, a, sender_address=self.accounts[0])
    153         r = self.conn.do(o)
    154         self.assertEqual(int(r, 16), 0)
    155 
    156         (tx_hash, o) = c.activate(self.address, self.accounts[0], a)
    157         r = self.conn.do(o)
    158         o = receipt(r)
    159         r = self.conn.do(o)
    160         self.assertEqual(r['status'], 1)
    161 
    162         o = c.have(self.address, a, sender_address=self.accounts[0])
    163         r = self.conn.do(o)
    164         self.assertEqual(int(r, 16), 1)
    165 
    166         o = c.is_active(self.address, a, sender_address=self.accounts[0])
    167         r = self.conn.do(o)
    168         self.assertEqual(int(r, 16), 1)
    169 
    170 
    171     def test_6_remove(self):
    172         nonce_oracle = RPCNonceOracle(self.accounts[0], self.conn)
    173         c = AccountsIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
    174 
    175         b = os.urandom(20)
    176         a = to_checksum_address(b.hex())
    177         (tx_hash, o) = c.add(self.address, self.accounts[0], a)
    178         r = self.conn.do(o)
    179 
    180         o = c.entry_count(self.address, sender_address=self.accounts[0])
    181         r = self.conn.do(o)
    182         self.assertEqual(int(r, 16), 1)
    183 
    184         (tx_hash, o) = c.remove(self.address, self.accounts[0], a)
    185         r = self.conn.do(o)
    186         o = receipt(r)
    187         r = self.conn.do(o)
    188         self.assertEqual(r['status'], 1)
    189 
    190         o = c.entry_count(self.address, sender_address=self.accounts[0])
    191         r = self.conn.do(o)
    192         self.assertEqual(int(r, 16), 0)
    193 
    194         addrs = []
    195         for i in range(8):
    196             b = os.urandom(20)
    197             addrs.append(to_checksum_address(b.hex()))
    198             (tx_hash, o) = c.add(self.address, self.accounts[0], addrs[i])
    199             r = self.conn.do(o)
    200             o = receipt(tx_hash)
    201             r = self.conn.do(o)
    202             self.assertEqual(r['status'], 1)
    203 
    204         o = c.entry_count(self.address, sender_address=self.accounts[0])
    205         r = self.conn.do(o)
    206         self.assertEqual(int(r, 16), 8)
    207 
    208         (tx_hash, o) = c.remove(self.address, self.accounts[0], addrs[4])
    209         r = self.conn.do(o)
    210         o = receipt(r)
    211         r = self.conn.do(o)
    212         self.assertEqual(r['status'], 1)
    213 
    214         r = []
    215         for i in range(7):
    216             o = c.entry(self.address, i, sender_address=self.accounts[0])
    217             r.append(self.conn.do(o))
    218 
    219         self.assertEqual(len(r), 7)
    220         self.assertNotIn(addrs[4], r)
    221 
    222 
    223     def test_7_time(self):
    224         nonce_oracle = RPCNonceOracle(self.accounts[0], self.conn)
    225         c = AccountsIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
    226 
    227         b = os.urandom(20)
    228         a = to_checksum_address(b.hex())
    229         (tx_hash, o) = c.add(self.address, self.accounts[0], a)
    230         self.conn.do(o)
    231 
    232         o = block_latest()
    233         r = self.conn.do(o)
    234         o = block_by_number(r)
    235         r = self.conn.do(o)
    236         t = r['timestamp']
    237 
    238         o = c.time(self.address, a, sender_address=self.accounts[0])
    239         r = self.conn.do(o)
    240         self.assertEqual(t, int(r, 16))
    241 
    242 
    243 if __name__ == '__main__':
    244     unittest.main()