Merge branch 'master' into lash/check-import-ussd

Refactor meta ci-cd pipeline.

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

.gitignore

@@ -8,3 +8,6 @@ gmon.out
 *.egg-info
 dist/
 build/
+**/*sqlite
+**/.nyc_output
+**/coverage

.gitlab-ci.yml

@@ -6,6 +6,7 @@ include:
   - local: 'apps/cic-notify/.gitlab-ci.yml'
   - local: 'apps/cic-meta/.gitlab-ci.yml'
   - local: 'apps/cic-cache/.gitlab-ci.yml'
+  - local: 'apps/data-seeding/.gitlab-ci.yml'
 
 stages:
   - build

apps/cic-cache/cic_cache/cache.py

@@ -125,6 +125,7 @@ class DataCache(Cache):
                 'to_value': int(r['to_value']),
                 'source_token': r['source_token'],
                 'destination_token': r['destination_token'],
+                'success': r['success'],
                 'tx_type': tx_type,
             }
 

apps/cic-cache/cic_cache/db/list.py

@@ -42,7 +42,7 @@ def list_transactions_mined_with_data(
     :result: Result set
     :rtype: SQLAlchemy.ResultProxy
     """
-    s = "SELECT tx_hash, block_number, date_block, sender, recipient, from_value, to_value, source_token, destination_token, domain, value FROM tx LEFT JOIN tag_tx_link ON tx.id = tag_tx_link.tx_id LEFT JOIN tag ON tag_tx_link.tag_id = tag.id WHERE block_number >= {} AND block_number <= {} ORDER BY block_number ASC, tx_index ASC".format(offset, end)
+    s = "SELECT tx_hash, block_number, date_block, sender, recipient, from_value, to_value, source_token, destination_token, success, domain, value FROM tx LEFT JOIN tag_tx_link ON tx.id = tag_tx_link.tx_id LEFT JOIN tag ON tag_tx_link.tag_id = tag.id WHERE block_number >= {} AND block_number <= {} ORDER BY block_number ASC, tx_index ASC".format(offset, end)
 
     r = session.execute(s)
     return r

apps/cic-cache/cic_cache/runnable/daemons/query.py

@@ -89,7 +89,7 @@ def process_transactions_all_data(session, env):
 
     offset = r[1]
     end = r[2]
-    if r[2] < r[1]:
+    if int(r[2]) < int(r[1]):
         raise ValueError('cart before the horse, dude')
 
     c = DataCache(session)

apps/cic-cache/config/test/syncer.ini

@@ -0,0 +1,2 @@
+[syncer]
+loop_interval = 1

apps/cic-eth/.coveragerc

@@ -5,3 +5,5 @@ omit =
 	cic_eth/db/migrations/*
 	cic_eth/sync/head.py
 	cic_eth/sync/mempool.py
+	cic_eth/queue/state.py
+	*redis*.py

apps/cic-eth/.gitlab-ci.yml

@@ -5,18 +5,29 @@
 
 .cic_eth_changes_target:
     rules:
-        - changes:
+    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
-            - $CONTEXT/$APP_NAME/*
+      #changes:
+          #- $CONTEXT/$APP_NAME/**/*
+      when: always
 
 build-mr-cic-eth:
     extends:
-        - .cic_eth_changes_target
-        - .py_build_merge_request
         - .cic_eth_variables
+        - .cic_eth_changes_target
+        - .py_build_target_test
+
+test-mr-cic-eth:
+    extends:
+        - .cic_eth_variables
+        - .cic_eth_changes_target
+    stage: test
+    image: $CI_REGISTRY_IMAGE/$APP_NAME-test:latest
+    script:
+        - cd apps/$APP_NAME/
+        - pytest -x --cov=cic_eth --cov-fail-under=90 --cov-report term-missing tests
+    needs: ["build-mr-cic-eth"]
 
 build-push-cic-eth:
     extends:
         - .py_build_push
         - .cic_eth_variables
-        
-

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

@@ -4,11 +4,18 @@ import logging
 # external imports
 import celery
 from chainlib.chain import ChainSpec
-from chainlib.eth.tx import unpack
+from chainlib.connection import RPCConnection
-from chainqueue.query import get_tx
+from chainlib.eth.tx import (
-from chainqueue.state import set_cancel
+        unpack,
+        TxFactory,
+        )
+from chainlib.eth.gas import OverrideGasOracle
+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 potaahto.symbols import snake_and_camel
 
 # local imports
 from cic_eth.db.models.base import SessionBase
@@ -21,13 +28,14 @@ from cic_eth.admin.ctrl import (
         )
 from cic_eth.queue.tx import queue_create
 from cic_eth.eth.gas import create_check_gas_task
+from cic_eth.task import BaseTask
 
 celery_app = celery.current_app
 logg = logging.getLogger()
 
 
-@celery_app.task(bind=True)
+@celery_app.task(bind=True, base=BaseTask)
-def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
+def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
     """Shift all transactions with nonces higher than the offset by the provided position delta.
 
     Transactions who are replaced by transactions that move nonces will be marked as OVERRIDDEN.
@@ -38,25 +46,29 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     :type tx_hash_orig_hex: str, 0x-hex
     :param delta: Amount
     """
+    chain_spec = ChainSpec.from_dict(chainspec_dict)
+    rpc = RPCConnection.connect(chain_spec, 'default')
+    rpc_signer = RPCConnection.connect(chain_spec, 'signer')
     queue = None
     try:
         queue = self.request.delivery_info.get('routing_key')
     except AttributeError:
         pass
 
-    chain_spec = ChainSpec.from_chain_str(chain_str)
+    session = BaseTask.session_func()
-    tx_brief = get_tx(tx_hash_orig_hex)
+    tx_brief = get_tx(chain_spec, tx_hash_orig_hex, session=session)
-    tx_raw = bytes.fromhex(strip_0x(tx_brief['signed_tx'][2:]))
+    tx_raw = bytes.fromhex(strip_0x(tx_brief['signed_tx']))
     tx = unpack(tx_raw, chain_spec)
     nonce = tx_brief['nonce']
     address = tx['from']
 
-    logg.debug('shifting nonce {} position(s) for address {}, offset {}'.format(delta, address, nonce))
+    logg.debug('shifting nonce {} position(s) for address {}, offset {}, hash {}'.format(delta, address, nonce, tx['hash']))
 
-    lock_queue(None, chain_str, address)
+    lock_queue(None, chain_spec.asdict(), address=address)
-    lock_send(None, chain_str, address)
+    lock_send(None, chain_spec.asdict(), address=address)
+
+    set_cancel(chain_spec, strip_0x(tx['hash']), manual=True, session=session)
 
-    session = SessionBase.create_session()
     q = session.query(Otx)
     q = q.join(TxCache)
     q = q.filter(TxCache.sender==address)
@@ -69,49 +81,57 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     for otx in otxs:
         tx_raw = bytes.fromhex(strip_0x(otx.signed_tx))
         tx_new = unpack(tx_raw, chain_spec)
+        tx_new = snake_and_camel(tx_new)
 
         tx_previous_hash_hex = tx_new['hash']
         tx_previous_nonce = tx_new['nonce']
 
-        del(tx_new['hash'])
+        tx_new['gas_price'] += 1
-        del(tx_new['hash_unsigned'])
+        tx_new['gasPrice'] = tx_new['gas_price']
         tx_new['nonce'] -= delta
 
-        (tx_hash_hex, tx_signed_raw_hex) = sign_tx(tx_new, chain_str)
+        logg.debug('tx_new {}'.format(tx_new))
+
+        del(tx_new['hash'])
+        del(tx_new['hash_unsigned'])
+        del(tx_new['hashUnsigned'])
+
+        gas_oracle = OverrideGasOracle(limit=tx_new['gas'], price=tx_new['gas_price'] + 1) # TODO: it should be possible to merely set this price here and if missing in the existing struct then fill it in (chainlib.eth.tx)
+        c = TxFactory(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.build_raw(tx_new)
         logg.debug('tx {} -> {} nonce {} -> {}'.format(tx_previous_hash_hex, tx_hash_hex, tx_previous_nonce, tx_new['nonce']))
 
         otx = Otx(
-            nonce=tx_new['nonce'],
+            tx_new['nonce'],
-            address=tx_new['from'],
+            tx_hash_hex,
-            tx_hash=tx_hash_hex,
+            tx_signed_raw_hex,
-            signed_tx=tx_signed_raw_hex,
+            )
-                )
         session.add(otx)
-        session.commit()
 
         # TODO: cancel all first, then replace. Otherwise we risk two non-locked states for two different nonces.
-        set_cancel(tx_previous_hash_hex, True)
+        set_cancel(chain_spec, strip_0x(tx_previous_hash_hex), manual=True, session=session)
 
-        TxCache.clone(tx_previous_hash_hex, tx_hash_hex)
+        TxCache.clone(tx_previous_hash_hex, tx_hash_hex, session=session)
 
         tx_hashes.append(tx_hash_hex)
         txs.append(tx_signed_raw_hex)
+        session.commit()
 
     session.close()
 
-    s = create_check_gas_and_send_task(
+    s = create_check_gas_task(
          txs, 
-         chain_str,
+         chain_spec,
          tx_new['from'],
-         tx_new['gas'],
+         gas=tx_new['gas'],
-         tx_hashes, 
+         tx_hashes_hex=tx_hashes, 
-         queue,
+         queue=queue,
         )
 
     s_unlock_send = celery.signature(
         'cic_eth.admin.ctrl.unlock_send',
         [
-            chain_str,
+            chain_spec.asdict(),
             tx_new['from'],
             ],
         queue=queue,
@@ -119,7 +139,7 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     s_unlock_direct = celery.signature(
         'cic_eth.admin.ctrl.unlock_queue',
         [
-            chain_str,
+            chain_spec.asdict(),
             tx_new['from'],
             ],
         queue=queue,

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

@@ -8,6 +8,7 @@ from chainlib.eth.constant import (
         ZERO_ADDRESS,
         )
 from cic_eth_registry import CICRegistry
+from cic_eth_registry.erc20 import ERC20Token
 from cic_eth_registry.error import UnknownContractError
 from chainlib.eth.address import to_checksum_address
 from chainlib.eth.contract import code
@@ -30,13 +31,14 @@ from chainqueue.db.enum import (
         status_str,
     )
 from chainqueue.error import TxStateChangeError
+from chainqueue.sql.query import get_tx
+from eth_erc20 import ERC20
 
 # local imports
 from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.role import AccountRole
 from cic_eth.db.models.nonce import Nonce
 from cic_eth.error import InitializationError
-from cic_eth.queue.query import get_tx
 
 app = celery.current_app
 
@@ -188,6 +190,7 @@ class AdminApi:
         s_manual = celery.signature(
             'cic_eth.queue.state.set_manual',
             [
+                chain_spec.asdict(),
                 tx_hash_hex,
                 ],
             queue=self.queue,
@@ -206,8 +209,9 @@ class AdminApi:
             s.link(s_gas)
 
         return s_manual.apply_async()
-                        
+                    
-    def check_nonce(self, address):
+
+    def check_nonce(self, chain_spec, address):
         s = celery.signature(
                 'cic_eth.queue.query.get_account_tx',
                 [
@@ -228,13 +232,12 @@ class AdminApi:
             s_get_tx = celery.signature(
                     'cic_eth.queue.query.get_tx',
                     [
-                    chain_spec.asdict(),
+                        chain_spec.asdict(),
                         k,
                         ],
                     queue=self.queue,
                     )
             tx = s_get_tx.apply_async().get()
-            #tx = get_tx(k)
             logg.debug('checking nonce {} (previous {})'.format(tx['nonce'], last_nonce))
             nonce_otx = tx['nonce']
             if not is_alive(tx['status']) and tx['status'] & local_fail > 0:
@@ -242,7 +245,9 @@ class AdminApi:
                 blocking_tx = k
                 blocking_nonce = nonce_otx
             elif nonce_otx - last_nonce > 1:
-                logg.error('nonce gap; {} followed {} for account {}'.format(nonce_otx, last_nonce, tx['from']))
+                logg.debug('tx {}'.format(tx))
+                tx_obj = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec)
+                logg.error('nonce gap; {} followed {} for account {}'.format(nonce_otx, last_nonce, tx_obj['from']))
                 blocking_tx = k
                 blocking_nonce = nonce_otx
                 break
@@ -256,12 +261,13 @@ class AdminApi:
                 'blocking': blocking_nonce,
             },
             'tx': {
-                'blocking': blocking_tx,
+                'blocking': add_0x(blocking_tx),
-                }
             }
+        }
 
 
-    def fix_nonce(self, address, nonce, chain_spec):
+    # TODO: is risky since it does not validate that there is actually a nonce problem?
+    def fix_nonce(self, chain_spec, address, nonce):
         s = celery.signature(
                 'cic_eth.queue.query.get_account_tx',
                 [
@@ -275,15 +281,17 @@ class AdminApi:
         txs = s.apply_async().get()
 
         tx_hash_hex = None
+        session = SessionBase.create_session()
         for k in txs.keys():
-            tx_dict = get_tx(k)
+            tx_dict = get_tx(chain_spec, k, session=session)
             if tx_dict['nonce'] == nonce:
                 tx_hash_hex = k
+        session.close()
 
         s_nonce = celery.signature(
                 'cic_eth.admin.nonce.shift_nonce',
                 [
-                    self.rpc.chain_spec.asdict(),
+                    chain_spec.asdict(),
                     tx_hash_hex, 
                 ],
                 queue=self.queue
@@ -388,12 +396,13 @@ class AdminApi:
     
         t = s.apply_async()
         tx = t.get()
-  
+
         source_token = None
         if tx['source_token'] != ZERO_ADDRESS:
+            source_token_declaration = None
             if registry != None:
                 try:
-                    source_token = registry.by_address(tx['source_token'])
+                    source_token_declaration = registry.by_address(tx['source_token'], sender_address=self.call_address)
                 except UnknownContractError:
                     logg.warning('unknown source token contract {} (direct)'.format(tx['source_token']))
             else:
@@ -406,16 +415,21 @@ class AdminApi:
                         queue=self.queue
                         )
                 t = s.apply_async()
-                source_token = t.get()
+                source_token_declaration = t.get()
-                if source_token == None:
+
-                    logg.warning('unknown source token contract {} (task pool)'.format(tx['source_token']))
+            if source_token_declaration != None:
+                logg.warning('found declarator record for source token {} but not checking validity'.format(tx['source_token']))
+                source_token = ERC20Token(chain_spec, self.rpc, tx['source_token'])
+                logg.debug('source token set tup {}'.format(source_token))
+
 
 
         destination_token = None
         if tx['destination_token'] != ZERO_ADDRESS:
+            destination_token_declaration = None
             if registry != None:
                 try:
-                    destination_token = registry.by_address(tx['destination_token'])
+                    destination_token_declaration = registry.by_address(tx['destination_token'], sender_address=self.call_address)
                 except UnknownContractError:
                     logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
             else:
@@ -428,10 +442,10 @@ class AdminApi:
                         queue=self.queue
                         )
                 t = s.apply_async()
-                destination_token = t.get()
+                destination_token_declaration = t.get()
-                if destination_token == None:
+            if destination_token_declaration != None:
-                    logg.warning('unknown destination token contract {} (task pool)'.format(tx['destination_token']))
+                logg.warning('found declarator record for destination token {} but not checking validity'.format(tx['destination_token']))
-
+                destination_token = ERC20Token(chain_spec, self.rpc, tx['destination_token'])
 
         tx['sender_description'] = 'Custodial account'
         tx['recipient_description'] = 'Custodial account'
@@ -543,13 +557,19 @@ class AdminApi:
                 if role != None:
                     tx['recipient_description'] = role
 
+        erc20_c = ERC20(chain_spec)
         if source_token != None:
-            tx['source_token_symbol'] = source_token.symbol()
+            tx['source_token_symbol'] = source_token.symbol
-            tx['sender_token_balance'] = source_token.function('balanceOf')(tx['sender']).call()
+            o = erc20_c.balance_of(tx['source_token'], tx['sender'], sender_address=self.call_address)
+            r = self.rpc.do(o)
+            tx['sender_token_balance'] = erc20_c.parse_balance_of(r)
 
         if destination_token != None:
-            tx['destination_token_symbol'] = destination_token.symbol()
+            tx['destination_token_symbol'] = destination_token.symbol
-            tx['recipient_token_balance'] = source_token.function('balanceOf')(tx['recipient']).call()
+            o = erc20_c.balance_of(tx['destination_token'], tx['recipient'], sender_address=self.call_address)
+            r = self.rpc.do(o)
+            tx['recipient_token_balance'] = erc20_c.parse_balance_of(r)
+            #tx['recipient_token_balance'] = destination_token.function('balanceOf')(tx['recipient']).call()
 
         # TODO: this can mean either not subitted or culled, need to check other txs with same nonce to determine which
         tx['network_status'] = 'Not in node' 

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

@@ -74,134 +74,134 @@ class Api:
         return s_token.apply_async()
 
 
-    def convert_transfer(self, from_address, to_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
+#    def convert_transfer(self, from_address, to_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
-        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens, and transfers to a specified receipient after convert has completed.
+#        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens, and transfers to a specified receipient after convert has completed.
-
+#
-        :param from_address: Ethereum address of sender
+#        :param from_address: Ethereum address of sender
-        :type from_address: str, 0x-hex
+#        :type from_address: str, 0x-hex
-        :param to_address: Ethereum address of receipient
+#        :param to_address: Ethereum address of receipient
-        :type to_address: str, 0x-hex
+#        :type to_address: str, 0x-hex
-        :param target_return: Estimated return from conversion
+#        :param target_return: Estimated return from conversion
-        :type  target_return: int
+#        :type  target_return: int
-        :param minimum_return: The least value of destination token return to allow
+#        :param minimum_return: The least value of destination token return to allow
-        :type minimum_return: int
+#        :type minimum_return: int
-        :param from_token_symbol: ERC20 token symbol of token being converted
+#        :param from_token_symbol: ERC20 token symbol of token being converted
-        :type from_token_symbol: str
+#        :type from_token_symbol: str
-        :param to_token_symbol: ERC20 token symbol of token to receive
+#        :param to_token_symbol: ERC20 token symbol of token to receive
-        :type to_token_symbol: str
+#        :type to_token_symbol: str
-        :returns: uuid of root task
+#        :returns: uuid of root task
-        :rtype: celery.Task
+#        :rtype: celery.Task
-        """
+#        """
-        raise NotImplementedError('out of service until new DEX migration is done')
+#        raise NotImplementedError('out of service until new DEX migration is done')
-        s_check = celery.signature(
+#        s_check = celery.signature(
-                'cic_eth.admin.ctrl.check_lock',
+#                'cic_eth.admin.ctrl.check_lock',
-                [
+#                [
-                    [from_token_symbol, to_token_symbol],
+#                    [from_token_symbol, to_token_symbol],
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    LockEnum.QUEUE,
+#                    LockEnum.QUEUE,
-                    from_address,
+#                    from_address,
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_nonce = celery.signature(
+#        s_nonce = celery.signature(
-                'cic_eth.eth.nonce.reserve_nonce',
+#                'cic_eth.eth.nonce.reserve_nonce',
-                [
+#                [
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_tokens = celery.signature(
+#        s_tokens = celery.signature(
-                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
+#                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
-                [
+#                [
-                    self.chain_str,
+#                    self.chain_str,
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_convert = celery.signature(
+#        s_convert = celery.signature(
-                'cic_eth.eth.bancor.convert_with_default_reserve',
+#                'cic_eth.eth.bancor.convert_with_default_reserve',
-                [
+#                [
-                    from_address,
+#                    from_address,
-                    target_return,
+#                    target_return,
-                    minimum_return,
+#                    minimum_return,
-                    to_address,
+#                    to_address,
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_nonce.link(s_tokens)
+#        s_nonce.link(s_tokens)
-        s_check.link(s_nonce)
+#        s_check.link(s_nonce)
-        if self.callback_param != None:
+#        if self.callback_param != None:
-            s_convert.link(self.callback_success)
+#            s_convert.link(self.callback_success)
-            s_tokens.link(s_convert).on_error(self.callback_error)
+#            s_tokens.link(s_convert).on_error(self.callback_error)
-        else:
+#        else:
-            s_tokens.link(s_convert)
+#            s_tokens.link(s_convert)
-
+#
-        t = s_check.apply_async(queue=self.queue)
+#        t = s_check.apply_async(queue=self.queue)
-        return t
+#        return t
-
+#
-
+#
-    def convert(self, from_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
+#    def convert(self, from_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
-        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens.
+#        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens.
-
+#
-        :param from_address: Ethereum address of sender
+#        :param from_address: Ethereum address of sender
-        :type from_address: str, 0x-hex
+#        :type from_address: str, 0x-hex
-        :param target_return: Estimated return from conversion
+#        :param target_return: Estimated return from conversion
-        :type  target_return: int
+#        :type  target_return: int
-        :param minimum_return: The least value of destination token return to allow
+#        :param minimum_return: The least value of destination token return to allow
-        :type minimum_return: int
+#        :type minimum_return: int
-        :param from_token_symbol: ERC20 token symbol of token being converted
+#        :param from_token_symbol: ERC20 token symbol of token being converted
-        :type from_token_symbol: str
+#        :type from_token_symbol: str
-        :param to_token_symbol: ERC20 token symbol of token to receive
+#        :param to_token_symbol: ERC20 token symbol of token to receive
-        :type to_token_symbol: str
+#        :type to_token_symbol: str
-        :returns: uuid of root task
+#        :returns: uuid of root task
-        :rtype: celery.Task
+#        :rtype: celery.Task
-        """
+#        """
-        raise NotImplementedError('out of service until new DEX migration is done')
+#        raise NotImplementedError('out of service until new DEX migration is done')
-        s_check = celery.signature(
+#        s_check = celery.signature(
-                'cic_eth.admin.ctrl.check_lock',
+#                'cic_eth.admin.ctrl.check_lock',
-                [
+#                [
-                    [from_token_symbol, to_token_symbol],
+#                    [from_token_symbol, to_token_symbol],
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    LockEnum.QUEUE,
+#                    LockEnum.QUEUE,
-                    from_address,
+#                    from_address,
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_nonce = celery.signature(
+#        s_nonce = celery.signature(
-                'cic_eth.eth.nonce.reserve_nonce',
+#                'cic_eth.eth.nonce.reserve_nonce',
-                [
+#                [
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_tokens = celery.signature(
+#        s_tokens = celery.signature(
-                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
+#                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
-                [
+#                [
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_convert = celery.signature(
+#        s_convert = celery.signature(
-                'cic_eth.eth.bancor.convert_with_default_reserve',
+#                'cic_eth.eth.bancor.convert_with_default_reserve',
-                [
+#                [
-                    from_address,
+#                    from_address,
-                    target_return,
+#                    target_return,
-                    minimum_return,
+#                    minimum_return,
-                    from_address,
+#                    from_address,
-                    self.chain_spec.asdict(),
+#                    self.chain_spec.asdict(),
-                    ],
+#                    ],
-                queue=self.queue,
+#                queue=self.queue,
-                )
+#                )
-        s_nonce.link(s_tokens)
+#        s_nonce.link(s_tokens)
-        s_check.link(s_nonce)
+#        s_check.link(s_nonce)
-        if self.callback_param != None:
+#        if self.callback_param != None:
-            s_convert.link(self.callback_success)
+#            s_convert.link(self.callback_success)
-            s_tokens.link(s_convert).on_error(self.callback_error)
+#            s_tokens.link(s_convert).on_error(self.callback_error)
-        else:
+#        else:
-            s_tokens.link(s_convert)
+#            s_tokens.link(s_convert)
-
+#
-        t = s_check.apply_async(queue=self.queue)
+#        t = s_check.apply_async(queue=self.queue)
-        return t
+#        return t
 
 
     def transfer(self, from_address, to_address, value, token_symbol):

apps/cic-eth/cic_eth/db/util.py

@@ -1,8 +0,0 @@
-import math
-
-def num_serialize(n):
-    if n == 0:
-        return b'\x00'
-    binlog = math.log2(n)
-    bytelength = int(binlog / 8 + 1)
-    return n.to_bytes(bytelength, 'big')

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

@@ -57,10 +57,12 @@ celery_app = celery.current_app
 logg = logging.getLogger()
 
 
+MAXIMUM_FEE_UNITS = 8000000
+
 class MaxGasOracle:
 
     def gas(code=None):
-        return 8000000
+        return MAXIMUM_FEE_UNITS
 
 
 def create_check_gas_task(tx_signed_raws_hex, chain_spec, holder_address, gas=None, tx_hashes_hex=None, queue=None):
@@ -150,7 +152,7 @@ def cache_gas_data(
 
 
 @celery_app.task(bind=True, throws=(OutOfGasError), base=CriticalSQLAlchemyAndWeb3Task)
-def check_gas(self, tx_hashes, chain_spec_dict, txs=[], address=None, gas_required=None):
+def check_gas(self, tx_hashes, chain_spec_dict, txs=[], address=None, gas_required=MAXIMUM_FEE_UNITS):
     """Check the gas level of the sender address of a transaction.
 
     If the account balance is not sufficient for the required gas, gas refill is requested and OutOfGasError raiser.
@@ -170,24 +172,30 @@ def check_gas(self, tx_hashes, chain_spec_dict, txs=[], address=None, gas_requir
     :return: Signed raw transaction data list
     :rtype: param txs, unchanged
     """
-    if len(txs) == 0:
+    chain_spec = ChainSpec.from_dict(chain_spec_dict)
-        for i in range(len(tx_hashes)):
+    logg.debug('txs {} tx_hashes {}'.format(txs, tx_hashes))
-            o = get_tx(tx_hashes[i])
+
-            txs.append(o['signed_tx'])
+    addresspass = None
-            if address == None:
+    if len(txs) == 0:
-                address = o['address']
+        addresspass = []
+        for i in range(len(tx_hashes)):
+            o = get_tx(chain_spec_dict, tx_hashes[i])
+            txs.append(o['signed_tx'])
+            logg.debug('sender {}'.format(o))
+            tx = unpack(bytes.fromhex(strip_0x(o['signed_tx'])), chain_spec)
+            if address == None:
+                address = tx['from']
+            elif address != tx['from']:
+                raise ValueError('txs passed to check gas must all have same sender; had {} got {}'.format(address, tx['from']))
+            addresspass.append(address)
 
-    #if not web3.Web3.isChecksumAddress(address):
     if not is_checksum_address(address):
         raise ValueError('invalid address {}'.format(address))
 
-    chain_spec = ChainSpec.from_dict(chain_spec_dict)
-
     queue = self.request.delivery_info.get('routing_key')
 
     conn = RPCConnection.connect(chain_spec)
 
-    # TODO: it should not be necessary to pass address explicitly, if not passed should be derived from the tx
     gas_balance = 0
     try:
         o = balance(address)
@@ -198,6 +206,9 @@ def check_gas(self, tx_hashes, chain_spec_dict, txs=[], address=None, gas_requir
         conn.disconnect()
         raise EthError('gas_balance call for {}: {}'.format(address, e))
 
+    if gas_required == None:
+        gas_required = MAXIMUM_FEE_UNITS
+
     logg.debug('address {} has gas {} needs {}'.format(address, gas_balance, gas_required))
     session = SessionBase.create_session()
     gas_provider = AccountRole.get_address('GAS_GIFTER', session=session)
@@ -268,7 +279,8 @@ def check_gas(self, tx_hashes, chain_spec_dict, txs=[], address=None, gas_requir
             queue=queue,
             )
         ready_tasks.append(s)
-    celery.group(ready_tasks)()
+    t = celery.group(ready_tasks)()
+    logg.debug('group {}'.format(t))
 
     return txs
 

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

@@ -21,6 +21,7 @@ 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
 
 # local imports
 from cic_eth.db import SessionBase
@@ -58,6 +59,9 @@ def hashes_to_txs(self, tx_hashes):
     if len(tx_hashes) == 0:
         raise ValueError('no transaction to send')
 
+    for i in range(len(tx_hashes)):
+        tx_hashes[i] = strip_0x(tx_hashes[i])
+
     queue = self.request.delivery_info['routing_key']
 
     session = SessionBase.create_session()
@@ -148,7 +152,7 @@ def send(self, txs, chain_spec_dict):
 
 @celery_app.task(bind=True, throws=(NotFoundEthException,), base=CriticalWeb3Task)
 def sync_tx(self, tx_hash_hex, chain_spec_dict):
-    """Force update of network status of a simgle transaction
+    """Force update of network status of a single transaction
 
     :param tx_hash_hex: Transaction hash
     :type tx_hash_hex: str, 0x-hex
@@ -173,12 +177,14 @@ def sync_tx(self, tx_hash_hex, chain_spec_dict):
 
     # TODO: apply receipt in tx object to validate and normalize input
     if rcpt != None:
+        rcpt = snake_and_camel(rcpt)
         success = rcpt['status'] == 1
-        logg.debug('sync tx {} mined block {} success {}'.format(tx_hash_hex, rcpt['blockNumber'], success))
+        logg.debug('sync tx {} mined block {} tx index {} success {}'.format(tx_hash_hex, rcpt['blockNumber'], rcpt['transactionIndex'], success))
 
         s = celery.signature(
             'cic_eth.queue.state.set_final',
             [
+                chain_spec_dict,
                 tx_hash_hex,
                 rcpt['blockNumber'],
                 rcpt['transactionIndex'],
@@ -186,12 +192,14 @@ def sync_tx(self, tx_hash_hex, chain_spec_dict):
                 ],
                 queue=queue,
             )
+    # TODO: it's not entirely clear how we can reliable determine that its in mempool without explicitly checking
     else:
         logg.debug('sync tx {} mempool'.format(tx_hash_hex))
 
         s = celery.signature(
             'cic_eth.queue.state.set_sent',
             [
+                chain_spec_dict,
                 tx_hash_hex,
                 ],
                 queue=queue,

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

@@ -19,7 +19,7 @@ from cic_eth_registry import CICRegistry
 from cic_eth_registry.erc20 import ERC20Token
 from chainqueue.db.models.otx import Otx
 from chainqueue.db.enum import StatusEnum
-from chainqueue.query import get_tx_cache
+from chainqueue.sql.query import get_tx_cache
 from eth_erc20 import ERC20
 
 # local imports

apps/cic-eth/cic_eth/queue/query.py

@@ -5,7 +5,7 @@ import datetime
 import celery
 from chainlib.chain import ChainSpec
 from chainlib.eth.tx import unpack
-import chainqueue.query
+import chainqueue.sql.query
 from chainqueue.db.enum import (
         StatusEnum,
         is_alive,
@@ -28,7 +28,7 @@ celery_app = celery.current_app
 def get_tx_cache(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.query.get_tx_cache(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.query.get_tx_cache(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -37,7 +37,7 @@ def get_tx_cache(chain_spec_dict, tx_hash):
 def get_tx(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r =  chainqueue.query.get_tx(chain_spec, tx_hash)
+    r =  chainqueue.sql.query.get_tx(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -46,7 +46,7 @@ def get_tx(chain_spec_dict, tx_hash):
 def get_account_tx(chain_spec_dict, address, as_sender=True, as_recipient=True, counterpart=None):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.query.get_account_tx(chain_spec, address, as_sender=True, as_recipient=True, counterpart=None, session=session)
+    r = chainqueue.sql.query.get_account_tx(chain_spec, address, as_sender=True, as_recipient=True, counterpart=None, session=session)
     session.close()
     return r
 
@@ -55,17 +55,17 @@ def get_account_tx(chain_spec_dict, address, as_sender=True, as_recipient=True,
 def get_upcoming_tx_nolock(chain_spec_dict, status=StatusEnum.READYSEND, not_status=None, recipient=None, before=None, limit=0, session=None):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.query.get_upcoming_tx(chain_spec, status, not_status=not_status, recipient=recipient, before=before, limit=limit, session=session, decoder=unpack)
+    r = chainqueue.sql.query.get_upcoming_tx(chain_spec, status, not_status=not_status, recipient=recipient, before=before, limit=limit, session=session, decoder=unpack)
     session.close()
     return r
 
 
 def get_status_tx(chain_spec, status, not_status=None, before=None, exact=False, limit=0, session=None):
-    return chainqueue.query.get_status_tx_cache(chain_spec, status, not_status=not_status, before=before, exact=exact, limit=limit, session=session, decoder=unpack)
+    return chainqueue.sql.query.get_status_tx_cache(chain_spec, status, not_status=not_status, before=before, exact=exact, limit=limit, session=session, decoder=unpack)
 
 
 def get_paused_tx(chain_spec, status=None, sender=None, session=None, decoder=None):
-    return chainqueue.query.get_paused_tx_cache(chain_spec, status=status, sender=sender, session=session, decoder=unpack)
+    return chainqueue.sql.query.get_paused_tx_cache(chain_spec, status=status, sender=sender, session=session, decoder=unpack)
 
 
 def get_nonce_tx(chain_spec, nonce, sender):

apps/cic-eth/cic_eth/queue/state.py

@@ -1,6 +1,6 @@
 # external imports
 from chainlib.chain import ChainSpec
-import chainqueue.state
+import chainqueue.sql.state
 
 # local imports
 import celery
@@ -14,7 +14,7 @@ celery_app = celery.current_app
 def set_sent(chain_spec_dict, tx_hash, fail=False):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_sent(chain_spec, tx_hash, fail, session=session)
+    r = chainqueue.sql.state.set_sent(chain_spec, tx_hash, fail, session=session)
     session.close()
     return r
 
@@ -23,7 +23,7 @@ def set_sent(chain_spec_dict, tx_hash, fail=False):
 def set_final(chain_spec_dict, tx_hash, block=None, tx_index=None, fail=False):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_final(chain_spec, tx_hash, block=block, tx_index=tx_index, fail=fail, session=session)
+    r = chainqueue.sql.state.set_final(chain_spec, tx_hash, block=block, tx_index=tx_index, fail=fail, session=session)
     session.close()
     return r
 
@@ -32,7 +32,7 @@ def set_final(chain_spec_dict, tx_hash, block=None, tx_index=None, fail=False):
 def set_cancel(chain_spec_dict, tx_hash, manual=False):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_cancel(chain_spec, tx_hash, manual, session=session)
+    r = chainqueue.sql.state.set_cancel(chain_spec, tx_hash, manual, session=session)
     session.close()
     return r
 
@@ -41,7 +41,7 @@ def set_cancel(chain_spec_dict, tx_hash, manual=False):
 def set_rejected(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_rejected(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.set_rejected(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -50,7 +50,7 @@ def set_rejected(chain_spec_dict, tx_hash):
 def set_fubar(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_fubar(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.set_fubar(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -59,7 +59,7 @@ def set_fubar(chain_spec_dict, tx_hash):
 def set_manual(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_manual(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.set_manual(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -68,7 +68,7 @@ def set_manual(chain_spec_dict, tx_hash):
 def set_ready(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_ready(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.set_ready(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -77,7 +77,7 @@ def set_ready(chain_spec_dict, tx_hash):
 def set_reserved(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_reserved(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.set_reserved(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -86,7 +86,7 @@ def set_reserved(chain_spec_dict, tx_hash):
 def set_waitforgas(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.set_waitforgas(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.set_waitforgas(chain_spec, tx_hash, session=session)
     session.close()
     return r
     
@@ -95,7 +95,7 @@ def set_waitforgas(chain_spec_dict, tx_hash):
 def get_state_log(chain_spec_dict, tx_hash):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.get_state_log(chain_spec, tx_hash, session=session)
+    r = chainqueue.sql.state.get_state_log(chain_spec, tx_hash, session=session)
     session.close()
     return r
 
@@ -104,6 +104,6 @@ def get_state_log(chain_spec_dict, tx_hash):
 def obsolete(chain_spec_dict, tx_hash, final):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     session = SessionBase.create_session()
-    r = chainqueue.state.obsolete_by_cache(chain_spec, tx_hash, final, session=session)
+    r = chainqueue.sql.state.obsolete_by_cache(chain_spec, tx_hash, final, session=session)
     session.close()
     return r

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

@@ -15,14 +15,14 @@ from sqlalchemy import tuple_
 from sqlalchemy import func
 from chainlib.chain import ChainSpec
 from chainlib.eth.tx import unpack
-import chainqueue.state
+import chainqueue.sql.state
 from chainqueue.db.enum import (
         StatusEnum,
         StatusBits,
         is_alive,
         dead,
         )
-from chainqueue.tx import create
+from chainqueue.sql.tx import create
 from chainqueue.error import NotLocalTxError
 from chainqueue.db.enum import status_str
 

apps/cic-eth/cic_eth/registry.py

@@ -5,29 +5,30 @@ import logging
 from cic_eth_registry import CICRegistry
 from cic_eth_registry.lookup.declarator import AddressDeclaratorLookup
 from cic_eth_registry.lookup.tokenindex import TokenIndexLookup
+from chainlib.eth.constant import ZERO_ADDRESS
 
 logg = logging.getLogger()
 
 
-def connect_token_registry(rpc, chain_spec):
+def connect_token_registry(rpc, chain_spec, sender_address=ZERO_ADDRESS):
     registry = CICRegistry(chain_spec, rpc)
-    token_registry_address = registry.by_name('TokenRegistry')
+    token_registry_address = registry.by_name('TokenRegistry', sender_address=sender_address)
     logg.debug('using token registry address {}'.format(token_registry_address))
     lookup = TokenIndexLookup(chain_spec, token_registry_address)
     CICRegistry.add_lookup(lookup)
 
 
-def connect_declarator(rpc, chain_spec, trusted_addresses):
+def connect_declarator(rpc, chain_spec, trusted_addresses, sender_address=ZERO_ADDRESS):
     registry = CICRegistry(chain_spec, rpc)
-    declarator_address = registry.by_name('AddressDeclarator')
+    declarator_address = registry.by_name('AddressDeclarator', sender_address=sender_address)
     logg.debug('using declarator address {}'.format(declarator_address))
     lookup = AddressDeclaratorLookup(chain_spec, declarator_address, trusted_addresses)
     CICRegistry.add_lookup(lookup)
 
 
-def connect(rpc, chain_spec, registry_address):
+def connect(rpc, chain_spec, registry_address, sender_address=ZERO_ADDRESS):
     CICRegistry.address = registry_address
     registry = CICRegistry(chain_spec, rpc)
-    registry_address = registry.by_name('ContractRegistry')
+    registry_address = registry.by_name('ContractRegistry', sender_address=sender_address)
     return registry
 

apps/cic-eth/cic_eth/runnable/daemons/dispatcher.py

@@ -21,7 +21,7 @@ from chainqueue.db.enum import (
     StatusBits,
     )
 from chainqueue.error import NotLocalTxError
-from chainqueue.state import set_reserved
+from chainqueue.sql.state import set_reserved
 
 # local imports
 import cic_eth

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

@@ -72,7 +72,9 @@ class CallbackFilter(SyncFilter):
         #transfer_data['token_address'] = tx.inputs[0]
         faucet_contract = tx.inputs[0]
 
-        o = Faucet.token(faucet_contract, sender_address=self.caller_address)
+        c = Faucet(self.chain_spec)
+
+        o = c.token(faucet_contract, sender_address=self.caller_address)
         r = conn.do(o)
         transfer_data['token_address'] = add_0x(c.parse_token(r))
 

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

@@ -10,14 +10,15 @@ from chainlib.eth.tx import unpack
 from chainqueue.db.enum import StatusBits
 from chainqueue.db.models.tx import TxCache
 from chainqueue.db.models.otx import Otx
-from chainqueue.query import get_paused_tx_cache as get_paused_tx
+from chainqueue.sql.query import get_paused_tx_cache as get_paused_tx
 
 # local imports
 from cic_eth.db.models.base import SessionBase
 from cic_eth.eth.gas import create_check_gas_task
 from .base import SyncFilter
 
-logg = logging.getLogger().getChild(__name__)
+#logg = logging.getLogger().getChild(__name__)
+logg = logging.getLogger()
 
 
 class GasFilter(SyncFilter):
@@ -27,11 +28,11 @@ class GasFilter(SyncFilter):
         self.chain_spec = chain_spec
 
 
-    def filter(self, conn, block, tx, session):
+    def filter(self, conn, block, tx, db_session):
         if tx.value > 0:
             tx_hash_hex = add_0x(tx.hash)
             logg.debug('gas refill tx {}'.format(tx_hash_hex))
-            session = SessionBase.bind_session(session)
+            session = SessionBase.bind_session(db_session)
             q = session.query(TxCache.recipient)
             q = q.join(Otx)
             q = q.filter(Otx.tx_hash==strip_0x(tx_hash_hex))
@@ -56,7 +57,7 @@ class GasFilter(SyncFilter):
                         tx_hashes_hex=list(txs.keys()),
                         queue=self.queue,
                 )
-                s.apply_async()
+                return s.apply_async()
 
 
     def __str__(self):

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

@@ -50,7 +50,8 @@ class RegistrationFilter(SyncFilter):
                     queue=self.queue,
                     )
                 s_nonce.link(s_gift)
-                s_nonce.apply_async()
+                t = s_nonce.apply_async()
+                return t
 
 
     def __str__(self):

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

@@ -3,7 +3,7 @@ import logging
 
 # external imports
 import celery
-from chainqueue.state import obsolete_by_cache
+from chainqueue.sql.state import obsolete_by_cache
 
 logg = logging.getLogger()
 

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

@@ -32,7 +32,7 @@ class TransferAuthFilter(SyncFilter):
         self.transfer_request_contract = registry.by_name('TransferAuthorization', sender_address=call_address)
 
 
-    def filter(self, conn, block, tx, session): #rcpt, chain_str, session=None):
+    def filter(self, conn, block, tx, db_session): #rcpt, chain_str, session=None):
 
         if tx.payload == None:
             logg.debug('no payload')
@@ -45,16 +45,17 @@ class TransferAuthFilter(SyncFilter):
             return False
 
         recipient = tx.inputs[0]
-        if recipient != self.transfer_request_contract.address():
+        #if recipient != self.transfer_request_contract.address():
+        if recipient != self.transfer_request_contract:
             logg.debug('not our transfer auth contract address {}'.format(recipient))
             return False
 
         r = TransferAuthorization.parse_create_request_request(tx.payload) 
-           
+          
-        sender = abi_decode_single(ABIContractType.ADDRESS, r[0]) 
+        sender = r[0]
-        recipient = abi_decode_single(ABIContractType.ADDRESS, r[1])
+        recipient = r[1]
-        token = abi_decode_single(ABIContractType.ADDRESS, r[2]) 
+        token = r[2]
-        value = abi_decode_single(ABIContractType.UINT256, r[3])
+        value = r[3]
 
         token_data = {
             'address': token,
@@ -64,6 +65,7 @@ class TransferAuthFilter(SyncFilter):
             'cic_eth.eth.nonce.reserve_nonce',
             [
                 [token_data],
+                self.chain_spec.asdict(), 
                 sender,
                 ],
             queue=self.queue,
@@ -80,7 +82,7 @@ class TransferAuthFilter(SyncFilter):
             )
         s_nonce.link(s_approve)
         t = s_nonce.apply_async()
-        return True
+        return t
 
 
     def __str__(self):

apps/cic-eth/cic_eth/stat.py

@@ -20,7 +20,11 @@ def init_chain_stat(rpc, block_start=0):
     if block_start == 0:
         o = block_latest()
         r = rpc.do(o)
-        block_start = int(r, 16)
+        try:
+            block_start = int(r, 16)
+        except TypeError:
+            block_start = int(r)
+        logg.debug('blockstart {}'.format(block_start))
 
     for i in range(BLOCK_SAMPLES):
         o = block_by_number(block_start-10+i)

apps/cic-eth/cic_eth/task.py

@@ -20,7 +20,8 @@ import liveness.linux
 from cic_eth.error import SeppukuError
 from cic_eth.db.models.base import SessionBase
 
-logg = logging.getLogger().getChild(__name__)
+#logg = logging.getLogger().getChild(__name__)
+logg = logging.getLogger()
 
 celery_app = celery.current_app
 
@@ -118,12 +119,13 @@ def registry():
     return CICRegistry.address
 
 
-@celery_app.task()
+@celery_app.task(bind=True, base=BaseTask)
-def registry_address_lookup(chain_spec_dict, address, connection_tag='default'):
+def registry_address_lookup(self, chain_spec_dict, address, connection_tag='default'):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     conn = RPCConnection.connect(chain_spec, tag=connection_tag)
     registry = CICRegistry(chain_spec, conn)
-    return registry.by_address(address)
+    r = registry.by_address(address, sender_address=self.call_address)
+    return r
 
 
 @celery_app.task(throws=(UnknownContractError,))
@@ -131,7 +133,7 @@ def registry_name_lookup(chain_spec_dict, name, connection_tag='default'):
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     conn = RPCConnection.connect(chain_spec, tag=connection_tag)
     registry = CICRegistry(chain_spec, conn)
-    return registry.by_name(name)
+    return registry.by_name(name, sender_address=self.call_address)
 
 
 @celery_app.task()

apps/cic-eth/cic_eth/version.py

@@ -10,7 +10,7 @@ version = (
         0,
         11,
         0,
-        'beta.13',
+        'beta.16',
         )
 
 version_object = semver.VersionInfo(

apps/cic-eth/docker/Dockerfile

@@ -1,48 +1,62 @@
-# FROM grassrootseconomics:cic
+FROM python:3.8.6-slim-buster as compile
-
-#FROM python:3.8.6-alpine
-FROM python:3.8.6-slim-buster
-
-#COPY --from=0 /usr/local/share/cic/solidity/ /usr/local/share/cic/solidity/
 
 WORKDIR /usr/src/cic-eth
 
-ARG pip_extra_index_url_flag='--index https://pypi.org/simple --extra-index-url https://pip.grassrootseconomics.net:8433'
-ARG root_requirement_file='requirements.txt'
-
-#RUN apk update && \
-#	apk add gcc musl-dev gnupg libpq
-#RUN apk add postgresql-dev
-#RUN apk add linux-headers
-#RUN apk add libffi-dev
 RUN apt-get update && \
 	apt install -y gcc gnupg libpq-dev wget make g++ gnupg bash procps git
 
-# Copy shared requirements from top of mono-repo
+#RUN python -m venv venv && . venv/bin/activate
-RUN echo "copying root req file:  ${root_requirement_file}"
-#COPY $root_requirement_file .
-#RUN pip install -r $root_requirement_file $pip_extra_index_url_flag
-RUN /usr/local/bin/python -m pip install --upgrade pip
-#RUN git clone https://gitlab.com/grassrootseconomics/cic-base.git && \
-#	cd cic-base && \
-#	git checkout 7ae1f02efc206b13a65873567b0f6d1c3b7f9bc0 && \
-#	python merge_requirements.py | tee merged_requirements.txt 
-#RUN cd cic-base && \
-#	pip install $pip_extra_index_url_flag -r ./merged_requirements.txt 
-RUN  pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2b9
 
-COPY cic-eth/scripts/ scripts/
+ARG pip_extra_index_url_flag='--index https://pypi.org/simple --extra-index-url https://pip.grassrootseconomics.net:8433'
-COPY cic-eth/setup.cfg cic-eth/setup.py ./
+RUN /usr/local/bin/python -m pip install --upgrade pip
-COPY cic-eth/cic_eth/ cic_eth/
+RUN pip install semver
-# Copy app specific requirements
+
-COPY cic-eth/requirements.txt .
+# TODO use a packaging style that lets us copy requirments only ie. pip-tools 
-COPY cic-eth/test_requirements.txt .
+COPY cic-eth/ .
 RUN	pip install $pip_extra_index_url_flag .
 
+# --- TEST IMAGE ---
+FROM python:3.8.6-slim-buster as test 
+
+RUN apt-get update && \
+	apt install -y gcc gnupg libpq-dev wget make g++ gnupg bash procps git
+
+WORKDIR /usr/src/cic-eth
+
+RUN /usr/local/bin/python -m pip install --upgrade pip
+
+COPY --from=compile /usr/local/bin/ /usr/local/bin/
+COPY --from=compile /usr/local/lib/python3.8/site-packages/ \
+                          /usr/local/lib/python3.8/site-packages/
+# TODO we could use venv inside container to isolate the system and app deps further
+# COPY --from=compile /usr/src/cic-eth/ .
+# RUN . venv/bin/activate
+
+COPY cic-eth/test_requirements.txt .
+RUN	pip install $pip_extra_index_url_flag -r test_requirements.txt 
+
+COPY cic-eth . 
+
+ENV PYTHONPATH .
+
+ENTRYPOINT ["pytest"]
+
+# --- RUNTIME ---
+FROM python:3.8.6-slim-buster as runtime 
+
+RUN apt-get update && \
+	apt install -y gnupg libpq-dev procps
+
+WORKDIR /usr/src/cic-eth
+
+COPY --from=compile /usr/local/bin/ /usr/local/bin/
+COPY --from=compile /usr/local/lib/python3.8/site-packages/ \
+                          /usr/local/lib/python3.8/site-packages/
+
 COPY cic-eth/docker/* ./ 
 RUN chmod 755 *.sh
-COPY cic-eth/tests/ tests/
 
+COPY cic-eth/scripts/ scripts/
 # # ini files in config directory defines the configurable parameters for the application
 # # they can all be overridden by environment variables
 # # to generate a list of environment variables from configuration, use: confini-dump -z <dir> (executable provided by confini package)
@@ -51,3 +65,4 @@ COPY cic-eth/cic_eth/db/migrations/ /usr/local/share/cic-eth/alembic/
 COPY cic-eth/crypto_dev_signer_config/ /usr/local/etc/crypto-dev-signer/
 
 COPY util/liveness/health.sh /usr/local/bin/health.sh
+

apps/cic-eth/requirements.txt

@@ -1,24 +1,25 @@
-cic-base~=0.1.2b11
+cic-base~=0.1.2b15
 celery==4.4.7
 crypto-dev-signer~=0.4.14b3
 confini~=0.3.6rc3
-cic-eth-registry~=0.5.5a4
+cic-eth-registry~=0.5.5a7
 redis==3.5.3
 alembic==1.4.2
 websockets==8.1
 requests~=2.24.0
 eth_accounts_index~=0.0.11a12
-erc20-transfer-authorization~=0.3.1a6
+erc20-transfer-authorization~=0.3.1a7
 uWSGI==2.0.19.1
 semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
 eth-address-index~=0.1.1a11
-chainlib~=0.0.3a2
+chainlib~=0.0.3rc2
 hexathon~=0.0.1a7
-chainsyncer[sql]~=0.0.2a4
+chainsyncer[sql]~=0.0.2a5
-chainqueue~=0.0.2a2
+chainqueue~=0.0.2b3
 sarafu-faucet==0.0.3a3
 erc20-faucet==0.2.1a4
 coincurve==15.0.0
 potaahto~=0.0.1a2
+pycryptodome==3.10.1

apps/cic-eth/setup.py

@@ -11,17 +11,6 @@ while True:
     requirements.append(l.rstrip())
 f.close()
 
-test_requirements = []
-f = open('test_requirements.txt', 'r')
-while True:
-    l = f.readline()
-    if l == '':
-        break
-    test_requirements.append(l.rstrip())
-f.close()
-
-
 setup(
-    install_requires=requirements,
+    install_requires=requirements
-    tests_require=test_requirements,
         )

apps/cic-eth/test_requirements.txt

@@ -4,4 +4,3 @@ pytest-mock==3.3.1
 pytest-cov==2.10.1
 eth-tester==0.5.0b3
 py-evm==0.3.0a20
-giftable-erc20-token==0.0.8a9

apps/cic-eth/tests/check/test_check_db.py

@@ -0,0 +1,8 @@
+# local imports
+from cic_eth.check.db import health
+
+def test_check_health(
+    init_database,
+        ):
+
+    assert health()

apps/cic-eth/tests/check/test_check_gas.py

@@ -0,0 +1,20 @@
+# local imports
+from cic_eth.check.gas import health
+from cic_eth.db.models.role import AccountRole
+
+def test_check_gas(
+    config,
+    init_database,
+    default_chain_spec,
+    eth_rpc,
+    custodial_roles,
+    whoever,
+        ):
+
+    config.add(str(default_chain_spec), 'CIC_CHAIN_SPEC', exists_ok=True)
+    config.add(100, 'ETH_GAS_GIFTER_MINIMUM_BALANCE', exists_ok=True)
+    assert health(config=config)
+
+    AccountRole.set('GAS_GIFTER', whoever, session=init_database)
+    init_database.commit()
+    assert not health(config=config)

apps/cic-eth/tests/check/test_check_redis.py

@@ -0,0 +1,16 @@
+# external imports
+import pytest
+
+# local imports
+from cic_eth.check.redis import health
+
+
+def test_check_redis(
+        config,
+        have_redis,
+        ):
+
+    if have_redis != None:
+        pytest.skip('cannot connect to redis, skipping test: {}'.format(have_redis))
+
+    assert health(unit='test', config=config)

apps/cic-eth/tests/check/test_check_signer.py

@@ -0,0 +1,13 @@
+# local imports
+from cic_eth.check.signer import health
+
+
+def test_check_signer(
+        default_chain_spec,
+        config,
+        eth_signer,
+        eth_rpc,
+        ):
+
+    config.add(str(default_chain_spec), 'CIC_CHAIN_SPEC', exists_ok=True)
+    assert health(config=config)

apps/cic-eth/tests/conftest.py

@@ -2,9 +2,11 @@
 import os
 import sys
 import logging
+import uuid
 
 # external imports
-from chainlib.eth.erc20 import ERC20
+from eth_erc20 import ERC20
+import redis
 
 # local imports
 from cic_eth.api import Api
@@ -19,6 +21,7 @@ from tests.fixtures_config import *
 from tests.fixtures_database import *
 from tests.fixtures_celery import *
 from tests.fixtures_role import *
+from tests.fixtures_contract import *
 from chainlib.eth.pytest import *
 from eth_contract_registry.pytest import *
 from cic_eth_registry.pytest.fixtures_contracts import *
@@ -55,3 +58,28 @@ def default_token(
         ):
     BaseTask.default_token_symbol = foo_token_symbol
     BaseTask.default_token_address = foo_token   
+
+
+
+@pytest.fixture(scope='session')
+def have_redis(
+        config,
+        ):
+
+    r = redis.Redis(
+            host = config.get('REDIS_HOST'),
+            port = config.get('REDIS_PORT'),
+            db = config.get('REDIS_DB'),
+            ) 
+    k = str(uuid.uuid4())
+    try:
+        r.set(k, 'foo')
+        r.delete(k)
+    except redis.exceptions.ConnectionError as e:
+        return e
+    except TypeError as e:
+        return e 
+
+    return None
+
+

apps/cic-eth/tests/filters/test_callback_filter.py

@@ -14,9 +14,9 @@ from chainlib.eth.tx import (
         Tx,
         )
 from chainlib.eth.block import Block
-from chainlib.eth.erc20 import ERC20
+from eth_erc20 import ERC20
 from sarafu_faucet import MinterFaucet
-from eth_accounts_index import AccountRegistry
+from eth_accounts_index.registry import AccountRegistry
 from potaahto.symbols import snake_and_camel
 from hexathon import add_0x
 
@@ -26,7 +26,6 @@ from cic_eth.runnable.daemons.filters.callback import CallbackFilter
 logg = logging.getLogger()
 
 
-@pytest.mark.skip()
 def test_transfer_tx(
         default_chain_spec,
         init_database,
@@ -66,7 +65,6 @@ def test_transfer_tx(
     assert transfer_type == 'transfer'
 
 
-@pytest.mark.skip()
 def test_transfer_from_tx(
         default_chain_spec,
         init_database,

apps/cic-eth/tests/filters/test_filter_bogus.py

@@ -0,0 +1,38 @@
+# local imports
+from cic_eth.runnable.daemons.filters.gas import GasFilter
+from cic_eth.runnable.daemons.filters.transferauth import TransferAuthFilter
+from cic_eth.runnable.daemons.filters.callback import CallbackFilter
+from cic_eth.runnable.daemons.filters.straggler import StragglerFilter
+from cic_eth.runnable.daemons.filters.tx import TxFilter
+from cic_eth.runnable.daemons.filters.register import RegistrationFilter
+
+
+# Hit tx mismatch paths on all filters
+def test_filter_bogus(
+        init_database,
+        bogus_tx_block,
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        transfer_auth,
+        cic_registry,
+        contract_roles,
+        register_lookups,
+        ):
+
+    fltrs = [
+        TransferAuthFilter(cic_registry, default_chain_spec, eth_rpc, call_address=contract_roles['CONTRACT_DEPLOYER']),
+        GasFilter(default_chain_spec, queue=None),
+        TxFilter(default_chain_spec, None),
+        CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER']),
+        StragglerFilter(default_chain_spec, None),
+        RegistrationFilter(default_chain_spec, queue=None),
+        ]
+      
+    for fltr in fltrs:
+        r = None
+        try:
+            r = fltr.filter(eth_rpc, bogus_tx_block[0], bogus_tx_block[1], db_session=init_database)
+        except:
+            pass
+        assert not r

apps/cic-eth/tests/filters/test_gas_filter.py

@@ -0,0 +1,101 @@
+# external imports
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import OverrideNonceOracle
+from chainqueue.sql.tx import create as queue_create
+from chainlib.eth.tx import (
+        TxFormat,
+        unpack,
+        Tx,
+        )
+from chainlib.eth.gas import (
+        Gas,
+        OverrideGasOracle,
+        )
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+from chainqueue.sql.state import (
+        set_waitforgas,
+        )
+from hexathon import strip_0x
+from chainqueue.db.models.otx import Otx
+from chainqueue.db.enum import StatusBits
+
+# local imports
+from cic_eth.runnable.daemons.filters.gas import GasFilter
+from cic_eth.eth.gas import cache_gas_data
+
+
+def test_filter_gas(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        celery_session_worker,
+    ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+    set_waitforgas(default_chain_spec, tx_hash_hex, session=init_database)
+    init_database.commit()
+    tx_hash_hex_wait = tx_hash_hex
+    otx = Otx.load(tx_hash_hex_wait, session=init_database)
+    assert otx.status & StatusBits.GAS_ISSUES == StatusBits.GAS_ISSUES
+
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['BOB'], agent_roles['ALICE'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            43,
+            agent_roles['BOB'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    fltr = GasFilter(default_chain_spec, queue=None)
+
+    o = block_latest()
+    r = eth_rpc.do(o)
+    o = block_by_number(r, include_tx=False)
+    r = eth_rpc.do(o)
+    block = Block(r)
+    block.txs = [tx_hash_hex]
+
+    tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
+    tx_src = unpack(tx_signed_raw_bytes, default_chain_spec)
+    tx = Tx(tx_src, block=block)
+    t = fltr.filter(eth_rpc, block, tx, db_session=init_database)
+
+    t.get_leaf()
+    assert t.successful()
+    init_database.commit()
+
+    otx = Otx.load(tx_hash_hex_wait, session=init_database)
+    assert otx.status & StatusBits.QUEUED == StatusBits.QUEUED

apps/cic-eth/tests/filters/test_register_filter.py

@@ -0,0 +1,78 @@
+# external imports
+from eth_accounts_index.registry import AccountRegistry
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import RPCNonceOracle
+from chainlib.eth.gas import OverrideGasOracle
+from chainlib.eth.tx import(
+        receipt,
+        unpack,
+        Tx,
+        )
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+from erc20_faucet import Faucet
+from hexathon import strip_0x
+from chainqueue.sql.query import get_account_tx
+
+# local imports
+from cic_eth.runnable.daemons.filters.register import RegistrationFilter
+
+
+def test_register_filter(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        account_registry,
+        faucet,
+        register_lookups,
+        contract_roles,
+        agent_roles,
+        cic_registry,
+        init_celery_tasks,
+        celery_session_worker,
+        caplog,
+        ):
+
+    nonce_oracle = RPCNonceOracle(contract_roles['ACCOUNT_REGISTRY_WRITER'], conn=eth_rpc)
+    gas_oracle = OverrideGasOracle(limit=AccountRegistry.gas(), conn=eth_rpc)
+
+    c = AccountRegistry(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = c.add(account_registry, contract_roles['ACCOUNT_REGISTRY_WRITER'], agent_roles['ALICE'])
+    r = eth_rpc.do(o)
+    tx_signed_raw_bytes = bytes.fromhex(strip_0x(o['params'][0]))
+
+    o = receipt(tx_hash_hex)
+    rcpt = eth_rpc.do(o)
+    assert rcpt['status'] == 1
+
+    o = block_latest()
+    r = eth_rpc.do(o)
+    o = block_by_number(r, include_tx=False)
+    r = eth_rpc.do(o)
+    block = Block(r)
+    block.txs = [tx_hash_hex]
+
+    tx_src = unpack(tx_signed_raw_bytes, default_chain_spec)
+    tx = Tx(tx_src, block=block, rcpt=rcpt)
+    tx.apply_receipt(rcpt)
+
+    fltr = RegistrationFilter(default_chain_spec, queue=None)
+    t = fltr.filter(eth_rpc, block, tx, db_session=init_database)
+
+    t.get_leaf()
+    assert t.successful()
+
+    gift_txs = get_account_tx(default_chain_spec.asdict(), agent_roles['ALICE'], as_sender=True, session=init_database)
+    ks = list(gift_txs.keys())
+    assert len(ks) == 1
+
+    tx_raw_signed_hex = strip_0x(gift_txs[ks[0]])
+    tx_raw_signed_bytes = bytes.fromhex(tx_raw_signed_hex)
+    gift_tx = unpack(tx_raw_signed_bytes, default_chain_spec)
+
+    gift = Faucet.parse_give_to_request(gift_tx['data'])
+    assert gift[0] == agent_roles['ALICE']

apps/cic-eth/tests/filters/test_straggler_filter.py

@@ -17,8 +17,8 @@ from chainlib.eth.block import (
         )
 from chainqueue.db.models.otx import Otx
 from chainqueue.db.enum import StatusBits
-from chainqueue.tx import create as queue_create
+from chainqueue.sql.tx import create as queue_create
-from chainqueue.state import (
+from chainqueue.sql.state import (
         set_reserved,
         set_ready,
         set_sent,

apps/cic-eth/tests/filters/test_transferauth_filter.py

@@ -0,0 +1,79 @@
+# external imports
+from erc20_transfer_authorization import TransferAuthorization
+from eth_erc20 import ERC20
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import RPCNonceOracle
+from chainlib.eth.gas import OverrideGasOracle
+from chainlib.eth.tx import (
+        receipt,
+        unpack,
+        Tx,
+        )
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+from hexathon import strip_0x
+from chainqueue.sql.query import get_account_tx
+
+# local imports
+from cic_eth.runnable.daemons.filters.transferauth import TransferAuthFilter
+
+
+def test_filter_transferauth(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        transfer_auth,
+        foo_token,
+        celery_session_worker,
+        register_lookups,
+        init_custodial,
+        cic_registry,
+    ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(contract_roles['CONTRACT_DEPLOYER'], eth_rpc)
+    gas_oracle = OverrideGasOracle(limit=200000, conn=eth_rpc)
+    c = TransferAuthorization(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = c.create_request(transfer_auth, contract_roles['CONTRACT_DEPLOYER'], agent_roles['ALICE'], agent_roles['BOB'], foo_token, 1024)
+    
+    r = rpc.do(o)
+    tx_signed_raw_bytes = bytes.fromhex(strip_0x(o['params'][0]))
+
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+
+    o = block_latest()
+    r = eth_rpc.do(o)
+    o = block_by_number(r, include_tx=False)
+    r = eth_rpc.do(o)
+    block = Block(r)
+    block.txs = [tx_hash_hex]
+
+    #tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
+    tx_src = unpack(tx_signed_raw_bytes, default_chain_spec)
+    tx = Tx(tx_src, block=block)
+
+    fltr = TransferAuthFilter(cic_registry, default_chain_spec, eth_rpc, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    t = fltr.filter(eth_rpc, block, tx, db_session=init_database)
+
+    t.get_leaf()
+    assert t.successful()
+
+    approve_txs = get_account_tx(default_chain_spec.asdict(), agent_roles['ALICE'], as_sender=True, session=init_database)
+    ks = list(approve_txs.keys())
+    assert len(ks) == 1
+
+    tx_raw_signed_hex = strip_0x(approve_txs[ks[0]])
+    tx_raw_signed_bytes = bytes.fromhex(tx_raw_signed_hex)
+    approve_tx = unpack(tx_raw_signed_bytes, default_chain_spec)
+
+    c = ERC20(default_chain_spec)
+    approve = c.parse_approve_request(approve_tx['data']) 
+    assert approve[0] == agent_roles['BOB']

apps/cic-eth/tests/filters/test_tx_filter.py

@@ -17,13 +17,12 @@ from chainlib.eth.block import (
         )
 from chainqueue.db.models.otx import Otx
 from chainqueue.db.enum import StatusBits
-from chainqueue.tx import create as queue_create
+from chainqueue.sql.tx import create as queue_create
-from chainqueue.state import (
+from chainqueue.sql.state import (
         set_reserved,
         set_ready,
         set_sent,
         )
-
 from hexathon import strip_0x
 
 # local imports
@@ -31,7 +30,7 @@ from cic_eth.runnable.daemons.filters.tx import TxFilter
 from cic_eth.eth.gas import cache_gas_data
 
 
-def test_tx(
+def test_filter_tx(
         default_chain_spec,
         init_database,
         eth_rpc,

apps/cic-eth/tests/fixtures_celery.py

@@ -22,7 +22,6 @@ def init_celery_tasks(
 @pytest.fixture(scope='session')
 def celery_includes():
     return [
-#        'cic_eth.eth.bancor',
         'cic_eth.eth.erc20',
         'cic_eth.eth.tx',
         'cic_eth.ext.tx',
@@ -47,8 +46,8 @@ def celery_config():
     bq = tempfile.mkdtemp()
     bp = tempfile.mkdtemp()
     rq = tempfile.mkdtemp()
-    logg.debug('celery broker queue {} processed {}'.format(bq, bp))
+    logg.debug('celery broker session queue {} processed {}'.format(bq, bp))
-    logg.debug('celery backend store {}'.format(rq))
+    logg.debug('celery backend session store {}'.format(rq))
     yield {
         'broker_url': 'filesystem://',
         'broker_transport_options': {
@@ -58,12 +57,11 @@ def celery_config():
             },
         'result_backend': 'file://{}'.format(rq),
             }
-    logg.debug('cleaning up celery filesystem backend files {} {} {}'.format(bq, bp, rq))
+    logg.debug('cleaning up celery session filesystem backend files {} {} {}'.format(bq, bp, rq))
     shutil.rmtree(bq)
     shutil.rmtree(bp)
     shutil.rmtree(rq)
 
-
 @pytest.fixture(scope='session')
 def celery_worker_parameters():
     return {

apps/cic-eth/tests/fixtures_contract.py

@@ -0,0 +1,77 @@
+# standard imports
+import os
+
+# external imports
+import pytest
+from chainlib.eth.contract import (
+        ABIContractEncoder,
+        ABIContractType,
+        )
+from chainlib.eth.nonce import RPCNonceOracle
+from chainlib.eth.gas import OverrideGasOracle
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+from chainlib.eth.tx import (
+        receipt,
+        TxFactory,
+        TxFormat,
+        unpack,
+        Tx,
+        )
+from hexathon import strip_0x
+
+script_dir = os.path.dirname(os.path.realpath(__file__))
+root_dir = os.path.dirname(script_dir)
+
+
+@pytest.fixture(scope='function')
+def bogus_tx_block(
+    default_chain_spec,
+    eth_rpc,
+    eth_signer,
+    contract_roles,
+        ):
+
+    nonce_oracle = RPCNonceOracle(contract_roles['CONTRACT_DEPLOYER'], conn=eth_rpc)
+    gas_oracle = OverrideGasOracle(limit=2000000, conn=eth_rpc)
+
+    f = open(os.path.join(script_dir, 'testdata', 'Bogus.bin'), 'r')
+    bytecode = f.read()
+    f.close()
+
+    c = TxFactory(default_chain_spec, signer=eth_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
+    tx = c.template(contract_roles['CONTRACT_DEPLOYER'], None, use_nonce=True)
+    tx = c.set_code(tx, bytecode)
+    (tx_hash_hex, o) = c.build(tx)
+
+    r = eth_rpc.do(o)
+
+    o = receipt(tx_hash_hex)
+    r = eth_rpc.do(o)
+
+    contract_address = r['contract_address']
+
+    enc = ABIContractEncoder()
+    enc.method('poke')
+    data = enc.get()
+    tx = c.template(contract_roles['CONTRACT_DEPLOYER'], contract_address, use_nonce=True)
+    tx = c.set_code(tx, data)
+    (tx_hash_hex, o) = c.finalize(tx, TxFormat.JSONRPC)
+    r = eth_rpc.do(o)
+    tx_signed_raw_hex = strip_0x(o['params'][0])
+
+    o = block_latest()
+    r = eth_rpc.do(o)
+    o = block_by_number(r, include_tx=False)
+    r = eth_rpc.do(o)
+    block = Block(r)
+    block.txs = [tx_hash_hex]
+
+    tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
+    tx_src = unpack(tx_signed_raw_bytes, default_chain_spec)
+    tx = Tx(tx_src, block=block)
+
+    return (block, tx)

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

@@ -9,8 +9,14 @@ from chainlib.eth.tx import (
         unpack,
         TxFormat,
         )
-from chainlib.eth.nonce import RPCNonceOracle
+from chainlib.eth.nonce import (
-from chainlib.eth.gas import Gas
+        RPCNonceOracle,
+        OverrideNonceOracle,
+        )
+from chainlib.eth.gas import (
+        Gas,
+        OverrideGasOracle,
+        )
 from chainlib.eth.address import to_checksum_address
 from hexathon import (
         strip_0x,
@@ -23,7 +29,15 @@ from chainqueue.db.enum import (
         StatusBits,
         status_str,
         )
-from chainqueue.query import get_tx
+from chainqueue.sql.state import (
+        set_fubar,
+        set_ready,
+        set_reserved,
+        )
+from chainqueue.sql.query import (
+        get_tx,
+        get_nonce_tx_cache,
+        )
 
 # local imports
 from cic_eth.api import AdminApi
@@ -36,150 +50,6 @@ from cic_eth.queue.tx import queue_create
 logg = logging.getLogger()
 
 
-#def test_resend_inplace(
-#    default_chain_spec,
-#    init_database,
-#    init_w3,
-#    celery_session_worker,
-#    ):
-#    
-#    chain_str = str(default_chain_spec)
-#    c = RpcClient(default_chain_spec)
-#    
-#    sigs = []
-#
-#    gas_provider = c.gas_provider()
-#
-#    s_nonce = celery.signature(
-#        'cic_eth.eth.nonce.reserve_nonce',
-#        [
-#            init_w3.eth.accounts[0],
-#            gas_provider,
-#            ],
-#        queue=None,
-#        )
-#    s_refill = celery.signature(
-#        'cic_eth.eth.gas.refill_gas',
-#        [
-#            chain_str,
-#            ],
-#        queue=None,
-#            )
-#    s_nonce.link(s_refill)
-#    t = s_nonce.apply_async()
-#    t.get()
-#    for r in t.collect():
-#        pass
-#    assert t.successful()
-#
-#    q = init_database.query(Otx)
-#    q = q.join(TxCache)
-#    q = q.filter(TxCache.recipient==init_w3.eth.accounts[0])
-#    o = q.first()
-#    tx_raw = o.signed_tx
-#
-#    tx_dict = unpack(bytes.fromhex(tx_raw), default_chain_spec)
-#    gas_price_before = tx_dict['gasPrice'] 
-#
-#    s = celery.signature(
-#        'cic_eth.admin.ctrl.lock_send',
-#        [
-#            chain_str,
-#            init_w3.eth.accounts[0],
-#            ],
-#        queue=None,
-#            )
-#    t = s.apply_async()
-#    t.get()
-#    assert t.successful()
-#
-#    api = AdminApi(c, queue=None)
-#    t = api.resend(tx_dict['hash'], chain_str, unlock=True)
-#    t.get()
-#    i = 0
-#    tx_hash_new_hex = None
-#    for r in t.collect():
-#        tx_hash_new_hex = r[1]
-#    assert t.successful()
-#  
-#    tx_raw_new = get_tx(tx_hash_new_hex) 
-#    logg.debug('get {}'.format(tx_raw_new))
-#    tx_dict_new = unpack(bytes.fromhex(tx_raw_new['signed_tx']), default_chain_spec)
-#    assert tx_hash_new_hex != tx_dict['hash']
-#    assert tx_dict_new['gasPrice'] > gas_price_before
-#
-#    tx_dict_after = get_tx(tx_dict['hash'])
-#
-#    logg.debug('logggg {}'.format(status_str(tx_dict_after['status'])))
-#    assert tx_dict_after['status'] & StatusBits.MANUAL
-
-
-#def test_check_fix_nonce(
-#    default_chain_spec,
-#    init_database,
-#    init_eth_account_roles,
-#    init_w3,
-#    eth_empty_accounts,
-#    celery_session_worker,
-#    ):
-#
-#    chain_str = str(default_chain_spec)
-#    
-#    sigs = []
-#    for i in range(5):
-#        s = celery.signature(
-#            'cic_eth.eth.gas.refill_gas',
-#            [
-#                eth_empty_accounts[i],
-#                chain_str,
-#                ],
-#            queue=None,
-#                )
-#        sigs.append(s)
-#
-#    t = celery.group(sigs)()
-#    txs = t.get()
-#    assert t.successful()
-#
-#    tx_hash = web3.Web3.keccak(hexstr=txs[2])
-#    c = RpcClient(default_chain_spec)
-#    api = AdminApi(c, queue=None)
-#    address = init_eth_account_roles['eth_account_gas_provider']
-#    nonce_spec = api.check_nonce(address)
-#    assert nonce_spec['nonce']['network'] == 0
-#    assert nonce_spec['nonce']['queue'] == 4
-#    assert nonce_spec['nonce']['blocking'] == None
-#
-#    s_set = celery.signature(
-#            'cic_eth.queue.tx.set_rejected',
-#            [
-#                tx_hash.hex(),
-#                ],
-#            queue=None,
-#            )
-#    t = s_set.apply_async()
-#    t.get()
-#    t.collect()
-#    assert t.successful()
-#
-#
-#    nonce_spec = api.check_nonce(address)
-#    assert nonce_spec['nonce']['blocking'] == 2
-#    assert nonce_spec['tx']['blocking'] == tx_hash.hex()
-#
-#    t = api.fix_nonce(address, nonce_spec['nonce']['blocking'])
-#    t.get()
-#    t.collect()
-#    assert t.successful()
-#
-#    for tx in txs[3:]:
-#        tx_hash = web3.Web3.keccak(hexstr=tx)
-#        tx_dict = get_tx(tx_hash.hex())
-#        assert tx_dict['status'] == StatusEnum.OVERRIDDEN
-#
-#
-
-
 def test_have_account(
     default_chain_spec,
     custodial_roles,
@@ -243,28 +113,6 @@ def test_tag_account(
     assert AccountRole.get_address('bar', init_database) == agent_roles['CAROL']
 
 
-#def test_ready(
-#    init_database,
-#    agent_roles,
-#    eth_rpc,
-#    ):
-#
-#    api = AdminApi(eth_rpc)
-#   
-#    with pytest.raises(InitializationError):
-#        api.ready()
-#
-#    bogus_account = os.urandom(20)
-#    bogus_account_hex = '0x' + bogus_account.hex()
-#
-#    api.tag_account('ETH_GAS_PROVIDER_ADDRESS', web3.Web3.toChecksumAddress(bogus_account_hex))
-#    with pytest.raises(KeyError):
-#        api.ready()
-#
-#    api.tag_account('ETH_GAS_PROVIDER_ADDRESS', eth_empty_accounts[0])
-#    api.ready()
-
-
 def test_tx(
     default_chain_spec,
     cic_registry,
@@ -286,3 +134,168 @@ def test_tx(
     api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['DEFAULT'])
     tx = api.tx(default_chain_spec, tx_hash=tx_hash_hex)
     logg.warning('code missing to verify tx contents {}'.format(tx))
+
+
+def test_check_nonce_gap(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        caplog,
+        ):
+
+    # NOTE: this only works as long as agents roles start at nonce 0
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 0)
+    gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
+
+    tx_hashes = []
+    txs = []
+
+    j = 0
+    for i in range(10):
+        c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+        if i == 3:
+            j = 1
+            nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], i+1)
+
+        queue_create(
+                default_chain_spec,
+                i+j,
+                agent_roles['ALICE'],
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                session=init_database,
+                )
+        cache_gas_data(
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                default_chain_spec.asdict(),
+                )
+        tx_hashes.append(tx_hash_hex)
+        txs.append(tx_signed_raw_hex)
+
+
+    init_database.commit()
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['DEFAULT'])
+    r = api.check_nonce(default_chain_spec, agent_roles['ALICE'])
+
+    assert r['nonce']['blocking'] == 4
+    assert r['tx']['blocking'] == tx_hashes[3] # one less because there is a gap
+
+
+def test_check_nonce_localfail(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        caplog,
+        ):
+
+    # NOTE: this only works as long as agents roles start at nonce 0
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 0)
+    gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
+
+    tx_hashes = []
+    txs = []
+
+    j = 0
+    for i in range(10):
+        c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+        queue_create(
+                default_chain_spec,
+                i,
+                agent_roles['ALICE'],
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                session=init_database,
+                )
+        cache_gas_data(
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                default_chain_spec.asdict(),
+                )
+        tx_hashes.append(tx_hash_hex)
+        txs.append(tx_signed_raw_hex)
+
+    set_ready(default_chain_spec, tx_hashes[4], session=init_database)
+    set_reserved(default_chain_spec, tx_hashes[4], session=init_database)
+    set_fubar(default_chain_spec, tx_hashes[4], session=init_database)
+
+    init_database.commit()
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['DEFAULT'])
+    r = api.check_nonce(default_chain_spec, agent_roles['ALICE'])
+
+    assert r['nonce']['blocking'] == 4
+    assert r['tx']['blocking'] == tx_hashes[4]
+
+
+def test_fix_nonce(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        init_celery_tasks,
+        caplog,
+        ):
+
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 0)
+    gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
+
+    tx_hashes = []
+    txs = []
+
+    for i in range(10):
+        c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+        queue_create(
+                default_chain_spec,
+                i,
+                agent_roles['ALICE'],
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                session=init_database,
+                )
+        cache_gas_data(
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                default_chain_spec.asdict(),
+                )
+        tx_hashes.append(tx_hash_hex)
+        txs.append(tx_signed_raw_hex)
+
+    init_database.commit()
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['DEFAULT'])
+    t = api.fix_nonce(default_chain_spec, agent_roles['ALICE'], 3)
+    r = t.get_leaf()
+    assert t.successful()
+
+    init_database.commit()
+    
+    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)
+        init_database.refresh(otx)
+        logg.debug('checking nonce {} tx {} status {}'.format(3, otx.tx_hash, otx.status))
+        if add_0x(k) == tx_hashes[3]:
+            assert otx.status & StatusBits.OBSOLETE == StatusBits.OBSOLETE
+        else:
+            assert otx.status == 1

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

@@ -0,0 +1,373 @@
+# standard imports
+import logging
+import io
+import json
+
+# external imports
+import pytest
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import (
+        nonce,
+        OverrideNonceOracle,
+        RPCNonceOracle,
+        )
+from chainqueue.sql.tx import create as queue_create
+from chainlib.eth.tx import (
+        raw,
+        receipt,
+        TxFormat,
+        Tx,
+        )
+from chainlib.eth.block import block_latest
+from chainlib.eth.gas import (
+        Gas,
+        OverrideGasOracle,
+        )
+from chainqueue.sql.state import (
+        set_reserved,
+        set_sent,
+        set_ready,
+        )
+from chainqueue.db.models.otx import Otx
+from chainqueue.db.enum import StatusBits
+from chainqueue.sql.query import get_nonce_tx_cache
+from eth_erc20 import ERC20
+from cic_eth_registry import CICRegistry
+
+# local imports
+from cic_eth.api.api_admin import AdminApi
+from cic_eth.eth.gas import cache_gas_data
+from cic_eth.eth.erc20 import cache_transfer_data
+
+logg = logging.getLogger()
+
+
+def test_admin_api_tx(
+        default_chain_spec,
+        init_database,
+        init_celery_tasks,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        custodial_roles,
+        celery_session_worker,
+        foo_token,
+        address_declarator,
+        cic_registry,
+        register_tokens,
+        register_lookups,
+        caplog,
+        ):
+
+    nonce_oracle = RPCNonceOracle(custodial_roles['FOO_TOKEN_GIFTER'], conn=eth_rpc)
+    gas_oracle = OverrideGasOracle(limit=100000, conn=eth_rpc)
+
+    o = nonce(custodial_roles['FOO_TOKEN_GIFTER'])
+    r = eth_rpc.do(o)
+    gifter_nonce = int(r, 16)
+
+    #c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    c = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.transfer(foo_token, custodial_roles['FOO_TOKEN_GIFTER'], agent_roles['ALICE'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+    queue_create(
+            default_chain_spec,
+            gifter_nonce, # will only work if agent starts at 0
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_transfer_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    init_database.commit()
+
+    o = raw(tx_signed_raw_hex)
+    eth_rpc.do(o)
+
+    o = receipt(tx_hash_hex)
+    r = eth_rpc.do(o)
+    assert r['status'] == 1
+
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+
+    # lookup by transaction hash, without registry
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    tx = api.tx(default_chain_spec, tx_hash=tx_hash_hex)
+    logg.debug('deployed {}'.format(contract_roles['CONTRACT_DEPLOYER']))
+    assert tx['tx_hash'] == tx_hash_hex 
+
+    # lookup by RLP transaction, without registry
+    tx = api.tx(default_chain_spec, tx_raw=tx_signed_raw_hex)
+    assert tx['tx_hash'] == tx_hash_hex 
+
+    # lookup by transaction hash, with registry
+    registry = CICRegistry(default_chain_spec, eth_rpc)
+    tx = api.tx(default_chain_spec, tx_hash=tx_hash_hex, registry=registry)
+    assert tx['tx_hash'] == tx_hash_hex 
+
+    # lookup by transaction hash, using writer
+    buf = io.StringIO()
+    api.tx(default_chain_spec, tx_hash=tx_hash_hex, renderer=json.dumps, w=buf)
+    tx = json.loads(buf.getvalue())
+    assert tx['tx_hash'] == tx_hash_hex 
+
+
+def test_admin_api_account(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        caplog,
+        ):
+
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
+
+    tx_hashes_alice = []
+    txs_alice = []
+
+    for i in range(3):
+        c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+        queue_create(
+                default_chain_spec,
+                42+i,
+                agent_roles['ALICE'],
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                session=init_database,
+                )
+        cache_gas_data(
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                default_chain_spec.asdict(),
+                )
+        tx_hashes_alice.append(tx_hash_hex)
+        txs_alice.append(tx_signed_raw_hex)
+
+    init_database.commit()
+
+    nonce_oracle = OverrideNonceOracle(agent_roles['BOB'], 13)
+    tx_hashes_bob = []
+    txs_bob = []
+
+    for i in range(2):
+        c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['BOB'], agent_roles['ALICE'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+        queue_create(
+                default_chain_spec,
+                13+i,
+                agent_roles['BOB'],
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                session=init_database,
+                )
+        cache_gas_data(
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                default_chain_spec.asdict(),
+                )
+        tx_hashes_bob.append(tx_hash_hex)
+        txs_bob.append(tx_signed_raw_hex)
+
+    init_database.commit()
+
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    r = api.account(default_chain_spec, agent_roles['ALICE'])
+    assert len(r) == 5
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    r = api.account(default_chain_spec, agent_roles['ALICE'], include_sender=False)
+    assert len(r) == 2
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    r = api.account(default_chain_spec, agent_roles['ALICE'], include_recipient=False)
+    assert len(r) == 3
+
+
+def test_admin_api_account_writer(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        caplog,
+        ):
+
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
+
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    init_database.commit()
+
+    buf = io.StringIO()
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    api.account(default_chain_spec, agent_roles['ALICE'], renderer=json.dumps, w=buf)
+
+    # TODO: improve eval
+    tx = json.loads(buf.getvalue())
+    assert tx['tx_hash'] == tx_hash_hex
+
+
+def test_registry(
+        eth_rpc,
+        cic_registry,
+        contract_roles,
+        celery_session_worker,
+        ):
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    t = api.registry()
+    r = t.get_leaf()
+    assert r == cic_registry
+
+
+def test_proxy_do(
+        default_chain_spec,
+        eth_rpc,
+        contract_roles,
+        celery_session_worker,
+        ):
+
+    o = block_latest()
+    r = eth_rpc.do(o)
+    
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    t = api.proxy_do(default_chain_spec, o)
+    rr = t.get_leaf()
+
+    assert r == rr
+
+
+def test_resend_inplace(
+        init_database,
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+
+    init_database.commit()
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    t = api.resend(tx_hash_hex, default_chain_spec, unlock=True)
+    r = t.get_leaf()
+    assert t.successful()
+
+
+    otx = Otx.load(tx_hash_hex, session=init_database)
+    assert otx.status & StatusBits.OBSOLETE == StatusBits.OBSOLETE
+
+    txs = get_nonce_tx_cache(default_chain_spec, otx.nonce, agent_roles['ALICE'], session=init_database)
+    assert len(txs) == 2
+
+
+
+@pytest.mark.xfail()
+def test_resend_clone(
+        init_database,
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        contract_roles,
+        celery_session_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+
+    init_database.commit()
+
+    api = AdminApi(eth_rpc, queue=None, call_address=contract_roles['CONTRACT_DEPLOYER'])
+    t = api.resend(tx_hash_hex, default_chain_spec, in_place=False)
+    r = t.get_leaf()
+    assert t.successful()
+
+    otx = Otx.load(tx_hash_hex, session=init_database)
+    assert otx.status & StatusBits.IN_NETWORK == StatusBits.IN_NETWORK
+    assert otx.status & StatusBits.OBSOLETE == StatusBits.OBSOLETE
+
+    txs = get_nonce_tx_cache(default_chain_spec, otx.nonce, agent_roles['ALICE'], session=init_database)
+    assert len(txs) == 1
+
+    txs = get_nonce_tx_cache(default_chain_spec, otx.nonce + 1, agent_roles['ALICE'], session=init_database)
+    assert len(txs) == 1
+
+    otx = Otx.load(txs[0], session=init_database)
+    assert otx.status == 0

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

@@ -8,11 +8,20 @@ import pytest
 import celery
 from cic_eth_registry.erc20 import ERC20Token
 from chainlib.chain import ChainSpec
+from eth_accounts_index import AccountsIndex
+from chainlib.eth.tx import (
+        transaction,
+        )
+from chainqueue.sql.state import (
+        set_reserved,
+        )
 
 # local imports
 from cic_eth.api import Api
+from cic_eth.queue.query import get_tx
 
-logg = logging.getLogger(__name__)
+#logg = logging.getLogger(__name__)
+logg = logging.getLogger()
 
 
 def test_account_api(
@@ -29,6 +38,47 @@ def test_account_api(
     assert t.successful()
 
 
+def test_account_api_register(
+        default_chain_spec,
+        init_database,
+        account_registry,
+        faucet,
+        custodial_roles,
+        cic_registry,
+        register_lookups,
+        eth_rpc,
+        celery_session_worker,
+        ):
+    api = Api(str(default_chain_spec), callback_param='accounts', callback_task='cic_eth.callbacks.noop.noop', queue=None)
+    t = api.create_account('')
+    register_tx_hash = t.get_leaf()
+    assert t.successful()
+
+    set_reserved(default_chain_spec, register_tx_hash, session=init_database)
+
+    tx = get_tx(default_chain_spec.asdict(), register_tx_hash)
+    s = celery.signature(
+            'cic_eth.eth.tx.send',
+            [
+                [tx['signed_tx']],
+                default_chain_spec.asdict(),
+                ],
+            queue=None
+            )
+    t = s.apply_async()
+    r = t.get_leaf()
+    assert t.successful()
+
+    o = transaction(register_tx_hash)
+    tx_src = eth_rpc.do(o)
+
+    c = AccountsIndex(default_chain_spec)
+    address = c.parse_add_request(tx_src['data'])
+    o = c.have(account_registry, address[0], sender_address=custodial_roles['CONTRACT_DEPLOYER'])
+    r = eth_rpc.do(o)
+    assert c.parse_have(r)
+
+
 def test_transfer_api(
         default_chain_spec,
         eth_rpc,
@@ -37,16 +87,15 @@ def test_transfer_api(
         custodial_roles,
         agent_roles,
         cic_registry,
-        register_tokens,
+        token_registry,
         register_lookups,
         celery_session_worker,
+        register_tokens,
+        foo_token_symbol,
         ):
 
-    #token = CICRegistry.get_address(default_chain_spec, bancor_tokens[0])
-    foo_token_cache = ERC20Token(default_chain_spec, eth_rpc, foo_token)
-
     api = Api(str(default_chain_spec), callback_param='transfer', callback_task='cic_eth.callbacks.noop.noop', queue=None)
-    t = api.transfer(custodial_roles['FOO_TOKEN_GIFTER'], agent_roles['ALICE'], 1024, foo_token_cache.symbol)
+    t = api.transfer(custodial_roles['FOO_TOKEN_GIFTER'], agent_roles['ALICE'], 1, foo_token_symbol) 
     t.get_leaf()
     assert t.successful()
 

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

@@ -0,0 +1,19 @@
+# local imports
+from cic_eth.api.api_task import Api
+from cic_eth.task import BaseTask
+
+def test_default_token(
+        default_chain_spec,
+        foo_token,
+        default_token,
+        token_registry,
+        register_tokens,
+        register_lookups,
+        cic_registry,
+        celery_session_worker,
+        ):
+
+    api = Api(str(default_chain_spec), queue=None)     
+    t = api.default_token()
+    r = t.get_leaf()
+    assert r['address'] == foo_token

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

@@ -4,7 +4,7 @@ import logging
 # external imports
 import pytest
 from chainlib.eth.nonce import RPCNonceOracle
-from chainlib.eth.erc20 import ERC20
+from eth_erc20 import ERC20
 from chainlib.eth.tx import receipt
 
 # local imports

apps/cic-eth/tests/task/test_task_account.py

@@ -9,7 +9,7 @@ import celery
 from chainlib.connection import RPCConnection
 from chainlib.eth.nonce import RPCNonceOracle
 from chainlib.eth.tx import receipt
-from eth_accounts_index import AccountRegistry
+from eth_accounts_index.registry import AccountRegistry
 from hexathon import strip_0x
 from chainqueue.db.enum import StatusEnum
 from chainqueue.db.models.otx import Otx
@@ -156,6 +156,7 @@ def test_gift(
     eth_signer,
     init_celery_tasks,
     cic_registry,
+    register_lookups,
     celery_session_worker,
     ):
 

apps/cic-eth/tests/task/test_task_admin.py

@@ -0,0 +1,88 @@
+# standard imports
+import logging
+
+# external imports
+import celery
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import OverrideNonceOracle
+from chainqueue.sql.tx import (
+        create as queue_create,
+        )
+from chainlib.eth.gas import (
+        Gas,
+        OverrideGasOracle,
+        )
+from chainlib.eth.tx import TxFormat
+from chainqueue.sql.query import get_nonce_tx_cache
+from chainqueue.db.models.otx import Otx
+from chainqueue.db.enum import StatusBits
+from hexathon import add_0x
+
+# local imports
+from cic_eth.admin.nonce import shift_nonce
+from cic_eth.eth.gas import cache_gas_data
+
+logg = logging.getLogger()
+
+
+def test_shift_nonce(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        celery_session_worker,
+        caplog,
+        ):
+
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
+
+    tx_hashes = []
+    txs = []
+
+    for i in range(10):
+        c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+        queue_create(
+                default_chain_spec,
+                42+i,
+                agent_roles['ALICE'],
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                session=init_database,
+                )
+        cache_gas_data(
+                tx_hash_hex,
+                tx_signed_raw_hex,
+                default_chain_spec.asdict(),
+                )
+        tx_hashes.append(tx_hash_hex)
+        txs.append(tx_signed_raw_hex)
+
+    init_database.commit()
+
+    s = celery.signature(
+            'cic_eth.admin.nonce.shift_nonce',
+            [
+                default_chain_spec.asdict(),
+                tx_hashes[3],
+                ],
+            queue=None
+            )
+    t = s.apply_async()
+    r = t.get_leaf()
+    assert t.successful()
+    init_database.commit()
+
+
+    for i in range(42+3, 42+10):
+        txs = get_nonce_tx_cache(default_chain_spec, i, agent_roles['ALICE'], session=init_database)
+        for k in txs.keys():
+            hsh = add_0x(k)
+            otx = Otx.load(hsh, session=init_database)
+            logg.debug('checking nonce {} tx {} status {}'.format(i, otx.tx_hash, otx.status))
+            if add_0x(k) == tx_hashes[i-42]:
+                assert otx.status & StatusBits.OBSOLETE == StatusBits.OBSOLETE
+            else:
+                assert otx.status == 1

apps/cic-eth/tests/task/test_task_erc20.py

@@ -4,7 +4,7 @@ import logging
 # external imports
 import pytest
 import celery
-from chainlib.eth.erc20 import ERC20
+from eth_erc20 import ERC20
 from chainlib.eth.nonce import RPCNonceOracle
 from chainlib.eth.tx import (
         receipt,

apps/cic-eth/tests/task/test_task_gas.py

@@ -0,0 +1,286 @@
+# standard imports
+import logging
+
+# external imports
+import celery
+import pytest
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import (
+        OverrideNonceOracle,
+        RPCNonceOracle,
+        )
+from chainlib.eth.gas import (
+        OverrideGasOracle,
+        Gas,
+        )
+from chainlib.eth.tx import (
+        unpack,
+        TxFormat,
+        )
+from chainlib.eth.constant import (
+        MINIMUM_FEE_UNITS,
+        MINIMUM_FEE_PRICE,
+        )
+from chainqueue.sql.tx import create as queue_create
+from chainqueue.sql.query import get_tx
+from chainqueue.db.enum import StatusBits
+from chainqueue.sql.state import (
+        set_ready,
+        set_reserved,
+        set_sent,
+        )
+from chainqueue.db.models.otx import Otx
+from hexathon import strip_0x
+
+# local imports
+from cic_eth.eth.gas import cache_gas_data
+from cic_eth.error import OutOfGasError
+
+logg = logging.getLogger()
+
+
+def test_task_check_gas_ok(
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        init_database,
+        agent_roles,
+        custodial_roles,
+        celery_session_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], conn=eth_rpc) 
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            0,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    init_database.commit()
+
+    s = celery.signature(
+            'cic_eth.eth.gas.check_gas',
+            [
+                [
+                    tx_hash_hex,
+                    ],
+                default_chain_spec.asdict(),
+                [],
+                None,
+                8000000,
+                ],
+            queue=None
+            )
+    t = s.apply_async()
+    t.get_leaf()
+    assert t.successful()
+
+    init_database.commit()
+
+    tx = get_tx(default_chain_spec, tx_hash_hex, session=init_database)
+    assert tx['status'] & StatusBits.QUEUED == StatusBits.QUEUED
+
+
+def test_task_check_gas_insufficient(
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        init_database,
+        agent_roles,
+        custodial_roles,
+        celery_session_worker,
+        whoever,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = OverrideNonceOracle(whoever, 42)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(whoever, agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            42,
+            whoever,
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    init_database.commit()
+
+    s = celery.signature(
+            'cic_eth.eth.gas.check_gas',
+            [
+                [
+                    tx_hash_hex,
+                    ],
+                default_chain_spec.asdict(),
+                [],
+                None,
+                None,
+                ],
+            queue=None
+            )
+    t = s.apply_async()
+    try:
+        r = t.get_leaf()
+    except OutOfGasError:
+        pass
+
+    init_database.commit()
+
+    tx = get_tx(default_chain_spec, tx_hash_hex, session=init_database)
+    assert tx['status'] & StatusBits.GAS_ISSUES == StatusBits.GAS_ISSUES
+
+
+def test_task_check_gas_low(
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        init_database,
+        agent_roles,
+        custodial_roles,
+        celery_session_worker,
+        whoever,
+        ):
+
+    gas_oracle = OverrideGasOracle(price=MINIMUM_FEE_PRICE, limit=MINIMUM_FEE_UNITS)
+    nonce_oracle = RPCNonceOracle(custodial_roles['GAS_GIFTER'], conn=eth_rpc)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = c.create(custodial_roles['GAS_GIFTER'], whoever, 100 * (10 ** 6))
+    r = eth_rpc.do(o)
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(whoever, conn=eth_rpc) 
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(whoever, agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            0,
+            whoever,
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    init_database.commit()
+
+    s = celery.signature(
+            'cic_eth.eth.gas.check_gas',
+            [
+                [
+                    tx_hash_hex,
+                    ],
+                default_chain_spec.asdict(),
+                ],
+                [],
+                None,
+                None,
+            queue=None
+            )
+    t = s.apply_async()
+    t.get_leaf()
+    assert t.successful()
+
+    init_database.commit()
+
+    tx = get_tx(default_chain_spec, tx_hash_hex, session=init_database)
+    assert tx['status'] & StatusBits.QUEUED == StatusBits.QUEUED
+
+
+@pytest.mark.parametrize(
+        '_gas_price,_gas_factor',
+        [
+        (None, 1.1),
+        (MINIMUM_FEE_PRICE * 1.1, 0.9),
+        (None, 1.3),
+        ]
+        )
+def test_task_resend_explicit(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        custodial_roles,
+        celery_session_worker,
+        _gas_price,
+        _gas_factor,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], conn=eth_rpc) 
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            0,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+    tx_before = unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), default_chain_spec)
+
+    init_database.commit()
+
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+
+    s = celery.signature(
+            'cic_eth.eth.gas.resend_with_higher_gas',
+            [
+                tx_hash_hex,
+                default_chain_spec.asdict(),
+                _gas_price,
+                _gas_factor,
+                ],
+            queue=None
+            )
+    t = s.apply_async()
+    r = t.get_leaf()
+    assert t.successful()
+
+    q = init_database.query(Otx)
+    q = q.filter(Otx.tx_hash==strip_0x(r))
+    otx = q.first()
+    if otx == None:
+        raise NotLocalTxError(r)
+
+    tx_after = unpack(bytes.fromhex(strip_0x(otx.signed_tx)), default_chain_spec)
+    logg.debug('gasprices before {} after {}'.format(tx_before['gasPrice'], tx_after['gasPrice']))
+    assert tx_after['gasPrice'] > tx_before['gasPrice']
+

apps/cic-eth/tests/task/test_task_tx.py

@@ -4,16 +4,27 @@ import logging
 # external imports
 import pytest
 import celery
-from chainlib.eth.gas import Gas
+from chainlib.eth.gas import (
+        OverrideGasOracle,
+        Gas,
+        )
 from chainlib.eth.nonce import RPCNonceOracle
 from chainlib.eth.tx import (
         TxFormat,
         unpack,
         transaction,
         receipt,
+        raw,
         )
 from hexathon import strip_0x
 from chainqueue.db.models.otx import Otx
+from chainqueue.sql.tx import create as queue_create
+from chainqueue.sql.state import (
+        set_reserved,
+        set_ready,
+        set_sent,
+        )
+from chainqueue.db.enum import StatusBits
 
 # local imports
 from cic_eth.queue.tx import register_tx
@@ -60,15 +71,6 @@ def test_tx_send(
 
 
 def test_sync_tx(
-        default_chain_spec,
-        eth_rpc,
-        eth_signer,
-        celery_session_worker,
-        ):
-    pass
-
-
-def test_resend_with_higher_gas(
         init_database,
         default_chain_spec,
         eth_rpc,
@@ -77,31 +79,48 @@ def test_resend_with_higher_gas(
         celery_session_worker,
         ):
 
-    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], conn=eth_rpc)
-    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
-    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 1024, tx_format=TxFormat.RLP_SIGNED)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
-    register_tx(tx_hash_hex, tx_signed_raw_hex, default_chain_spec, None, session=init_database)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
-    cache_gas_data(tx_hash_hex, tx_signed_raw_hex, default_chain_spec.asdict())
+
-    tx_before = unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), default_chain_spec)
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+
+    o = raw(tx_signed_raw_hex)
+    r = eth_rpc.do(o)
+
+    o = receipt(tx_hash_hex)
+    r = eth_rpc.do(o)
+    assert r['status'] == 1
 
     s = celery.signature(
-            'cic_eth.eth.gas.resend_with_higher_gas',
+            'cic_eth.eth.tx.sync_tx',
             [
                 tx_hash_hex,
                 default_chain_spec.asdict(),
                 ],
-            queue=None,
+            queue=None
             )
     t = s.apply_async()
     r = t.get_leaf()
+    assert t.successful() 
 
-    q = init_database.query(Otx)
+    init_database.commit()
-    q = q.filter(Otx.tx_hash==strip_0x(r))
-    otx = q.first()
-    if otx == None:
-        raise NotLocalTxError(r)
-
-    tx_after = unpack(bytes.fromhex(strip_0x(otx.signed_tx)), default_chain_spec)
-    logg.debug('gasprices before {} after {}'.format(tx_before['gasPrice'], tx_after['gasPrice']))
-    assert tx_after['gasPrice'] > tx_before['gasPrice']
 
+    o = Otx.load(tx_hash_hex, session=init_database)
+    assert o.status & StatusBits.FINAL == StatusBits.FINAL

apps/cic-eth/tests/task/test_task_tx_misc.py

@@ -0,0 +1,170 @@
+# standard imports
+import os
+import logging
+
+# external imports
+import pytest
+import celery
+from chainqueue.sql.tx import create as queue_create
+from chainlib.eth.nonce import (
+        RPCNonceOracle,
+        OverrideNonceOracle,
+        )
+from chainlib.eth.gas import (
+        OverrideGasOracle,
+        Gas,
+        )
+from chainlib.eth.tx import (
+        TxFormat,
+        unpack,
+        receipt,
+        )
+from hexathon import (
+        add_0x,
+        strip_0x,
+        )
+from chainqueue.sql.state import (
+        set_reserved,
+        set_ready,
+        )
+
+logg = logging.getLogger()
+
+
+def test_hashes_to_txs(
+        init_database,
+        default_chain_spec,
+        agent_roles,            
+        eth_rpc,
+        eth_signer,
+        celery_session_worker,
+        ):
+
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex_one, tx_signed_raw_hex_one) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex_one,
+            tx_signed_raw_hex_one,
+            session=init_database,
+            )
+
+    #nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 43)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex_two, tx_signed_raw_hex_two) = c.create(agent_roles['ALICE'], agent_roles['CAROL'], 200 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            43,
+            agent_roles['ALICE'],
+            tx_hash_hex_two,
+            tx_signed_raw_hex_two,
+            session=init_database,
+            )
+
+    init_database.commit()
+
+    bogus_one = add_0x(os.urandom(32).hex())
+    bogus_two = add_0x(os.urandom(32).hex())
+
+    yarrgs = [
+                bogus_one,
+                tx_hash_hex_two,
+                bogus_two,
+                tx_hash_hex_one,
+            ]   
+    s = celery.signature(
+        'cic_eth.eth.tx.hashes_to_txs',
+        [
+            yarrgs,
+                ],
+        queue=None,
+        )
+    t = s.apply_async()
+    r = t.get_leaf()
+    assert t.successful()
+    assert len(r) == 2
+
+    logg.debug('r {}'.format(r))
+    txs = [
+        tx_signed_raw_hex_two,
+        tx_signed_raw_hex_one,
+            ]
+    for tx in r:
+        txs.remove(add_0x(tx))
+    assert len(txs) == 0
+
+
+
+def test_double_send(
+        init_database,
+        default_chain_spec,
+        agent_roles,            
+        eth_rpc,
+        eth_signer,
+        celery_session_worker,
+        ):
+
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], conn=eth_rpc)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex_one, tx_signed_raw_hex_one) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex_one,
+            tx_signed_raw_hex_one,
+            session=init_database,
+            )
+    set_ready(default_chain_spec, tx_hash_hex_one, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex_one, session=init_database)
+
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex_two, tx_signed_raw_hex_two) = c.create(agent_roles['ALICE'], agent_roles['CAROL'], 200 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+
+    queue_create(
+            default_chain_spec,
+            43,
+            agent_roles['ALICE'],
+            tx_hash_hex_two,
+            tx_signed_raw_hex_two,
+            session=init_database,
+            )
+
+    set_ready(default_chain_spec, tx_hash_hex_two, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex_two, session=init_database)
+    init_database.commit()
+
+    yarrgs = [
+        tx_signed_raw_hex_one,
+        tx_signed_raw_hex_two,
+            ]
+    s = celery.signature(
+            'cic_eth.eth.tx.send',
+            [
+                yarrgs,
+                default_chain_spec.asdict(),
+                ],
+            queue=None
+            )
+    t = s.apply_async()
+    r = t.get_leaf()
+    assert t.successful()
+
+    o = receipt(tx_hash_hex_one)
+    r = eth_rpc.do(o)
+    assert r['status'] == 1
+
+    o = receipt(tx_hash_hex_two)
+    r = eth_rpc.do(o)
+    assert r['status'] == 1
+
+
+

apps/cic-eth/tests/testdata/Bogus.bin

@@ -0,0 +1 @@
+60806040526000805534801561001457600080fd5b50610181806100246000396000f3fe608060405234801561001057600080fd5b5060043610610053576000357c0100000000000000000000000000000000000000000000000000000000900480630dbe671f146100585780631817835814610076575b600080fd5b610060610080565b60405161006d91906100ae565b60405180910390f35b61007e610086565b005b60005481565b600080815480929190610098906100d3565b9190505550565b6100a8816100c9565b82525050565b60006020820190506100c3600083018461009f565b92915050565b6000819050919050565b60006100de826100c9565b91507fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff8214156101115761011061011c565b5b600182019050919050565b7f4e487b7100000000000000000000000000000000000000000000000000000000600052601160045260246000fdfea264697066735822122034ad8e91e864f030d47f5b93e281869206c1b203c36dc79a209ac9c9c16e577564736f6c63430008040033

apps/cic-eth/tests/testdata/Bogus.sol

@@ -0,0 +1,10 @@
+pragma solidity ^0.8.0;
+
+contract Bogus {
+
+	uint256 public a = 0;
+
+	function poke() public {
+		a++;
+	}
+}

apps/cic-eth/tests/unit/ext/test_address.py

@@ -3,7 +3,7 @@ from chainlib.eth.nonce import RPCNonceOracle
 from chainlib.eth.tx import (
         receipt,
         )
-from eth_address_declarator import AddressDeclarator
+from eth_address_declarator import Declarator
 from hexathon import add_0x
 
 # local imports
@@ -19,11 +19,12 @@ def test_translate(
         agent_roles,
         cic_registry,
         init_celery_tasks,
+        register_lookups,
     ):
 
     nonce_oracle = RPCNonceOracle(contract_roles['CONTRACT_DEPLOYER'], eth_rpc)
 
-    c = AddressDeclarator(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle)
+    c = Declarator(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle)
 
     description = 'alice'.encode('utf-8').ljust(32, b'\x00').hex()
     (tx_hash_hex, o) = c.add_declaration(address_declarator, contract_roles['CONTRACT_DEPLOYER'], agent_roles['ALICE'], add_0x(description))

apps/cic-eth/tests/unit/ext/test_ext_tx.py

@@ -8,7 +8,7 @@ from chainlib.eth.tx import (
         count,
         receipt,
         )
-from chainlib.eth.erc20 import ERC20
+from eth_erc20 import ERC20
 from chainlib.eth.nonce import RPCNonceOracle
 
 # local imports

apps/cic-eth/tests/unit/test_registry_connect.py

@@ -0,0 +1,22 @@
+# local imports
+from cic_eth.registry import *
+
+def test_registry_connect(
+        eth_rpc,
+        default_chain_spec,
+        address_declarator,
+        token_registry,
+        contract_roles,
+        purge_lookups,
+        registry,
+        agent_roles,
+        ):
+
+    r = connect(eth_rpc, default_chain_spec, registry, sender_address=contract_roles['CONTRACT_DEPLOYER'])
+
+    connect_declarator(eth_rpc, default_chain_spec, [agent_roles['ALICE']], sender_address=contract_roles['CONTRACT_DEPLOYER'])
+    r.by_name('AddressDeclarator', sender_address=contract_roles['CONTRACT_DEPLOYER'])
+
+    connect_token_registry(eth_rpc, default_chain_spec, sender_address=contract_roles['CONTRACT_DEPLOYER'])
+    r.by_name('TokenRegistry', sender_address=contract_roles['CONTRACT_DEPLOYER'])
+

apps/cic-eth/tests/unit/test_stat.py

@@ -0,0 +1,18 @@
+# standard imports
+import datetime
+
+# local imports
+from cic_eth.stat import init_chain_stat
+
+
+def test_chain_stat(
+        eth_rpc,
+        init_eth_tester,
+        ):
+
+    now = int(datetime.datetime.now().timestamp()) + 1
+    for i in range(11):
+        init_eth_tester.time_travel(now + (i * 2))
+    
+    s = init_chain_stat(eth_rpc, block_start=0)
+    assert s.block_average() == 2

apps/cic-meta/.gitignore

@@ -3,3 +3,5 @@ dist
 dist-web
 dist-server
 scratch
+coverage
+.nyc_output

apps/cic-meta/.gitlab-ci.yml

@@ -3,17 +3,38 @@
     variables:
         APP_NAME: cic-meta
         DOCKERFILE_PATH: $APP_NAME/docker/Dockerfile
+        IMAGE_TAG: $CI_REGISTRY_IMAGE/$APP_NAME:unittest-$CI_COMMIT_SHORT_SHA
 
 .cic_meta_changes_target:
     rules:
-        - changes:
+        - if: $CI_PIPELINE_SOURCE == "merge_request_event"
-            - $CONTEXT/$APP_NAME/*
+#        - changes:
+#            - $CONTEXT/$APP_NAME/*
+        - when: always
 
-build-mr-cic-meta:
+cic-meta-build-mr:
+  stage: build
+  extends:
+    - .cic_meta_variables
+    - .cic_meta_changes_target
+  script:
+    - mkdir -p /kaniko/.docker
+    - echo "{\"auths\":{\"$CI_REGISTRY\":{\"username\":\"$CI_REGISTRY_USER\",\"password\":\"$CI_REGISTRY_PASSWORD\"}}}" > "/kaniko/.docker/config.json"
+    # - /kaniko/executor --context $CONTEXT --dockerfile $DOCKERFILE_PATH $KANIKO_CACHE_ARGS --destination $IMAGE_TAG
+    - /kaniko/executor --context $CONTEXT --dockerfile $DOCKERFILE_PATH $KANIKO_CACHE_ARGS --destination $IMAGE_TAG 
+
+test-mr-cic-meta:
     extends:
-        - .cic_meta_changes_target
-        - .py_build_merge_request
         - .cic_meta_variables
+        - .cic_meta_changes_target
+    stage: test
+    image: $IMAGE_TAG 
+    script:
+        - cd /tmp/src/cic-meta
+        - npm install --dev 
+        - npm run test
+        - npm run test:coverage
+    needs: ["cic-meta-build-mr"]
 
 build-push-cic-meta:
     extends:

apps/cic-meta/bin/get.js

@@ -0,0 +1,51 @@
+#!/usr/bin/env node
+const colors = require('colors');
+const {Meta} = require("../dist");
+
+let { argv } = require('yargs')
+    .usage('Usage: $0 -m http://localhost:63380 -n publickeys')
+    .example(
+        '$0 -m http://localhost:63380 -n publickeys',
+        'Fetches the public keys blob from the meta server'
+    )
+    .option('m', {
+        alias: 'metaurl',
+        describe: 'The URL for the meta service',
+        demandOption: 'The meta url is required',
+        type: 'string',
+        nargs: 1,
+    })
+    .option('n', {
+        alias: 'name',
+        describe: 'The name of the resource to be fetched from the meta service',
+        demandOption: 'The name of the resource is required',
+        type: 'string',
+        nargs: 1,
+    })
+    .option('t', {
+        alias: 'type',
+        describe: 'The type of resource to be fetched from the meta service\n' +
+            'Options: `user`, `phone` and `custom`\n' +
+            'Defaults to `custom`',
+        type: 'string',
+        nargs: 1,
+    })
+    .epilog('Grassroots Economics (c) 2021')
+    .wrap(null);
+
+const metaUrl = argv.m;
+const resourceName = argv.n;
+let type = argv.t;
+if (type === undefined) {
+    type = 'custom'
+}
+
+(async () => {
+    const identifier = await Meta.getIdentifier(resourceName, type);
+    console.log(colors.cyan(`Meta server storage identifier: ${identifier}`));
+    const metaResponse = await Meta.get(identifier, metaUrl);
+    if (typeof metaResponse !== "object") {
+        console.error(colors.red('Metadata get failed!'));
+    }
+    console.log(colors.green(metaResponse));
+})();

apps/cic-meta/bin/set.js

@@ -0,0 +1,81 @@
+#!/usr/bin/env node
+const fs = require("fs");
+const colors = require('colors');
+const {Meta} = require("../dist");
+
+let { argv } = require('yargs')
+    .usage('Usage: $0 -m http://localhost:63380 -k ./privatekeys.asc -n publickeys -r ./publickeys.asc')
+    .example(
+        '$0 -m http://localhost:63380 -k ./privatekeys.asc -n publickeys -r ./publickeys.asc',
+        'Updates the public keys blob to the meta server'
+    )
+    .option('m', {
+        alias: 'metaurl',
+        describe: 'The URL for the meta service',
+        demandOption: 'The meta url is required',
+        type: 'string',
+        nargs: 1,
+    })
+    .option('k', {
+        alias: 'privatekey',
+        describe: 'The PGP private key blob file used to sign the changes to the meta service',
+        demandOption: 'The private key file is required',
+        type: 'string',
+        nargs: 1,
+    })
+    .option('n', {
+        alias: 'name',
+        describe: 'The name of the resource to be set or updated to the meta service',
+        demandOption: 'The name of the resource is required',
+        type: 'string',
+        nargs: 1,
+    })
+    .option('r', {
+        alias: 'resource',
+        describe: 'The resource file to be set or updated to the meta service',
+        demandOption: 'The resource file is required',
+        type: 'string',
+        nargs: 1,
+    })
+    .option('t', {
+        alias: 'type',
+        describe: 'The type of resource to be set or updated to the meta service\n' +
+            'Options: `user`, `phone` and `custom`\n' +
+            'Defaults to `custom`',
+        type: 'string',
+        nargs: 1,
+    })
+    .epilog('Grassroots Economics (c) 2021')
+    .wrap(null);
+
+const metaUrl = argv.m;
+const privateKeyFile = argv.k;
+const resourceName = argv.n;
+const resourceFile = argv.r;
+let type = argv.t;
+if (type === undefined) {
+    type = 'custom'
+}
+
+const privateKey = readFile(privateKeyFile);
+const resource = readFile(resourceFile);
+
+(async () => {
+    if (privateKey && resource) {
+        const identifier = await Meta.getIdentifier(resourceName, type);
+        console.log(colors.cyan(`Meta server storage identifier: ${identifier}`));
+        const meta = new Meta(metaUrl, privateKey);
+        meta.onload = async (status) => {
+            const response = await meta.set(identifier, resource)
+            console.log(colors.green(response));
+        }
+    }
+})();
+
+function readFile(filename) {
+    if(!fs.existsSync(filename)) {
+        console.log(colors.red(`File ${filename} not found`));
+        return;
+    }
+    return fs.readFileSync(filename, {encoding: 'utf8', flag: 'r'});
+}

apps/cic-meta/docker/Dockerfile

@@ -4,29 +4,28 @@ WORKDIR /tmp/src/cic-meta
 
 RUN apk add --no-cache postgresql bash
 
-COPY cic-meta/package.json \
+# required to build the cic-client-meta module
-     ./
-
 COPY cic-meta/src/ src/
-COPY cic-meta/tests/ tests/
 COPY cic-meta/scripts/ scripts/
 
+# copy the dependencies
+COPY cic-meta/package.json .
+COPY cic-meta/tsconfig.json . 
+COPY cic-meta/webpack.config.js .
+     
 RUN npm install
 
-# see exports_dir gpg.ini
+COPY cic-meta/tests/ tests/
 COPY cic-meta/tests/*.asc /root/pgp/
-RUN alias tsc=node_modules/typescript/bin/tsc 
-
 
+# copy runtime configs
 COPY cic-meta/.config/ /usr/local/etc/cic-meta/
-# COPY cic-meta/scripts/server/initdb/server.postgres.sql /usr/local/share/cic-meta/sql/server.sql
 
+# db migrations
 COPY cic-meta/docker/db.sh ./db.sh
 RUN chmod 755 ./db.sh
 
-#RUN alias ts-node=/tmp/src/cic-meta/node_modules/ts-node/dist/bin.js
+RUN alias tsc=node_modules/typescript/bin/tsc 
-#ENTRYPOINT [ "./node_modules/ts-node/dist/bin.js", "./scripts/server/server.ts" ]
-
 COPY cic-meta/docker/start_server.sh ./start_server.sh
 RUN chmod 755 ./start_server.sh
 ENTRYPOINT ["sh", "./start_server.sh"]

apps/cic-meta/package.json

@@ -1,22 +1,30 @@
 {
-	"name": "cic-client-meta",
+	"name": "@cicnet/cic-client-meta",
-	"version": "0.0.7-alpha.8",
+	"version": "0.0.11",
 	"description": "Signed CRDT metadata graphs for the CIC network",
 	"main": "dist/index.js",
 	"types": "dist/index.d.ts",
+	"bin": {
+		"meta-set": "bin/set.js",
+		"meta-get": "bin/get.js"
+	},
+	"preferGlobal": true,
 	"scripts": {
 		"test": "mocha -r node_modules/node-localstorage/register -r ts-node/register tests/*.ts",
+		"test:coverage": "nyc mocha tests/*.ts --timeout 3000 --check-coverage=true",
 		"build": "node_modules/typescript/bin/tsc -d --outDir dist src/index.ts",
 		"build-server": "tsc -d --outDir dist-server scripts/server/*.ts",
 		"pack": "node_modules/typescript/bin/tsc -d --outDir dist && webpack",
 		"clean": "rm -rf dist",
 		"prepare": "npm run build && npm run build-server",
-                "start": "./node_modules/ts-node/dist/bin.js ./scripts/server/server.ts" 
+		"start": "./node_modules/ts-node/dist/bin.js ./scripts/server/server.ts",
+		"publish": "npm publish --access public"
 	},
 	"dependencies": {
+		"@cicnet/crdt-meta": "^0.0.10",
 		"@ethereumjs/tx": "^3.0.0-beta.1",
 		"automerge": "^0.14.1",
-		"crdt-meta": "0.0.8",
+		"colors": "^1.4.0",
 		"ethereumjs-wallet": "^1.0.1",
 		"ini": "^1.3.8",
 		"openpgp": "^4.10.8",
@@ -27,7 +35,9 @@
 	"devDependencies": {
 		"@types/mocha": "^8.0.3",
 		"mocha": "^8.2.0",
+		"nock": "^13.1.0",
 		"node-localstorage": "^2.1.6",
+		"nyc": "^15.1.0",
 		"ts-node": "^9.0.0",
 		"typescript": "^4.0.5",
 		"webpack": "^5.4.0",
@@ -43,5 +53,26 @@
 	"license": "GPL-3.0-or-later",
 	"engines": {
 		"node": ">=14.16.1"
+	},
+	"nyc": {
+		"include": [
+			"src/**/*.ts"
+		],
+		"extension": [
+			".ts"
+		],
+		"require": [
+			"ts-node/register"
+		],
+		"reporter": [
+			"text",
+			"html"
+		],
+		"sourceMap": true,
+		"instrument": true,
+		"branches": ">80",
+		"lines": ">80",
+		"functions": ">80",
+		"statements": ">80"
 	}
 }

apps/cic-meta/scripts/dumpconfig.js

@@ -1,4 +1,4 @@
-import { Config } from 'crdt-meta';
+import { Config } from '@cicnet/crdt-meta';
 const fs = require('fs');
 
 if (process.argv[2] === undefined) {

apps/cic-meta/scripts/server/handlers.ts

@@ -1,7 +1,7 @@
 import * as Automerge from 'automerge';
 import * as pgp from 'openpgp';
 
-import { Envelope, Syncable } from 'crdt-meta';
+import { Envelope, Syncable } from '@cicnet/crdt-meta';
 
 
 function handleNoMergeGet(db, digest, keystore) {
@@ -31,7 +31,7 @@ function handleNoMergeGet(db, digest, keystore) {
 					doh(e);
 				});
 			}).catch((e) => {
-				console.error('mesage', e);
+				console.error('message', e);
 				doh(e);
 			});
 		})
@@ -46,7 +46,7 @@ function handleServerMergePost(data, db, digest, keystore, signer) {
 			let e = undefined;
 			let s = undefined;
 			if (v === undefined) {
-				s = new Syncable(digest, data);
+				s = new Syncable(digest, o);
 				s.onwrap = (e) => {
 					whohoo(e.toJSON());
 				};

apps/cic-meta/scripts/server/server.ts

@@ -3,7 +3,8 @@ import * as fs from 'fs';
 import * as path from 'path';
 
 import * as handlers from './handlers';
-import { PGPKeyStore, PGPSigner, Config, SqliteAdapter, PostgresAdapter } from 'crdt-meta';
+import { PGPKeyStore, PGPSigner, Config } from '@cicnet/crdt-meta';
+import { SqliteAdapter, PostgresAdapter } from '../../src/db';
 
 import { standardArgs } from './args';
 
@@ -203,7 +204,7 @@ async function processRequest(req, res) {
 		}
 
 		if (content === undefined) {
-			console.error('empty onctent', data);
+			console.error('empty content', data);
 			res.writeHead(400, {"Content-Type": "text/plain"});
 			res.end();
 			return;

apps/cic-meta/src/custom.ts

@@ -0,0 +1,27 @@
+import {Addressable, mergeKey, Syncable} from "@cicnet/crdt-meta";
+
+class Custom extends Syncable implements Addressable {
+
+    name:	string
+    value:	Object
+
+    constructor(name:string, v:Object={}) {
+        super('', v);
+        Custom.toKey(name).then((cid) => {
+            this.id = cid;
+            this.name = name;
+        });
+    }
+
+    public static async toKey(item:string, identifier: string = ':cic.custom') {
+        return await mergeKey(Buffer.from(item), Buffer.from(identifier));
+    }
+
+    public key(): string {
+        return this.id;
+    }
+}
+
+export {
+    Custom,
+}

apps/cic-meta/src/db.ts

@@ -0,0 +1,90 @@
+import * as pg from 'pg';
+import * as sqlite from 'sqlite3';
+
+type DbConfig = {
+	name:		string
+	host:		string
+	port:		number
+	user:		string
+	password:	string	
+}
+
+interface DbAdapter {
+	query:	(s:string, callback:(e:any, rs:any) => void) => void
+	close: 	() => void
+}
+
+const re_creatematch = /^(CREATE)/i
+const re_getmatch = /^(SELECT)/i;
+const re_setmatch = /^(INSERT|UPDATE)/i;
+
+class SqliteAdapter implements DbAdapter {
+
+	db:		any
+
+	constructor(dbConfig:DbConfig, callback?:(any) => void) {
+		this.db = new sqlite.Database(dbConfig.name); //, callback);
+	}
+
+	public query(s:string, callback:(e:any, rs?:any) => void): void {
+		const local_callback = (e, rs) => {
+			let r = undefined;
+			if (rs !== undefined) {
+				r = {
+					rowCount: rs.length,
+					rows: rs,
+				}
+			}
+			callback(e, r);
+		};
+		if (s.match(re_getmatch)) {
+			this.db.all(s, local_callback);
+		} else if (s.match(re_setmatch)) {
+			this.db.run(s, local_callback);
+		} else if (s.match(re_creatematch)) {
+			this.db.run(s, callback);
+		} else {
+			throw 'unhandled query';
+		}
+	}
+
+	public close() {
+		this.db.close();
+	}
+}
+
+class PostgresAdapter implements DbAdapter {
+
+	db:		any
+
+	constructor(dbConfig:DbConfig) {
+		let o = dbConfig;
+		o['database'] = o.name;
+		this.db = new pg.Pool(o);
+		return this.db;
+	}
+
+	public query(s:string, callback:(e:any, rs:any) => void): void {
+		this.db.query(s, (e, rs) => {
+			let r = {
+				length: rs.rowCount,
+			}
+			rs.length = rs.rowCount;
+			if (e === undefined) {
+				e = null;
+			}
+			console.debug(e, rs);
+			callback(e, rs);
+		});
+	}
+
+	public close() {
+		this.db.end();
+	}
+}
+
+export {
+	DbConfig,
+	SqliteAdapter,
+	PostgresAdapter,
+}

apps/cic-meta/src/index.ts

@@ -1,2 +1,4 @@
 export { User } from './user';
 export { Phone } from './phone';
+export { Custom } from './custom';
+export { Meta } from './meta';

apps/cic-meta/src/meta.ts

@@ -0,0 +1,128 @@
+import {ArgPair, Envelope, Syncable, MutablePgpKeyStore, PGPSigner} from "@cicnet/crdt-meta";
+import {User} from "./user";
+import {Phone} from "./phone";
+import {Custom} from "./custom";
+const fetch = require("node-fetch");
+
+const headers =  {
+    'Content-Type': 'application/json;charset=utf-8',
+    'x-cic-automerge': 'client'
+};
+const options = {
+    headers: headers,
+};
+
+class Meta {
+    keystore: MutablePgpKeyStore = new MutablePgpKeyStore();
+    signer: PGPSigner = new PGPSigner(this.keystore);
+    metaUrl: string;
+    private privateKey: string;
+    onload: (status: boolean) => void;
+
+    constructor(metaUrl: string, privateKey: any) {
+        this.metaUrl = metaUrl;
+        this.privateKey = privateKey;
+        this.keystore.loadKeyring().then(() => {
+            this.keystore.importPrivateKey(privateKey).then(() => this.onload(true));
+        });
+    }
+
+    async set(identifier: string, data: Object): Promise<any> {
+        let syncable: Syncable;
+        const response = await Meta.get(identifier, this.metaUrl);
+        if (response === `Request to ${this.metaUrl}/${identifier} failed. Connection error.`) {
+            return response;
+        } else if (typeof response !== "object" || typeof data !== "object") {
+            syncable = new Syncable(identifier, data);
+            const res = await this.updateMeta(syncable, identifier);
+            return `${res.status}: ${res.statusText}`;
+        } else {
+            syncable = await Meta.get(identifier, this.metaUrl);
+            let update: Array<ArgPair> = [];
+            for (const prop in data) {
+                update.push(new ArgPair(prop, data[prop]));
+            }
+            syncable.update(update, 'client-branch');
+            const res = await this.updateMeta(syncable, identifier);
+            return `${res.status}: ${res.statusText}`;
+        }
+    }
+
+    async updateMeta(syncable: Syncable, identifier: string): Promise<any> {
+        const envelope: Envelope = await this.wrap(syncable);
+        const reqBody: string = envelope.toJSON();
+        const putOptions = {
+            method: 'PUT',
+            headers: headers,
+            body: reqBody
+        };
+        return await fetch(`${this.metaUrl}/${identifier}`, putOptions).then(async response => {
+            if (response.ok) {
+                return Promise.resolve({
+                    status: response.status,
+                    statusText: response.statusText + ', Metadata updated successfully!'
+                });
+            } else {
+                return Promise.reject({
+                    status: response.status,
+                    statusText: response.statusText
+                });
+            }
+        });
+    }
+
+    static async get(identifier: string, metaUrl: string): Promise<any> {
+        const response = await fetch(`${metaUrl}/${identifier}`, options).then(response => {
+            if (response.ok) {
+                return (response.json());
+            } else {
+                return Promise.reject({
+                    status: response.status,
+                    statusText: response.statusText
+                });
+            }
+        }).catch(error => {
+            if (error.code === 'ECONNREFUSED') {
+                return `Request to ${metaUrl}/${identifier} failed. Connection error.`
+            }
+            return `${error.status}: ${error.statusText}`;
+        });
+        if (typeof response !== "object") {
+            return response;
+        }
+        return Envelope.fromJSON(JSON.stringify(response)).unwrap();
+    }
+
+    static async getIdentifier(name: string, type: string = 'custom'): Promise<string> {
+        let identifier: string;
+        type = type.toLowerCase();
+        if (type === 'user') {
+            identifier = await User.toKey(name);
+        } else if (type === 'phone') {
+            identifier = await Phone.toKey(name);
+        } else if (type === 'custom') {
+            identifier = await Custom.toKey(name);
+        } else {
+            identifier = await Custom.toKey(name, type);
+        }
+        return identifier;
+    }
+
+    wrap(syncable: Syncable): Promise<Envelope> {
+        return new Promise<Envelope>(async (resolve, reject) => {
+            syncable.setSigner(this.signer);
+            syncable.onwrap = async (env) => {
+                if (env === undefined) {
+                    reject();
+                    return;
+                }
+                resolve(env);
+            };
+            syncable.sign();
+        });
+    }
+}
+
+export {
+    Meta,
+}

apps/cic-meta/src/phone.ts

@@ -1,4 +1,4 @@
-import { Syncable, Addressable, mergeKey } from 'crdt-meta';
+import { Syncable, Addressable, mergeKey } from '@cicnet/crdt-meta';
 
 class Phone extends Syncable implements Addressable {
 

apps/cic-meta/src/user.ts

@@ -1,4 +1,4 @@
-import { Syncable, Addressable, toAddressKey } from 'crdt-meta';
+import { Syncable, Addressable, toAddressKey } from '@cicnet/crdt-meta';
 
 const keySalt = new TextEncoder().encode(':cic.person');
 class User extends Syncable implements Addressable {

apps/cic-meta/tests/custom.ts

@@ -0,0 +1,49 @@
+import * as assert from 'assert';
+import {Custom} from "../src";
+
+const testName = 'areas';
+const testObject = {
+    area: ['Nairobi', 'Mombasa', 'Kilifi']
+}
+const testNameKey = '8f3da0c90ba2b89ff217da96f6088cbaf987a1b58bc33c3a5e526e53cec7cfed';
+const testIdentifier = ':cic.area'
+const testIdentifierKey = 'da6194e6f33726546e82c328df4c120b844d6427859156518bd600765bf8b2b7';
+
+describe('custom', () => {
+
+    context('with predefined data', () => {
+        it('should create a custom object', () => {
+            const custom = new Custom(testName, testObject);
+            setTimeout(() => {
+                assert.strictEqual(custom.name, testName);
+                assert.deepStrictEqual(custom.m.data, testObject);
+                assert.strictEqual(custom.key(), testNameKey)
+            }, 0);
+        });
+    });
+
+    context('without predefined data', () => {
+        it('should create a custom object', () => {
+            const custom = new Custom(testName);
+            setTimeout(() => {
+                assert.strictEqual(custom.name, testName);
+                assert.deepStrictEqual(custom.m.data, {});
+                assert.strictEqual(custom.key(), testNameKey)
+            }, 0);
+        });
+    });
+
+    describe('#toKey()', () => {
+        context('without a custom identifier', () => {
+            it('should generate a key from the custom name', async () => {
+                assert.strictEqual(await Custom.toKey(testName), testNameKey);
+            });
+        });
+
+        context('with a custom identifier', () => {
+            it('should generate a key from the custom name with a custom identifier', async () => {
+                assert.strictEqual(await Custom.toKey(testName, testIdentifier), testIdentifierKey);
+            });
+        });
+    });
+});

apps/cic-meta/tests/meta.ts

@@ -0,0 +1,176 @@
+import * as assert from 'assert';
+import * as fs from 'fs';
+const nock = require('nock');
+import {Meta} from "../src";
+import {getResponse, metaData, networkErrorResponse, notFoundResponse, putResponse} from "./response";
+import {Syncable} from "@cicnet/crdt-meta";
+
+const metaUrl = 'https://meta.dev.grassrootseconomics.net';
+const testAddress = '0xc1912fee45d61c87cc5ea59dae31190fffff232d';
+const testAddressKey = 'a51472cb4df63b199a4de01335b1b4d1bbee27ff4f03340aa1d592f26c6acfe2';
+const testPhone = '+254123456789';
+const testPhoneKey = 'be3cc8212b7eb57c6217ddd42230bd8ccd2f01382bf8c1c77d3a683fa5a9bb16';
+const testName = 'areas'
+const testNameKey = '8f3da0c90ba2b89ff217da96f6088cbaf987a1b58bc33c3a5e526e53cec7cfed';
+const testIdentifier = ':cic.area'
+const testIdentifierKey = 'da6194e6f33726546e82c328df4c120b844d6427859156518bd600765bf8b2b7';
+
+function readFile(filename) {
+    if(!fs.existsSync(filename)) {
+        console.error(`File ${filename} not found`);
+        return;
+    }
+    return fs.readFileSync(filename, {encoding: 'utf8', flag: 'r'});
+}
+
+const privateKey = readFile('./privatekeys.asc');
+
+describe('meta', () => {
+    beforeEach(() => {
+        nock(metaUrl)
+            .get(`/${testAddressKey}`)
+            .reply(200, getResponse);
+
+        nock(metaUrl)
+            .get(`/${testPhoneKey}`)
+            .reply(200, getResponse);
+
+        nock(metaUrl)
+            .get(`/${testAddress}`)
+            .reply(404);
+
+        nock(metaUrl)
+            .get(`/${testIdentifier}`)
+            .replyWithError(networkErrorResponse);
+
+        nock(metaUrl)
+            .put(`/${testAddressKey}`)
+            .reply(200, putResponse);
+
+        nock(metaUrl)
+            .put(`/${testAddress}`)
+            .reply(404);
+
+        nock(metaUrl)
+            .post('/post')
+            .reply(500);
+    });
+
+    describe('#get()', () => {
+        it('should fetch data from the meta service', async () => {
+            const account = await Meta.get(testAddressKey, metaUrl);
+            assert.strictEqual(account.toJSON(account), getResponse.payload);
+        });
+
+        context('if item is not found', () => {
+            it('should respond with an error', async () => {
+                const account = await Meta.get(testAddress, metaUrl);
+                assert.strictEqual(account, `404: Not Found`);
+            });
+        });
+
+        context('in case of network error', () => {
+            it('should respond with an error', async () => {
+                const account = await Meta.get(testIdentifier, metaUrl);
+                assert.strictEqual(account, `Request to ${metaUrl}/${testIdentifier} failed. Connection error.`);
+            });
+        });
+    })
+
+    describe('#set()', () => {
+        context('object data', () => {
+            it('should set data to the meta server', () => {
+                const meta = new Meta(metaUrl, privateKey);
+                meta.onload = async (status) => {
+                    const response = await meta.set(testAddressKey, metaData);
+                    assert.strictEqual(response, `${putResponse.status}: ${putResponse.statusText}`);
+                }
+            });
+        });
+
+        context('string data', () => {
+            it('should set data to the meta server', () => {
+                const meta = new Meta(metaUrl, privateKey);
+                meta.onload = async (status) => {
+                    const response = await meta.set(testPhoneKey, testAddress);
+                    assert.strictEqual(response, `${putResponse.status}: ${putResponse.statusText}`);
+                }
+            });
+        });
+
+        context('in case of network error', () => {
+            it('should respond with an error', () => {
+                const meta = new Meta(metaUrl, privateKey);
+                meta.onload = async (status) => {
+                    const response = await meta.set(testIdentifier, metaData);
+                    assert.strictEqual(response, `Request to ${metaUrl}/${testIdentifier} failed. Connection error.`);
+                }
+            });
+        });
+    });
+
+    describe('#updateMeta()', () => {
+        it('should update data in the meta server', async () => {
+            const syncable = new Syncable(testAddressKey, metaData);
+            const meta = new Meta(metaUrl, privateKey);
+            meta.onload = async (status) => {
+                const response = await meta.updateMeta(syncable, testAddressKey);
+                assert.strictEqual(response, putResponse);
+            }
+        });
+
+        context('if item is not found', () => {
+            it('should respond with an error', () => {
+                const syncable = new Syncable(testAddress, metaData);
+                const meta = new Meta(metaUrl, privateKey);
+                meta.onload = async (status) => {
+                    const response = await meta.updateMeta(syncable, testAddress);
+                    assert.strictEqual(response, notFoundResponse);
+                }
+            });
+        });
+    });
+
+    describe('#wrap()', () => {
+        it('should sign a syncable object', function () {
+            const syncable = new Syncable(testAddressKey, metaData);
+            const meta = new Meta(metaUrl, privateKey);
+            meta.onload = async (status) => {
+                const response = await meta.wrap(syncable);
+                assert.strictEqual(response.toJSON(), getResponse);
+            }
+        });
+    })
+
+    describe('#getIdentifier()', () => {
+        context('without type', () => {
+            it('should return an identifier', async () => {
+                assert.strictEqual(await Meta.getIdentifier(testName), testNameKey);
+            });
+        });
+
+        context('with user type', () => {
+            it('should return an identifier', async () => {
+                assert.strictEqual(await Meta.getIdentifier(testAddress, 'user'), testAddressKey);
+            });
+        });
+
+        context('with phone type', () => {
+            it('should return an identifier', async () => {
+                assert.strictEqual(await Meta.getIdentifier(testPhone, 'phone'), testPhoneKey);
+            });
+        });
+
+        context('with custom type', () => {
+            it('should return an identifier', async () => {
+                assert.strictEqual(await Meta.getIdentifier(testName, 'custom'), testNameKey);
+            });
+        });
+
+        context('with unrecognised type', () => {
+            it('should return an identifier', async () => {
+                assert.strictEqual(await Meta.getIdentifier(testName, testIdentifier), testIdentifierKey);
+            });
+        });
+    });
+});

apps/cic-meta/tests/phone.ts

@@ -0,0 +1,24 @@
+import * as assert from 'assert';
+import {Phone} from "../src";
+
+const testAddress = '0xc1912fee45d61c87cc5ea59dae31190fffff232d';
+const testPhone = '+254123456789';
+const testPhoneKey = 'be3cc8212b7eb57c6217ddd42230bd8ccd2f01382bf8c1c77d3a683fa5a9bb16';
+
+describe('phone', () => {
+
+    it('should create a phone object', () => {
+        const phone = new Phone(testAddress, testPhone);
+        setTimeout(() => {
+            assert.strictEqual(phone.address, testAddress);
+            assert.strictEqual(phone.m.data.msisdn, testPhone);
+            assert.strictEqual(phone.key(), testPhoneKey)
+        }, 0);
+    });
+
+    describe('#toKey()', () => {
+        it('should generate a key from the phone number', async () => {
+            assert.strictEqual(await Phone.toKey(testPhone), testPhoneKey);
+        });
+    });
+});

apps/cic-meta/tests/server.ts

@@ -4,7 +4,8 @@ import pgp = require('openpgp');
 import sqlite = require('sqlite3');
 
 import * as handlers from '../scripts/server/handlers';
-import { Envelope, Syncable, ArgPair, PGPKeyStore, PGPSigner, KeyStore, Signer, SqliteAdapter } from 'crdt-meta';
+import { Envelope, Syncable, ArgPair, PGPKeyStore, PGPSigner, KeyStore, Signer } from '@cicnet/crdt-meta';
+import { SqliteAdapter } from '../src/db';
 
 function createKeystore() {
 	const pksa = fs.readFileSync(__dirname + '/privatekeys.asc', 'utf-8');

apps/cic-meta/tests/user.ts

@@ -0,0 +1,54 @@
+import * as assert from 'assert';
+
+import { User } from "../src";
+
+const testAddress = '0xc1912fee45d61c87cc5ea59dae31190fffff232d';
+const testAddressKey = 'a51472cb4df63b199a4de01335b1b4d1bbee27ff4f03340aa1d592f26c6acfe2';
+const testUser = {
+    user: {
+        firstName: 'Test',
+        lastName: 'User'
+    }
+}
+
+describe('user', () => {
+
+    context('without predefined data', () => {
+        it('should create a user object', () => {
+            const user = new User(testAddress);
+            setTimeout(() => {
+                assert.strictEqual(user.address, testAddress);
+                assert.strictEqual(user.key(), testAddressKey);
+                assert.strictEqual(user.m.data.user.firstName, '');
+                assert.strictEqual(user.m.data.user.lastName, '');
+            }, 0);
+        });
+    });
+
+    context('with predefined data', () => {
+        it('should create a user object', () => {
+            const user = new User(testAddress, testUser);
+            setTimeout(() => {
+                assert.strictEqual(user.address, testAddress);
+                assert.strictEqual(user.key(), testAddressKey);
+                assert.strictEqual(user.m.data.user.firstName, testUser.user.firstName);
+                assert.strictEqual(user.m.data.user.lastName, testUser.user.lastName);
+            }, 0);
+        });
+    });
+
+    describe('#setName()', () => {
+        it('should set user\'s names to metadata', () => {
+            const user = new User(testAddress);
+            user.setName(testUser.user.firstName, testUser.user.lastName);
+            assert.strictEqual(user.m.data.user.firstName, testUser.user.firstName);
+            assert.strictEqual(user.m.data.user.lastName, testUser.user.lastName);
+        });
+    });
+
+    describe('#toKey()', () => {
+        it('should generate a key from the user\'s address', async () => {
+            assert.strictEqual(await User.toKey(testAddress), testAddressKey);
+        });
+    });
+});

apps/cic-meta/tsconfig.json

@@ -2,7 +2,7 @@
   "compilerOptions": {
     "baseUrl": ".",
     "outDir": "./dist.browser",
-    "target": "es5",
+    "target": "es2015",
     "module": "commonjs",
     "moduleResolution": "node",
     "lib": ["es2016", "dom", "es5"],
@@ -19,6 +19,7 @@
   "include": [
     "src/**/*",
     "scripts/server/*",
-    "index.ts"
+    "index.ts",
+    "bin"
   ]
 }

apps/cic-notify/cic_notify/error.py

@@ -9,3 +9,7 @@ class AlreadyInitializedError(Exception):
 class PleaseCommitFirstError(Exception):
     """Raised when there exists uncommitted changes in the code while trying to build out the package."""
     pass
+
+
+class NotificationSendError(Exception):
+    """Raised when a notification failed to due to some error as per the service responsible for dispatching the notification."""

apps/cic-notify/cic_notify/ext/enums.py

@@ -0,0 +1,19 @@
+# standard imports
+from enum import IntEnum
+
+
+class AfricasTalkingStatusCodes(IntEnum):
+    PROCESSED = 100
+    SENT = 101
+    QUEUED = 102
+    RISK_HOLD = 401
+    INVALID_SENDER_ID = 402
+    INVALID_PHONE_NUMBER = 403
+    UNSUPPORTED_NUMBER_TYPE = 404
+    INSUFFICIENT_BALANCE = 405
+    USER_IN_BLACKLIST = 406
+    COULD_NOT_ROUTE = 407
+    INTERNAL_SERVER_ERROR = 500
+    GATEWAY_ERROR = 501
+    REJECTED_BY_GATEWAY = 502
+

apps/cic-notify/cic_notify/tasks/sms/africastalking.py

@@ -6,7 +6,8 @@ import celery
 import africastalking
 
 # local imports
-from cic_notify.error import NotInitializedError, AlreadyInitializedError
+from cic_notify.error import NotInitializedError, AlreadyInitializedError, NotificationSendError
+from cic_notify.ext.enums import AfricasTalkingStatusCodes
 
 logg = logging.getLogger()
 celery_app = celery.current_app
@@ -50,10 +51,27 @@ class AfricasTalkingNotifier:
         if self.sender_id:
             response = self.api_client.send(message=message, recipients=[recipient], sender_id=self.sender_id)
             logg.debug(f'Africastalking response sender-id {response}')
+
         else:
             response = self.api_client.send(message=message, recipients=[recipient])
             logg.debug(f'africastalking response no-sender-id {response}')
 
+        recipients = response.get('SMSMessageData').get('Recipients')
+
+        if len(recipients) != 1:
+            status = response.get('SMSMessageData').get('Message')
+            raise NotificationSendError(f'Unexpected number of recipients: {len(recipients)}. Status: {status}')
+
+        status_code = recipients[0].get('statusCode')
+        status = recipients[0].get('status')
+
+        if status_code not in [
+            AfricasTalkingStatusCodes.PROCESSED.value,
+            AfricasTalkingStatusCodes.SENT.value,
+            AfricasTalkingStatusCodes.QUEUED.value
+        ]:
+            raise NotificationSendError(f'Sending notification failed due to: {status}')
+
 
 @celery_app.task
 def send(message, recipient):

apps/cic-notify/cic_notify/version.py

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

apps/cic-notify/setup.cfg

@@ -28,6 +28,7 @@ packages =
 	cic_notify
 	cic_notify.db
 	cic_notify.db.models
+	cic_notify.ext
 	cic_notify.tasks.sms
 	cic_notify.runnable
 scripts =

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

@@ -5,6 +5,7 @@ LOCALE_PATH=/usr/src/cic-ussd/var/lib/locale/
 MAX_BODY_LENGTH=1024
 PASSWORD_PEPPER=QYbzKff6NhiQzY3ygl2BkiKOpER8RE/Upqs/5aZWW+I=
 SERVICE_CODE=*483*46#,*483*061#,*384*96#
+SUPPORT_PHONE_NUMBER=0757628885
 
 [phone_number]
 REGION=KE

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

@@ -5,6 +5,7 @@ LOCALE_PATH=var/lib/locale/
 MAX_BODY_LENGTH=1024
 PASSWORD_PEPPER=QYbzKff6NhiQzY3ygl2BkiKOpER8RE/Upqs/5aZWW+I=
 SERVICE_CODE=*483*46#
+SUPPORT_PHONE_NUMBER=0757628885
 
 [ussd]
 MENU_FILE=/usr/local/lib/python3.8/site-packages/cic_ussd/db/ussd_menu.json

apps/cic-ussd/.config/test/integration.ini

@@ -0,0 +1,4 @@
+[test]
+gift_value = 50.00
+server_url = http://localhost:63315/
+token_symbol = GFT

apps/cic-ussd/cic_ussd/db/ussd_menu.json

@@ -238,13 +238,43 @@
             "description": "Menu to display a user's entire profile",
             "display_key": "ussd.kenya.display_user_metadata",
             "name": "display_user_metadata",
-            "parent": "account_management"
+            "parent": "metadata_management"
         },
         "41": {
             "description": "The recipient is not in the system",
             "display_key": "ussd.kenya.exit_invalid_recipient",
             "name": "exit_invalid_recipient",
             "parent": null
+        },
+        "42": {
+            "description": "Pin entry menu for changing name data.",
+            "display_key": "ussd.kenya.name_edit_pin_authorization",
+            "name": "name_edit_pin_authorization",
+            "parent": "metadata_management"
+        },
+        "43": {
+            "description": "Pin entry menu for changing gender data.",
+            "display_key": "ussd.kenya.gender_edit_pin_authorization",
+            "name": "gender_edit_pin_authorization",
+            "parent": "metadata_management"
+        },
+        "44": {
+            "description": "Pin entry menu for changing location data.",
+            "display_key": "ussd.kenya.location_edit_pin_authorization",
+            "name": "location_edit_pin_authorization",
+            "parent": "metadata_management"
+        },
+        "45": {
+            "description": "Pin entry menu for changing products data.",
+            "display_key": "ussd.kenya.products_edit_pin_authorization",
+            "name": "products_edit_pin_authorization",
+            "parent": "metadata_management"
+        },
+        "46": {
+            "description": "Pin confirmation for pin change.",
+            "display_key": "ussd.kenya.new_pin_confirmation",
+            "name": "new_pin_confirmation",
+            "parent": "metadata_management"
         }
 
     }

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

@@ -78,31 +78,30 @@ class MetadataRequestsHandler(Metadata):
         :param data: The data to be stored in the metadata server.
         :type data: dict|str
         """
-        data = json.dumps(data).encode('utf-8')
+        data = json.dumps(data)
         result = make_request(method='POST', url=self.url, data=data, headers=self.headers)
         metadata_http_error_handler(result=result)
-        metadata = result.content
+        metadata = result.json()
         self.edit(data=metadata)
 
-    def edit(self, data: bytes):
+    def edit(self, data: Union[Dict, str]):
         """ This function is responsible for editing data in the metadata server corresponding to a unique pointer.
         :param data: The data to be edited in the metadata server.
-        :type data: bytes
+        :type data: dict
         """
         cic_meta_signer = Signer()
         signature = cic_meta_signer.sign_digest(data=data)
         algorithm = cic_meta_signer.get_operational_key().get('algo')
-        decoded_data = data.decode('utf-8')
         formatted_data = {
-            'm': data.decode('utf-8'),
+            'm': json.dumps(data),
             's': {
                 'engine': self.engine,
                 'algo': algorithm,
                 'data': signature,
-                'digest': json.loads(data).get('digest'),
+                'digest': data.get('digest'),
             }
         }
-        formatted_data = json.dumps(formatted_data).encode('utf-8')
+        formatted_data = json.dumps(formatted_data)
         result = make_request(method='PUT', url=self.url, data=formatted_data, headers=self.headers)
         logg.info(f'signed metadata submission status: {result.status_code}.')
         metadata_http_error_handler(result=result)
@@ -110,17 +109,32 @@ class MetadataRequestsHandler(Metadata):
             decoded_identifier = self.identifier.decode("utf-8")
         except UnicodeDecodeError:
             decoded_identifier = self.identifier.hex()
-        logg.info(f'identifier: {decoded_identifier}. metadata pointer: {self.metadata_pointer} set to: {decoded_data}.')
+        logg.info(f'identifier: {decoded_identifier}. metadata pointer: {self.metadata_pointer} set to: {data}.')
 
     def query(self):
-        """This function is responsible for querying the metadata server for data corresponding to a unique pointer."""
+        """
+        :return:
+        :rtype:
+        """
+        # retrieve the  metadata
         result = make_request(method='GET', url=self.url)
         metadata_http_error_handler(result=result)
-        response_data = result.content
+
-        data = json.loads(response_data.decode('utf-8'))
+        # json serialize retrieved data
+        result_data = result.json()
+
+        # validate result data format
+        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()
-            deserialized_person = person.deserialize(person_data=data)
+            person_data = person.deserialize(person_data=result_data)
-            data = json.dumps(deserialized_person.serialize())
+
-            cache_data(self.metadata_pointer, data=data)
+            # format new person data for caching
+            data = json.dumps(person_data.serialize())
+
+            # cache metadata
+            cache_data(key=self.metadata_pointer, data=data)
             logg.debug(f'caching: {data} with key: {self.metadata_pointer}')

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

@@ -47,14 +47,13 @@ class Signer:
         logg.debug(f'using signing key: {key_id}, algorithm: {key_algorithm}')
         return gpg_keys[0]
 
-    def sign_digest(self, data: bytes):
+    def sign_digest(self, data: dict):
         """
         :param data:
         :type data:
         :return:
         :rtype:
         """
-        data = json.loads(data)
         digest = data['digest']
         key_id = self.get_operational_key().get('keyid')
         signature = self.gpg.sign(digest, passphrase=self.gpg_passphrase, keyid=key_id)

apps/cic-ussd/cic_ussd/phone_number.py

@@ -41,3 +41,7 @@ def get_user_by_phone_number(phone_number: str) -> Optional[Account]:
     phone_number = process_phone_number(phone_number=phone_number, region='KE')
     user = Account.session.query(Account).filter_by(phone_number=phone_number).first()
     return user
+
+
+class Support:
+    phone_number = None

apps/cic-ussd/cic_ussd/processor.py

@@ -19,7 +19,7 @@ from cic_ussd.db.models.ussd_session import UssdSession
 from cic_ussd.error import MetadataNotFoundError, SeppukuError
 from cic_ussd.menu.ussd_menu import UssdMenu
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
-from cic_ussd.phone_number import get_user_by_phone_number
+from cic_ussd.phone_number import get_user_by_phone_number, Support
 from cic_ussd.redis import cache_data, create_cached_data_key, get_cached_data
 from cic_ussd.state_machine import UssdStateMachine
 from cic_ussd.conversions import to_wei, from_wei
@@ -251,9 +251,9 @@ def process_display_user_metadata(user: Account, display_key: str):
         identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
         cic_type=':cic.person'
     )
-    user_metadata = get_cached_data(key)
+    cached_metadata = get_cached_data(key)
-    if user_metadata:
+    if cached_metadata:
-        user_metadata = json.loads(user_metadata)
+        user_metadata = json.loads(cached_metadata)
         contact_data = get_contact_data_from_vcard(vcard=user_metadata.get('vcard'))
         logg.debug(f'{contact_data}')
         full_name = f'{contact_data.get("given")} {contact_data.get("family")}'
@@ -270,7 +270,24 @@ def process_display_user_metadata(user: Account, display_key: str):
             products=products
         )
     else:
-        raise MetadataNotFoundError(f'Expected person metadata but found none in cache for key: {key}')
+        # TODO [Philip]: All these translations could be moved to translation files.
+        logg.warning(f'Expected person metadata but found none in cache for key: {key}')
+
+        absent = ''
+        if user.preferred_language == 'en':
+            absent = 'Not provided'
+        elif user.preferred_language == 'sw':
+            absent = 'Haijawekwa'
+
+        return translation_for(
+            key=display_key,
+            preferred_language=user.preferred_language,
+            full_name=absent,
+            gender=absent,
+            location=absent,
+            products=absent
+        )
+
 
 
 def process_account_statement(user: Account, display_key: str, ussd_session: dict):
@@ -406,12 +423,6 @@ def process_request(user_input: str, user: Account, ussd_session: Optional[dict]
     :return: A ussd menu's corresponding text value.
     :rtype: Document
     """
-    # retrieve metadata before any transition
-    key = generate_metadata_pointer(
-        identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
-        cic_type=':cic.person'
-    )
-    person_metadata = get_cached_data(key=key)
 
     if ussd_session:
         if user_input == "0":
@@ -433,8 +444,9 @@ def process_request(user_input: str, user: Account, ussd_session: Optional[dict]
                     'exit_invalid_pin',
                     'exit_invalid_new_pin',
                     'exit_pin_mismatch',
-                    'exit_invalid_request'
+                    'exit_invalid_request',
-                ] and person_metadata is not None:
+                    'exit_successful_transaction'
+                ]:
                     return UssdMenu.find_by_name(name='start')
                 else:
                     return UssdMenu.find_by_name(name=last_state)
@@ -466,6 +478,14 @@ def next_state(ussd_session: dict, user: Account, user_input: str) -> str:
     return new_state
 
 
+def process_exit_invalid_menu_option(display_key: str, preferred_language: str):
+    return translation_for(
+        key=display_key,
+        preferred_language=preferred_language,
+        support_phone=Support.phone_number
+    )
+
+
 def custom_display_text(
         display_key: str,
         menu_name: str,
@@ -502,5 +522,7 @@ def custom_display_text(
         return process_account_statement(display_key=display_key, user=user, ussd_session=ussd_session)
     elif menu_name == 'display_user_metadata':
         return process_display_user_metadata(display_key=display_key, user=user)
+    elif menu_name == 'exit_invalid_menu_option':
+        return process_exit_invalid_menu_option(display_key=display_key, preferred_language=user.preferred_language)
     else:
         return translation_for(key=display_key, preferred_language=user.preferred_language)

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

@@ -26,7 +26,7 @@ from cic_ussd.metadata.base import Metadata
 from cic_ussd.operations import (define_response_with_content,
                                  process_menu_interaction_requests,
                                  define_multilingual_responses)
-from cic_ussd.phone_number import process_phone_number
+from cic_ussd.phone_number import process_phone_number, Support
 from cic_ussd.processor import get_default_token_data
 from cic_ussd.redis import cache_data, create_cached_data_key, InMemoryStore
 from cic_ussd.requests import (get_request_endpoint,
@@ -126,6 +126,8 @@ else:
 
 valid_service_codes = config.get('APP_SERVICE_CODE').split(",")
 
+Support.phone_number = config.get('APP_SUPPORT_PHONE_NUMBER')
+
 
 def application(env, start_response):
     """Loads python code for application to be accessible over web server
@@ -143,7 +145,7 @@ def application(env, start_response):
     if get_request_method(env=env) == 'POST' and get_request_endpoint(env=env) == '/':
 
         if env.get('CONTENT_TYPE') != 'application/x-www-form-urlencoded':
-            start_response('405 Play by the rules', errors_headers)
+            start_response('405 Urlencoded, please', errors_headers)
             return []
 
         post_data = env.get('wsgi.input').read()
@@ -211,6 +213,9 @@ def application(env, start_response):
         return [response_bytes]
 
     else:
+        logg.error('invalid query {}'.format(env))
+        for r in env:
+            logg.debug('{}: {}'.format(r, env))
         start_response('405 Play by the rules', errors_headers)
         return []