cicnet/erc20-demurrage-token
5
0
Fork 0
mirror of git://holbrook.no/erc20-demurrage-token synced 2026-04-28 19:01:03 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch lash/readmemore

Browse Source
This commit is contained in:
lash
2023-03-22 10:17:40 +00:00
parent 3ebb645077 f052529935
commit acd5eb3d7b
44 changed files with 2703 additions and 3086 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
python/erc20_demurrage_token/__init__.py
View File

@@ -1,4 +1,8 @@
from .token import (
DemurrageToken,
DemurrageTokenSettings,
DemurrageRedistribution,
)
from .token import create
from .token import bytecode
from .token import args

0
python/erc20_demurrage_token/data/.chainlib Normal file
View File

2
python/erc20_demurrage_token/data/DemurrageTokenSingleNocap.bin
View File

File diff suppressed because one or more lines are too long

2
python/erc20_demurrage_token/data/DemurrageTokenSingleNocap.json
View File

File diff suppressed because one or more lines are too long

1
python/erc20_demurrage_token/data/DemurrageTokenSingleNocap.metadata.json Normal file
View File

File diff suppressed because one or more lines are too long

36
python/erc20_demurrage_token/expiry.py Normal file
View File

@@ -0,0 +1,36 @@
# external imports
from chainlib.eth.tx import (
TxFactory,
TxFormat,
)
from chainlib.eth.contract import (
ABIContractEncoder,
ABIContractType,
abi_decode_single,
)
from chainlib.eth.constant import ZERO_ADDRESS
class ExpiryContract(TxFactory):
def set_expire_period(self, contract_address, sender_address, expire_timestamp, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('setExpirePeriod')
enc.typ(ABIContractType.UINT256)
enc.uint256(expire_timestamp)
data = enc.get()
tx = self.template(sender_address, contract_address, use_nonce=True)
tx = self.set_code(tx, data)
tx = self.finalize(tx, tx_format)
return tx
def expires(self, contract_address, sender_address=ZERO_ADDRESS):
return self.call_noarg('expires', contract_address, sender_address=sender_address)
@classmethod
def parse_expires(self, v):
return abi_decode_single(ABIContractType.UINT256, v)

102
python/erc20_demurrage_token/runnable/deploy.py → python/erc20_demurrage_token/runnable/#publish.py#
View File

@@ -14,12 +14,21 @@ import logging
# external imports
import confini
from funga.eth.signer import EIP155Signer
from funga.eth.keystore.dict import DictKeystore
from chainlib.chain import ChainSpec
from chainlib.eth.nonce import (
RPCNonceOracle,
OverrideNonceOracle,
)
from chainlib.eth.gas import (
RPCGasOracle,
OverrideGasOracle,
)
from chainlib.eth.connection import EthHTTPConnection
from chainlib.eth.tx import receipt
from chainlib.eth.constant import ZERO_ADDRESS
import chainlib.eth.cli
from chainlib.eth.settings import process_settings
from chainlib.settings import ChainSettings
from chainlib.eth.cli.arg import (
Arg,
ArgFlag,
@@ -30,6 +39,11 @@ from chainlib.eth.cli.config import (
process_config,
)
from chainlib.eth.cli.log import process_log
from chainlib.eth.settings import process_settings
from chainlib.eth.address import to_checksum_address
from chainlib.settings import ChainSettings
from dexif import to_fixed
# local imports
import erc20_demurrage_token
@@ -40,32 +54,41 @@ from erc20_demurrage_token import (
logg = logging.getLogger()
script_dir = os.path.dirname(__file__)
data_dir = os.path.join(script_dir, '..', 'data')
config_dir = os.path.join(data_dir, 'config')
def process_config_local(config, arg, args, flags):
config.add(args.token_name, 'TOKEN_NAME', False)
config.add(args.token_symbol, 'TOKEN_SYMBOL', False)
config.add(args.token_decimals, 'TOKEN_DECIMALS', False)
config.add(args.sink_address, 'TOKEN_SINK_ADDRESS', False)
config.add(args.redistribution_period, 'TOKEN_REDISTRIBUTION_PERIOD', False)
config.add(args.demurrage_level, 'TOKEN_DEMURRAGE_LEVEL', False)
config.add(0, 'TOKEN_SUPPLY_LIMIT', False)
config.add(args.token_name, 'TOKEN_NAME')
config.add(args.token_symbol, 'TOKEN_SYMBOL')
config.add(args.token_decimals, 'TOKEN_DECIMALS')
sink_address = to_checksum_address(args.sink_address)
config.add(sink_address, 'TOKEN_SINK_ADDRESS')
config.add(args.redistribution_period, 'TOKEN_REDISTRIBUTION_PERIOD')
v = (1 - (args.demurrage_level / 1000000)) ** (1 / config.get('TOKEN_REDISTRIBUTION_PERIOD'))
if v >= 1.0:
raise ValueError('demurrage level must be less than 100%')
demurrage_level = to_fixed(v)
logg.info('v {} demurrage level {}'.format(v, demurrage_level))
config.add(demurrage_level, 'TOKEN_DEMURRAGE_LEVEL')
return config
arg_flags = ArgFlag()
arg = Arg(arg_flags)
flags = arg_flags.STD_WRITE | arg_flags.EXEC | arg_flags.WALLET
flags = arg_flags.STD_WRITE | arg_flags.WALLET
argparser = chainlib.eth.cli.ArgumentParser()
argparser = chainlib.eth.cli.ArgumentParser(arg_flags)
argparser = process_args(argparser, arg, flags)
argparser.add_argument('--name', dest='token_name', type=str, help='Token name')
argparser.add_argument('--symbol', dest='token_symbol', required=True, type=str, help='Token symbol')
argparser.add_argument('--decimals', dest='token_decimals', type=int, help='Token decimals')
argparser.add_argument('--sink-address', dest='sink_address', type=str, help='demurrage level,ppm per minute')
#argparser.add_argument('--supply-limit', dest='supply_limit', type=int, help='token supply limit (0 = no limit)')
argparser.add_argument('--redistribution-period', dest='redistribution_period', type=int, help='redistribution period, minutes (0 = deactivate)') # default 10080 = week
#argparser.add_argument('--multi', action='store_true', help='automatic redistribution')
argparser.add_argument('--demurrage-level', dest='demurrage_level', type=int, help='demurrage level, ppm per minute')
argparser.add_argument('--redistribution-period', type=int, help='redistribution period, minutes (0 = deactivate)') # default 10080 = week
argparser.add_argument('--demurrage-level', dest='demurrage_level', type=int, help='demurrage level, ppm per period')
args = argparser.parse_args()
logg = process_log(args, logg)
@@ -80,16 +103,45 @@ settings = process_settings(settings, config)
logg.debug('settings loaded:\n{}'.format(settings))
#extra_args = {
# 'redistribution_period': 'TOKEN_REDISTRIBUTION_PERIOD',
# 'demurrage_level': 'TOKEN_DEMURRAGE_LEVEL',
# 'supply_limit': 'TOKEN_SUPPLY_LIMIT',
# 'token_name': 'TOKEN_NAME',
# 'token_symbol': 'TOKEN_SYMBOL',
# 'token_decimals': 'TOKEN_DECIMALS',
# 'sink_address': 'TOKEN_SINK_ADDRESS',
# 'multi': None,
# }
#config = chainlib.eth.cli.Config.from_args(args, arg_flags, extra_args=extra_args, default_fee_limit=DemurrageToken.gas(), base_config_dir=config_dir)
#
#if not bool(config.get('TOKEN_NAME')):
# logg.info('token name not set, using symbol {} as name'.format(config.get('TOKEN_SYMBOL')))
# config.add(config.get('TOKEN_SYMBOL'), 'TOKEN_NAME', True)
#
#if config.get('TOKEN_SUPPLY_LIMIT') == None:
# config.add(0, 'TOKEN_SUPPLY_LIMIT', True)
#
#if config.get('TOKEN_REDISTRIBUTION_PERIOD') == None:
# config.add(10800, 'TOKEN_REDISTRIBUTION_PERIOD', True)
#logg.debug('config loaded:\n{}'.format(config))
#
#wallet = chainlib.eth.cli.Wallet()
#wallet.from_config(config)
#
#rpc = chainlib.eth.cli.Rpc(wallet=wallet)
#conn = rpc.connect_by_config(config)
#
#chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
def main():
chain_spec = settings.get('CHAIN_SPEC')
conn = settings.get('CONN')
signer = settings.get('SIGNER')
signer_address = settings.get('SENDER_ADDRESS')
gas_oracle = settings.get('FEE_ORACLE')
nonce_oracle = settings.get('NONCE_ORACLE')
c = DemurrageToken(chain_spec, signer=signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
c = DemurrageToken(
settings.get('CHAIN_SPEC'),
signer=settings.get('SIGNER'),
gas_oracle=settings.get('FEE_ORACLE'),
nonce_oracle=settings.get('NONCE_ORACLE'),
)
token_settings = DemurrageTokenSettings()
token_settings.name = config.get('TOKEN_NAME')
token_settings.symbol = config.get('TOKEN_SYMBOL')
@@ -99,10 +151,8 @@ def main():
token_settings.sink_address = config.get('TOKEN_SINK_ADDRESS')
(tx_hash_hex, o) = c.constructor(
signer_address,
settings.get('SENDER_ADDRESS'),
token_settings,
redistribute=config.true('_MULTI'),
cap=int(config.get('TOKEN_SUPPLY_LIMIT')),
)
if settings.get('RPC_SEND'):
conn.do(o)

1
python/erc20_demurrage_token/runnable/.#publish.py Symbolic link
View File

@@ -0,0 +1 @@
wor@gecon.733148:1676287007

152
python/erc20_demurrage_token/runnable/apply.py
View File

@@ -1,152 +0,0 @@
"""Deploy sarafu token
.. moduleauthor:: Louis Holbrook <dev@holbrook.no>
.. pgp:: 0826EDA1702D1E87C6E2875121D2E7BB88C2A746
"""
# standard imports
import sys
import os
import json
import argparse
import logging
import datetime
import math
# external imports
import confini
import chainlib.eth.cli
from chainlib.eth.block import (
block_latest,
block_by_number,
Block,
)
from chainlib.eth.connection import EthHTTPConnection
from chainlib.eth.tx import receipt
from chainlib.eth.constant import ZERO_ADDRESS
from hexathon import to_int as hex_to_int
import chainlib.eth.cli
from chainlib.eth.settings import process_settings
from chainlib.settings import ChainSettings
from chainlib.eth.cli.arg import (
Arg,
ArgFlag,
process_args,
)
from chainlib.eth.cli.config import (
Config,
process_config,
)
from chainlib.eth.cli.log import process_log
# local imports
import erc20_demurrage_token
from erc20_demurrage_token import (
DemurrageToken,
DemurrageTokenSettings,
)
logg = logging.getLogger()
def process_config_local(config, arg, args, flags):
config.add(args.steps, '_STEPS', False)
return config
arg_flags = ArgFlag()
arg = Arg(arg_flags)
flags = arg_flags.STD_WRITE | arg_flags.EXEC | arg_flags.WALLET
argparser = chainlib.eth.cli.ArgumentParser()
argparser = process_args(argparser, arg, flags)
argparser.add_argument('--steps', type=int, default=0, help='Max demurrage steps to apply per round')
args = argparser.parse_args()
logg = process_log(args, logg)
config = Config()
config = process_config(config, arg, args, flags)
config = process_config_local(config, arg, args, flags)
logg.debug('config loaded:\n{}'.format(config))
settings = ChainSettings()
settings = process_settings(settings, config)
logg.debug('settings loaded:\n{}'.format(settings))
def main():
chain_spec = settings.get('CHAIN_SPEC')
conn = settings.get('CONN')
o = block_latest()
r = conn.do(o)
block_start_number = None
try:
block_start_number = hex_to_int(r)
except TypeError:
block_start_number = int(r)
o = block_by_number(block_start_number)
r = conn.do(o)
block_start = Block(r)
block_start_timestamp = block_start.timestamp
block_start_datetime = datetime.datetime.fromtimestamp(block_start_timestamp)
gas_oracle = settings.get('FEE_ORACLE')
c = DemurrageToken(chain_spec, gas_oracle=gas_oracle)
o = c.demurrage_timestamp(settings.get('EXEC'))
r = conn.do(o)
demurrage_timestamp = None
try:
demurrage_timestamp = hex_to_int(r)
except TypeError:
demurrage_timestamp = int(r)
demurrage_datetime = datetime.datetime.fromtimestamp(demurrage_timestamp)
total_seconds = block_start_timestamp - demurrage_timestamp
total_steps = total_seconds / 60
if total_steps < 1.0:
logg.error('only {} seconds since last demurrage application, skipping'.format(total_seconds))
return
logg.debug('block start is at {} demurrage is at {} -> {} minutes'.format(
block_start_datetime,
demurrage_datetime,
total_steps,
))
rounds = 1
if config.get('_STEPS') > 0:
rounds = math.ceil(total_steps / config.get('_STEPS'))
logg.info('will perform {} rounds of {} steps'.format(rounds, config.get('_STEPS')))
last_tx_hash = None
for i in range(rounds):
signer = settings.get('SIGNER')
signer_address = settings.get('SENDER_ADDRESS')
nonce_oracle = settings.get('NONCE_ORACLE')
c = DemurrageToken(chain_spec, signer=signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
(tx_hash_hex, o) = c.apply_demurrage(config.get('_EXEC_ADDRESS'), signer_address, limit=config.get('_STEPS'))
if settings.get('RPC_SEND'):
print(tx_hash_hex)
conn.do(o)
if config.true('_WAIT_ALL') or (i == rounds - 1 and config.true('_WAIT')):
r = conn.wait(tx_hash_hex)
if r['status'] == 0:
sys.stderr.write('EVM revert while deploying contract. Wish I had more to tell you')
sys.exit(1)
else:
print(o)
if __name__ == '__main__':
main()

176
python/erc20_demurrage_token/runnable/publish.py Normal file
View File

@@ -0,0 +1,176 @@
"""Deploy sarafu token
.. moduleauthor:: Louis Holbrook <dev@holbrook.no>
.. pgp:: 0826EDA1702D1E87C6E2875121D2E7BB88C2A746
"""
# standard imports
import sys
import os
import json
import argparse
import logging
# external imports
import confini
from funga.eth.signer import EIP155Signer
from funga.eth.keystore.dict import DictKeystore
from chainlib.chain import ChainSpec
from chainlib.eth.nonce import (
RPCNonceOracle,
OverrideNonceOracle,
)
from chainlib.eth.gas import (
RPCGasOracle,
OverrideGasOracle,
)
from chainlib.eth.connection import EthHTTPConnection
from chainlib.eth.tx import receipt
from chainlib.eth.constant import ZERO_ADDRESS
import chainlib.eth.cli
from chainlib.eth.cli.arg import (
Arg,
ArgFlag,
process_args,
)
from chainlib.eth.cli.config import (
Config,
process_config,
)
from chainlib.eth.cli.log import process_log
from chainlib.eth.settings import process_settings
from chainlib.eth.address import to_checksum_address
from chainlib.settings import ChainSettings
from dexif import to_fixed
# local imports
import erc20_demurrage_token
from erc20_demurrage_token import (
DemurrageToken,
DemurrageTokenSettings,
)
logg = logging.getLogger()
script_dir = os.path.dirname(__file__)
data_dir = os.path.join(script_dir, '..', 'data')
config_dir = os.path.join(data_dir, 'config')
def process_config_local(config, arg, args, flags):
config.add(args.token_name, 'TOKEN_NAME')
config.add(args.token_symbol, 'TOKEN_SYMBOL')
config.add(args.token_decimals, 'TOKEN_DECIMALS')
sink_address = to_checksum_address(args.sink_address)
config.add(sink_address, 'TOKEN_SINK_ADDRESS')
config.add(args.redistribution_period, 'TOKEN_REDISTRIBUTION_PERIOD')
v = (1 - (args.demurrage_level / 1000000)) ** (1 / config.get('TOKEN_REDISTRIBUTION_PERIOD'))
if v >= 1.0:
raise ValueError('demurrage level must be less than 100%')
demurrage_level = to_fixed(v)
logg.info('v {} demurrage level {}'.format(v, demurrage_level))
config.add(demurrage_level, 'TOKEN_DEMURRAGE_LEVEL')
return config
arg_flags = ArgFlag()
arg = Arg(arg_flags)
flags = arg_flags.STD_WRITE | arg_flags.WALLET
argparser = chainlib.eth.cli.ArgumentParser(arg_flags)
argparser = process_args(argparser, arg, flags)
argparser.add_argument('--name', dest='token_name', type=str, help='Token name')
argparser.add_argument('--symbol', dest='token_symbol', required=True, type=str, help='Token symbol')
argparser.add_argument('--decimals', dest='token_decimals', type=int, help='Token decimals')
argparser.add_argument('--sink-address', dest='sink_address', type=str, help='demurrage level,ppm per minute')
argparser.add_argument('--redistribution-period', type=int, help='redistribution period, minutes (0 = deactivate)') # default 10080 = week
argparser.add_argument('--demurrage-level', dest='demurrage_level', type=int, help='demurrage level, ppm per period')
args = argparser.parse_args()
logg = process_log(args, logg)
config = Config()
config = process_config(config, arg, args, flags)
config = process_config_local(config, arg, args, flags)
logg.debug('config loaded:\n{}'.format(config))
settings = ChainSettings()
settings = process_settings(settings, config)
logg.debug('settings loaded:\n{}'.format(settings))
#extra_args = {
# 'redistribution_period': 'TOKEN_REDISTRIBUTION_PERIOD',
# 'demurrage_level': 'TOKEN_DEMURRAGE_LEVEL',
# 'supply_limit': 'TOKEN_SUPPLY_LIMIT',
# 'token_name': 'TOKEN_NAME',
# 'token_symbol': 'TOKEN_SYMBOL',
# 'token_decimals': 'TOKEN_DECIMALS',
# 'sink_address': 'TOKEN_SINK_ADDRESS',
# 'multi': None,
# }
#config = chainlib.eth.cli.Config.from_args(args, arg_flags, extra_args=extra_args, default_fee_limit=DemurrageToken.gas(), base_config_dir=config_dir)
#
#if not bool(config.get('TOKEN_NAME')):
# logg.info('token name not set, using symbol {} as name'.format(config.get('TOKEN_SYMBOL')))
# config.add(config.get('TOKEN_SYMBOL'), 'TOKEN_NAME', True)
#
#if config.get('TOKEN_SUPPLY_LIMIT') == None:
# config.add(0, 'TOKEN_SUPPLY_LIMIT', True)
#
#if config.get('TOKEN_REDISTRIBUTION_PERIOD') == None:
# config.add(10800, 'TOKEN_REDISTRIBUTION_PERIOD', True)
#logg.debug('config loaded:\n{}'.format(config))
#
#wallet = chainlib.eth.cli.Wallet()
#wallet.from_config(config)
#
#rpc = chainlib.eth.cli.Rpc(wallet=wallet)
#conn = rpc.connect_by_config(config)
#
#chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
def main():
conn = settings.get('CONN')
c = DemurrageToken(
settings.get('CHAIN_SPEC'),
signer=settings.get('SIGNER'),
gas_oracle=settings.get('FEE_ORACLE'),
nonce_oracle=settings.get('NONCE_ORACLE'),
)
token_settings = DemurrageTokenSettings()
token_settings.name = config.get('TOKEN_NAME')
token_settings.symbol = config.get('TOKEN_SYMBOL')
token_settings.decimals = int(config.get('TOKEN_DECIMALS'))
token_settings.demurrage_level = int(config.get('TOKEN_DEMURRAGE_LEVEL'))
token_settings.period_minutes = int(config.get('TOKEN_REDISTRIBUTION_PERIOD'))
token_settings.sink_address = config.get('TOKEN_SINK_ADDRESS')
(tx_hash_hex, o) = c.constructor(
settings.get('SENDER_ADDRESS'),
token_settings,
)
if settings.get('RPC_SEND'):
conn.do(o)
if config.true('_WAIT'):
r = conn.wait(tx_hash_hex)
if r['status'] == 0:
sys.stderr.write('EVM revert while deploying contract. Wish I had more to tell you')
sys.exit(1)
# TODO: pass through translator for keys (evm tester uses underscore instead of camelcase)
address = r['contractAddress']
print(address)
else:
print(tx_hash_hex)
else:
print(o)
if __name__ == '__main__':
main()

66
python/erc20_demurrage_token/seal.py Normal file
View File

@@ -0,0 +1,66 @@
# standard imports
import enum
# external imports
from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.jsonrpc import JSONRPCRequest
from chainlib.eth.tx import (
TxFactory,
TxFormat,
)
from chainlib.eth.contract import (
ABIContractEncoder,
ABIContractType,
abi_decode_single,
)
from hexathon import (
add_0x,
)
class ContractState(enum.IntEnum):
MINTER_STATE = 1
SINK_STATE = 2
EXPIRY_STATE = 4
CAP_STATE = 8
CONTRACT_SEAL_STATE_MAX = 0
for v in dir(ContractState):
if len(v) > 6 and v[-6:] == '_STATE':
CONTRACT_SEAL_STATE_MAX += getattr(ContractState, v).value
class SealedContract(TxFactory):
def seal(self, contract_address, sender_address, seal, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('seal')
enc.typ(ABIContractType.UINT256)
enc.uint256(seal)
data = enc.get()
tx = self.template(sender_address, contract_address, use_nonce=True)
tx = self.set_code(tx, data)
tx = self.finalize(tx, tx_format)
return tx
def is_sealed(self, contract_address, v, sender_address=ZERO_ADDRESS, id_generator=None):
j = JSONRPCRequest(id_generator)
o = j.template()
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('isSealed')
enc.typ(ABIContractType.UINT256)
enc.uint256(v)
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
o['params'].append(self.normalize(tx))
o['params'].append('latest')
o = j.finalize(o)
return o
@classmethod
def parse_is_sealed(self, v):
return abi_decode_single(ABIContractType.BOOLEAN, v)

290
python/erc20_demurrage_token/token.py
View File

@@ -10,6 +10,7 @@ from chainlib.eth.tx import (
from chainlib.hash import keccak256_string_to_hex
from chainlib.eth.contract import (
ABIContractEncoder,
ABIContractDecoder,
ABIContractType,
abi_decode_single,
)
@@ -20,13 +21,38 @@ from hexathon import (
add_0x,
strip_0x,
)
from dexif import from_fixed
# local imports
from erc20_demurrage_token.data import data_dir
from erc20_demurrage_token.seal import SealedContract
from erc20_demurrage_token.expiry import ExpiryContract
logg = logging.getLogger(__name__)
class DemurrageRedistribution:
def __init__(self, v):
d = ABIContractDecoder()
v = strip_0x(v)
d.typ(ABIContractType.UINT256)
d.typ(ABIContractType.UINT256)
d.typ(ABIContractType.BYTES32)
d.val(v[:64])
d.val(v[64:128])
d.val(v[128:192])
r = d.decode()
self.period = r[0]
self.value = r[1]
self.demurrage = from_fixed(r[2])
def __str__(self):
return 'period {} value {} demurrage {}'.format(self.period, self.value, self.demurrage)
class DemurrageTokenSettings:
def __init__(self):
@@ -47,59 +73,40 @@ class DemurrageTokenSettings:
)
class DemurrageToken(ERC20):
class DemurrageToken(ERC20, SealedContract, ExpiryContract):
__abi = {}
__bytecode = {}
valid_modes = [
'MultiNocap',
'SingleNocap',
'MultiCap',
'SingleCap',
]
def constructor(self, sender_address, settings, redistribute=True, cap=0, tx_format=TxFormat.JSONRPC):
if int(cap) < 0:
raise ValueError('cap must be 0 or positive integer')
code = DemurrageToken.bytecode(multi=redistribute, cap=cap>0)
enc = ABIContractEncoder()
enc.string(settings.name)
enc.string(settings.symbol)
enc.uint256(settings.decimals)
enc.uint256(settings.demurrage_level)
enc.uint256(settings.period_minutes)
enc.address(settings.sink_address)
if cap > 0:
enc.uint256(cap)
code += enc.get()
def constructor(self, sender_address, settings, tx_format=TxFormat.JSONRPC, version=None):
code = self.cargs(settings.name, settings.symbol, settings.decimals, settings.demurrage_level, settings.period_minutes, settings.sink_address, version=version)
tx = self.template(sender_address, None, use_nonce=True)
tx = self.set_code(tx, code)
return self.finalize(tx, tx_format)
@staticmethod
def cargs(name, symbol, decimals, demurrage_level, period_minutes, sink_address, version=None):
code = DemurrageToken.bytecode()
enc = ABIContractEncoder()
enc.string(name)
enc.string(symbol)
enc.uint256(decimals)
enc.uint256(demurrage_level)
enc.uint256(period_minutes)
enc.address(sink_address)
code += enc.get()
return code
@staticmethod
def gas(code=None):
return 4000000
return 7000000
@staticmethod
def __to_contract_name(multi, cap):
name = 'DemurrageToken'
if multi:
name += 'Multi'
else:
name += 'Single'
if cap:
name += 'Cap'
else:
name += 'Nocap'
logg.debug('bytecode name {}'.format(name))
return name
@staticmethod
def abi(multi=True, cap=False):
name = DemurrageToken.__to_contract_name(multi, cap)
def abi():
name = 'DemurrageTokenSingleNocap'
if DemurrageToken.__abi.get(name) == None:
f = open(os.path.join(data_dir, name + '.json'), 'r')
DemurrageToken.__abi[name] = json.load(f)
@@ -108,8 +115,8 @@ class DemurrageToken(ERC20):
@staticmethod
def bytecode(multi=True, cap=False):
name = DemurrageToken.__to_contract_name(multi, cap)
def bytecode(version=None):
name = 'DemurrageTokenSingleNocap'
if DemurrageToken.__bytecode.get(name) == None:
f = open(os.path.join(data_dir, name + '.bin'), 'r')
DemurrageToken.__bytecode[name] = f.read()
@@ -145,9 +152,14 @@ class DemurrageToken(ERC20):
return tx
# backwards compatibility
def add_minter(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
return self.add_writer(contract_address, sender_address, address, tx_format=tx_format)
def add_writer(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('addMinter')
enc.method('addWriter')
enc.typ(ABIContractType.ADDRESS)
enc.address(address)
data = enc.get()
@@ -157,9 +169,26 @@ class DemurrageToken(ERC20):
return tx
def remove_minter(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
def set_max_supply(self, contract_address, sender_address, cap, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('removeMinter')
enc.method('setMaxSupply')
enc.typ(ABIContractType.UINT256)
enc.uint256(cap)
data = enc.get()
tx = self.template(sender_address, contract_address, use_nonce=True)
tx = self.set_code(tx, data)
tx = self.finalize(tx, tx_format)
return tx
# backwards compatibility
def remove_minter(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
return self.delete_writer(contract_address, sender_address, address, tx_format=tx_format)
def delete_writer(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('deleteWriter')
enc.typ(ABIContractType.ADDRESS)
enc.address(address)
data = enc.get()
@@ -280,18 +309,18 @@ class DemurrageToken(ERC20):
return o
def to_redistribution(self, contract_address, participants, demurrage_modifier_ppm, value, period, sender_address=ZERO_ADDRESS, id_generator=None):
def to_redistribution(self, contract_address, participants, demurrage_modifier, value, period, sender_address=ZERO_ADDRESS, id_generator=None):
j = JSONRPCRequest(id_generator)
o = j.template()
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('toRedistribution')
enc.typ(ABIContractType.UINT256)
enc.typ(ABIContractType.UINT256)
enc.typ_literal('int128')
enc.typ(ABIContractType.UINT256)
enc.typ(ABIContractType.UINT256)
enc.uint256(participants)
enc.uint256(demurrage_modifier_ppm)
enc.uint256(demurrage_modifier)
enc.uint256(value)
enc.uint256(period)
data = add_0x(enc.get())
@@ -310,8 +339,11 @@ class DemurrageToken(ERC20):
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('toRedistributionPeriod')
enc.typ(ABIContractType.BYTES32)
enc.bytes32(redistribution)
v = strip_0x(redistribution)
enc.typ_literal('(uint32,uint72,uint64)')
enc.bytes32(v[:64])
enc.bytes32(v[64:128])
enc.bytes32(v[128:192])
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
@@ -321,22 +353,26 @@ class DemurrageToken(ERC20):
return o
def to_redistribution_participants(self, contract_address, redistribution, sender_address=ZERO_ADDRESS, id_generator=None):
j = JSONRPCRequest(id_generator)
o = j.template()
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('toRedistributionParticipants')
enc.typ(ABIContractType.BYTES32)
enc.bytes32(redistribution)
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
o['params'].append(self.normalize(tx))
o['params'].append('latest')
o = j.finalize(o)
return o
# def to_redistribution_participants(self, contract_address, redistribution, sender_address=ZERO_ADDRESS, id_generator=None):
# j = JSONRPCRequest(id_generator)
# o = j.template()
# o['method'] = 'eth_call'
# enc = ABIContractEncoder()
# enc.method('toRedistributionParticipants')
# v = strip_0x(redistribution)
# enc.typ_literal('(uint32,uint72,uint104)')
# #enc.typ(ABIContractType.BYTES32)
# enc.bytes32(v[:64])
# enc.bytes32(v[64:128])
# enc.bytes32(v[128:192])
# data = add_0x(enc.get())
# tx = self.template(sender_address, contract_address)
# tx = self.set_code(tx, data)
# o['params'].append(self.normalize(tx))
# o['params'].append('latest')
# o = j.finalize(o)
# return o
#
def to_redistribution_supply(self, contract_address, redistribution, sender_address=ZERO_ADDRESS, id_generator=None):
j = JSONRPCRequest(id_generator)
@@ -344,8 +380,11 @@ class DemurrageToken(ERC20):
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('toRedistributionSupply')
enc.typ(ABIContractType.BYTES32)
enc.bytes32(redistribution)
v = strip_0x(redistribution)
enc.typ_literal('(uint32,uint72,uint64)')
enc.bytes32(v[:64])
enc.bytes32(v[64:128])
enc.bytes32(v[128:192])
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
@@ -361,8 +400,11 @@ class DemurrageToken(ERC20):
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('toRedistributionDemurrageModifier')
enc.typ(ABIContractType.BYTES32)
enc.bytes32(redistribution)
v = strip_0x(redistribution)
enc.typ_literal('(uint32,uint72,uint64)')
enc.bytes32(v[:64])
enc.bytes32(v[64:128])
enc.bytes32(v[128:192])
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
@@ -389,6 +431,31 @@ class DemurrageToken(ERC20):
return o
def set_sink_address(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('setSinkAddress')
enc.typ(ABIContractType.ADDRESS)
enc.address(address)
data = enc.get()
tx = self.template(sender_address, contract_address, use_nonce=True)
tx = self.set_code(tx, data)
tx = self.finalize(tx, tx_format)
return tx
def sweep(self, contract_address, sender_address, recipient_address, tx_format=TxFormat.JSONRPC):
enc = ABIContractEncoder()
enc.method('sweep')
enc.typ(ABIContractType.ADDRESS)
enc.address(recipient_address)
data = enc.get()
tx = self.template(sender_address, contract_address, use_nonce=True)
tx = self.set_code(tx, data)
tx = self.finalize(tx, tx_format)
return tx
def apply_demurrage(self, contract_address, sender_address, limit=0, tx_format=TxFormat.JSONRPC):
if limit == 0:
return self.transact_noarg('applyDemurrage', contract_address, sender_address)
@@ -420,8 +487,8 @@ class DemurrageToken(ERC20):
return tx
def tax_level(self, contract_address, sender_address=ZERO_ADDRESS):
return self.call_noarg('taxLevel', contract_address, sender_address=sender_address)
def decay_level(self, contract_address, sender_address=ZERO_ADDRESS):
return self.call_noarg('decayLevel', contract_address, sender_address=sender_address)
def resolution_factor(self, contract_address, sender_address=ZERO_ADDRESS):
@@ -448,28 +515,28 @@ class DemurrageToken(ERC20):
return self.call_noarg('demurrageTimestamp', contract_address, sender_address=sender_address)
def supply_cap(self, contract_address, sender_address=ZERO_ADDRESS):
return self.call_noarg('supplyCap', contract_address, sender_address=sender_address)
def max_supply(self, contract_address, sender_address=ZERO_ADDRESS):
return self.call_noarg('maxSupply', contract_address, sender_address=sender_address)
def grow_by(self, contract_address, value, period, sender_address=ZERO_ADDRESS, id_generator=None):
j = JSONRPCRequest(id_generator)
o = j.template()
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('growBy')
enc.typ(ABIContractType.UINT256)
enc.typ(ABIContractType.UINT256)
enc.uint256(value)
enc.uint256(period)
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
o['params'].append(self.normalize(tx))
o['params'].append('latest')
o = j.finalize(o)
return o
# def grow_by(self, contract_address, value, period, sender_address=ZERO_ADDRESS, id_generator=None):
# j = JSONRPCRequest(id_generator)
# o = j.template()
# o['method'] = 'eth_call'
# enc = ABIContractEncoder()
# enc.method('growBy')
# enc.typ(ABIContractType.UINT256)
# enc.typ(ABIContractType.UINT256)
# enc.uint256(value)
# enc.uint256(period)
# data = add_0x(enc.get())
# tx = self.template(sender_address, contract_address)
# tx = self.set_code(tx, data)
# o['params'].append(self.normalize(tx))
# o['params'].append('latest')
# o = j.finalize(o)
# return o
#
def decay_by(self, contract_address, value, period, sender_address=ZERO_ADDRESS, id_generator=None):
j = JSONRPCRequest(id_generator)
@@ -497,7 +564,7 @@ class DemurrageToken(ERC20):
enc = ABIContractEncoder()
enc.method('getDistribution')
enc.typ(ABIContractType.UINT256)
enc.typ(ABIContractType.UINT256)
enc.typ_literal('int128')
enc.uint256(supply)
enc.uint256(demurrage_amount)
data = add_0x(enc.get())
@@ -515,8 +582,11 @@ class DemurrageToken(ERC20):
o['method'] = 'eth_call'
enc = ABIContractEncoder()
enc.method('getDistributionFromRedistribution')
enc.typ(ABIContractType.BYTES32)
enc.bytes32(redistribution)
v = strip_0x(redistribution)
enc.typ_literal('(uint32,uint72,uint64)')
enc.bytes32(v[:64])
enc.bytes32(v[64:128])
enc.bytes32(v[128:192])
data = add_0x(enc.get())
tx = self.template(sender_address, contract_address)
tx = self.set_code(tx, data)
@@ -544,7 +614,8 @@ class DemurrageToken(ERC20):
@classmethod
def parse_demurrage_amount(self, v):
return abi_decode_single(ABIContractType.UINT256, v)
# return abi_decode_single(ABIContractType.UINT256, v)
return from_fixed(v)
@classmethod
@@ -559,8 +630,8 @@ class DemurrageToken(ERC20):
@classmethod
def parse_redistributions(self, v):
return abi_decode_single(ABIContractType.BYTES32, v)
return strip_0x(v)
@classmethod
def parse_account_period(self, v):
@@ -598,7 +669,7 @@ class DemurrageToken(ERC20):
@classmethod
def parse_tax_level(self, v):
def parse_decay_level(self, v):
return abi_decode_single(ABIContractType.UINT256, v)
@@ -610,3 +681,26 @@ class DemurrageToken(ERC20):
@classmethod
def parse_total_burned(self, v):
return abi_decode_single(ABIContractType.UINT256, v)
def bytecode(**kwargs):
return DemurrageToken.bytecode(version=kwargs.get('version'))
def create(**kwargs):
return DemurrageToken.cargs(
kwargs['name'],
kwargs['symbol'],
kwargs['decimals'],
kwargs['demurragelevel'],
kwargs['redistributionperiod'],
kwargs['sinkaddress'],
version=kwargs.get('version'))
def args(v):
if v == 'create':
return (['name', 'symbol', 'decimals', 'demurragelevel', 'redistributionperiod', 'sinkaddress'], ['version'],)
elif v == 'default' or v == 'bytecode':
return ([], ['version'],)
raise ValueError('unknown command: ' + v)

192
python/erc20_demurrage_token/unittest/base.py
View File

@@ -1,6 +1,7 @@
# standard imports
import logging
import os
import math
# external imports
from chainlib.eth.unittest.ethtester import EthTesterCase
@@ -19,20 +20,21 @@ from erc20_demurrage_token import (
DemurrageTokenSettings,
DemurrageToken,
)
from dexif import *
logg = logging.getLogger(__name__)
logg = logging.getLogger()
#BLOCKTIME = 5 # seconds
TAX_LEVEL = int(10000 * 2) # 2%
# calc "1-(0.98)^(1/518400)" <- 518400 = 30 days of blocks
# 0.00000003897127107225
#PERIOD = int(60/BLOCKTIME) * 60 * 24 * 30 # month
PERIOD = 10
PERIOD = 43200
class TestTokenDeploy:
def __init__(self, rpc, token_symbol='FOO', token_name='Foo Token', sink_address=ZERO_ADDRESS, supply=10**12, tax_level=TAX_LEVEL, period=PERIOD):
"""tax level is ppm, 1000000 = 100%"""
def __init__(self, rpc, token_symbol='FOO', token_name='Foo Token', sink_address=ZERO_ADDRESS, tax_level=TAX_LEVEL, period=PERIOD):
self.tax_level = tax_level
self.period_seconds = period * 60
@@ -40,7 +42,8 @@ class TestTokenDeploy:
self.settings.name = token_name
self.settings.symbol = token_symbol
self.settings.decimals = 6
self.settings.demurrage_level = tax_level * (10 ** 32)
tax_level_input = to_fixed((1 - (tax_level / 1000000)) ** (1 / period))
self.settings.demurrage_level = tax_level_input
self.settings.period_minutes = period
self.settings.sink_address = sink_address
self.sink_address = self.settings.sink_address
@@ -57,25 +60,11 @@ class TestTokenDeploy:
except TypeError:
self.start_time = int(r['timestamp'])
self.default_supply = supply
self.default_supply_cap = int(self.default_supply * 10)
def deploy(self, rpc, deployer_address, interface, mode, supply_cap=10**12):
def publish(self, rpc, publisher_address, interface, supply_cap=0):
tx_hash = None
o = None
logg.debug('mode {} {}'.format(mode, self.settings))
self.mode = mode
if mode == 'MultiNocap':
(tx_hash, o) = interface.constructor(deployer_address, self.settings, redistribute=True, cap=0)
elif mode == 'SingleNocap':
(tx_hash, o) = interface.constructor(deployer_address, self.settings, redistribute=False, cap=0)
elif mode == 'MultiCap':
(tx_hash, o) = interface.constructor(deployer_address, self.settings, redistribute=True, cap=supply_cap)
elif mode == 'SingleCap':
(tx_hash, o) = interface.constructor(deployer_address, self.settings, redistribute=False, cap=supply_cap)
else:
raise ValueError('Invalid mode "{}", valid are {}'.format(mode, DemurrageToken.valid_modes))
(tx_hash, o) = interface.constructor(publisher_address, self.settings)
r = rpc.do(o)
o = receipt(tx_hash)
@@ -100,25 +89,43 @@ class TestDemurrage(EthTesterCase):
period = getattr(self, 'period')
except AttributeError as e:
pass
self.deployer = TestTokenDeploy(self.rpc, period=period)
self.default_supply = self.deployer.default_supply
self.default_supply_cap = self.deployer.default_supply_cap
self.publisher = TestTokenDeploy(self.rpc, period=period, sink_address=self.accounts[9])
self.default_supply = 0
self.default_supply_cap = 0
self.start_block = None
self.address = None
self.start_time = None
def deploy(self, interface, mode):
self.address = self.deployer.deploy(self.rpc, self.accounts[0], interface, mode, supply_cap=self.default_supply_cap)
self.start_block = self.deployer.start_block
self.start_time = self.deployer.start_time
self.tax_level = self.deployer.tax_level
self.period_seconds = self.deployer.period_seconds
self.sink_address = self.deployer.sink_address
def publish(self, interface):
self.address = self.publisher.publish(self.rpc, self.accounts[0], interface, supply_cap=self.default_supply_cap)
self.start_block = self.publisher.start_block
self.start_time = self.publisher.start_time
self.tax_level = self.publisher.tax_level
self.period_seconds = self.publisher.period_seconds
self.sink_address = self.publisher.sink_address
logg.debug('contract address {} start block {} start time {}'.format(self.address, self.start_block, self.start_time))
def assert_within(self, v, target, tolerance_ppm):
lower_target = target - (target * (tolerance_ppm / 1000000))
higher_target = target + (target * (tolerance_ppm / 1000000))
#self.assertGreaterEqual(v, lower_target)
#self.assertLessEqual(v, higher_target)
if v >= lower_target and v <= higher_target:
logg.debug('asserted within {} <= {} <= {}'.format(lower_target, v, higher_target))
return
raise AssertionError('{} not within lower {} and higher {}'.format(v, lower_target, higher_target))
def assert_within_greater(self, v, target, tolerance_ppm):
greater_target = target + (target * (tolerance_ppm / 1000000))
self.assertLessEqual(v, greater_target)
self.assertGreaterEqual(v, target)
logg.debug('asserted within greater {} >= {} >= {}'.format(greater_target, v, target))
def assert_within_lower(self, v, target, tolerance_ppm):
lower_target = target - (target * (tolerance_ppm / 1000000))
self.assertGreaterEqual(v, lower_target)
@@ -126,11 +133,10 @@ class TestDemurrage(EthTesterCase):
logg.debug('asserted within lower {} <= {} <= {}'.format(lower_target, v, target))
def assert_within_greater(self, v, target, tolerance_ppm):
higher_target = target + (target * (tolerance_ppm / 1000000))
self.assertLessEqual(v, higher_target)
self.assertGreaterEqual(v, target)
logg.debug('asserted within lower {} <= {} <= {}'.format(target, v, higher_target))
def assert_equal_decimals(self, v, target, precision):
target = int(target * (10 ** precision))
target = target / (10 ** precision)
self.assertEqual(v, target)
def tearDown(self):
@@ -145,115 +151,7 @@ class TestDemurrageDefault(TestDemurrage):
nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
self.mode = os.environ.get('ERC20_DEMURRAGE_TOKEN_TEST_MODE')
if self.mode == None:
self.mode = 'MultiNocap'
logg.debug('executing test setup default mode {}'.format(self.mode))
self.publish(c)
self.deploy(c, self.mode)
logg.info('deployed with mode {}'.format(self.mode))
class TestDemurrageSingle(TestDemurrage):
def setUp(self):
super(TestDemurrageSingle, self).setUp()
nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
self.mode = os.environ.get('ERC20_DEMURRAGE_TOKEN_TEST_MODE')
single_valid_modes = [
'SingleNocap',
'SingleCap',
]
if self.mode != None:
if self.mode not in single_valid_modes:
raise ValueError('Invalid mode "{}" for "single" contract tests, valid are {}'.format(self.mode, single_valid_modes))
else:
self.mode = 'SingleNocap'
logg.debug('executing test setup demurragesingle mode {}'.format(self.mode))
self.deploy(c, self.mode)
logg.info('deployed with mode {}'.format(self.mode))
class TestDemurrageCap(TestDemurrage):
def setUp(self):
super(TestDemurrageCap, self).setUp()
nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
self.mode = os.environ.get('ERC20_DEMURRAGE_TOKEN_TEST_MODE')
cap_valid_modes = [
'MultiCap',
'SingleCap',
]
if self.mode != None:
if self.mode not in cap_valid_modes:
raise ValueError('Invalid mode "{}" for "cap" contract tests, valid are {}'.format(self.mode, cap_valid_modes))
else:
self.mode = 'MultiCap'
logg.debug('executing test setup demurragecap mode {}'.format(self.mode))
self.deploy(c, self.mode)
logg.info('deployed with mode {}'.format(self.mode))
class TestDemurrageUnit(TestDemurrage):
def setUp(self):
self.period = 1
self.period_seconds = self.period * 60
self.tax_level = TAX_LEVEL
super(TestDemurrageUnit, self).setUp()
nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
self.settings = DemurrageTokenSettings()
self.settings.name = 'Foo Token'
self.settings.symbol = 'FOO'
self.settings.decimals = 6
self.settings.demurrage_level = self.tax_level * (10 ** 32)
self.settings.period_minutes = int(self.period_seconds/60)
self.settings.sink_address = self.accounts[9]
self.sink_address = self.settings.sink_address
o = block_latest()
self.start_block = self.rpc.do(o)
o = block_by_number(self.start_block, include_tx=False)
r = self.rpc.do(o)
try:
self.start_time = int(r['timestamp'], 16)
except TypeError:
self.start_time = int(r['timestamp'])
self.default_supply = 1000000000000
self.default_supply_cap = int(self.default_supply * 10)
nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
self.mode = os.environ.get('ERC20_DEMURRAGE_TOKEN_TEST_MODE')
unit_valid_modes = [
'SingleNocap',
'SingleCap',
]
if self.mode != None:
if self.mode not in unit_valid_modes:
raise ValueError('Invalid mode "{}" for "unit" contract tests, valid are {}'.format(self.mode, unit_valid_modes))
else:
self.mode = 'SingleNocap'
logg.debug('executing test setup unit mode {}'.format(self.mode))
self.deploy(c, self.mode)
logg.info('deployed with mode {}'.format(self.mode))
self.default_supply = 10**12
self.default_supply_cap = self.default_supply

146
python/erc20_demurrage_token/unittest/newbase.py Normal file
View File

@@ -0,0 +1,146 @@
# standard imports
import logging
import os
import math
# external imports
from chainlib.eth.unittest.ethtester import EthTesterCase
from chainlib.eth.tx import (
receipt,
)
from chainlib.eth.block import (
block_latest,
block_by_number,
)
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.constant import ZERO_ADDRESS
# local imports
from erc20_demurrage_token import (
DemurrageTokenSettings,
DemurrageToken,
)
from dexif import *
logg = logging.getLogger()
TAX_LEVEL = int(10000 * 2) # 2%
PERIOD = 43200 # 30 days in minutes
class TestTokenDeploy:
"""tax level is ppm, 1000000 = 100%"""
def __init__(self, rpc, token_symbol='FOO', token_name='Foo Token', sink_address=ZERO_ADDRESS, supply=10**12, tax_level=TAX_LEVEL, period=PERIOD):
self.tax_level = tax_level
self.period_seconds = period * 60
self.settings = DemurrageTokenSettings()
self.settings.name = token_name
self.settings.symbol = token_symbol
self.settings.decimals = 6
tax_level_input = to_fixed((1 - (tax_level / 1000000)) ** (1 / period))
self.settings.demurrage_level = tax_level_input
self.settings.period_minutes = period
self.settings.sink_address = sink_address
self.sink_address = self.settings.sink_address
logg.debug('using demurrage token settings: {}'.format(self.settings))
o = block_latest()
self.start_block = rpc.do(o)
o = block_by_number(self.start_block, include_tx=False)
r = rpc.do(o)
try:
self.start_time = int(r['timestamp'], 16)
except TypeError:
self.start_time = int(r['timestamp'])
self.default_supply = supply
self.default_supply_cap = 0
def deploy(self, rpc, deployer_address, interface, supply_cap=0):
tx_hash = None
o = None
(tx_hash, o) = interface.constructor(deployer_address, self.settings, redistribute=False, cap=0)
r = rpc.do(o)
o = receipt(tx_hash)
r = rpc.do(o)
assert r['status'] == 1
self.start_block = r['block_number']
self.address = r['contract_address']
o = block_by_number(r['block_number'])
r = rpc.do(o)
self.start_time = r['timestamp']
return self.address
class TestDemurrage(EthTesterCase):
def setUp(self):
super(TestDemurrage, self).setUp()
period = PERIOD
try:
period = getattr(self, 'period')
except AttributeError as e:
pass
self.deployer = TestTokenDeploy(self.rpc, period=period)
self.default_supply = self.deployer.default_supply
self.default_supply_cap = self.deployer.default_supply_cap
self.start_block = None
self.address = None
self.start_time = None
def deploy(self, interface):
self.address = self.deployer.deploy(self.rpc, self.accounts[0], interface, supply_cap=self.default_supply_cap)
self.start_block = self.deployer.start_block
self.start_time = self.deployer.start_time
self.tax_level = self.deployer.tax_level
self.period_seconds = self.deployer.period_seconds
self.sink_address = self.deployer.sink_address
logg.debug('contract address {} start block {} start time {}'.format(self.address, self.start_block, self.start_time))
def assert_within(self, v, target, tolerance_ppm):
lower_target = target - (target * (tolerance_ppm / 1000000))
higher_target = target + (target * (tolerance_ppm / 1000000))
#self.assertGreaterEqual(v, lower_target)
#self.assertLessEqual(v, higher_target)
if v >= lower_target and v <= higher_target:
logg.debug('asserted within {} <= {} <= {}'.format(lower_target, v, higher_target))
return
raise AssertionError('{} not within lower {} and higher {}'.format(v, lower_target, higher_target))
def assert_within_lower(self, v, target, tolerance_ppm):
lower_target = target - (target * (tolerance_ppm / 1000000))
self.assertGreaterEqual(v, lower_target)
self.assertLessEqual(v, target)
logg.debug('asserted within lower {} <= {} <= {}'.format(lower_target, v, target))
def assert_equal_decimals(self, v, target, precision):
target = int(target * (10 ** precision))
target = target / (10 ** precision)
self.assertEqual(v, target)
def tearDown(self):
pass
class TestDemurrageDefault(TestDemurrage):
def setUp(self):
super(TestDemurrageDefault, self).setUp()
nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
c = DemurrageToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
self.deploy(c)