test_addressdeclarator_kv.py (11196B)
1 # standard imports 2 import os 3 import unittest 4 import json 5 import logging 6 import hashlib 7 8 # external imports 9 from chainlib.eth.unittest.ethtester import EthTesterCase 10 from chainlib.eth.contract import ( 11 ABIContractEncoder, 12 ABIContractType, 13 ) 14 from chainlib.eth.nonce import RPCNonceOracle 15 from chainlib.eth.tx import receipt 16 from giftable_erc20_token import GiftableToken 17 from hexathon import ( 18 add_0x, 19 strip_0x, 20 ) 21 22 # local imports 23 from eth_address_declarator.declarator import AddressDeclarator 24 from eth_address_declarator import Declarator 25 from eth_address_declarator.unittest import TestAddressDeclaratorBase 26 27 logging.basicConfig(level=logging.DEBUG) 28 logg = logging.getLogger() 29 30 testdir = os.path.dirname(__file__) 31 32 description = '0x{:<064s}'.format(b'foo'.hex()) 33 34 35 class TestAddressDeclarator(TestAddressDeclaratorBase): 36 37 def setUp(self): 38 super(TestAddressDeclarator, self).setUp() 39 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 40 41 #c = GiftableToken(signer=self.signer, nonce_oracle=nonce_oracle, chain_id=self.chain_spec.chain_id()) 42 c = GiftableToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 43 (tx_hash_hex, o) = c.constructor(self.accounts[0], 'BarToken', 'BAR', 6) 44 self.rpc.do(o) 45 46 o = receipt(tx_hash_hex) 47 r = self.rpc.do(o) 48 self.assertEqual(r['status'], 1) 49 50 self.bar_token_address = r['contract_address'] 51 52 53 def test_basic(self): 54 k = add_0x(os.urandom(32).hex()) 55 v = add_0x(os.urandom(32).hex()) 56 57 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 58 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 59 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[0], self.foo_token_address, v, topic=k) 60 self.rpc.do(o) 61 o = receipt(tx_hash_hex) 62 r = self.rpc.do(o) 63 self.assertEqual(r['status'], 1) 64 65 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 66 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 67 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, v, topic=k) 68 self.rpc.do(o) 69 o = receipt(tx_hash_hex) 70 r = self.rpc.do(o) 71 self.assertEqual(r['status'], 1) 72 73 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 74 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 75 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[0], self.bar_token_address, v, topic=k) 76 self.rpc.do(o) 77 78 o = c.declarator_count(self.address, self.foo_token_address, sender_address=self.accounts[0]) 79 r = self.rpc.do(o) 80 self.assertEqual(c.parse_declarator_count(r), 2) 81 82 o = c.declarator_count(self.address, self.bar_token_address, sender_address=self.accounts[0]) 83 r = self.rpc.do(o) 84 self.assertEqual(c.parse_declarator_count(r), 1) 85 86 87 def test_get_single_declaration(self): 88 k = add_0x(os.urandom(32).hex()) 89 v = add_0x(os.urandom(32).hex()) 90 91 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 92 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 93 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, v, topic=k) 94 self.rpc.do(o) 95 o = receipt(tx_hash_hex) 96 r = self.rpc.do(o) 97 self.assertEqual(r['status'], 1) 98 99 o = c.declaration(self.address, self.accounts[1], self.foo_token_address, topic=k, sender_address=self.accounts[0]) 100 r = self.rpc.do(o) 101 self.assertEqual(strip_0x(r)[128:], strip_0x(v)) 102 103 104 def test_kv_declarations(self): 105 106 k = add_0x(os.urandom(32).hex()) 107 v = add_0x(os.urandom(32).hex()) 108 w = add_0x(os.urandom(32).hex()) 109 110 k_two = add_0x(os.urandom(32).hex()) 111 v_two = add_0x(os.urandom(32).hex()) 112 113 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 114 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 115 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, v, topic=k) 116 self.rpc.do(o) 117 o = receipt(tx_hash_hex) 118 r = self.rpc.do(o) 119 self.assertEqual(r['status'], 1) 120 121 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 122 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 123 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, w, topic=k) 124 self.rpc.do(o) 125 o = receipt(tx_hash_hex) 126 r = self.rpc.do(o) 127 self.assertEqual(r['status'], 1) 128 129 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, v_two, topic=k_two) 130 self.rpc.do(o) 131 o = receipt(tx_hash_hex) 132 r = self.rpc.do(o) 133 self.assertEqual(r['status'], 1) 134 # nonce_oracle = RPCNonceOracle(self.accounts[2], self.rpc) 135 # c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 136 # (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[2], self.foo_token_address, k, v) 137 # self.rpc.do(o) 138 # 139 # c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 140 # (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[2], self.bar_token_address, k, v) 141 # self.rpc.do(o) 142 143 o = c.declaration(self.address, self.accounts[1], self.foo_token_address, topic=k, sender_address=self.accounts[0]) 144 r = self.rpc.do(o) 145 self.assertEqual(strip_0x(r)[128:], strip_0x(v) + strip_0x(w)) 146 147 o = c.declaration(self.address, self.accounts[1], self.foo_token_address, topic=k_two, sender_address=self.accounts[0]) 148 r = self.rpc.do(o) 149 self.assertEqual(strip_0x(r)[128:], strip_0x(v_two)) 150 151 152 def test_declarator_to_subject(self): 153 k = add_0x(os.urandom(32).hex()) 154 v = add_0x(os.urandom(32).hex()) 155 156 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 157 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 158 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, v, topic=k) 159 self.rpc.do(o) 160 161 nonce_oracle = RPCNonceOracle(self.accounts[2], self.rpc) 162 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 163 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[2], self.foo_token_address, v, topic=k) 164 self.rpc.do(o) 165 166 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 167 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 168 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.bar_token_address, v, topic=k) 169 self.rpc.do(o) 170 171 o = c.declaration_address_at(self.address, self.accounts[1], 0, sender_address=self.accounts[0]) 172 r = self.rpc.do(o) 173 self.assertEqual(c.parse_declaration_address_at(r), strip_0x(self.foo_token_address)) 174 175 o = c.declaration_address_at(self.address, self.accounts[2], 0, sender_address=self.accounts[0]) 176 r = self.rpc.do(o) 177 self.assertEqual(c.parse_declaration_address_at(r), strip_0x(self.foo_token_address)) 178 179 o = c.declaration_address_at(self.address, self.accounts[1], 1, sender_address=self.accounts[0]) 180 r = self.rpc.do(o) 181 self.assertEqual(c.parse_declaration_address_at(r), strip_0x(self.bar_token_address)) 182 183 184 def test_subject_to_declarator(self): 185 k = add_0x(os.urandom(32).hex()) 186 v = add_0x(os.urandom(32).hex()) 187 188 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 189 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 190 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.foo_token_address, v, topic=k) 191 self.rpc.do(o) 192 193 nonce_oracle = RPCNonceOracle(self.accounts[2], self.rpc) 194 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 195 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[2], self.foo_token_address, v, topic=k) 196 self.rpc.do(o) 197 198 nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc) 199 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 200 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[1], self.bar_token_address, v, topic=k) 201 self.rpc.do(o) 202 203 o = c.declarator_address_at(self.address, self.foo_token_address, 0, sender_address=self.accounts[0]) 204 r = self.rpc.do(o) 205 self.assertEqual(c.parse_declaration_address_at(r), strip_0x(self.accounts[1])) 206 207 o = c.declarator_address_at(self.address, self.foo_token_address, 1, sender_address=self.accounts[0]) 208 r = self.rpc.do(o) 209 self.assertEqual(c.parse_declaration_address_at(r), strip_0x(self.accounts[2])) 210 211 212 def test_three_first(self): 213 k = add_0x(os.urandom(32).hex()) 214 v = [] 215 for i in range(3): 216 v.append(add_0x(os.urandom(32).hex())) 217 218 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 219 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 220 221 for proof in v: 222 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[0], self.foo_token_address, proof, topic=k) 223 self.rpc.do(o) 224 225 o = c.declarator_count(self.address, self.foo_token_address, sender_address=self.accounts[0]) 226 r = self.rpc.do(o) 227 self.assertEqual(c.parse_declarator_count(r), 1) 228 229 o = c.declaration(self.address, self.accounts[0], self.foo_token_address, topic=k, sender_address=self.accounts[0]) 230 r = self.rpc.do(o) 231 proofs = c.parse_declaration(r) 232 self.assertEqual(len(proofs), 3) 233 234 for i in range(3): 235 self.assertEqual(proofs[i], strip_0x(v[i])) 236 237 238 def test_three_first_different(self): 239 k = add_0x(os.urandom(32).hex()) 240 v = [] 241 a = [] 242 for i in range(3): 243 v.append(add_0x(os.urandom(32).hex())) 244 a.append(add_0x(os.urandom(20).hex())) 245 246 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 247 c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 248 249 for i in range(3): 250 (tx_hash_hex, o) = c.add_declaration(self.address, self.accounts[0], a[i], v[i], topic=k) 251 self.rpc.do(o) 252 253 for i in range(3): 254 o = c.declarator_count(self.address, a[i], sender_address=self.accounts[0]) 255 r = self.rpc.do(o) 256 self.assertEqual(c.parse_declarator_count(r), 1) 257 258 o = c.declaration(self.address, self.accounts[0], a[i], topic=k, sender_address=self.accounts[0]) 259 r = self.rpc.do(o) 260 proofs = c.parse_declaration(r) 261 self.assertEqual(len(proofs), 1) 262 self.assertEqual(proofs[0], strip_0x(v[i])) 263 264 265 if __name__ == '__main__': 266 unittest.main()