test_request.py (9880B)
1 # standard imports 2 import os 3 import unittest 4 import logging 5 import random 6 7 # external imports 8 from chainlib.eth.nonce import RPCNonceOracle 9 from chainlib.eth.address import is_same_address 10 from giftable_erc20_token import GiftableToken 11 from eth_erc20 import ERC20 12 from chainlib.eth.tx import receipt 13 14 # local imports 15 from erc20_transfer_authorization import TransferAuthorization 16 from erc20_transfer_authorization.unittest import TestBase 17 18 #logging.basicConfig(level=logging.DEBUG) 19 logg = logging.getLogger() 20 21 testdir = os.path.dirname(__file__) 22 23 24 class TestRequest(TestBase): 25 26 def setUp(self): 27 super(TestRequest, self).setUp() 28 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 29 self.c = TransferAuthorization(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 30 31 c = GiftableToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 32 (tx_hash_hex, o) = c.mint_to(self.token_address, self.accounts[0], self.accounts[9], 10000) 33 self.rpc.do(o) 34 o = receipt(tx_hash_hex) 35 r = self.rpc.do(o) 36 self.assertEqual(r['status'], 1) 37 38 nonce_oracle = RPCNonceOracle(self.accounts[9], self.rpc) 39 c = ERC20(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 40 (tx_hash_hex, o) = c.approve(self.token_address, self.accounts[9], self.address, 10000) 41 self.rpc.do(o) 42 o = receipt(tx_hash_hex) 43 r = self.rpc.do(o) 44 self.assertEqual(r['status'], 1) 45 46 47 def test_basic(self): 48 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 49 r = self.rpc.do(o) 50 serial = self.c.parse_count(r) 51 self.assertEqual(serial, 0) 52 53 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 54 r = self.rpc.do(o) 55 serial = self.c.parse_count(r) 56 self.assertEqual(serial, 0) 57 58 (tx_hash_hex, o) = self.c.create_request(self.address, self.accounts[0], self.accounts[9], self.accounts[2], self.token_address, 1) 59 self.rpc.do(o) 60 61 (tx_hash_hex, o) = self.c.create_request(self.address, self.accounts[0], self.accounts[9], self.accounts[3], self.token_address, 2) 62 self.rpc.do(o) 63 64 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 65 r = self.rpc.do(o) 66 serial = self.c.parse_count(r) 67 self.assertEqual(serial, 2) 68 69 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 70 r = self.rpc.do(o) 71 serial = self.c.parse_count(r) 72 self.assertEqual(serial, 1) 73 74 (tx_hash_hex, o) = self.c.yay(self.address, self.accounts[0], 1) 75 self.rpc.do(o) 76 77 (tx_hash_hex, o) = self.c.check_result(self.address, self.accounts[0], 1) 78 self.rpc.do(o) 79 80 (tx_hash_hex, o) = self.c.execute_request(self.address, self.accounts[0], 1) 81 self.rpc.do(o) 82 o = receipt(tx_hash_hex) 83 r = self.rpc.do(o) 84 self.assertEqual(r['status'], 1) 85 86 o = self.c.count(self.address, sender_address=self.accounts[0]) 87 r = self.rpc.do(o) 88 count = self.c.parse_count(r) 89 self.assertEqual(count, 1) 90 91 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 92 r = self.rpc.do(o) 93 serial = self.c.parse_count(r) 94 self.assertEqual(serial, 2) 95 96 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 97 r = self.rpc.do(o) 98 serial = self.c.parse_count(r) 99 self.assertEqual(serial, 2) 100 101 (tx_hash_hex, o) = self.c.nay(self.address, self.accounts[0], 2) 102 self.rpc.do(o) 103 104 (tx_hash_hex, o) = self.c.check_result(self.address, self.accounts[0], 2) 105 self.rpc.do(o) 106 107 o = self.c.count(self.address, sender_address=self.accounts[0]) 108 r = self.rpc.do(o) 109 count = self.c.parse_count(r) 110 self.assertEqual(count, 0) 111 112 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 113 r = self.rpc.do(o) 114 serial = self.c.parse_count(r) 115 self.assertEqual(serial, 2) 116 117 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 118 r = self.rpc.do(o) 119 serial = self.c.parse_count(r) 120 self.assertEqual(serial, 0) 121 122 123 def test_hi_lo(self): 124 for i in range(5): 125 (tx_hash_hex, o) = self.c.create_request(self.address, self.accounts[0], self.accounts[9], self.accounts[i+1], self.token_address, i+1) 126 self.rpc.do(o) 127 128 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 129 r = self.rpc.do(o) 130 serial = self.c.parse_count(r) 131 self.assertEqual(serial, 5) 132 133 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 134 r = self.rpc.do(o) 135 serial = self.c.parse_count(r) 136 self.assertEqual(serial, 1) 137 138 (tx_hash_hex, o) = self.c.yay(self.address, self.accounts[0], 6) 139 self.rpc.do(o) 140 o = receipt(tx_hash_hex) 141 r = self.rpc.do(o) 142 self.assertEqual(r['status'], 0) 143 144 (tx_hash_hex, o) = self.c.yay(self.address, self.accounts[0], 5) 145 self.rpc.do(o) 146 147 (tx_hash_hex, o) = self.c.check_result(self.address, self.accounts[0], 5) 148 self.rpc.do(o) 149 150 (tx_hash_hex, o) = self.c.execute_request(self.address, self.accounts[0], 5) 151 self.rpc.do(o) 152 153 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 154 r = self.rpc.do(o) 155 serial = self.c.parse_count(r) 156 self.assertEqual(serial, 5) 157 158 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 159 r = self.rpc.do(o) 160 serial = self.c.parse_count(r) 161 self.assertEqual(serial, 1) 162 163 o = self.c.requests(self.address, 5, sender_address=self.accounts[0]) 164 r = self.rpc.do(o) 165 request = self.c.parse_request(r) 166 self.assertTrue(request.is_executed()) 167 168 o = self.c.count(self.address, sender_address=self.accounts[0]) 169 r = self.rpc.do(o) 170 count = self.c.parse_count(r) 171 self.assertEqual(count, 4) 172 173 (tx_hash_hex, o) = self.c.create_request(self.address, self.accounts[0], self.accounts[9], self.accounts[0], self.token_address, 100) 174 self.rpc.do(o) 175 176 o = self.c.count(self.address, sender_address=self.accounts[0]) 177 r = self.rpc.do(o) 178 count = self.c.parse_count(r) 179 self.assertEqual(count, 5) 180 181 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 182 r = self.rpc.do(o) 183 serial = self.c.parse_count(r) 184 self.assertEqual(serial, 6) 185 186 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 187 r = self.rpc.do(o) 188 serial = self.c.parse_count(r) 189 self.assertEqual(serial, 1) 190 191 (tx_hash_hex, o) = self.c.yay(self.address, self.accounts[0], 6) 192 self.rpc.do(o) 193 194 (tx_hash_hex, o) = self.c.check_result(self.address, self.accounts[0], 6) 195 self.rpc.do(o) 196 197 (tx_hash_hex, o) = self.c.execute_request(self.address, self.accounts[0], 6) 198 self.rpc.do(o) 199 200 o = self.c.requests(self.address, 6, sender_address=self.accounts[0]) 201 r = self.rpc.do(o) 202 request = self.c.parse_request(r) 203 self.assertTrue(request.is_executed()) 204 205 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 206 r = self.rpc.do(o) 207 serial = self.c.parse_count(r) 208 self.assertEqual(serial, 6) 209 210 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 211 r = self.rpc.do(o) 212 serial = self.c.parse_count(r) 213 self.assertEqual(serial, 1) 214 215 c = ERC20(self.chain_spec) 216 o = c.balance(self.token_address, self.accounts[0], sender_address=self.accounts[0]) 217 r = self.rpc.do(o) 218 balance = c.parse_balance(r) 219 self.assertEqual(balance, 100) 220 221 o = c.balance(self.token_address, self.accounts[5], sender_address=self.accounts[0]) 222 r = self.rpc.do(o) 223 balance = c.parse_balance(r) 224 self.assertEqual(balance, 5) 225 226 o = c.balance(self.token_address, self.accounts[9], sender_address=self.accounts[0]) 227 r = self.rpc.do(o) 228 balance = c.parse_balance(r) 229 self.assertEqual(balance, 10000-100-5) 230 231 232 def test_hi_lo_shuffle(self): 233 for i in range(100): 234 (tx_hash_hex, o) = self.c.create_request(self.address, self.accounts[0], self.accounts[9], self.accounts[(i+1)%8], self.token_address, i+1) 235 self.rpc.do(o) 236 237 v = list(range(1, 101)) 238 c = 0 239 random.shuffle(v) 240 for i in v: 241 logg.info('processing request number {} at index {}'.format(c, i)) 242 (tx_hash_hex, o) = self.c.nay(self.address, self.accounts[0], i) 243 self.rpc.do(o) 244 245 (tx_hash_hex, o) = self.c.check_result(self.address, self.accounts[0], i) 246 self.rpc.do(o) 247 o = receipt(tx_hash_hex) 248 r = self.rpc.do(o) 249 self.assertEqual(r['status'], 1) 250 251 o = self.c.requests(self.address, i, sender_address=self.accounts[0]) 252 r = self.rpc.do(o) 253 request = self.c.parse_request(r) 254 self.assertTrue(request.is_rejected()) 255 256 c += 1 257 o = self.c.count(self.address, sender_address=self.accounts[0]) 258 r = self.rpc.do(o) 259 count = self.c.parse_count(r) 260 self.assertEqual(count, 100-c) 261 262 263 o = self.c.last_serial(self.address, sender_address=self.accounts[0]) 264 r = self.rpc.do(o) 265 serial = self.c.parse_count(r) 266 self.assertEqual(serial, 100) 267 268 o = self.c.next_serial(self.address, sender_address=self.accounts[0]) 269 r = self.rpc.do(o) 270 serial = self.c.parse_count(r) 271 self.assertEqual(serial, 0) 272 273 274 if __name__ == '__main__': 275 unittest.main()