kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Philip/fix filter callbacks

Browse Source
This commit is contained in:
Philip Wafula 2021-04-16 20:24:07 +00:00
parent 01454c9ac0
commit f410e8b7e3
20 changed files with 388 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/cic-cache/docker/Dockerfile
View File

@ -17,7 +17,7 @@ RUN apt-get update && \
# Copy shared requirements from top of mono-repo
RUN echo "copying root req file ${root_requirement_file}"
RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a58
RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a76
COPY cic-cache/requirements.txt ./
COPY cic-cache/setup.cfg \

20
apps/cic-eth/cic_eth/eth/meta.py
View File

@ -1,6 +1,10 @@
# extended imports
# external imports
from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.status import Status as TxStatus
from cic_eth_registry.erc20 import ERC20Token
# local imports
from cic_eth.ext.address import translate_address
class ExtendedTx:
@ -27,12 +31,12 @@ class ExtendedTx:
self.status_code = TxStatus.PENDING.value
def set_actors(self, sender, recipient, trusted_declarator_addresses=None):
def set_actors(self, sender, recipient, trusted_declarator_addresses=None, caller_address=ZERO_ADDRESS):
self.sender = sender
self.recipient = recipient
if trusted_declarator_addresses != None:
self.sender_label = translate_address(sender, trusted_declarator_addresses, self.chain_spec)
self.recipient_label = translate_address(recipient, trusted_declarator_addresses, self.chain_spec)
self.sender_label = translate_address(sender, trusted_declarator_addresses, self.chain_spec, sender_address=caller_address)
self.recipient_label = translate_address(recipient, trusted_declarator_addresses, self.chain_spec, sender_address=caller_address)
def set_tokens(self, source, source_value, destination=None, destination_value=None):
@ -40,8 +44,8 @@ class ExtendedTx:
destination = source
if destination_value == None:
destination_value = source_value
st = ERC20Token(self.rpc, source)
dt = ERC20Token(self.rpc, destination)
st = ERC20Token(self.chain_spec, self.rpc, source)
dt = ERC20Token(self.chain_spec, self.rpc, destination)
self.source_token = source
self.source_token_symbol = st.symbol
self.source_token_name = st.name
@ -62,10 +66,10 @@ class ExtendedTx:
self.status_code = n
def to_dict(self):
def asdict(self):
o = {}
for attr in dir(self):
if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'to_dict']:
if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'asdict', 'rpc']:
continue
o[attr] = getattr(self, attr)
return o

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

@ -1,7 +1,7 @@
# standard imports
import logging
# third-party imports
# external imports
import celery
from cic_eth_registry.error import UnknownContractError
from chainlib.status import Status as TxStatus
@ -9,7 +9,13 @@ from chainlib.eth.address import to_checksum_address
from chainlib.eth.error import RequestMismatchException
from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.eth.erc20 import ERC20
from hexathon import strip_0x
from hexathon import (
strip_0x,
add_0x,
)
# TODO: use sarafu_Faucet for both when inheritance has been implemented
from erc20_single_shot_faucet import SingleShotFaucet
from sarafu_faucet import MinterFaucet as Faucet
# local imports
from .base import SyncFilter
@ -18,65 +24,73 @@ from cic_eth.eth.meta import ExtendedTx
logg = logging.getLogger().getChild(__name__)
def parse_transfer(tx):
r = ERC20.parse_transfer_request(tx.payload)
transfer_data = {}
transfer_data['to'] = r[0]
transfer_data['value'] = r[1]
transfer_data['from'] = tx['from']
transfer_data['token_address'] = tx['to']
return ('transfer', transfer_data)
def parse_transferfrom(tx):
r = ERC20.parse_transfer_request(tx.payload)
transfer_data = unpack_transferfrom(tx.payload)
transfer_data['from'] = r[0]
transfer_data['to'] = r[1]
transfer_data['value'] = r[2]
transfer_data['token_address'] = tx['to']
return ('transferfrom', transfer_data)
def parse_giftto(tx):
# TODO: broken
logg.error('broken')
return
transfer_data = unpack_gift(tx.payload)
transfer_data['from'] = tx.inputs[0]
transfer_data['value'] = 0
transfer_data['token_address'] = ZERO_ADDRESS
# TODO: would be better to query the gift amount from the block state
for l in tx.logs:
topics = l['topics']
logg.debug('topixx {}'.format(topics))
if strip_0x(topics[0]) == '45c201a59ac545000ead84f30b2db67da23353aa1d58ac522c48505412143ffa':
#transfer_data['value'] = web3.Web3.toInt(hexstr=strip_0x(l['data']))
transfer_data['value'] = int.from_bytes(bytes.fromhex(strip_0x(l_data)))
#token_address_bytes = topics[2][32-20:]
token_address = strip_0x(topics[2])[64-40:]
transfer_data['token_address'] = to_checksum_address(token_address)
return ('tokengift', transfer_data)
class CallbackFilter(SyncFilter):
trusted_addresses = []
def __init__(self, chain_spec, method, queue):
def __init__(self, chain_spec, method, queue, caller_address=ZERO_ADDRESS):
self.queue = queue
self.method = method
self.chain_spec = chain_spec
self.caller_address = caller_address
def parse_transfer(self, tx, conn):
if not tx.payload:
return (None, None)
r = ERC20.parse_transfer_request(tx.payload)
transfer_data = {}
transfer_data['to'] = r[0]
transfer_data['value'] = r[1]
transfer_data['from'] = tx.outputs[0]
transfer_data['token_address'] = tx.inputs[0]
return ('transfer', transfer_data)
def parse_transferfrom(self, tx, conn):
if not tx.payload:
return (None, None)
r = ERC20.parse_transfer_from_request(tx.payload)
transfer_data = {}
transfer_data['from'] = r[0]
transfer_data['to'] = r[1]
transfer_data['value'] = r[2]
transfer_data['token_address'] = tx.inputs[0]
return ('transferfrom', transfer_data)
def parse_giftto(self, tx, conn):
if not tx.payload:
return (None, None)
r = Faucet.parse_give_to_request(tx.payload)
transfer_data = {}
transfer_data['to'] = r[0]
transfer_data['value'] = tx.value
transfer_data['from'] = tx.outputs[0]
#transfer_data['token_address'] = tx.inputs[0]
faucet_contract = tx.inputs[0]
c = SingleShotFaucet(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))
o = c.amount(faucet_contract, sender_address=self.caller_address)
r = conn.do(o)
transfer_data['value'] = c.parse_amount(r)
return ('tokengift', transfer_data)
def call_back(self, transfer_type, result):
logg.debug('result {}'.format(result))
result['chain_spec'] = result['chain_spec'].asdict()
s = celery.signature(
self.method,
[
result,
transfer_type,
int(result['status_code'] == 0),
int(result['status_code'] != 0),
],
queue=self.queue,
)
@ -92,26 +106,29 @@ class CallbackFilter(SyncFilter):
# s_translate.link(s)
# s_translate.apply_async()
t = s.apply_async()
return s
return t
def parse_data(self, tx):
def parse_data(self, tx, conn):
transfer_type = None
transfer_data = None
# TODO: what's with the mix of attributes and dict keys
logg.debug('have payload {}'.format(tx.payload))
method_signature = tx.payload[:8]
logg.debug('tx status {}'.format(tx.status))
for parser in [
parse_transfer,
parse_transferfrom,
parse_giftto,
self.parse_transfer,
self.parse_transferfrom,
self.parse_giftto,
]:
try:
(transfer_type, transfer_data) = parser(tx)
break
if tx:
(transfer_type, transfer_data) = parser(tx, conn)
if transfer_type == None:
continue
else:
pass
except RequestMismatchException:
continue
@ -128,7 +145,7 @@ class CallbackFilter(SyncFilter):
transfer_data = None
transfer_type = None
try:
(transfer_type, transfer_data) = self.parse_data(tx)
(transfer_type, transfer_data) = self.parse_data(tx, conn)
except TypeError:
logg.debug('invalid method data length for tx {}'.format(tx.hash))
return
@ -144,16 +161,17 @@ class CallbackFilter(SyncFilter):
result = None
try:
tokentx = ExtendedTx(conn, tx.hash, self.chain_spec)
tokentx.set_actors(transfer_data['from'], transfer_data['to'], self.trusted_addresses)
tokentx.set_actors(transfer_data['from'], transfer_data['to'], self.trusted_addresses, caller_address=self.caller_address)
tokentx.set_tokens(transfer_data['token_address'], transfer_data['value'])
if transfer_data['status'] == 0:
tokentx.set_status(1)
else:
tokentx.set_status(0)
t = self.call_back(transfer_type, tokentx.to_dict())
logg.info('callback success task id {} tx {}'.format(t, tx.hash))
result = tokentx.asdict()
t = self.call_back(transfer_type, result)
logg.info('callback success task id {} tx {} queue {}'.format(t, tx.hash, t.queue))
except UnknownContractError:
logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tc.queue, tc.method, transfer_data['to'], tx.hash))
logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tx.queue, tx.method, transfer_data['to'], tx.hash))
def __str__(self):

1
apps/cic-eth/cic_eth/runnable/daemons/tasker.py
View File

@ -39,6 +39,7 @@ from cic_eth.queue import (
from cic_eth.callbacks import (
Callback,
http,
noop,
#tcp,
redis,
)

11
apps/cic-eth/cic_eth/runnable/daemons/tracker.py
View File

@ -15,7 +15,6 @@ import cic_base.config
import cic_base.log
import cic_base.argparse
import cic_base.rpc
from cic_eth_registry import CICRegistry
from cic_eth_registry.error import UnknownContractError
from chainlib.chain import ChainSpec
from chainlib.eth.constant import ZERO_ADDRESS
@ -43,6 +42,12 @@ from cic_eth.runnable.daemons.filters import (
TransferAuthFilter,
)
from cic_eth.stat import init_chain_stat
from cic_eth.registry import (
connect as connect_registry,
connect_declarator,
connect_token_registry,
)
script_dir = os.path.realpath(os.path.dirname(__file__))
@ -109,6 +114,8 @@ def main():
logg.info('Initializing HEAD syncer on backend {}'.format(syncer_backend))
syncers.append(HeadSyncer(syncer_backend))
connect_registry(rpc, chain_spec, config.get('CIC_REGISTRY_ADDRESS'))
trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
if trusted_addresses_src == None:
logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
@ -116,6 +123,8 @@ def main():
trusted_addresses = trusted_addresses_src.split(',')
for address in trusted_addresses:
logg.info('using trusted address {}'.format(address))
connect_declarator(rpc, chain_spec, trusted_addresses)
connect_token_registry(rpc, chain_spec)
CallbackFilter.trusted_addresses = trusted_addresses
callback_filters = []

2
apps/cic-eth/cic_eth/version.py
View File

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

3
apps/cic-eth/docker/Dockerfile
View File

@ -29,8 +29,7 @@ RUN /usr/local/bin/python -m pip install --upgrade pip
# 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.2a76
RUN pip install $pip_extra_index_url_flag chainsyncer~=0.0.2a2
RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a77
COPY cic-eth/scripts/ scripts/
COPY cic-eth/setup.cfg cic-eth/setup.py ./

18
apps/cic-eth/requirements.txt
View File

@ -1,25 +1,25 @@
cic-base~=0.1.2a67
cic-base~=0.1.2a76
celery==4.4.7
crypto-dev-signer~=0.4.14a17
crypto-dev-signer~=0.4.14b2
confini~=0.3.6rc3
cic-eth-registry~=0.5.4a13
cic-eth-registry~=0.5.4a16
#cic-bancor~=0.0.6
redis==3.5.3
alembic==1.4.2
websockets==8.1
requests~=2.24.0
eth_accounts_index~=0.0.11a7
erc20-transfer-authorization~=0.3.1a3
#simple-rlp==0.1.2
eth_accounts_index~=0.0.11a9
erc20-transfer-authorization~=0.3.1a5
uWSGI==2.0.19.1
semver==2.13.0
websocket-client==0.57.0
moolb~=0.1.1b2
eth-address-index~=0.1.1a7
chainlib~=0.0.2a5
eth-address-index~=0.1.1a9
chainlib~=0.0.2a13
hexathon~=0.0.1a7
chainsyncer[sql]~=0.0.2a2
chainqueue~=0.0.1a7
pysha3==1.0.2
coincurve==15.0.0
sarafu-faucet~=0.0.2a19
sarafu-faucet==0.0.2a28
potaahto~=0.0.1a1

2
apps/cic-eth/test_requirements.txt
View File

@ -4,4 +4,4 @@ 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.8a4
giftable-erc20-token==0.0.8a9

223
apps/cic-eth/tests/filters/test_callback_filter.py Normal file
View File

@ -0,0 +1,223 @@
# standard import
import logging
import datetime
import os
# external imports
import pytest
from chainlib.connection import RPCConnection
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.gas import OverrideGasOracle
from chainlib.eth.tx import (
receipt,
transaction,
Tx,
)
from chainlib.eth.block import Block
from chainlib.eth.erc20 import ERC20
from sarafu_faucet import MinterFaucet
from eth_accounts_index import AccountRegistry
from potaahto.symbols import snake_and_camel
from hexathon import add_0x
# local imports
from cic_eth.runnable.daemons.filters.callback import CallbackFilter
logg = logging.getLogger()
@pytest.mark.skip()
def test_transfer_tx(
default_chain_spec,
init_database,
eth_rpc,
eth_signer,
foo_token,
agent_roles,
token_roles,
contract_roles,
celery_session_worker,
):
rpc = RPCConnection.connect(default_chain_spec, 'default')
nonce_oracle = RPCNonceOracle(token_roles['FOO_TOKEN_OWNER'], rpc)
gas_oracle = OverrideGasOracle(conn=rpc, limit=200000)
txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, o) = txf.transfer(foo_token, token_roles['FOO_TOKEN_OWNER'], agent_roles['ALICE'], 1024)
r = rpc.do(o)
o = transaction(tx_hash_hex)
r = rpc.do(o)
logg.debug(r)
tx_src = snake_and_camel(r)
tx = Tx(tx_src)
o = receipt(tx_hash_hex)
r = rpc.do(o)
assert r['status'] == 1
rcpt = snake_and_camel(r)
tx.apply_receipt(rcpt)
fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
(transfer_type, transfer_data) = fltr.parse_transfer(tx, eth_rpc)
assert transfer_type == 'transfer'
@pytest.mark.skip()
def test_transfer_from_tx(
default_chain_spec,
init_database,
eth_rpc,
eth_signer,
foo_token,
agent_roles,
token_roles,
contract_roles,
celery_session_worker,
):
rpc = RPCConnection.connect(default_chain_spec, 'default')
nonce_oracle = RPCNonceOracle(token_roles['FOO_TOKEN_OWNER'], rpc)
gas_oracle = OverrideGasOracle(conn=rpc, limit=200000)
txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, o) = txf.approve(foo_token, token_roles['FOO_TOKEN_OWNER'], agent_roles['ALICE'], 1024)
r = rpc.do(o)
o = receipt(tx_hash_hex)
r = rpc.do(o)
assert r['status'] == 1
nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], rpc)
txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, o) = txf.transfer_from(foo_token, agent_roles['ALICE'], token_roles['FOO_TOKEN_OWNER'], agent_roles['BOB'], 1024)
r = rpc.do(o)
o = transaction(tx_hash_hex)
r = rpc.do(o)
tx_src = snake_and_camel(r)
tx = Tx(tx_src)
o = receipt(tx_hash_hex)
r = rpc.do(o)
assert r['status'] == 1
rcpt = snake_and_camel(r)
tx.apply_receipt(rcpt)
fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
(transfer_type, transfer_data) = fltr.parse_transferfrom(tx, eth_rpc)
assert transfer_type == 'transferfrom'
def test_faucet_gift_to_tx(
default_chain_spec,
init_database,
eth_rpc,
eth_signer,
foo_token,
agent_roles,
contract_roles,
faucet,
account_registry,
celery_session_worker,
):
rpc = RPCConnection.connect(default_chain_spec, 'default')
gas_oracle = OverrideGasOracle(conn=rpc, limit=800000)
nonce_oracle = RPCNonceOracle(contract_roles['ACCOUNT_REGISTRY_WRITER'], rpc)
txf = AccountRegistry(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, o) = txf.add(account_registry, contract_roles['ACCOUNT_REGISTRY_WRITER'], agent_roles['ALICE'])
r = rpc.do(o)
o = receipt(tx_hash_hex)
r = rpc.do(o)
assert r['status'] == 1
nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], rpc)
txf = MinterFaucet(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, o) = txf.give_to(faucet, agent_roles['ALICE'], agent_roles['ALICE'])
r = rpc.do(o)
o = transaction(tx_hash_hex)
r = rpc.do(o)
tx_src = snake_and_camel(r)
tx = Tx(tx_src)
o = receipt(tx_hash_hex)
r = rpc.do(o)
assert r['status'] == 1
rcpt = snake_and_camel(r)
tx.apply_receipt(rcpt)
fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
(transfer_type, transfer_data) = fltr.parse_giftto(tx, eth_rpc)
assert transfer_type == 'tokengift'
assert transfer_data['token_address'] == foo_token
def test_callback_filter(
default_chain_spec,
init_database,
eth_rpc,
eth_signer,
foo_token,
token_roles,
agent_roles,
contract_roles,
register_lookups,
):
rpc = RPCConnection.connect(default_chain_spec, 'default')
nonce_oracle = RPCNonceOracle(token_roles['FOO_TOKEN_OWNER'], rpc)
gas_oracle = OverrideGasOracle(conn=rpc, limit=200000)
txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, o) = txf.transfer(foo_token, token_roles['FOO_TOKEN_OWNER'], agent_roles['ALICE'], 1024)
r = rpc.do(o)
o = transaction(tx_hash_hex)
r = rpc.do(o)
logg.debug(r)
mockblock_src = {
'hash': add_0x(os.urandom(32).hex()),
'number': '0x2a',
'transactions': [tx_hash_hex],
'timestamp': datetime.datetime.utcnow().timestamp(),
}
mockblock = Block(mockblock_src)
tx_src = snake_and_camel(r)
tx = Tx(tx_src, block=mockblock)
o = receipt(tx_hash_hex)
r = rpc.do(o)
assert r['status'] == 1
rcpt = snake_and_camel(r)
tx.apply_receipt(rcpt)
fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
class CallbackMock:
def __init__(self):
self.results = {}
def call_back(self, transfer_type, result):
self.results[transfer_type] = result
mock = CallbackMock()
fltr.call_back = mock.call_back
fltr.filter(eth_rpc, mockblock, tx, init_database)
assert mock.results.get('transfer') != None
assert mock.results['transfer']['destination_token'] == foo_token

2
apps/cic-notify/cic_notify/runnable/tasker.py
View File

@ -33,7 +33,9 @@ elif args.v:
config = confini.Config(args.c, args.env_prefix)
config.process()
config.add(args.q, '_CELERY_QUEUE', True)
config.censor('PASSWORD', 'DATABASE')
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
# connect to database
dsn = dsn_from_config(config)

26
apps/cic-notify/cic_notify/version.py
View File

@ -6,11 +6,10 @@ import time
import semver
# local imports
from cic_notify.error import PleaseCommitFirstError
logg = logging.getLogger()
version = (0, 4, 0, 'alpha.3')
version = (0, 4, 0, 'alpha.4')
version_object = semver.VersionInfo(
major=version[0],
@ -18,27 +17,4 @@ version_object = semver.VersionInfo(
patch=version[2],
prerelease=version[3],
)
version_string = str(version_object)
def git_hash():
import subprocess
git_hash = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True)
git_hash_brief = git_hash.stdout.decode('utf-8')[:8]
return git_hash_brief
try:
version_git = git_hash()
version_string += '+build.{}'.format(version_git)
except FileNotFoundError:
time_string_pair = str(time.time()).split('.')
version_string += '+build.{}{:<09d}'.format(
time_string_pair[0],
int(time_string_pair[1]),
)
logg.info(f'Final version string will be {version_string}')
__version_string__ = version_string

1
apps/cic-notify/setup.cfg
View File

@ -1,6 +1,5 @@
[metadata]
name = cic-notify
version= 0.4.0a3
description = CIC notifications service
author = Louis Holbrook
author_email = dev@holbrook.no

30
apps/cic-notify/setup.py
View File

@ -1,9 +1,31 @@
# standard imports
import logging
import subprocess
import time
from setuptools import setup
# third-party imports
# local imports
from cic_notify.version import version_string
logg = logging.getLogger()
def git_hash():
git_hash = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True)
git_hash_brief = git_hash.stdout.decode('utf-8')[:8]
return git_hash_brief
try:
version_git = git_hash()
version_string += '+build.{}'.format(version_git)
except FileNotFoundError:
time_string_pair = str(time.time()).split('.')
version_string += '+build.{}{:<09d}'.format(
time_string_pair[0],
int(time_string_pair[1]),
)
logg.info(f'Final version string will be {version_string}')
requirements = []
@ -25,6 +47,6 @@ while True:
test_requirements_file.close()
setup(
version=version_string,
install_requires=requirements,
tests_require=test_requirements,
)
tests_require=test_requirements)

4
apps/cic-ussd/.config/redis.ini
View File

@ -1,6 +1,6 @@
[celery]
BROKER_URL=redis://redis:6379
RESULT_URL=redis://redis:6379
BROKER_URL=redis://
RESULT_URL=redis://
[redis]
HOSTNAME=redis

12
apps/cic-ussd/cic_ussd/runnable/server.py
View File

@ -57,19 +57,17 @@ arg_parser.add_argument('--env-prefix',
help='environment prefix for variables to overwrite configuration')
args = arg_parser.parse_args()
# parse config
config = Config(config_dir=args.c, env_prefix=args.env_prefix)
config.process()
config.censor('PASSWORD', 'DATABASE')
# define log levels
if args.vv:
logging.getLogger().setLevel(logging.DEBUG)
elif args.v:
logging.getLogger().setLevel(logging.INFO)
# log config vars
logg.debug(config)
# parse config
config = Config(config_dir=args.c, env_prefix=args.env_prefix)
config.process()
config.censor('PASSWORD', 'DATABASE')
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
# initialize elements
# set up translations

17
apps/cic-ussd/cic_ussd/runnable/tasker.py
View File

@ -6,6 +6,7 @@ import tempfile
# third party imports
import celery
import i18n
import redis
from confini import Config
@ -33,18 +34,18 @@ arg_parser.add_argument('-vv', action='store_true', help='be more verbose')
arg_parser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
args = arg_parser.parse_args()
# parse config
config = Config(config_dir=args.c, env_prefix=args.env_prefix)
config.process()
config.censor('PASSWORD', 'DATABASE')
# define log levels
if args.vv:
logging.getLogger().setLevel(logging.DEBUG)
elif args.v:
logging.getLogger().setLevel(logging.INFO)
logg.debug(config)
# parse config
config = Config(args.c, args.env_prefix)
config.process()
config.add(args.q, '_CELERY_QUEUE', True)
config.censor('PASSWORD', 'DATABASE')
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
# connect to database
data_source_name = dsn_from_config(config)
@ -77,6 +78,10 @@ if key_file_path:
validate_presence(path=key_file_path)
Signer.key_file_path = key_file_path
# set up translations
i18n.load_path.append(config.get('APP_LOCALE_PATH'))
i18n.set('fallback', config.get('APP_LOCALE_FALLBACK'))
# set up celery
current_app = celery.Celery(__name__)

4
apps/cic-ussd/cic_ussd/tasks/callback_handler.py
View File

@ -83,8 +83,8 @@ def process_incoming_transfer_callback(result: dict, param: str, status_code: in
# collect result data
recipient_blockchain_address = result.get('recipient')
sender_blockchain_address = result.get('sender')
token_symbol = result.get('token_symbol')
value = result.get('destination_value')
token_symbol = result.get('destination_token_symbol')
value = result.get('destination_token_value')
# try to find users in system
recipient_user = session.query(User).filter_by(blockchain_address=recipient_blockchain_address).first()

2
apps/cic-ussd/requirements.txt
View File

@ -1,4 +1,4 @@
cic_base[full_graph]~=0.1.2a68
cic-eth~=0.11.0b3
cic-notify~=0.4.0a3
cic-notify~=0.4.0a4
cic-types~=0.1.0a10

4
docker-compose.yml
View File

@ -446,7 +446,7 @@ services:
deploy:
restart_policy:
condition: on-failure
command: "/root/start_tasker.sh -q cic-notify"
command: "/root/start_tasker.sh -q cic-notify -vv"
cic-meta-server:
@ -494,6 +494,8 @@ services:
DATABASE_NAME: cic_ussd
DATABASE_ENGINE: postgresql
DATABASE_DRIVER: psycopg2
CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis}
CELERY_RESULT_URL: ${CELERY_BROKER_URL:-redis://redis}
PGP_PASSPHRASE: merman
SERVER_PORT: 9000
CIC_META_URL: ${CIC_META_URL:-http://meta:8000}