test_period.py (9148B)
1 # standard imports 2 import os 3 import unittest 4 import json 5 import logging 6 7 # external imports 8 from chainlib.eth.constant import ZERO_ADDRESS 9 from chainlib.eth.nonce import RPCNonceOracle 10 from chainlib.eth.tx import ( 11 receipt, 12 TxFactory, 13 TxFormat, 14 ) 15 from chainlib.eth.contract import ( 16 ABIContractEncoder, 17 ABIContractType, 18 ) 19 from hexathon import same as hex_same 20 from hexathon import strip_0x 21 from dexif import from_fixed 22 23 # local imports 24 from erc20_demurrage_token import DemurrageToken 25 from erc20_demurrage_token import DemurrageRedistribution 26 27 # test imports 28 from erc20_demurrage_token.unittest import TestDemurrageDefault 29 30 logging.basicConfig(level=logging.DEBUG) 31 logg = logging.getLogger() 32 33 testdir = os.path.dirname(__file__) 34 35 class TestPeriod(TestDemurrageDefault): 36 37 def test_period_and_amount(self): 38 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 39 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 40 (tx_hash, o) = c.mint_to(self.address, self.accounts[0], self.accounts[1], 1024) 41 r = self.rpc.do(o) 42 o = receipt(tx_hash) 43 r = self.rpc.do(o) 44 self.assertEqual(r['status'], 1) 45 46 for i in range(100): 47 self.backend.time_travel(self.start_time + int((self.period_seconds / 100) * (i + 1))) 48 (tx_hash, o) = c.apply_demurrage(self.address, self.accounts[0], 0) 49 self.rpc.do(o) 50 o = receipt(tx_hash) 51 r = self.rpc.do(o) 52 self.assertEqual(r['status'], 1) 53 54 for lg in r['logs']: 55 if hex_same(lg['topics'][0], '1c9c74563c32efd114cb36fb5e432d9386c8254d08456614804a33a3088ab736'): 56 self.assert_equal_decimals(0.98, from_fixed(strip_0x(lg['data'])), 2) 57 58 59 def test_period_demurrage(self): 60 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 61 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 62 (tx_hash, o) = c.mint_to(self.address, self.accounts[0], self.accounts[1], 1024) 63 r = self.rpc.do(o) 64 o = receipt(tx_hash) 65 r = self.rpc.do(o) 66 self.assertEqual(r['status'], 1) 67 68 self.backend.time_travel(self.start_time + self.period_seconds + int(self.period_seconds / 2)) 69 (tx_hash, o) = c.change_period(self.address, self.accounts[0]) 70 self.rpc.do(o) 71 o = receipt(tx_hash) 72 r = self.rpc.do(o) 73 self.assertEqual(r['status'], 1) 74 75 for lg in r['logs']: 76 if hex_same(lg['topics'][0], '1c9c74563c32efd114cb36fb5e432d9386c8254d08456614804a33a3088ab736'): 77 self.assert_equal_decimals(0.9701, from_fixed(strip_0x(lg['data'])), 4) 78 79 o = c.redistributions(self.address, 1, sender_address=self.accounts[0]) 80 r = self.rpc.do(o) 81 redistribution_data = c.parse_redistributions(r) 82 redistribution = DemurrageRedistribution(redistribution_data) 83 logg.debug('fixxx {}'.format(redistribution.demurrage)) 84 85 86 def test_period(self): 87 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 88 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 89 (tx_hash, o) = c.mint_to(self.address, self.accounts[0], self.accounts[1], 1024) 90 r = self.rpc.do(o) 91 o = receipt(tx_hash) 92 r = self.rpc.do(o) 93 self.assertEqual(r['status'], 1) 94 95 self.backend.time_travel(self.start_time + self.period_seconds) 96 97 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 98 (tx_hash, o) = c.change_period(self.address, self.accounts[0]) 99 r = self.rpc.do(o) 100 o = receipt(tx_hash) 101 r = self.rpc.do(o) 102 self.assertEqual(r['status'], 1) 103 104 o = c.redistributions(self.address, 1, sender_address=self.accounts[0]) 105 r = self.rpc.do(o) 106 redistribution = c.parse_redistributions(r) 107 108 o = c.to_redistribution_period(self.address, redistribution, sender_address=self.accounts[0]) 109 r = self.rpc.do(o) 110 period = c.parse_to_redistribution_period(r) 111 self.assertEqual(2, period) 112 113 o = c.actual_period(self.address, sender_address=self.accounts[0]) 114 r = self.rpc.do(o) 115 period = c.parse_actual_period(r) 116 self.assertEqual(2, period) 117 118 o = c.to_redistribution_demurrage_modifier(self.address, redistribution, sender_address=self.accounts[0]) 119 r = self.rpc.do(o) 120 period = from_fixed(r) 121 redistro = DemurrageRedistribution(redistribution) 122 logg.debug('redistro {} {}'.format(redistro, period)) 123 124 self.backend.time_travel(self.start_time + self.period_seconds * 2) 125 126 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 127 (tx_hash, o) = c.change_period(self.address, self.accounts[0]) 128 r = self.rpc.do(o) 129 o = receipt(tx_hash) 130 r = self.rpc.do(o) 131 self.assertEqual(r['status'], 1) 132 133 o = c.redistributions(self.address, 2, sender_address=self.accounts[0]) 134 r = self.rpc.do(o) 135 redistribution = c.parse_redistributions(r) 136 137 o = c.to_redistribution_demurrage_modifier(self.address, redistribution, sender_address=self.accounts[0]) 138 r = self.rpc.do(o) 139 period = from_fixed(r) 140 141 # allow test code float rounding error to billionth 142 modifier = (1 - (self.tax_level / 1000000)) ** ((self.period_seconds * 2) / 60) 143 modifier *= 10 ** 9 144 modifier = int(modifier) * (10 ** (28 - 9)) 145 146 period /= (10 ** (28 - 9)) 147 period = int(period) * (10 ** (28 - 9)) 148 self.assertEqual(modifier, period) 149 150 151 def test_change_sink(self): 152 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 153 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 154 155 o = c.balance_of(self.address, ZERO_ADDRESS, sender_address=self.accounts[0]) 156 r = self.rpc.do(o) 157 balance = c.parse_balance_of(r) 158 self.assertEqual(balance, 0) 159 160 (tx_hash, o) = c.mint_to(self.address, self.accounts[0], self.accounts[1], 102400000000) 161 r = self.rpc.do(o) 162 o = receipt(tx_hash) 163 r = self.rpc.do(o) 164 self.assertEqual(r['status'], 1) 165 166 self.backend.time_travel(self.start_time + self.period_seconds + 1) 167 168 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 169 (tx_hash, o) = c.change_period(self.address, self.accounts[0]) 170 r = self.rpc.do(o) 171 o = receipt(tx_hash) 172 r = self.rpc.do(o) 173 self.assertEqual(r['status'], 1) 174 175 o = c.balance_of(self.address, self.sink_address, sender_address=self.accounts[0]) 176 r = self.rpc.do(o) 177 balance = c.parse_balance_of(r) 178 self.assertGreater(balance, 0) 179 old_sink_balance = balance 180 181 o = c.balance_of(self.address, self.accounts[3], sender_address=self.accounts[0]) 182 r = self.rpc.do(o) 183 balance = c.parse_balance_of(r) 184 self.assertEqual(balance, 0) 185 186 nonce_oracle = RPCNonceOracle(self.accounts[5], self.rpc) 187 c = TxFactory(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 188 enc = ABIContractEncoder() 189 enc.method('setSinkAddress') 190 enc.typ(ABIContractType.ADDRESS) 191 enc.address(self.accounts[3]) 192 data = enc.get() 193 o = c.template(self.accounts[5], self.address, use_nonce=True) 194 o = c.set_code(o, data) 195 (tx_hash, o) = c.finalize(o, TxFormat.JSONRPC) 196 r = self.rpc.do(o) 197 o = receipt(tx_hash) 198 r = self.rpc.do(o) 199 self.assertEqual(r['status'], 0) 200 201 nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc) 202 c = TxFactory(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 203 enc = ABIContractEncoder() 204 enc.method('setSinkAddress') 205 enc.typ(ABIContractType.ADDRESS) 206 enc.address(self.accounts[3]) 207 data = enc.get() 208 o = c.template(self.accounts[0], self.address, use_nonce=True) 209 o = c.set_code(o, data) 210 (tx_hash, o) = c.finalize(o, TxFormat.JSONRPC) 211 r = self.rpc.do(o) 212 o = receipt(tx_hash) 213 r = self.rpc.do(o) 214 self.assertEqual(r['status'], 1) 215 216 self.backend.time_travel(self.start_time + (self.period_seconds * 2) + 1) 217 218 c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle) 219 (tx_hash, o) = c.change_period(self.address, self.accounts[0]) 220 r = self.rpc.do(o) 221 o = receipt(tx_hash) 222 r = self.rpc.do(o) 223 self.assertEqual(r['status'], 1) 224 225 o = c.balance_of(self.address, self.sink_address, sender_address=self.accounts[0]) 226 r = self.rpc.do(o) 227 balance = c.parse_balance_of(r) 228 self.assertLess(balance, old_sink_balance) 229 230 o = c.balance_of(self.address, self.accounts[3], sender_address=self.accounts[0]) 231 r = self.rpc.do(o) 232 balance = c.parse_balance_of(r) 233 self.assertGreater(balance, 0) 234 235 236 if __name__ == '__main__': 237 unittest.main()