init

Bvander/fix notify migrations

See merge request grassrootseconomics/cic-internal-integration!238

apps/cic-eth/admin_requirements.txt

@@ -1,5 +1,5 @@
 SQLAlchemy==1.3.20
-cic-eth-registry~=0.5.6a2
+cic-eth-registry>=0.5.6a2,<0.6.0
 hexathon~=0.0.1a7
-chainqueue~=0.0.2b6
-eth-erc20~=0.0.10a3
+chainqueue>=0.0.3a1,<0.1.0
+eth-erc20>=0.0.10a3,<0.1.0

apps/cic-eth/cic_eth/admin/ctrl.py

@@ -6,6 +6,11 @@ import logging
 import celery
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.chain import ChainSpec
+from hexathon import (
+        add_0x,
+        strip_0x,
+        uniform as hex_uniform,
+        )
 
 # local imports
 from cic_eth.db.enum import LockEnum
@@ -19,6 +24,12 @@ from cic_eth.error import LockedError
 celery_app = celery.current_app
 logg = logging.getLogger()
 
+
+def normalize_address(a):
+    if a == None:
+        return None
+    return add_0x(hex_uniform(strip_0x(a)))
+
 @celery_app.task(base=CriticalSQLAlchemyTask)
 def lock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.ALL, tx_hash=None):
     """Task wrapper to set arbitrary locks
@@ -32,6 +43,7 @@ def lock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.AL
     :returns: New lock state for address
     :rtype: number
     """
+    address = normalize_address(address)
     chain_str = '::'
     if chain_spec_dict != None:
         chain_str = str(ChainSpec.from_dict(chain_spec_dict))
@@ -53,6 +65,7 @@ def unlock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.
     :returns: New lock state for address
     :rtype: number
     """
+    address = normalize_address(address)
     chain_str = '::'
     if chain_spec_dict != None:
         chain_str = str(ChainSpec.from_dict(chain_spec_dict))
@@ -72,6 +85,7 @@ def lock_send(chained_input, chain_spec_dict, address=ZERO_ADDRESS, tx_hash=None
     :returns: New lock state for address
     :rtype: number
     """
+    address = normalize_address(address)
     chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, LockEnum.SEND, address=address, tx_hash=tx_hash)
     logg.debug('Send locked for {}, flag now {}'.format(address, r))
@@ -89,6 +103,7 @@ def unlock_send(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
     :returns: New lock state for address
     :rtype: number
     """
+    address = normalize_address(address)
     chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, LockEnum.SEND, address=address)
     logg.debug('Send unlocked for {}, flag now {}'.format(address, r))
@@ -106,6 +121,7 @@ def lock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS, tx_hash=Non
     :returns: New lock state for address
     :rtype: number
     """
+    address = normalize_address(address)
     chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, LockEnum.QUEUE, address=address, tx_hash=tx_hash)
     logg.debug('Queue direct locked for {}, flag now {}'.format(address, r))
@@ -123,6 +139,7 @@ def unlock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
     :returns: New lock state for address
     :rtype: number
     """
+    address = normalize_address(address)
     chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, LockEnum.QUEUE, address=address)
     logg.debug('Queue direct unlocked for {}, flag now {}'.format(address, r))
@@ -131,6 +148,7 @@ def unlock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
 
 @celery_app.task(base=CriticalSQLAlchemyTask)
 def check_lock(chained_input, chain_spec_dict, lock_flags, address=None):
+    address = normalize_address(address)
     chain_str = '::'
     if chain_spec_dict != None:
         chain_str = str(ChainSpec.from_dict(chain_spec_dict))

apps/cic-eth/cic_eth/admin/nonce.py

@@ -14,7 +14,11 @@ from chainqueue.sql.query import get_tx
 from chainqueue.sql.state import set_cancel
 from chainqueue.db.models.otx import Otx
 from chainqueue.db.models.tx import TxCache
-from hexathon import strip_0x
+from hexathon import (
+        strip_0x,
+        add_0x,
+        uniform as hex_uniform,
+        )
 from potaahto.symbols import snake_and_camel
 
 # local imports
@@ -69,15 +73,17 @@ def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
 
     set_cancel(chain_spec, strip_0x(tx['hash']), manual=True, session=session)
 
+    query_address = add_0x(hex_uniform(strip_0x(address))) # aaaaargh
     q = session.query(Otx)
     q = q.join(TxCache)
-    q = q.filter(TxCache.sender==address)
+    q = q.filter(TxCache.sender==query_address)
     q = q.filter(Otx.nonce>=nonce+delta)
     q = q.order_by(Otx.nonce.asc())
     otxs = q.all()
 
     tx_hashes = []
     txs = []
+    gas_total = 0
     for otx in otxs:
         tx_raw = bytes.fromhex(strip_0x(otx.signed_tx))
         tx_new = unpack(tx_raw, chain_spec)
@@ -89,8 +95,10 @@ def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
         tx_new['gas_price'] += 1
         tx_new['gasPrice'] = tx_new['gas_price']
         tx_new['nonce'] -= delta
+        gas_total += tx_new['gas_price'] * tx_new['gas']
 
         logg.debug('tx_new {}'.format(tx_new))
+        logg.debug('gas running total {}'.format(gas_total))
 
         del(tx_new['hash'])
         del(tx_new['hash_unsigned'])
@@ -122,8 +130,10 @@ def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
     s = create_check_gas_task(
          txs, 
          chain_spec,
-         tx_new['from'],
-         gas=tx_new['gas'],
+         #tx_new['from'],
+         address,
+         #gas=tx_new['gas'],
+         gas=gas_total,
          tx_hashes_hex=tx_hashes, 
          queue=queue,
         )
@@ -132,7 +142,8 @@ def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
         'cic_eth.admin.ctrl.unlock_send',
         [
             chain_spec.asdict(),
-            tx_new['from'],
+            address,
+            #tx_new['from'],
             ],
         queue=queue,
         )
@@ -140,7 +151,8 @@ def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
         'cic_eth.admin.ctrl.unlock_queue',
         [
             chain_spec.asdict(),
-            tx_new['from'],
+            address,
+            #tx_new['from'],
             ],
         queue=queue,
         )

apps/cic-eth/cic_eth/api/admin.py

@@ -21,6 +21,7 @@ from chainlib.hash import keccak256_hex_to_hex
 from hexathon import (
         strip_0x,
         add_0x,
+        uniform as hex_uniform,
         )
 from chainlib.eth.gas import balance
 from chainqueue.db.enum import (
@@ -307,6 +308,8 @@ class AdminApi:
         :param address: Ethereum address to return transactions for
         :type address: str, 0x-hex
         """
+
+        address = add_0x(hex_uniform(strip_0x(address)))
         last_nonce = -1
         s = celery.signature(
                 'cic_eth.queue.query.get_account_tx',

apps/cic-eth/cic_eth/db/migrations/default/versions/0ec0d6d1e785_add_chainqueue.py

@@ -8,7 +8,8 @@ Create Date: 2021-04-02 18:30:55.398388
 from alembic import op
 import sqlalchemy as sa
 
-from chainqueue.db.migrations.sqlalchemy import (
+#from chainqueue.db.migrations.sqlalchemy import (
+from chainqueue.db.migrations.default.export import (
         chainqueue_upgrade,
         chainqueue_downgrade,
         )

apps/cic-eth/cic_eth/db/migrations/default/versions/b125cbf81e32_add_chain_syncer.py

@@ -8,7 +8,8 @@ Create Date: 2021-04-02 18:36:44.459603
 from alembic import op
 import sqlalchemy as sa
 
-from chainsyncer.db.migrations.sqlalchemy import (
+#from chainsyncer.db.migrations.sqlalchemy import (
+from chainsyncer.db.migrations.default.export import (
         chainsyncer_upgrade,
         chainsyncer_downgrade,
         )

apps/cic-eth/cic_eth/eth/account.py

@@ -23,7 +23,7 @@ from chainlib.error import JSONRPCException
 from eth_accounts_index.registry import AccountRegistry
 from eth_accounts_index import AccountsIndex 
 from sarafu_faucet import MinterFaucet
-from chainqueue.db.models.tx import TxCache
+from chainqueue.sql.tx import cache_tx_dict
 
 # local import
 from cic_eth_registry import CICRegistry
@@ -300,20 +300,17 @@ def cache_gift_data(
 
     session = self.create_session()
 
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        tx['to'],
-        ZERO_ADDRESS,
-        ZERO_ADDRESS,
-        0,
-        0,
-        session=session,
-            )
+    tx_dict = {
+            'hash': tx_hash_hex,
+            'from': tx['from'],
+            'to': tx['to'],
+            'source_token': ZERO_ADDRESS,
+            'destination_token': ZERO_ADDRESS,
+            'from_value': 0,
+            'to_value': 0,
+            }
 
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
+    (tx_dict, cache_id) = cache_tx_dict(tx_dict, session=session)
     session.close()
     return (tx_hash_hex, cache_id)
 
@@ -342,18 +339,15 @@ def cache_account_data(
     tx_data = AccountsIndex.parse_add_request(tx['data'])
 
     session = SessionBase.create_session()
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        tx['to'],
-        ZERO_ADDRESS,
-        ZERO_ADDRESS,
-        0,
-        0,
-        session=session,
-            )
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
+    tx_dict = {
+            'hash': tx_hash_hex,
+            'from': tx['from'],
+            'to': tx['to'],
+            'source_token': ZERO_ADDRESS,
+            'destination_token': ZERO_ADDRESS,
+            'from_value': 0,
+            'to_value': 0,
+            }
+    (tx_dict, cache_id) = cache_tx_dict(tx_dict, session=session)
     session.close()
     return (tx_hash_hex, cache_id)

apps/cic-eth/cic_eth/eth/bancor.py.bak

@@ -1,385 +0,0 @@
-# standard imports
-import os
-import logging
-
-# third-party imports
-import celery
-import web3
-from cic_registry import CICRegistry
-from cic_registry.chain import ChainSpec
-
-# local imports
-from cic_eth.db import SessionBase
-from cic_eth.db.models.convert import TxConvertTransfer
-from cic_eth.db.models.otx import Otx
-from cic_eth.db.models.tx import TxCache
-from cic_eth.eth.task import sign_and_register_tx
-from cic_eth.eth.task import create_check_gas_and_send_task
-from cic_eth.eth.token import TokenTxFactory
-from cic_eth.eth.factory import TxFactory
-from cic_eth.eth.util import unpack_signed_raw_tx
-from cic_eth.eth.rpc import RpcClient
-
-celery_app = celery.current_app 
-#logg = celery_app.log.get_default_logger()
-logg = logging.getLogger()
-
-contract_function_signatures = {
-        'convert': 'f3898a97',
-        'convert2': '569706eb',
-        }
-
-
-class BancorTxFactory(TxFactory):
-
-    """Factory for creating Bancor network transactions.
-    """
-    def convert(
-            self,
-            source_token_address,
-            destination_token_address,
-            reserve_address,
-            source_amount,
-            minimum_return,
-            chain_spec,
-            fee_beneficiary='0x0000000000000000000000000000000000000000',
-            fee_ppm=0,
-            ):
-        """Create a BancorNetwork "convert" transaction.
-
-        :param source_token_address: ERC20 contract address for token to convert from
-        :type source_token_address: str, 0x-hex
-        :param destination_token_address: ERC20 contract address for token to convert to
-        :type destination_token_address: str, 0x-hex
-        :param reserve_address: ERC20 contract address of Common reserve token
-        :type reserve_address: str, 0x-hex
-        :param source_amount: Amount of source tokens to convert
-        :type source_amount: int
-        :param minimum_return: Minimum amount of destination tokens to accept as result for conversion
-        :type source_amount: int
-        :return: Unsigned "convert" transaction in standard Ethereum format
-        :rtype: dict
-        """
-        network_contract = CICRegistry.get_contract(chain_spec, 'BancorNetwork')
-        network_gas = network_contract.gas('convert')
-        tx_convert_buildable = network_contract.contract.functions.convert2(
-            [
-                source_token_address,
-                source_token_address,
-                reserve_address,
-                destination_token_address,
-                destination_token_address,
-            ],
-            source_amount,
-            minimum_return,
-            fee_beneficiary,
-            fee_ppm,
-            )
-        tx_convert = tx_convert_buildable.buildTransaction({
-                'from': self.address,
-                'gas': network_gas,
-                'gasPrice': self.gas_price,
-                'chainId': chain_spec.chain_id(),
-                'nonce': self.next_nonce(),
-                })
-        return tx_convert
-
-
-def unpack_convert(data):
-    f = data[2:10]
-    if f != contract_function_signatures['convert2']:
-        raise ValueError('Invalid convert data ({})'.format(f))
-
-    d = data[10:]
-    path = d[384:]
-    source = path[64-40:64]
-    destination = path[-40:]
-
-    amount = int(d[64:128], 16)
-    min_return = int(d[128:192], 16)
-    fee_recipient = d[192:256]
-    fee = int(d[256:320], 16)
-    return {
-        'amount': amount,
-        'min_return': min_return,
-        'source_token': web3.Web3.toChecksumAddress('0x' + source),
-        'destination_token': web3.Web3.toChecksumAddress('0x' + destination),
-        'fee_recipient': fee_recipient,
-        'fee': fee,
-        }
-
-
-
-# Kept for historical reference, it unpacks a convert call without fee parameters
-#def _unpack_convert_mint(data):
-#    f = data[2:10]
-#    if f != contract_function_signatures['convert2']:
-#        raise ValueError('Invalid convert data ({})'.format(f))
-#
-#    d = data[10:]
-#    path = d[256:]
-#    source = path[64-40:64]
-#    destination = path[-40:]
-#
-#    amount = int(d[64:128], 16)
-#    min_return = int(d[128:192], 16)
-#    return {
-#        'amount': amount,
-#        'min_return': min_return,
-#        'source_token': web3.Web3.toChecksumAddress('0x' + source),
-#        'destination_token': web3.Web3.toChecksumAddress('0x' + destination),
-#        }
-
-
-@celery_app.task(bind=True)
-def convert_with_default_reserve(self, tokens, from_address, source_amount, minimum_return, to_address, chain_str):
-    """Performs a conversion between two liquid tokens using Bancor network.
-
-    :param tokens: Token pair, source and destination respectively
-    :type tokens: list of str, 0x-hex
-    :param from_address: Ethereum address of sender
-    :type from_address: str, 0x-hex
-    :param source_amount: Amount of source tokens to convert
-    :type source_amount: int
-    :param minimum_return: Minimum about of destination tokens to receive
-    :type minimum_return: int
-    """
-
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-    queue = self.request.delivery_info['routing_key']
-
-    c = RpcClient(chain_spec, holder_address=from_address)
-
-    cr = CICRegistry.get_contract(chain_spec, 'BancorNetwork')
-    source_token = CICRegistry.get_address(chain_spec, tokens[0]['address'])
-    reserve_address = CICRegistry.get_contract(chain_spec, 'BNTToken', 'ERC20').address()
-
-    tx_factory = TokenTxFactory(from_address, c)
-   
-    tx_approve_zero = tx_factory.approve(source_token.address(), cr.address(), 0, chain_spec)
-    (tx_approve_zero_hash_hex, tx_approve_zero_signed_hex) = sign_and_register_tx(tx_approve_zero, chain_str, queue, 'cic_eth.eth.token.otx_cache_approve') 
-
-    tx_approve = tx_factory.approve(source_token.address(), cr.address(), source_amount, chain_spec)
-    (tx_approve_hash_hex, tx_approve_signed_hex) = sign_and_register_tx(tx_approve, chain_str, queue, 'cic_eth.eth.token.otx_cache_approve') 
-
-    tx_factory = BancorTxFactory(from_address, c)
-    tx_convert = tx_factory.convert(
-            tokens[0]['address'],
-            tokens[1]['address'],
-            reserve_address,
-            source_amount,
-            minimum_return,
-            chain_spec,
-            )
-    (tx_convert_hash_hex, tx_convert_signed_hex) = sign_and_register_tx(tx_convert, chain_str, queue, 'cic_eth.eth.bancor.otx_cache_convert')
-
-    # TODO: consider moving save recipient to async task / chain it before the tx send
-    if to_address != None:
-        save_convert_recipient(tx_convert_hash_hex, to_address, chain_str)
-
-    s = create_check_gas_and_send_task(
-            [tx_approve_zero_signed_hex, tx_approve_signed_hex, tx_convert_signed_hex],
-            chain_str,
-            from_address,
-            tx_approve_zero['gasPrice'] * tx_approve_zero['gas'],
-            tx_hashes_hex=[tx_approve_hash_hex],
-            queue=queue,
-            )
-    s.apply_async()
-    return tx_convert_hash_hex
-
-
-#@celery_app.task()
-#def process_approval(tx_hash_hex):
-#    t = session.query(TxConvertTransfer).query(TxConvertTransfer.approve_tx_hash==tx_hash_hex).first()
-#    c = session.query(Otx).query(Otx.tx_hash==t.convert_tx_hash)
-#    gas_limit = 8000000
-#    gas_price = GasOracle.gas_price()
-#
-#    # TODO: use celery group instead
-#    s_queue = celery.signature(
-#            'cic_eth.queue.tx.create',
-#            [
-#                nonce,
-#                c['address'], # TODO: check that this is in fact sender address
-#                c['tx_hash'],
-#                c['signed_tx'],
-#                ]
-#            )
-#    s_queue.apply_async()
-#
-#    s_check_gas = celery.signature(
-#            'cic_eth.eth.gas.check_gas',
-#            [
-#                c['address'],
-#                [c['signed_tx']],
-#                gas_limit * gas_price,
-#                ]
-#            )
-#    s_send = celery.signature(
-#            'cic_eth.eth.tx.send',
-#            [],
-#            )
-#
-#    s_set_sent = celery.signature(
-#            'cic_eth.queue.state.set_sent',
-#            [False],
-#            )
-#    s_send.link(s_set_sent)
-#    s_check_gas.link(s_send)
-#    s_check_gas.apply_async()
-#    return tx_hash_hex
-
-
-
-@celery_app.task()
-def save_convert_recipient(convert_hash, recipient_address, chain_str):
-    """Registers the recipient target for a convert-and-transfer operation.
-
-    :param convert_hash: Transaction hash of convert operation
-    :type convert_hash: str, 0x-hex
-    :param recipient_address: Address of consequtive transfer recipient
-    :type recipient_address: str, 0x-hex
-    """
-    session = SessionBase.create_session()
-    t = TxConvertTransfer(convert_hash, recipient_address, chain_str)
-    session.add(t)
-    session.commit()
-    session.close()
-
-
-@celery_app.task()
-def save_convert_transfer(convert_hash, transfer_hash):
-    """Registers that the transfer part of a convert-and-transfer operation has been executed.
-
-    :param convert_hash: Transaction hash of convert operation
-    :type convert_hash: str, 0x-hex
-    :param convert_hash: Transaction hash of transfer operation
-    :type convert_hash: str, 0x-hex
-    :returns: transfer_hash,
-    :rtype: list, single str, 0x-hex
-    """
-    session = SessionBase.create_session()
-    t = TxConvertTransfer.get(convert_hash)
-    t.transfer(transfer_hash)
-    session.add(t)
-    session.commit()
-    session.close()
-    return [transfer_hash]
-
-
-# TODO: seems unused, consider removing
-@celery_app.task()
-def resolve_converters_by_tokens(tokens, chain_str):
-    """Return converters for a list of tokens.
-
-    :param tokens: Token addresses to look up
-    :type tokens: list of str, 0x-hex
-    :return: Addresses of matching converters
-    :rtype: list of str, 0x-hex
-    """
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-    for t in tokens:
-        c = CICRegistry.get_contract(chain_spec, 'ConverterRegistry')
-        fn = c.function('getConvertersByAnchors')
-        try:
-            converters = fn([t['address']]).call()
-        except Exception as e:
-            raise e
-        t['converters'] = converters
-
-    return tokens
-
-
-@celery_app.task(bind=True)
-def transfer_converted(self, tokens, holder_address, receiver_address, value, tx_convert_hash_hex, chain_str):
-    """Execute the ERC20 transfer of a convert-and-transfer operation.
-
-    First argument is a list of tokens, to enable the task to be chained to the symbol to token address resolver function. However, it accepts only one token as argument.
-
-    :param tokens: Token addresses 
-    :type tokens: list of str, 0x-hex
-    :param holder_address: Token holder address
-    :type holder_address: str, 0x-hex
-    :param holder_address: Token receiver address
-    :type holder_address: str, 0x-hex
-    :param value: Amount of token, in 'wei'
-    :type value: int
-    :raises TokenCountError: Either none or more then one tokens have been passed as tokens argument
-    :return: Transaction hash
-    :rtype: str, 0x-hex
-    """
-    # we only allow one token, one transfer
-    if len(tokens) != 1:
-        raise TokenCountError
-
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-
-    queue = self.request.delivery_info['routing_key']
-
-    c = RpcClient(chain_spec, holder_address=holder_address)
-
-    # get transaction parameters
-    gas_price = c.gas_price()
-    tx_factory = TokenTxFactory(holder_address, c)
-
-    token_address = tokens[0]['address']
-    tx_transfer = tx_factory.transfer(
-        token_address,
-        receiver_address,
-        value,
-        chain_spec,
-            )
-    (tx_transfer_hash_hex, tx_transfer_signed_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, 'cic_eth.eth.token.otx_cache_transfer')
-
-    # send transaction
-    logg.info('transfer converted token {} from {} to {} value {} {}'.format(token_address, holder_address, receiver_address, value, tx_transfer_signed_hex))
-    s = create_check_gas_and_send_task(
-            [tx_transfer_signed_hex],
-            chain_str,
-            holder_address,
-            tx_transfer['gasPrice'] * tx_transfer['gas'],
-            None,
-            queue,
-            )
-    s_save = celery.signature(
-            'cic_eth.eth.bancor.save_convert_transfer',
-            [
-                tx_convert_hash_hex,
-                tx_transfer_hash_hex,
-                ],
-            queue=queue,
-            )
-    s_save.link(s)
-    s_save.apply_async()
-    return tx_transfer_hash_hex
-
-
-@celery_app.task()
-def otx_cache_convert(
-        tx_hash_hex,
-        tx_signed_raw_hex,
-        chain_str,
-        ):
-
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-    tx_signed_raw_bytes = bytes.fromhex(tx_signed_raw_hex[2:])
-    tx = unpack(tx_signed_raw_bytes, chain_spec)
-    tx_data = unpack_convert(tx['data'])
-    logg.debug('tx data {}'.format(tx_data))
-
-    session = TxCache.create_session()
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        tx['from'],
-        tx_data['source_token'],
-        tx_data['destination_token'],
-        tx_data['amount'],
-        tx_data['amount'],
-            )
-    session.add(tx_cache)
-    session.commit()
-    session.close()
-    return tx_hash_hex
-

apps/cic-eth/cic_eth/eth/erc20.py

@@ -13,9 +13,9 @@ from chainlib.eth.tx import (
 from cic_eth_registry import CICRegistry
 from cic_eth_registry.erc20 import ERC20Token
 from hexathon import strip_0x
-from chainqueue.db.models.tx import TxCache
 from chainqueue.error import NotLocalTxError
 from eth_erc20 import ERC20
+from chainqueue.sql.tx import cache_tx_dict
 
 # local imports
 from cic_eth.db.models.base import SessionBase
@@ -375,19 +375,16 @@ def cache_transfer_data(
     token_value = tx_data[1]
 
     session = SessionBase.create_session()
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        recipient_address,
-        tx['to'],
-        tx['to'],
-        token_value,
-        token_value,
-        session=session,
-            )
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
+    tx_dict = {
+            'hash': tx_hash_hex,
+            'from': tx['from'],
+            'to': recipient_address,
+            'source_token': tx['to'],
+            'destination_token': tx['to'],
+            'from_value': token_value,
+            'to_value': token_value,
+            }
+    (tx_dict, cache_id) = cache_tx_dict(tx_dict, session=session)
     session.close()
     return (tx_hash_hex, cache_id)
 
@@ -417,19 +414,16 @@ def cache_transfer_from_data(
     token_value = tx_data[2]
 
     session = SessionBase.create_session()
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        recipient_address,
-        tx['to'],
-        tx['to'],
-        token_value,
-        token_value,
-        session=session,
-            )
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
+    tx_dict = {
+            'hash': tx_hash_hex,
+            'from': tx['from'],
+            'to': recipient_address,
+            'source_token': tx['to'],
+            'destination_token': tx['to'],
+            'from_value': token_value,
+            'to_value': token_value,
+            }
+    (tx_dict, cache_id) = cache_tx_dict(tx_dict, session=session)
     session.close()
     return (tx_hash_hex, cache_id)
 
@@ -458,19 +452,16 @@ def cache_approve_data(
     token_value = tx_data[1]
 
     session = SessionBase.create_session()
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        recipient_address,
-        tx['to'],
-        tx['to'],
-        token_value,
-        token_value,
-        session=session,
-            )
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
+    tx_dict = {
+            'hash': tx_hash_hex,
+            'from': tx['from'],
+            'to': recipient_address,
+            'source_token': tx['to'],
+            'destination_token': tx['to'],
+            'from_value': token_value,
+            'to_value': token_value,
+            }
+    (tx_dict, cache_id) = cache_tx_dict(tx_dict, session=session)
     session.close()
     return (tx_hash_hex, cache_id)
 

apps/cic-eth/cic_eth/eth/gas.py

@@ -9,6 +9,7 @@ from chainlib.chain import ChainSpec
 from chainlib.eth.address import is_checksum_address
 from chainlib.connection import RPCConnection
 from chainqueue.db.enum import StatusBits
+from chainqueue.sql.tx import cache_tx_dict
 from chainlib.eth.gas import (
         balance,
         price,
@@ -133,20 +134,17 @@ def cache_gas_data(
 
     session = SessionBase.create_session()
 
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        tx['to'],
-        ZERO_ADDRESS,
-        ZERO_ADDRESS,
-        tx['value'],
-        tx['value'],
-        session=session,
-            )
+    tx_dict = {
+            'hash': tx_hash_hex,
+            'from': tx['from'],
+            'to': tx['to'],
+            'source_token': ZERO_ADDRESS,
+            'destination_token': ZERO_ADDRESS,
+            'from_value': tx['value'],
+            'to_value': tx['value'],
+            }
 
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
+    (tx_dict, cache_id) = cache_tx_dict(tx_dict, session=session)
     session.close()
     return (tx_hash_hex, cache_id)
 

apps/cic-eth/cic_eth/eth/tx.py

@@ -18,7 +18,6 @@ from hexathon import (
         strip_0x,
         )
 from chainqueue.db.models.tx import Otx
-from chainqueue.db.models.tx import TxCache
 from chainqueue.db.enum import StatusBits
 from chainqueue.error import NotLocalTxError
 from potaahto.symbols import snake_and_camel

apps/cic-eth/cic_eth/runnable/daemons/filters/gas.py

@@ -11,6 +11,7 @@ from chainqueue.db.enum import StatusBits
 from chainqueue.db.models.tx import TxCache
 from chainqueue.db.models.otx import Otx
 from chainqueue.sql.query import get_paused_tx_cache as get_paused_tx
+from chainlib.eth.address import to_checksum_address
 
 # local imports
 from cic_eth.db.models.base import SessionBase
@@ -47,12 +48,13 @@ class GasFilter(SyncFilter):
 
             SessionBase.release_session(session)
 
+            address = to_checksum_address(r[0])
             logg.info('resuming gas-in-waiting txs for {}'.format(r[0]))
             if len(txs) > 0:
                 s = create_check_gas_task(
                         list(txs.values()),
                         self.chain_spec,
-                        r[0],
+                        address,
                         0,
                         tx_hashes_hex=list(txs.keys()),
                         queue=self.queue,

apps/cic-eth/cic_eth/version.py

@@ -9,8 +9,8 @@ import semver
 version = (
         0,
         12,
-        1,
-        'alpha.2',
+        2,
+        'alpha.3',
         )
 
 version_object = semver.VersionInfo(

apps/cic-eth/docker/Dockerfile

@@ -13,7 +13,7 @@ ARG GITLAB_PYTHON_REGISTRY="https://gitlab.com/api/v4/projects/27624814/packages
 #    --force-reinstall \
 #	--extra-index-url $GITLAB_PYTHON_REGISTRY --extra-index-url $EXTRA_INDEX_URL \
 #    -r requirements.txt 
-COPY *requirements.txt . 
+COPY *requirements.txt ./
 RUN --mount=type=cache,mode=0755,target=/root/.cache/pip \
     pip install --index-url https://pypi.org/simple  \
 	  --extra-index-url $GITLAB_PYTHON_REGISTRY \

apps/cic-eth/requirements.txt

@@ -1,3 +1,3 @@
 celery==4.4.7
-chainlib~=0.0.5a2
+chainlib-eth>=0.0.6a1,<0.1.0
 semver==2.13.0

apps/cic-eth/services_requirements.txt

@@ -1,15 +1,15 @@
-chainsyncer[sql]~=0.0.3a5
-chainqueue~=0.0.2b7
+chainqueue>=0.0.3a2,<0.1.0
+chainsyncer[sql]>=0.0.5a1,<0.1.0
 alembic==1.4.2
 confini>=0.3.6rc4,<0.5.0
 redis==3.5.3
 hexathon~=0.0.1a7
 pycryptodome==3.10.1
 liveness~=0.0.1a7
-eth-address-index~=0.1.2a2
-eth-accounts-index~=0.0.12a2
-cic-eth-registry~=0.5.6a2
-erc20-faucet~=0.2.2a2
-erc20-transfer-authorization~=0.3.2a2
-sarafu-faucet~=0.0.4a3
+eth-address-index>=0.1.3a1,<0.2.0
+eth-accounts-index>=0.0.13a1,<0.1.0
+cic-eth-registry>=0.5.7a1,<0.6.0
+erc20-faucet>=0.2.3a1,<0.3.0
+erc20-transfer-authorization>=0.3.3a1,<0.4.0
+sarafu-faucet>=0.0.4a5,<0.1.0
 moolb~=0.1.1b2

apps/cic-eth/test_requirements.txt

@@ -6,4 +6,4 @@ pytest-redis==2.0.0
 redis==3.5.3
 eth-tester==0.5.0b3
 py-evm==0.3.0a20
-eth-erc20~=0.0.10a3
+eth-erc20~=0.0.11a1

apps/cic-eth/tests/task/api/test_admin.py

@@ -290,6 +290,7 @@ def test_fix_nonce(
     txs = get_nonce_tx_cache(default_chain_spec, 3, agent_roles['ALICE'], session=init_database)
     ks = txs.keys()
     assert len(ks) == 2
+
     for k in ks:
         hsh = add_0x(k)
         otx = Otx.load(hsh, session=init_database)

apps/cic-eth/tests/task/api/test_admin_noncritical.py

@@ -184,7 +184,7 @@ def test_admin_api_account(
 
 
     api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
-    r = api.account(default_chain_spec, agent_roles['ALICE'])
+    r = api.account(default_chain_spec, agent_roles['ALICE'], include_sender=True, include_recipient=True)
     assert len(r) == 5
 
     api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])

apps/cic-eth/tests/unit/queue/test_query.py

@@ -9,6 +9,11 @@ from chainlib.eth.gas import (
         Gas,
         )
 from chainlib.chain import ChainSpec
+from hexathon import (
+        add_0x,
+        strip_0x,
+        uniform as hex_uniform,
+        )
 
 # local imports
 from cic_eth.db.enum import LockEnum
@@ -34,7 +39,10 @@ def test_upcoming_with_lock(
     gas_oracle = RPCGasOracle(eth_rpc)
     c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
 
-    (tx_hash_hex, tx_rpc) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6))
+    alice_normal = add_0x(hex_uniform(strip_0x(agent_roles['ALICE'])))
+    bob_normal = add_0x(hex_uniform(strip_0x(agent_roles['BOB'])))
+
+    (tx_hash_hex, tx_rpc) = c.create(alice_normal, bob_normal, 100 * (10 ** 6))
     tx_signed_raw_hex = tx_rpc['params'][0]
 
     register_tx(tx_hash_hex, tx_signed_raw_hex, default_chain_spec, None, session=init_database)
@@ -43,12 +51,12 @@ def test_upcoming_with_lock(
     txs = get_upcoming_tx(default_chain_spec, StatusEnum.PENDING)
     assert len(txs.keys()) == 1
 
-    Lock.set(str(default_chain_spec), LockEnum.SEND, address=agent_roles['ALICE'])
+    Lock.set(str(default_chain_spec), LockEnum.SEND, address=alice_normal)
 
-    txs = get_upcoming_tx(default_chain_spec, StatusEnum.PENDING)
+    txs = get_upcoming_tx(default_chain_spec, status=StatusEnum.PENDING)
     assert len(txs.keys()) == 0
 
-    (tx_hash_hex, tx_rpc) = c.create(agent_roles['BOB'], agent_roles['ALICE'], 100 * (10 ** 6))
+    (tx_hash_hex, tx_rpc) = c.create(bob_normal, alice_normal, 100 * (10 ** 6))
     tx_signed_raw_hex = tx_rpc['params'][0]
 
     register_tx(tx_hash_hex, tx_signed_raw_hex, default_chain_spec, None, session=init_database)

apps/cic-eth/tools_requirements.txt

@@ -1,7 +1,7 @@
-crypto-dev-signer~=0.4.14b7
+crypto-dev-signer>=0.4.14b7,<=0.4.14
 chainqueue~=0.0.2b6
 confini>=0.3.6rc4,<0.5.0
-cic-eth-registry~=0.5.6a2
+cic-eth-registry>=0.5.7a1,<0.6.0
 redis==3.5.3
 hexathon~=0.0.1a7
 pycryptodome==3.10.1

apps/cic-notify/cic_notify/runnable/tasker.py

@@ -34,6 +34,8 @@ elif args.v:
 config = confini.Config(args.c, args.env_prefix)
 config.process()
 config.add(args.q, '_CELERY_QUEUE', True)
+config.censor('API_KEY', 'AFRICASTALKING')
+config.censor('API_USERNAME', 'AFRICASTALKING')
 config.censor('PASSWORD', 'DATABASE')
 logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 

apps/cic-notify/cic_notify/version.py

@@ -9,7 +9,7 @@ import semver
 
 logg = logging.getLogger()
 
-version = (0, 4, 0, 'alpha.8')
+version = (0, 4, 0, 'alpha.10')
 
 version_object = semver.VersionInfo(
         major=version[0],

apps/cic-notify/docker/Dockerfile

@@ -12,14 +12,11 @@ RUN --mount=type=cache,mode=0755,target=/root/.cache/pip \
 	  --extra-index-url $GITLAB_PYTHON_REGISTRY --extra-index-url $EXTRA_INDEX_URL \
     -r requirements.txt
 
-COPY . . 
-
+COPY . .
 RUN python setup.py install
 
-# TODO please review..can this go into requirements?
-RUN pip install $pip_extra_index_url_flag .[africastalking,notifylog]
-
 COPY docker/*.sh .
+RUN chmod +x *.sh 
 
 # ini files in config directory defines the configurable parameters for the application
 # they can all be overridden by environment variables
@@ -27,4 +24,5 @@ COPY docker/*.sh .
 COPY .config/ /usr/local/etc/cic-notify/
 COPY cic_notify/db/migrations/ /usr/local/share/cic-notify/alembic/
 
+
 ENTRYPOINT []

apps/cic-notify/docker/Dockerfile_ci

@@ -11,14 +11,12 @@ RUN pip install --index-url https://pypi.org/simple  \
 	  --extra-index-url $GITLAB_PYTHON_REGISTRY --extra-index-url $EXTRA_INDEX_URL \
     -r requirements.txt
 
-COPY . . 
+COPY . .
 
 RUN python setup.py install
 
-# TODO please review..can this go into requirements?
-RUN pip install $pip_extra_index_url_flag .[africastalking,notifylog]
-
 COPY docker/*.sh .
+RUN chmod +x *.sh 
 
 # ini files in config directory defines the configurable parameters for the application
 # they can all be overridden by environment variables

apps/cic-notify/docker/db.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
-
-migrate.py -c /usr/local/etc/cic-notify --migrations-dir /usr/local/share/cic-notify/alembic -vv
+set -e
+python scripts/migrate.py -c /usr/local/etc/cic-notify --migrations-dir /usr/local/share/cic-notify/alembic -vv

apps/cic-notify/docker/start_tasker.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
 
-. ./db.sh
+. /root/db.sh
 
 /usr/local/bin/cic-notify-tasker -vv $@

apps/cic-notify/requirements.txt

@@ -1 +1,7 @@
-cic_base[full_graph]~=0.2.0a3
+confini~=0.4.1a1
+africastalking==1.2.3
+SQLAlchemy==1.3.20
+alembic==1.4.2
+psycopg2==2.8.6
+celery==4.4.7
+redis==3.5.3

apps/cic-notify/scripts/migrate.py

@@ -35,9 +35,10 @@ elif args.v:
 
 config = confini.Config(args.c, args.env_prefix)
 config.process()
+config.censor('API_KEY', 'AFRICASTALKING')
+config.censor('API_USERNAME', 'AFRICASTALKING')
 config.censor('PASSWORD', 'DATABASE')
-#config.censor('PASSWORD', 'SSL')
-logg.debug('config:\n{}'.format(config))
+logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 
 migrations_dir = os.path.join(args.migrations_dir, config.get('DATABASE_ENGINE'))
 if not os.path.isdir(migrations_dir):

apps/cic-notify/setup.cfg

@@ -29,18 +29,11 @@ packages =
 	cic_notify.db
 	cic_notify.db.models
 	cic_notify.ext
+	cic_notify.tasks
 	cic_notify.tasks.sms
 	cic_notify.runnable
 scripts =
-	scripts/migrate.py
-[options.extras_require]
-africastalking = africastalking==1.2.3
-notifylog = psycopg2==2.8.6
-testing =
-	pytest==6.0.1
-	pytest-celery==0.0.0a1
-	pytest-mock==3.3.1
-	pysqlite3==0.4.3
+	./scripts/migrate.py
 
 [options.entry_points]
 console_scripts =

apps/cic-notify/test_requirements.txt

@@ -2,3 +2,4 @@ pytest~=6.0.1
 pytest-celery~=0.0.0a1
 pytest-mock~=3.3.1
 pysqlite3~=0.4.3
+pytest-cov==2.10.1

apps/cic-ussd/.config/app.ini

@@ -1,6 +1,6 @@
 [app]
 ALLOWED_IP=0.0.0.0/0
-LOCALE_FALLBACK=en
+LOCALE_FALLBACK=sw
 LOCALE_PATH=var/lib/locale/
 MAX_BODY_LENGTH=1024
 PASSWORD_PEPPER=QYbzKff6NhiQzY3ygl2BkiKOpER8RE/Upqs/5aZWW+I=

apps/cic-ussd/cic_ussd/balance.py

@@ -15,45 +15,47 @@ from cic_ussd.conversions import from_wei
 logg = logging.getLogger()
 
 
-class BalanceManager:
-
-    def __init__(self, address: str, chain_str: str, token_symbol: str):
-        """
-        :param address: Ethereum address of account whose balance is being queried
-        :type address: str, 0x-hex
-        :param chain_str: The chain name and network id.
-        :type chain_str: str
-        :param token_symbol: ERC20 token symbol of whose balance is being queried
-        :type token_symbol: str
-        """
-        self.address = address
-        self.chain_str = chain_str
-        self.token_symbol = token_symbol
-
-    def get_balances(self, asynchronous: bool = False) -> Union[celery.Task, dict]:
-        """
-        This function queries cic-eth for an account's balances, It provides a means to receive the balance either
-        asynchronously or synchronously depending on the provided value for teh asynchronous parameter. It returns a
-        dictionary containing network, outgoing and incoming balances.
-        :param asynchronous: Boolean value checking whether to return balances asynchronously
-        :type asynchronous: bool
-        :return:
-        :rtype:
-        """
-        if asynchronous:
-            cic_eth_api = Api(
-                chain_str=self.chain_str,
-                callback_queue='cic-ussd',
-                callback_task='cic_ussd.tasks.callback_handler.process_balances_callback',
-                callback_param=''
-            )
-            cic_eth_api.balance(address=self.address, token_symbol=self.token_symbol)
-        else:
-            cic_eth_api = Api(chain_str=self.chain_str)
-            balance_request_task = cic_eth_api.balance(
-                address=self.address,
-                token_symbol=self.token_symbol)
-            return balance_request_task.get()[0]
+def get_balances(
+        address: str,
+        chain_str: str,
+        token_symbol: str,
+        asynchronous: bool = False,
+        callback_param: any = None,
+        callback_task='cic_ussd.tasks.callback_handler.process_balances_callback') -> Union[celery.Task, dict]:
+    """
+    This function queries cic-eth for an account's balances, It provides a means to receive the balance either
+    asynchronously or synchronously depending on the provided value for teh asynchronous parameter. It returns a
+    dictionary containing network, outgoing and incoming balances.
+    :param address: Ethereum address of the recipient
+    :type address: str, 0x-hex
+    :param chain_str: The chain name and network id.
+    :type chain_str: str
+    :param callback_param:
+    :type callback_param:
+    :param callback_task:
+    :type callback_task:
+    :param token_symbol: ERC20 token symbol of the account whose balance is being queried.
+    :type token_symbol: str
+    :param asynchronous: Boolean value checking whether to return balances asynchronously.
+    :type asynchronous: bool
+    :return:
+    :rtype:
+    """
+    logg.debug(f'Retrieving balance for address: {address}')
+    if asynchronous:
+        cic_eth_api = Api(
+            chain_str=chain_str,
+            callback_queue='cic-ussd',
+            callback_task=callback_task,
+            callback_param=callback_param
+        )
+        cic_eth_api.balance(address=address, token_symbol=token_symbol)
+    else:
+        cic_eth_api = Api(chain_str=chain_str)
+        balance_request_task = cic_eth_api.balance(
+            address=address,
+            token_symbol=token_symbol)
+        return balance_request_task.get()[0]
 
 
 def compute_operational_balance(balances: dict) -> float:

apps/cic-ussd/cic_ussd/error.py

@@ -48,3 +48,6 @@ class InitializationError(Exception):
     pass
 
 
+class UnknownUssdRecipient(Exception):
+    """Raised when a recipient of a transaction is not known to the ussd application."""
+

apps/cic-ussd/cic_ussd/metadata/base.py

@@ -127,14 +127,19 @@ class MetadataRequestsHandler(Metadata):
         if not isinstance(result_data, dict):
             raise ValueError(f'Invalid result data object: {result_data}.')
 
-        if result.status_code == 200 and self.cic_type == ':cic.person':
-            # validate person metadata
-            person = Person()
-            person_data = person.deserialize(person_data=result_data)
+        if result.status_code == 200:
+            if self.cic_type == ':cic.person':
+                # validate person metadata
+                person = Person()
+                person_data = person.deserialize(person_data=result_data)
 
-            # format new person data for caching
-            data = json.dumps(person_data.serialize())
+                # format new person data for caching
+                serialized_person_data = person_data.serialize()
+                data = json.dumps(serialized_person_data)
+            else:
+                data = json.dumps(result_data)
 
             # cache metadata
             cache_data(key=self.metadata_pointer, data=data)
             logg.debug(f'caching: {data} with key: {self.metadata_pointer}')
+        return result_data

apps/cic-ussd/cic_ussd/metadata/preferences.py

@@ -1,6 +1,7 @@
 # standard imports
 
 # external imports
+import celery
 
 # local imports
 from .base import MetadataRequestsHandler

apps/cic-ussd/cic_ussd/processor.py

@@ -12,7 +12,7 @@ from tinydb.table import Document
 
 # local imports
 from cic_ussd.account import define_account_tx_metadata, retrieve_account_statement
-from cic_ussd.balance import BalanceManager, compute_operational_balance, get_cached_operational_balance
+from cic_ussd.balance import compute_operational_balance, get_balances, get_cached_operational_balance
 from cic_ussd.chain import Chain
 from cic_ussd.db.models.account import Account
 from cic_ussd.db.models.base import SessionBase
@@ -182,7 +182,6 @@ def process_transaction_pin_authorization(account: Account, display_key: str, se
     token_symbol = retrieve_token_symbol()
     user_input = ussd_session.get('session_data').get('transaction_amount')
     transaction_amount = to_wei(value=int(user_input))
-    logg.debug('Requires integration to determine user tokens.')
     return process_pin_authorization(
         account=account,
         display_key=display_key,
@@ -380,12 +379,9 @@ def process_start_menu(display_key: str, user: Account):
     token_symbol = retrieve_token_symbol()
     chain_str = Chain.spec.__str__()
     blockchain_address = user.blockchain_address
-    balance_manager = BalanceManager(address=blockchain_address,
-                                     chain_str=chain_str,
-                                     token_symbol=token_symbol)
 
     # get balances synchronously for display on start menu
-    balances_data = balance_manager.get_balances()
+    balances_data = get_balances(address=blockchain_address, chain_str=chain_str, token_symbol=token_symbol)
 
     key = create_cached_data_key(
         identifier=bytes.fromhex(blockchain_address[2:]),

apps/cic-ussd/cic_ussd/runnable/daemons/cic_user_tasker.py

@@ -8,13 +8,16 @@ import tempfile
 import celery
 import i18n
 import redis
+from chainlib.chain import ChainSpec
 from confini import Config
 
 # local imports
+from cic_ussd.chain import Chain
 from cic_ussd.db import dsn_from_config
 from cic_ussd.db.models.base import SessionBase
 from cic_ussd.metadata.signer import Signer
 from cic_ussd.metadata.base import Metadata
+from cic_ussd.phone_number import Support
 from cic_ussd.redis import InMemoryStore
 from cic_ussd.session.ussd_session import UssdSession as InMemoryUssdSession
 from cic_ussd.validator import validate_presence
@@ -82,6 +85,15 @@ Signer.key_file_path = key_file_path
 i18n.load_path.append(config.get('APP_LOCALE_PATH'))
 i18n.set('fallback', config.get('APP_LOCALE_FALLBACK'))
 
+chain_spec = ChainSpec(
+    common_name=config.get('CIC_COMMON_NAME'),
+    engine=config.get('CIC_ENGINE'),
+    network_id=config.get('CIC_NETWORK_ID')
+)
+
+Chain.spec = chain_spec
+Support.phone_number = config.get('APP_SUPPORT_PHONE_NUMBER')
+
 # set up celery
 current_app = celery.Celery(__name__)
 

apps/cic-ussd/cic_ussd/tasks/__init__.py

@@ -1,8 +1,4 @@
 # standard import
-import os
-import logging
-import urllib
-import json
 
 # third-party imports
 # this must be included for the package to be recognized as a tasks package
@@ -14,3 +10,5 @@ from .logger import *
 from .ussd_session import *
 from .callback_handler import *
 from .metadata import *
+from .notifications import *
+from .processor import *

apps/cic-ussd/cic_ussd/tasks/callback_handler.py

@@ -7,14 +7,14 @@ from datetime import datetime, timedelta
 import celery
 
 # local imports
+from cic_ussd.balance import compute_operational_balance, get_balances
+from cic_ussd.chain import Chain
 from cic_ussd.conversions import from_wei
 from cic_ussd.db.models.base import SessionBase
 from cic_ussd.db.models.account import Account
-from cic_ussd.account import define_account_tx_metadata
 from cic_ussd.error import ActionDataNotFoundError
-from cic_ussd.redis import InMemoryStore, cache_data, create_cached_data_key
+from cic_ussd.redis import InMemoryStore, cache_data, create_cached_data_key, get_cached_data
 from cic_ussd.tasks.base import CriticalSQLAlchemyTask
-from cic_ussd.transactions import IncomingTransactionProcessor
 
 logg = logging.getLogger(__file__)
 celery_app = celery.current_app
@@ -58,9 +58,9 @@ def process_account_creation_callback(self, result: str, url: str, status_code:
 
             # add phone number metadata lookup
             s_phone_pointer = celery.signature(
-                    'cic_ussd.tasks.metadata.add_phone_pointer',
-                    [result, phone_number]
-                    )
+                'cic_ussd.tasks.metadata.add_phone_pointer',
+                [result, phone_number]
+            )
             s_phone_pointer.apply_async(queue=queue)
 
             # add custom metadata tags
@@ -87,59 +87,106 @@ def process_account_creation_callback(self, result: str, url: str, status_code:
     session.close()
 
 
-@celery_app.task
-def process_incoming_transfer_callback(result: dict, param: str, status_code: int):
-    session = SessionBase.create_session()
+@celery_app.task(bind=True)
+def process_transaction_callback(self, result: dict, param: str, status_code: int):
     if status_code == 0:
+        chain_str = Chain.spec.__str__()
 
-        # collect result data
+        # collect transaction metadata
+        destination_token_symbol = result.get('destination_token_symbol')
+        destination_token_value = result.get('destination_token_value')
         recipient_blockchain_address = result.get('recipient')
         sender_blockchain_address = result.get('sender')
-        token_symbol = result.get('destination_token_symbol')
-        value = result.get('destination_token_value')
+        source_token_symbol = result.get('source_token_symbol')
+        source_token_value = result.get('source_token_value')
 
-        # try to find users in system
-        recipient_user = session.query(Account).filter_by(blockchain_address=recipient_blockchain_address).first()
-        sender_user = session.query(Account).filter_by(blockchain_address=sender_blockchain_address).first()
+        # build stakeholder callback params
+        recipient_metadata = {
+            "token_symbol": destination_token_symbol,
+            "token_value": destination_token_value,
+            "blockchain_address": recipient_blockchain_address,
+            "tag": "recipient",
+            "tx_param": param
+        }
 
-        # check whether recipient is in the system
-        if not recipient_user:
-            session.close()
-            raise ValueError(
-                f'Tx for recipient: {recipient_blockchain_address} was received but has no matching user in the system.'
-            )
+        # retrieve account balances
+        get_balances(
+            address=recipient_blockchain_address,
+            callback_param=recipient_metadata,
+            chain_str=chain_str,
+            callback_task='cic_ussd.tasks.callback_handler.process_transaction_balances_callback',
+            token_symbol=destination_token_symbol,
+            asynchronous=True)
 
-        # process incoming transactions
-        incoming_tx_processor = IncomingTransactionProcessor(phone_number=recipient_user.phone_number,
-                                                             preferred_language=recipient_user.preferred_language,
-                                                             token_symbol=token_symbol,
-                                                             value=value)
+        # only retrieve sender if transaction is a transfer
+        if param == 'transfer':
+            sender_metadata = {
+                "blockchain_address": sender_blockchain_address,
+                "token_symbol": source_token_symbol,
+                "token_value": source_token_value,
+                "tag": "sender",
+                "tx_param": param
+            }
 
-        if param == 'tokengift':
-            incoming_tx_processor.process_token_gift_incoming_transactions()
-        elif param == 'transfer':
-            if sender_user:
-                sender_information = define_account_tx_metadata(user=sender_user)
-                incoming_tx_processor.process_transfer_incoming_transaction(
-                    sender_information=sender_information,
-                    recipient_blockchain_address=recipient_blockchain_address
-                )
-            else:
-                logg.warning(
-                    f'Tx with sender: {sender_blockchain_address} was received but has no matching user in the system.'
-                )
-                incoming_tx_processor.process_transfer_incoming_transaction(
-                    sender_information='GRASSROOTS ECONOMICS',
-                    recipient_blockchain_address=recipient_blockchain_address
-                )
-        else:
-            session.close()
-            raise ValueError(f'Unexpected transaction param: {param}.')
+            get_balances(
+                address=sender_blockchain_address,
+                callback_param=sender_metadata,
+                chain_str=chain_str,
+                callback_task='cic_ussd.tasks.callback_handler.process_transaction_balances_callback',
+                token_symbol=source_token_symbol,
+                asynchronous=True)
+    else:
+        raise ValueError(f'Unexpected status code: {status_code}.')
+
+
+@celery_app.task(bind=True)
+def process_transaction_balances_callback(self, result: list, param: dict, status_code: int):
+    queue = self.request.delivery_info.get('routing_key')
+    if status_code == 0:
+        # retrieve balance data
+        balances_data = result[0]
+        operational_balance = compute_operational_balance(balances=balances_data)
+
+        # retrieve account's address
+        blockchain_address = param.get('blockchain_address')
+
+        # append balance to transaction metadata
+        transaction_metadata = param
+        transaction_metadata['operational_balance'] = operational_balance
+
+        # retrieve account's preferences
+        s_preferences_metadata = celery.signature(
+            'cic_ussd.tasks.metadata.query_preferences_metadata',
+            [blockchain_address],
+            queue=queue
+        )
+
+        # parse metadata and run validations
+        s_process_account_metadata = celery.signature(
+            'cic_ussd.tasks.processor.process_tx_metadata_for_notification',
+            [transaction_metadata],
+            queue=queue
+        )
+
+        # issue notification of transaction
+        s_notify_account = celery.signature(
+            'cic_ussd.tasks.notifications.notify_account_of_transaction',
+            queue=queue
+        )
+
+        if param.get('tx_param') == 'transfer':
+            celery.chain(s_preferences_metadata, s_process_account_metadata, s_notify_account).apply_async()
+
+        if param.get('tx_param') == 'tokengift':
+            s_process_account_metadata = celery.signature(
+                'cic_ussd.tasks.processor.process_tx_metadata_for_notification',
+                [{}, transaction_metadata],
+                queue=queue
+            )
+            celery.chain(s_process_account_metadata, s_notify_account).apply_async()
     else:
-        session.close()
         raise ValueError(f'Unexpected status code: {status_code}.')
 
-    session.close()
 
 @celery_app.task
 def process_balances_callback(result: list, param: str, status_code: int):
@@ -151,6 +198,7 @@ def process_balances_callback(result: list, param: str, status_code: int):
             salt=':cic.balances_data'
         )
         cache_data(key=key, data=json.dumps(balances_data))
+        logg.debug(f'caching: {balances_data} with key: {key}')
     else:
         raise ValueError(f'Unexpected status code: {status_code}.')
 

apps/cic-ussd/cic_ussd/tasks/metadata.py

@@ -26,6 +26,7 @@ def query_person_metadata(blockchain_address: str):
     :rtype:
     """
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
+    logg.debug(f'Retrieving person metadata for address: {blockchain_address}.')
     person_metadata_client = PersonMetadata(identifier=identifier)
     person_metadata_client.query()
 
@@ -72,3 +73,15 @@ def add_preferences_metadata(blockchain_address: str, data: dict):
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
     custom_metadata_client = PreferencesMetadata(identifier=identifier)
     custom_metadata_client.create(data=data)
+
+
+@celery_app.task()
+def query_preferences_metadata(blockchain_address: str):
+    """This method retrieves preferences metadata based on an account's blockchain address.
+    :param blockchain_address: Blockchain address of an account.
+    :type blockchain_address: str | Ox-hex
+    """
+    identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
+    logg.debug(f'Retrieving preferences metadata for address: {blockchain_address}.')
+    person_metadata_client = PreferencesMetadata(identifier=identifier)
+    return person_metadata_client.query()

apps/cic-ussd/cic_ussd/tasks/notifications.py

@@ -0,0 +1,70 @@
+# standard imports
+import datetime
+import logging
+
+# third-party imports
+import celery
+
+# local imports
+from cic_ussd.notifications import Notifier
+from cic_ussd.phone_number import Support
+
+celery_app = celery.current_app
+logg = logging.getLogger(__file__)
+notifier = Notifier()
+
+
+@celery_app.task
+def notify_account_of_transaction(notification_data: dict):
+    """
+    :param notification_data:
+    :type notification_data:
+    :return:
+    :rtype:
+    """
+
+    account_tx_role = notification_data.get('account_tx_role')
+    amount = notification_data.get('amount')
+    balance = notification_data.get('balance')
+    phone_number = notification_data.get('phone_number')
+    preferred_language = notification_data.get('preferred_language')
+    token_symbol = notification_data.get('token_symbol')
+    transaction_account_metadata = notification_data.get('transaction_account_metadata')
+    transaction_type = notification_data.get('transaction_type')
+
+    timestamp = datetime.datetime.now().strftime('%d-%m-%y, %H:%M %p')
+
+    if transaction_type == 'tokengift':
+        support_phone = Support.phone_number
+        notifier.send_sms_notification(
+            key='sms.account_successfully_created',
+            phone_number=phone_number,
+            preferred_language=preferred_language,
+            balance=balance,
+            support_phone=support_phone,
+            token_symbol=token_symbol
+        )
+
+    if transaction_type == 'transfer':
+        if account_tx_role == 'recipient':
+            notifier.send_sms_notification(
+                key='sms.received_tokens',
+                phone_number=phone_number,
+                preferred_language=preferred_language,
+                amount=amount,
+                token_symbol=token_symbol,
+                tx_sender_information=transaction_account_metadata,
+                timestamp=timestamp,
+                balance=balance
+            )
+        else:
+            notifier.send_sms_notification(
+                key='sms.sent_tokens',
+                phone_number=phone_number,
+                preferred_language=preferred_language,
+                amount=amount,
+                token_symbol=token_symbol,
+                tx_recipient_information=transaction_account_metadata,
+                timestamp=timestamp,
+                balance=balance
+            )

apps/cic-ussd/cic_ussd/tasks/processor.py

@@ -0,0 +1,88 @@
+# standard imports
+import logging
+
+# third-party imports
+import celery
+from i18n import config
+
+# local imports
+from cic_ussd.account import define_account_tx_metadata
+from cic_ussd.db.models.account import Account
+from cic_ussd.db.models.base import SessionBase
+from cic_ussd.error import UnknownUssdRecipient
+from cic_ussd.transactions import from_wei
+
+
+celery_app = celery.current_app
+logg = logging.getLogger(__file__)
+
+
+@celery_app.task
+def process_tx_metadata_for_notification(result: celery.Task, transaction_metadata: dict):
+    """
+    :param result:
+    :type result:
+    :param transaction_metadata:
+    :type transaction_metadata:
+    :return:
+    :rtype:
+    """
+    notification_data = {}
+
+    # get preferred language
+    preferred_language = result.get('preferred_language')
+    if not preferred_language:
+        preferred_language = config.get('fallback')
+    notification_data['preferred_language'] = preferred_language
+
+    # validate account information against present ussd storage data.
+    session = SessionBase.create_session()
+    blockchain_address = transaction_metadata.get('blockchain_address')
+    tag = transaction_metadata.get('tag')
+    account = session.query(Account).filter_by(blockchain_address=blockchain_address).first()
+    if not account and tag == 'recipient':
+        session.close()
+        raise UnknownUssdRecipient(
+            f'Tx for recipient: {blockchain_address} was received but has no matching user in the system.'
+        )
+
+    # get phone number associated with account
+    phone_number = account.phone_number
+    notification_data['phone_number'] = phone_number
+
+    # get account's role in transaction i.e sender / recipient
+    tx_param = transaction_metadata.get('tx_param')
+    notification_data['transaction_type'] = tx_param
+
+    # get token amount and symbol
+    if tag == 'recipient':
+        account_tx_role = tag
+        amount = transaction_metadata.get('token_value')
+        amount = from_wei(value=amount)
+        token_symbol = transaction_metadata.get('token_symbol')
+    else:
+        account_tx_role = tag
+        amount = transaction_metadata.get('token_value')
+        amount = from_wei(value=amount)
+        token_symbol = transaction_metadata.get('token_symbol')
+    notification_data['account_tx_role'] = account_tx_role
+    notification_data['amount'] = amount
+    notification_data['token_symbol'] = token_symbol
+
+    # get account's standard ussd identification pattern
+    if tx_param == 'transfer':
+        tx_account_metadata = define_account_tx_metadata(user=account)
+        notification_data['transaction_account_metadata'] = tx_account_metadata
+
+        if tag == 'recipient':
+            notification_data['notification_key'] = 'sms.received_tokens'
+        else:
+            notification_data['notification_key'] = 'sms.sent_tokens'
+
+    if tx_param == 'tokengift':
+        notification_data['notification_key'] = 'sms.account_successfully_created'
+
+    # get account's balance
+    notification_data['balance'] = transaction_metadata.get('operational_balance')
+
+    return notification_data

apps/cic-ussd/cic_ussd/transactions.py

@@ -7,9 +7,9 @@ from datetime import datetime
 from cic_eth.api import Api
 
 # local imports
-from cic_ussd.balance import get_cached_operational_balance
+from cic_ussd.balance import get_balances, get_cached_operational_balance
 from cic_ussd.notifications import Notifier
-
+from cic_ussd.phone_number import Support
 
 logg = logging.getLogger()
 notifier = Notifier()
@@ -50,61 +50,6 @@ def to_wei(value: int) -> int:
     return int(value * 1e+6)
 
 
-class IncomingTransactionProcessor:
-
-    def __init__(self, phone_number: str, preferred_language: str, token_symbol: str, value: int):
-        """
-        :param phone_number: The recipient's phone number.
-        :type phone_number: str
-        :param preferred_language: The user's preferred language.
-        :type preferred_language: str
-        :param token_symbol: The symbol for the token the recipient receives.
-        :type token_symbol: str
-        :param value: The amount of tokens received in the transactions.
-        :type value: int
-        """
-        self.phone_number = phone_number
-        self.preferred_language = preferred_language
-        self.token_symbol = token_symbol
-        self.value = value
-
-    def process_token_gift_incoming_transactions(self):
-        """This function processes incoming transactions with a "tokengift" param, it collects all appropriate data to
-        send out notifications to users when their accounts are successfully created.
-
-        """
-        balance = from_wei(value=self.value)
-        key = 'sms.account_successfully_created'
-        notifier.send_sms_notification(key=key,
-                                       phone_number=self.phone_number,
-                                       preferred_language=self.preferred_language,
-                                       balance=balance,
-                                       token_symbol=self.token_symbol)
-
-    def process_transfer_incoming_transaction(self, sender_information: str, recipient_blockchain_address: str):
-        """This function processes incoming transactions with the "transfer" param and issues notifications to users
-        about reception of funds into their accounts.
-        :param sender_information: A string with a user's full name and phone number.
-        :type sender_information: str
-        :param recipient_blockchain_address:
-        type recipient_blockchain_address: str
-        """
-        key = 'sms.received_tokens'
-        amount = from_wei(value=self.value)
-        timestamp = datetime.now().strftime('%d-%m-%y, %H:%M %p')
-
-        operational_balance = get_cached_operational_balance(blockchain_address=recipient_blockchain_address)
-
-        notifier.send_sms_notification(key=key,
-                                       phone_number=self.phone_number,
-                                       preferred_language=self.preferred_language,
-                                       amount=amount,
-                                       token_symbol=self.token_symbol,
-                                       tx_sender_information=sender_information,
-                                       timestamp=timestamp,
-                                       balance=operational_balance)
-
-
 class OutgoingTransactionProcessor:
 
     def __init__(self, chain_str: str, from_address: str, to_address: str):
@@ -116,6 +61,7 @@ class OutgoingTransactionProcessor:
         :param to_address: Ethereum address of the recipient
         :type to_address: str, 0x-hex
         """
+        self.chain_str = chain_str
         self.cic_eth_api = Api(chain_str=chain_str)
         self.from_address = from_address
         self.to_address = to_address

apps/cic-ussd/requirements.txt

@@ -1,4 +1,17 @@
-cic_base[full_graph]==0.2.0a3
-cic-eth~=0.12.0a1
-cic-notify~=0.4.0a8
-cic-types~=0.1.0a11
+cic-eth~=0.12.2a3
+cic-notify~=0.4.0a10
+cic-types~=0.1.0a14
+confini~=0.4.1a1
+semver==2.13.0
+alembic==1.4.2
+SQLAlchemy==1.3.20
+psycopg2==2.8.6
+tinydb==4.2.0
+phonenumbers==8.12.12
+redis==3.5.3
+celery==4.4.7
+python-i18n[YAML]==0.3.9
+pyxdg==0.27
+bcrypt==3.2.0
+uWSGI==2.0.19.1
+transitions==0.8.4

apps/cic-ussd/var/lib/locale/sms.en.yml

@@ -3,5 +3,7 @@ en:
     You have been registered on Sarafu Network! To use dial *384*96# on Safaricom and *483*96# on other networks. For help %{support_phone}.
   received_tokens: |-
     Successfully received %{amount} %{token_symbol} from %{tx_sender_information} %{timestamp}. New balance is %{balance} %{token_symbol}.
+  sent_tokens: |-
+    Successfully sent %{amount} %{token_symbol} to %{tx_recipient_information} %{timestamp}. New balance is %{balance} %{token_symbol}.
   terms: |-
     By using the service, you agree to the terms and conditions at http://grassecon.org/tos

apps/cic-ussd/var/lib/locale/sms.sw.yml

@@ -2,6 +2,8 @@ sw:
   account_successfully_created: |-
     Umesajiliwa kwa huduma ya Sarafu! Kutumia bonyeza *384*96# Safaricom ama *483*46# kwa utandao tofauti. Kwa Usaidizi %{support_phone}.
   received_tokens: |-
-    Umepokea %{amount} %{token_symbol} kutoka kwa %{tx_sender_information} %{timestamp}. Salio la %{token_symbol} ni %{balance}.
+    Umepokea %{amount} %{token_symbol} kutoka kwa %{tx_sender_information} %{timestamp}. Salio lako ni %{balance} %{token_symbol}.
+  sent_tokens: |-
+    Umetuma %{amount} %{token_symbol} kwa %{tx_recipient_information} %{timestamp}. Salio lako ni %{balance} %{token_symbol}.
   terms: |-
     Kwa kutumia hii huduma, umekubali sheria na masharti yafuatayo http://grassecon.org/tos

apps/contract-migration/.dockerignore

@@ -1,4 +1,6 @@
 .git
 .cache
 .dot
-**/doc
+**/doc
+**/.venv
+**/venv

apps/contract-migration/requirements.txt

@@ -1,11 +1,10 @@
-sarafu-faucet~=0.0.4a4
-cic-eth[tools]==0.12.1a2
-eth-erc20~=0.0.10a3
-erc20-demurrage-token==0.0.2a4
-eth-address-index~=0.1.2a2
-eth-accounts-index~=0.0.12a2
-cic-eth-registry~=0.5.6a2
-erc20-faucet~=0.2.2a2
-erc20-transfer-authorization~=0.3.2a2
-sarafu-faucet~=0.0.4a3
-chainlib-eth~=0.0.5a4
+cic-eth[tools]==0.12.2a3
+eth-erc20>=0.0.11a1,<0.1.0
+erc20-demurrage-token>=0.0.2a5,<0.1.0
+eth-address-index>=0.1.3a1,<0.2.0
+eth-accounts-index>=0.0.13a1,<0.1.0
+cic-eth-registry>=0.5.7a1,<=0.6.0
+erc20-faucet>=0.2.3a1,<0.3.0
+erc20-transfer-authorization>=0.3.3a1,<0.4.0
+sarafu-faucet>=0.0.4a4,<0.1.0
+chainlib-eth>=0.0.6a1,<0.1.0

apps/contract-migration/reset.sh

@@ -158,6 +158,7 @@ export CIC_DECLARATOR_ADDRESS=$CIC_DECLARATOR_ADDRESS
 EOF
 
 cat ./envlist | bash from_env.sh > $CIC_DATA_DIR/.env_all
+cat ./envlist
 # popd
 
 set +a

apps/contract-migration/seed_cic_eth.sh

@@ -72,6 +72,8 @@ cic-eth-tag -i $CIC_CHAIN_SPEC ACCOUNT_REGISTRY_WRITER $DEV_ETH_ACCOUNT_ACCOUNT_
 eth-accounts-index-writer -y $keystore_file -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -a $DEV_ACCOUNT_INDEX_ADDRESS -ww $debug $DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER
 
 # Transfer gas to custodial gas provider adddress
+_CONFINI_DIR=$CONFINI_DIR
+unset CONFINI_DIR
 >&2 echo gift gas to gas gifter
 >&2 eth-gas --send -y $keystore_file -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -w $debug -a $DEV_ETH_ACCOUNT_GAS_GIFTER $gas_amount
 
@@ -100,6 +102,7 @@ export DEV_ETH_SARAFU_TOKEN_ADDRESS=$DEV_ETH_RESERVE_ADDRESS
 
 #echo -n 0 > $init_level_file
 
+CONFINI_DIR=$_CONFINI_DIR
 # Remove the SEND (8), QUEUE (16) and INIT (2) locks (or'ed), set by default at migration
 cic-eth-ctl -i :: unlock INIT
 cic-eth-ctl -i :: unlock SEND

apps/data-seeding/.dockerignore

@@ -2,7 +2,7 @@
 .cache
 .dot
 **/doc
-**/node_modules
+node_modules/
 **/venv
 **/.venv
 

apps/data-seeding/docker/.env

@@ -0,0 +1,12 @@
+ETH_PROVIDER=http://eth:8545
+CELERY_BROKER_URL=redis://redis:6379
+CELERY_RESULT_URL=redis://redis:6379
+CIC_REGISTRY_ADDRESS=0xea6225212005e86a4490018ded4bf37f3e772161
+TOKEN_SYMBOL=GFT
+REDIS_HOST=redis
+REDIS_PORT=6379
+USER_USSD_HOST=cic-user-ussd-server
+USER_USSD_PORT=9000
+KEYSTORE_FILE_PATH=/root/keystore/UTC--2021-01-08T17-18-44.521011372Z--eb3907ecad74a0013c259d5874ae7f22dcbcc95c
+OUT_DIR=out
+NUMBER_OF_USERS=100

apps/data-seeding/docker/Dockerfile

@@ -8,11 +8,9 @@ RUN mkdir -vp /usr/local/etc/cic
 COPY package.json \
      package-lock.json \
      .
-
-RUN --mount=type=cache,mode=0755,target=/root/node_modules npm install
+RUN npm ci --production 
 
 COPY requirements.txt . 
-
 ARG EXTRA_INDEX_URL="https://pip.grassrootseconomics.net:8433"
 ARG GITLAB_PYTHON_REGISTRY="https://gitlab.com/api/v4/projects/27624814/packages/pypi/simple"
 RUN --mount=type=cache,mode=0755,target=/root/.cache/pip pip install \
@@ -21,4 +19,5 @@ RUN --mount=type=cache,mode=0755,target=/root/.cache/pip pip install \
 
 COPY . . 
 
+
 ENTRYPOINT [ ]

apps/data-seeding/scripts/run_ussd_user_imports.sh

@@ -0,0 +1,67 @@
+#! /bin/sh
+
+set -u
+set -e
+
+while getopts ":n:o:g:" opt; do
+  case $opt in
+    n) NUMBER_OF_USERS="$OPTARG"
+    ;;
+    o) OUT_DIR="$OPTARG"
+    ;;
+    \?) echo "Invalid option -$OPTARG" >&2
+    ;;
+  esac
+done
+
+# using timeout because the timeout flag for celery inspect does not work
+timeout 5 celery inspect ping -b $CELERY_BROKER_URL
+if [[ $? -eq 124 ]] 
+then
+  >&2 echo "Celery workers not available. Is the CELERY_BROKER_URL ($CELERY_BROKER_URL) correct?"
+  exit 1
+fi
+
+if [[ -d $OUT_DIR ]]
+then
+  echo "found existing OUT_DIR cleaning up..."
+  rm -rf $OUT_DIR
+  mkdir -p $OUT_DIR
+else
+  echo "OUT_DIR does not exist creating it."
+  mkdir -p $OUT_DIR
+fi
+
+
+echo "creating accounts"
+
+python create_import_users.py --dir $OUT_DIR $NUMBER_OF_USERS
+
+echo "purging existing ussd tasks..."
+
+celery -A cic_ussd.import_task purge -Q cic-import-ussd --broker $CELERY_BROKER_URL -f
+
+echo "running import_balance in the background..."
+
+python cic_ussd/import_balance.py -v -c config -p $ETH_PROVIDER \
+  -r $CIC_REGISTRY_ADDRESS --token-symbol $TOKEN_SYMBOL -y $KEYSTORE_FILE_PATH $OUT_DIR > import_task_log.log & 
+
+import_pid=$!
+echo "import_balance pid: $import_pid" 
+
+echo "importing accounts"
+
+python cic_ussd/import_users.py -vv -c config --ussd-host $USER_USSD_HOST --ussd-port $USER_USSD_PORT --ussd-no-ssl out
+
+echo "importing user meta data"
+node cic_meta/import_meta.js $OUT_DIR $NUMBER_OF_USERS 
+
+echo "import meta prefereneces"
+node cic_meta/import_meta_preferences.js $OUT_DIR $NUMBER_OF_USERS 
+
+echo "Running validation!"
+python verify.py -v -c config -r $CIC_REGISTRY_ADDRESS -p $ETH_PROVIDER \
+  --token-symbol $TOKEN_SYMBOL $OUT_DIR 
+
+kill $import_pid
+exit 0

docker-compose.yml

@@ -71,6 +71,8 @@ services:
       - bee-data:/tmp/cic/bee
 
   contract-migration:
+    profiles:
+      - migrations 
     build:
       context: apps/contract-migration
       dockerfile: docker/Dockerfile
@@ -79,6 +81,7 @@ services:
         pip_extra_args: $PIP_EXTRA_ARGS
     # image: registry.gitlab.com/grassrootseconomics/cic-internal-integration/contract-migration:latest
     environment:
+      CIC_REGISTRY_ADDRESS: $CIC_REGISTRY_ADDRESS
       # ETH_PROVIDER should be broken out into host/port but cic-eth expects this
       ETH_PROVIDER: http://eth:8545
       # And these two are for wait-for-it (could parse this)
@@ -123,6 +126,8 @@ services:
       - contract-config:/tmp/cic/config
 
   cic-cache-tracker:
+    profiles:
+      - cache 
     build:
       context: apps/cic-cache
       dockerfile: docker/Dockerfile
@@ -159,6 +164,8 @@ services:
       - contract-config:/tmp/cic/config/:ro
 
   cic-cache-tasker:
+    profiles:
+      - cache 
     build:
       context: apps/cic-cache
       dockerfile: docker/Dockerfile
@@ -196,6 +203,8 @@ services:
       - contract-config:/tmp/cic/config/:ro
 
   cic-cache-server:
+    profiles:
+      - cache 
     build:
       context: apps/cic-cache
       dockerfile: docker/Dockerfile
@@ -227,7 +236,6 @@ services:
 
 
   cic-eth-tasker:
-    # image: grassrootseconomics:cic-eth-service
     build: 
         context: apps/cic-eth
         dockerfile: docker/Dockerfile
@@ -338,7 +346,6 @@ services:
       TASKS_TRANSFER_CALLBACKS: $TASKS_TRANSFER_CALLBACKS
       DATABASE_DEBUG: ${DATABASE_DEBUG:-false}
         #DATABASE_DEBUG: 1
-
     depends_on:
       - eth
       - postgres
@@ -429,6 +436,8 @@ services:
 
 
   cic-meta-server:
+    profiles:
+      - custodial-meta
     hostname: meta
     build:
       context: apps/cic-meta
@@ -462,6 +471,8 @@ services:
     # command: "/root/start_server.sh -vv"
 
   cic-user-ussd-server:
+    profiles:
+      - custodial-ussd
     build:
       context: apps/cic-ussd
       dockerfile: docker/Dockerfile
@@ -491,6 +502,8 @@ services:
     command: "/root/start_cic_user_ussd_server.sh -vv"
 
   cic-user-server:
+    profiles:
+      - custodial-ussd
     build:
       context: apps/cic-ussd
       dockerfile: docker/Dockerfile
@@ -513,6 +526,8 @@ services:
     command: "/root/start_cic_user_server.sh -vv"
 
   cic-user-tasker:
+    profiles:
+      - custodial-ussd
     build:
       context: apps/cic-ussd/
       dockerfile: docker/Dockerfile