From 77fe41da4bdaa8b113cd0f10bec0e7220e0bec4b Mon Sep 17 00:00:00 2001 From: nolash Date: Sat, 9 Oct 2021 13:24:57 +0200 Subject: [PATCH] Revert to multi-token task queue entry point for token api call --- apps/cic-eth/cic_eth/api/api_task.py | 51 ++++------- apps/cic-eth/cic_eth/eth/erc20.py | 91 ++++++++----------- apps/cic-eth/cic_eth/pytest/mock/callback.py | 21 +++-- .../tests/task/api/test_app_noncritical.py | 1 + 4 files changed, 73 insertions(+), 91 deletions(-) diff --git a/apps/cic-eth/cic_eth/api/api_task.py b/apps/cic-eth/cic_eth/api/api_task.py index 31a18466..6c596a7b 100644 --- a/apps/cic-eth/cic_eth/api/api_task.py +++ b/apps/cic-eth/cic_eth/api/api_task.py @@ -45,28 +45,25 @@ class Api(ApiBase): proof = [proof] chain_spec_dict = self.chain_spec.asdict() - i = 0 - s_group = [] - for token_symbol in token_symbols: - s_token_resolve = celery.signature( - 'cic_eth.eth.erc20.resolve_token_by_symbol', - [ - token_symbol, - chain_spec_dict, - ], - queue=self.queue, - ) + s_token_resolve = celery.signature( + 'cic_eth.eth.erc20.resolve_tokens_by_symbol', + [ + token_symbols, + chain_spec_dict, + ], + queue=self.queue, + ) - s_token_info = celery.signature( - 'cic_eth.eth.erc20.token_info', - [ - chain_spec_dict, - proof[i], - ], - queue=self.queue, - ) + s_token_info = celery.signature( + 'cic_eth.eth.erc20.token_info', + [ + chain_spec_dict, + proof, + ], + queue=self.queue, + ) - s_token_verify = celery.signature( + s_token_verify = celery.signature( 'cic_eth.eth.erc20.verify_token_info', [ chain_spec_dict, @@ -75,18 +72,10 @@ class Api(ApiBase): ], queue=self.queue, ) - i += 1 - - s_token_info.link(s_token_verify) - s_token_resolve.link(s_token_info) - #if self.callback_param != None: - # s_token_verify.link(self.callback_success) - # s_token_verify.on_error(self.callback_error) - - s_group.append(s_token_resolve) - # return s_token_resolve.apply_async() - return celery.group(s_group)() + s_token_info.link(s_token_verify) + s_token_resolve.link(s_token_info) + return s_token_resolve.apply_async() # def convert_transfer(self, from_address, to_address, target_return, minimum_return, from_token_symbol, to_token_symbol): diff --git a/apps/cic-eth/cic_eth/eth/erc20.py b/apps/cic-eth/cic_eth/eth/erc20.py index a43f65cc..2c2685d3 100644 --- a/apps/cic-eth/cic_eth/eth/erc20.py +++ b/apps/cic-eth/cic_eth/eth/erc20.py @@ -328,26 +328,6 @@ def approve(self, tokens, holder_address, spender_address, value, chain_spec_dic return tx_hash_hex -@celery_app.task(bind=True, base=CriticalWeb3Task) -def resolve_token_by_symbol(self, token_symbol, chain_spec_dict): - chain_spec = ChainSpec.from_dict(chain_spec_dict) - rpc = RPCConnection.connect(chain_spec, 'default') - registry = CICRegistry(chain_spec, rpc) - - session = self.create_session() - sender_address = AccountRole.get_address('DEFAULT', session) - session.close() - - token_address = registry.by_name(token_symbol, sender_address=sender_address) - logg.debug('token {}'.format(token_address)) - token = { - 'address': token_address, - 'symbol': token_symbol, - 'converters': [], - } - return token - - @celery_app.task(bind=True, base=CriticalWeb3Task) def resolve_tokens_by_symbol(self, token_symbols, chain_spec_dict): """Returns contract addresses of an array of ERC20 token symbols @@ -498,52 +478,59 @@ def cache_approve_data( @celery_app.task(bind=True, base=BaseTask) -def token_info(self, token, chain_spec_dict, proofs=[]): +def token_info(self, tokens, chain_spec_dict, proofs=[]): chain_spec = ChainSpec.from_dict(chain_spec_dict) rpc = RPCConnection.connect(chain_spec, 'default') - result_data = [] - token_chain_object = ERC20Token(chain_spec, rpc, add_0x(token['address'])) - token_chain_object.load(rpc) + i = 0 - token_symbol_proof_hex = to_identifier(token_chain_object.symbol) - token_proofs = [token_symbol_proof_hex] - token_proofs += proofs + for token in tokens: + result_data = [] + token_chain_object = ERC20Token(chain_spec, rpc, add_0x(token['address'])) + token_chain_object.load(rpc) - token_data = { - 'decimals': token_chain_object.decimals, - 'name': token_chain_object.name, - 'symbol': token_chain_object.symbol, - 'address': tx_normalize.executable_address(token_chain_object.address), - 'proofs': token_proofs, - } + token_symbol_proof_hex = to_identifier(token_chain_object.symbol) + token_proofs = [token_symbol_proof_hex] + token_proofs += proofs[i] - return token_data + tokens[i] = { + 'decimals': token_chain_object.decimals, + 'name': token_chain_object.name, + 'symbol': token_chain_object.symbol, + 'address': tx_normalize.executable_address(token_chain_object.address), + 'proofs': token_proofs, + 'converters': tokens[i]['converters'], + } + i += 1 + + return tokens @celery_app.task(bind=True, base=BaseTask) -def verify_token_info(self, token, chain_spec_dict, success_callback, error_callback): +def verify_token_info(self, tokens, chain_spec_dict, success_callback, error_callback): queue = self.request.delivery_info.get('routing_key') - s = celery.signature( - 'cic_eth.eth.trust.verify_proofs', - [ - token, - token['address'], - token['proofs'], - chain_spec_dict, - success_callback, - error_callback, - ], - queue=queue, - ) - s.link(success_callback) - s.on_error(error_callback) - s.apply_async() + + for token in tokens: + s = celery.signature( + 'cic_eth.eth.trust.verify_proofs', + [ + token, + token['address'], + token['proofs'], + chain_spec_dict, + success_callback, + error_callback, + ], + queue=queue, + ) + s.link(success_callback) + s.on_error(error_callback) + s.apply_async() #s_group.append(s) #celery.group(s_group)() - return token + return tokens @celery_app.task(bind=True, base=BaseTask) diff --git a/apps/cic-eth/cic_eth/pytest/mock/callback.py b/apps/cic-eth/cic_eth/pytest/mock/callback.py index 8453c30f..d1d6fc4b 100644 --- a/apps/cic-eth/cic_eth/pytest/mock/callback.py +++ b/apps/cic-eth/cic_eth/pytest/mock/callback.py @@ -1,5 +1,9 @@ # standard imports import logging +import mmap + +# standard imports +import tempfile # external imports import celery @@ -10,19 +14,20 @@ logg = logging.getLogger() celery_app = celery.current_app class CallbackTask(celery.Task): - errs = {} - oks = {} + + mmap_path = tempfile.mkdtemp() @celery_app.task(bind=True, base=CallbackTask) def test_error_callback(self, a, b, c): - o = CallbackTask.oks s = 'ok' if c > 0: - o = CallbackTask.errs s = 'err' - if o.get(b) == None: - o[b] = [] - o[b].append(a) + fp = os.path.join(self.mmap_path, b) + f = open(fp, 'wb') + m = mmap.mmap(f.fileno(), access=mmap.ACCESS_WRITE, length=1) + m.write(c.to_bytes(1, 'big')) + m.close() + f.close() - logg.debug('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> test callback ({}): {} {} {}'.format(s, a, b, c)) + logg.debug('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> test callback ({}): {} {} {} -> {}'.format(s, a, b, c, fp)) diff --git a/apps/cic-eth/tests/task/api/test_app_noncritical.py b/apps/cic-eth/tests/task/api/test_app_noncritical.py index d5007010..af42b79c 100644 --- a/apps/cic-eth/tests/task/api/test_app_noncritical.py +++ b/apps/cic-eth/tests/task/api/test_app_noncritical.py @@ -78,6 +78,7 @@ def test_tokens( t = api.tokens(['FOO'], proof=[[bogus_proof]]) r = t.get() logg.debug('r {}'.format(r)) + time.sleep(0.1) assert len(CallbackTask.errs[api_param]) == 1 assert CallbackTask.oks.get(api_param) == None