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

Merge remote-tracking branch 'origin/master' into lash/cache-faucet

Browse Source
This commit is contained in:
nolash 2021-05-02 12:12:12 +02:00
commit f4fe62cd9e
20 changed files with 114 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -36,6 +36,7 @@ from cic_eth.eth import (
from cic_eth.admin import (
debug,
ctrl,
token
)
from cic_eth.queue import (
query,

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

@ -2,7 +2,7 @@ FROM node:15.3.0-alpine3.10
WORKDIR /tmp/src/cic-meta
RUN apk add --no-cache postgresql
RUN apk add --no-cache postgresql bash
COPY cic-meta/package.json \
./

5
apps/cic-meta/docker/db.sh
View File

@ -1,3 +1,6 @@
#!/bin/bash
set -e
PGPASSWORD=$DATABASE_PASSWORD psql -v ON_ERROR_STOP=1 -U $DATABASE_USER -h $DATABASE_HOST -p $DATABASE_PORT -d $DATABASE_NAME -f $SCHEMA_SQL_PATH
PGPASSWORD=$DATABASE_PASSWORD psql -U $DATABASE_USER -h $DATABASE_HOST -p $DATABASE_PORT -d $DATABASE_NAME -f $SCHEMA_SQL_PATH

4
apps/cic-meta/docker/start_server.sh
View File

@ -1,3 +1,7 @@
#!/bin/bash
set -euo pipefail
# db migration
sh ./db.sh
# /usr/local/bin/node /usr/local/bin/cic-meta-server $@

2
apps/cic-meta/scripts/initdb/server.postgres.sql
View File

@ -1,4 +1,4 @@
create table if not exists cic_meta.store (
create table if not exists store (
id serial primary key not null,
owner_fingerprint text not null,
hash char(64) not null unique,

2
apps/cic-ussd/cic_ussd/account.py
View File

@ -20,7 +20,7 @@ def define_account_tx_metadata(user: Account):
)
key = generate_metadata_pointer(
identifier=identifier,
cic_type='cic.person'
cic_type=':cic.person'
)
account_metadata = get_cached_data(key=key)

2
apps/cic-ussd/cic_ussd/balance.py
View File

@ -80,7 +80,7 @@ def get_cached_operational_balance(blockchain_address: str):
"""
key = create_cached_data_key(
identifier=bytes.fromhex(blockchain_address[2:]),
salt='cic.balances_data'
salt=':cic.balances_data'
)
cached_balance = get_cached_data(key=key)
if cached_balance:

10
apps/cic-ussd/cic_ussd/error.py
View File

@ -38,3 +38,13 @@ class MetadataStoreError(Exception):
pass
class SeppukuError(Exception):
"""Exception base class for all errors that should cause system shutdown"""
pass
class InitializationError(Exception):
"""Exception raised when initialization state is insufficient to run component"""
pass

2
apps/cic-ussd/cic_ussd/metadata/base.py
View File

@ -118,7 +118,7 @@ class MetadataRequestsHandler(Metadata):
metadata_http_error_handler(result=result)
response_data = result.content
data = json.loads(response_data.decode('utf-8'))
if result.status_code == 200 and self.cic_type == 'cic.person':
if result.status_code == 200 and self.cic_type == ':cic.person':
person = Person()
deserialized_person = person.deserialize(person_data=json.loads(data))
data = json.dumps(deserialized_person.serialize())

2
apps/cic-ussd/cic_ussd/metadata/person.py
View File

@ -9,4 +9,4 @@ from .base import MetadataRequestsHandler
class PersonMetadata(MetadataRequestsHandler):
def __init__(self, identifier: bytes):
super().__init__(cic_type='cic.person', identifier=identifier)
super().__init__(cic_type=':cic.person', identifier=identifier)

2
apps/cic-ussd/cic_ussd/metadata/phone.py
View File

@ -10,4 +10,4 @@ from .base import MetadataRequestsHandler
class PhonePointerMetadata(MetadataRequestsHandler):
def __init__(self, identifier: bytes):
super().__init__(cic_type='cic.msisdn', identifier=identifier)
super().__init__(cic_type=':cic.phone', identifier=identifier)

98
apps/cic-ussd/cic_ussd/processor.py
View File

@ -7,6 +7,7 @@ from typing import Optional
# third party imports
import celery
from sqlalchemy import desc
from cic_eth.api import Api
from tinydb.table import Document
# local imports
@ -15,7 +16,7 @@ from cic_ussd.balance import BalanceManager, compute_operational_balance, get_ca
from cic_ussd.chain import Chain
from cic_ussd.db.models.account import AccountStatus, Account
from cic_ussd.db.models.ussd_session import UssdSession
from cic_ussd.error import MetadataNotFoundError
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
@ -28,6 +29,38 @@ from cic_types.models.person import generate_metadata_pointer, get_contact_data_
logg = logging.getLogger(__name__)
def get_default_token_data():
chain_str = Chain.spec.__str__()
cic_eth_api = Api(chain_str=chain_str)
default_token_request_task = cic_eth_api.default_token()
default_token_data = default_token_request_task.get()
return default_token_data
def retrieve_token_symbol(chain_str: str = Chain.spec.__str__()):
"""
:param chain_str:
:type chain_str:
:return:
:rtype:
"""
cache_key = create_cached_data_key(
identifier=chain_str.encode('utf-8'),
salt=':cic.default_token_data'
)
cached_data = get_cached_data(key=cache_key)
if cached_data:
default_token_data = json.loads(cached_data)
return default_token_data.get('symbol')
else:
logg.warning('Cached default token data not found. Attempting retrieval from default token API')
default_token_data = get_default_token_data()
if default_token_data:
return default_token_data.get('symbol')
else:
raise SeppukuError(f'Could not retrieve default token for: {chain_str}')
def process_pin_authorization(display_key: str, user: Account, **kwargs) -> str:
"""
This method provides translation for all ussd menu entries that follow the pin authorization pattern.
@ -73,7 +106,9 @@ def process_exit_insufficient_balance(display_key: str, user: Account, ussd_sess
# compile response data
user_input = ussd_session.get('user_input').split('*')[-1]
transaction_amount = to_wei(value=int(user_input))
token_symbol = 'GFT'
# get default data
token_symbol = retrieve_token_symbol()
recipient_phone_number = ussd_session.get('session_data').get('recipient_phone_number')
recipient = get_user_by_phone_number(phone_number=recipient_phone_number)
@ -102,7 +137,7 @@ def process_exit_successful_transaction(display_key: str, user: Account, ussd_se
:rtype: str
"""
transaction_amount = to_wei(int(ussd_session.get('session_data').get('transaction_amount')))
token_symbol = 'GFT'
token_symbol = retrieve_token_symbol()
recipient_phone_number = ussd_session.get('session_data').get('recipient_phone_number')
recipient = get_user_by_phone_number(phone_number=recipient_phone_number)
tx_recipient_information = define_account_tx_metadata(user=recipient)
@ -137,7 +172,7 @@ def process_transaction_pin_authorization(user: Account, display_key: str, ussd_
tx_recipient_information = define_account_tx_metadata(user=recipient)
tx_sender_information = define_account_tx_metadata(user=user)
token_symbol = 'GFT'
token_symbol = retrieve_token_symbol()
user_input = ussd_session.get('session_data').get('transaction_amount')
transaction_amount = to_wei(value=int(user_input))
logg.debug('Requires integration to determine user tokens.')
@ -168,18 +203,18 @@ def process_account_balances(user: Account, display_key: str, ussd_session: dict
logg.debug('Requires call to retrieve tax and bonus amounts')
tax = ''
bonus = ''
token_symbol = retrieve_token_symbol()
return translation_for(
key=display_key,
preferred_language=user.preferred_language,
operational_balance=operational_balance,
tax=tax,
bonus=bonus,
token_symbol='GFT'
token_symbol=token_symbol
)
def format_transactions(transactions: list, preferred_language: str):
def format_transactions(transactions: list, preferred_language: str, token_symbol: str):
formatted_transactions = ''
if len(transactions) > 0:
@ -190,7 +225,7 @@ def format_transactions(transactions: list, preferred_language: str):
timestamp = transaction.get('timestamp')
action_tag = transaction.get('action_tag')
direction = transaction.get('direction')
token_symbol = 'GFT'
token_symbol = token_symbol
if action_tag == 'SENT' or action_tag == 'ULITUMA':
formatted_transactions += f'{action_tag} {value} {token_symbol} {direction} {recipient_phone_number} {timestamp}.\n'
@ -214,7 +249,7 @@ def process_display_user_metadata(user: Account, display_key: str):
"""
key = generate_metadata_pointer(
identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
cic_type='cic.person'
cic_type=':cic.person'
)
user_metadata = get_cached_data(key)
if user_metadata:
@ -251,9 +286,11 @@ def process_account_statement(user: Account, display_key: str, ussd_session: dic
"""
# retrieve cached statement
identifier = blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address)
key = create_cached_data_key(identifier=identifier, salt='cic.statement')
key = create_cached_data_key(identifier=identifier, salt=':cic.statement')
transactions = get_cached_data(key=key)
token_symbol = retrieve_token_symbol()
first_transaction_set = []
middle_transaction_set = []
last_transaction_set = []
@ -277,7 +314,8 @@ def process_account_statement(user: Account, display_key: str, ussd_session: dic
preferred_language=user.preferred_language,
first_transaction_set=format_transactions(
transactions=first_transaction_set,
preferred_language=user.preferred_language
preferred_language=user.preferred_language,
token_symbol=token_symbol
)
)
elif display_key == 'ussd.kenya.middle_transaction_set':
@ -286,7 +324,8 @@ def process_account_statement(user: Account, display_key: str, ussd_session: dic
preferred_language=user.preferred_language,
middle_transaction_set=format_transactions(
transactions=middle_transaction_set,
preferred_language=user.preferred_language
preferred_language=user.preferred_language,
token_symbol=token_symbol
)
)
@ -296,7 +335,8 @@ def process_account_statement(user: Account, display_key: str, ussd_session: dic
preferred_language=user.preferred_language,
last_transaction_set=format_transactions(
transactions=last_transaction_set,
preferred_language=user.preferred_language
preferred_language=user.preferred_language,
token_symbol=token_symbol
)
)
@ -312,18 +352,19 @@ def process_start_menu(display_key: str, user: Account):
:return: Corresponding translation text response
:rtype: str
"""
token_symbol = retrieve_token_symbol()
chain_str = Chain.spec.__str__()
blockchain_address = user.blockchain_address
balance_manager = BalanceManager(address=blockchain_address,
chain_str=chain_str,
token_symbol='GFT')
token_symbol=token_symbol)
# get balances synchronously for display on start menu
balances_data = balance_manager.get_balances()
key = create_cached_data_key(
identifier=bytes.fromhex(blockchain_address[2:]),
salt='cic.balances_data'
salt=':cic.balances_data'
)
cache_data(key=key, data=json.dumps(balances_data))
@ -340,9 +381,6 @@ def process_start_menu(display_key: str, user: Account):
# retrieve and cache account's statement
retrieve_account_statement(blockchain_address=blockchain_address)
# TODO [Philip]: figure out how to get token symbol from a metadata layer of sorts.
token_symbol = 'GFT'
return translation_for(
key=display_key,
preferred_language=user.preferred_language,
@ -375,6 +413,13 @@ 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":
return UssdMenu.parent_menu(menu_name=ussd_session.get('state'))
@ -382,29 +427,12 @@ def process_request(user_input: str, user: Account, ussd_session: Optional[dict]
successive_state = next_state(ussd_session=ussd_session, user=user, user_input=user_input)
return UssdMenu.find_by_name(name=successive_state)
else:
key = generate_metadata_pointer(
identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
cic_type='cic.person'
)
# retrieve and cache account's metadata
s_query_person_metadata = celery.signature(
'cic_ussd.tasks.metadata.query_person_metadata',
[user.blockchain_address]
)
s_query_person_metadata.apply_async(queue='cic-ussd')
if user.has_valid_pin():
last_ussd_session = retrieve_most_recent_ussd_session(phone_number=user.phone_number)
if last_ussd_session:
# get metadata
person_metadata = get_cached_data(key=key)
# get last state
last_state = last_ussd_session.state
# if last state is account_creation_prompt and metadata exists, show start menu
if last_state in [
'account_creation_prompt',

18
apps/cic-ussd/cic_ussd/runnable/daemons/cic_user_ussd_server.py
View File

@ -17,6 +17,7 @@ from cic_ussd.chain import Chain
from cic_ussd.db import dsn_from_config
from cic_ussd.db.models.base import SessionBase
from cic_ussd.encoder import PasswordEncoder
from cic_ussd.error import InitializationError
from cic_ussd.files.local_files import create_local_file_data_stores, json_file_parser
from cic_ussd.menu.ussd_menu import UssdMenu
from cic_ussd.metadata.signer import Signer
@ -25,7 +26,8 @@ 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.redis import InMemoryStore
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,
get_request_method)
from cic_ussd.runnable.server_base import exportable_parser, logg
@ -107,6 +109,20 @@ Chain.spec = chain_spec
UssdStateMachine.states = states
UssdStateMachine.transitions = transitions
# retrieve default token data
default_token_data = get_default_token_data()
chain_str = Chain.spec.__str__()
# cache default token for re-usability
if default_token_data:
cache_key = create_cached_data_key(
identifier=chain_str.encode('utf-8'),
salt=':cic.default_token_data'
)
cache_data(key=cache_key, data=json.dumps(default_token_data))
else:
raise InitializationError(f'Default token data for: {chain_str} not found.')
def application(env, start_response):
"""Loads python code for application to be accessible over web server

2
apps/cic-ussd/cic_ussd/state_machine/logic/transaction.py
View File

@ -64,7 +64,7 @@ def has_sufficient_balance(state_machine_data: Tuple[str, dict, Account]) -> boo
# get cached balance
key = create_cached_data_key(
identifier=bytes.fromhex(user.blockchain_address[2:]),
salt='cic.balances_data'
salt=':cic.balances_data'
)
cached_balance = get_cached_data(key=key)
operational_balance = compute_operational_balance(balances=json.loads(cached_balance))

2
apps/cic-ussd/cic_ussd/state_machine/logic/user.py
View File

@ -176,7 +176,7 @@ def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, Account]):
blockchain_address = user.blockchain_address
key = generate_metadata_pointer(
identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
cic_type='cic.person'
cic_type=':cic.person'
)
user_metadata = get_cached_data(key=key)

2
apps/cic-ussd/cic_ussd/state_machine/logic/validator.py
View File

@ -23,7 +23,7 @@ def has_cached_user_metadata(state_machine_data: Tuple[str, dict, Account]):
# check for user metadata in cache
key = generate_metadata_pointer(
identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
cic_type='cic.person'
cic_type=':cic.person'
)
user_metadata = get_cached_data(key=key)
return user_metadata is not None

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

@ -136,7 +136,7 @@ def process_balances_callback(result: list, param: str, status_code: int):
blockchain_address = balances_data.get('address')
key = create_cached_data_key(
identifier=bytes.fromhex(blockchain_address[2:]),
salt='cic.balances_data'
salt=':cic.balances_data'
)
cache_data(key=key, data=json.dumps(balances_data))
else:
@ -226,7 +226,7 @@ def process_statement_callback(result, param: str, status_code: int):
# cache account statement
identifier = bytes.fromhex(param[2:])
key = create_cached_data_key(identifier=identifier, salt='cic.statement')
key = create_cached_data_key(identifier=identifier, salt=':cic.statement')
data = json.dumps(processed_transactions)
# cache statement data

2
apps/cic-ussd/tests/cic_ussd/metadata/test_user_metadata.py
View File

@ -105,7 +105,7 @@ def test_get_user_metadata(caplog,
assert 'Get latest data status: 200' in caplog.text
key = generate_metadata_pointer(
identifier=identifier,
cic_type='cic.person'
cic_type=':cic.person'
)
cached_user_metadata = get_cached_data(key=key)
assert cached_user_metadata

2
apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_validator_logic.py
View File

@ -36,7 +36,7 @@ def test_has_cached_user_metadata(create_in_db_ussd_session,
user = create_activated_user
key = generate_metadata_pointer(
identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
cic_type='cic.person'
cic_type=':cic.person'
)
cache_data(key=key, data=json.dumps(person_metadata))
result = has_cached_user_metadata(state_machine_data=state_machine_data)

2
apps/cic-ussd/tests/fixtures/user.py vendored
View File

@ -115,6 +115,6 @@ def cached_user_metadata(create_activated_user, init_redis_cache, person_metadat
user_metadata = json.dumps(person_metadata)
key = generate_metadata_pointer(
identifier=blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address),
cic_type='cic.person'
cic_type=':cic.person'
)
cache_data(key=key, data=user_metadata)