cic-stack/apps/cic-eth-aux/erc20-demurrage-token/tests/test_demurrage_task.py

70 lines
1.8 KiB
Python

# standard imports
import logging
import copy
import datetime
# external imports
import celery
# cic-eth imports
from cic_eth_aux.erc20_demurrage_token import (
DemurrageCalculationTask,
aux_setup,
)
from cic_eth_aux.erc20_demurrage_token.api import Api as AuxApi
logg = logging.getLogger()
def test_demurrage_calulate_task(
default_chain_spec,
eth_rpc,
cic_registry,
celery_session_worker,
register_demurrage_token,
demurrage_token_symbol,
contract_roles,
load_config,
):
config = copy.copy(load_config)
config.add(str(default_chain_spec), 'CIC_CHAIN_SPEC', exists_ok=True)
config.add(demurrage_token_symbol, 'CIC_DEFAULT_TOKEN_SYMBOL', exists_ok=True)
aux_setup(eth_rpc, load_config, sender_address=contract_roles['CONTRACT_DEPLOYER'])
since = datetime.datetime.utcnow() - datetime.timedelta(minutes=1)
s = celery.signature(
'cic_eth_aux.erc20_demurrage_token.get_adjusted_balance',
[
demurrage_token_symbol,
1000,
since.timestamp(),
],
queue=None,
)
t = s.apply_async()
r = t.get_leaf()
assert t.successful()
assert r == 980
def test_demurrage_calculate_api(
default_chain_spec,
eth_rpc,
cic_registry,
celery_session_worker,
register_demurrage_token,
demurrage_token_symbol,
contract_roles,
load_config,
):
api = AuxApi(str(default_chain_spec), queue=None)
since = datetime.datetime.utcnow() - datetime.timedelta(minutes=1)
t = api.get_adjusted_balance(demurrage_token_symbol, 1000, since.timestamp())
r = t.get_leaf()
assert t.successful()
assert r == 980