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

Updates menu logic to adapt changes from multi-token env.

This commit is contained in:
PhilipWafula 2021-11-22 13:17:39 +03:00
parent c3ee108c8b
commit 8e365ea586
Signed by untrusted user: mango-habanero
GPG Key ID: B00CE9034DA19FB7
3 changed files with 25 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
apps/cic-ussd/cic_ussd/state_machine/logic/pin.py
View File

@ -3,7 +3,6 @@ user's pin.
""" """
# standard imports # standard imports
import json
import logging import logging
import re import re
from typing import Tuple from typing import Tuple
@ -16,6 +15,7 @@ from cic_ussd.db.models.account import Account
from cic_ussd.db.models.base import SessionBase from cic_ussd.db.models.base import SessionBase
from cic_ussd.db.enum import AccountStatus from cic_ussd.db.enum import AccountStatus
from cic_ussd.encoder import create_password_hash, check_password_hash from cic_ussd.encoder import create_password_hash, check_password_hash
from cic_ussd.processor.util import wait_for_session_data
from cic_ussd.session.ussd_session import create_or_update_session, persist_ussd_session from cic_ussd.session.ussd_session import create_or_update_session, persist_ussd_session
@ -31,11 +31,8 @@ def is_valid_pin(state_machine_data: Tuple[str, dict, Account, Session]) -> bool
:rtype: bool :rtype: bool
""" """
user_input, ussd_session, account, session = state_machine_data user_input, ussd_session, account, session = state_machine_data
pin_is_valid = False
matcher = r'^\d{4}$' matcher = r'^\d{4}$'
if re.match(matcher, user_input): return bool(re.match(matcher, user_input))
pin_is_valid = True
return pin_is_valid
def is_authorized_pin(state_machine_data: Tuple[str, dict, Account, Session]) -> bool: def is_authorized_pin(state_machine_data: Tuple[str, dict, Account, Session]) -> bool:
@ -68,7 +65,7 @@ def save_initial_pin_to_session_data(state_machine_data: Tuple[str, dict, Accoun
:param state_machine_data: A tuple containing user input, a ussd session and user object. :param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: tuple :type state_machine_data: tuple
""" """
user_input, ussd_session, user, session = state_machine_data user_input, ussd_session, account, session = state_machine_data
initial_pin = create_password_hash(user_input) initial_pin = create_password_hash(user_input)
if ussd_session.get('data'): if ussd_session.get('data'):
data = ussd_session.get('data') data = ussd_session.get('data')
@ -97,7 +94,8 @@ def pins_match(state_machine_data: Tuple[str, dict, Account, Session]) -> bool:
:return: A match between two pin values. :return: A match between two pin values.
:rtype: bool :rtype: bool
""" """
user_input, ussd_session, user, session = state_machine_data user_input, ussd_session, account, session = state_machine_data
wait_for_session_data('Initial pin', session_data_key='initial_pin', ussd_session=ussd_session)
initial_pin = ussd_session.get('data').get('initial_pin') initial_pin = ussd_session.get('data').get('initial_pin')
return check_password_hash(user_input, initial_pin) return check_password_hash(user_input, initial_pin)
@ -107,11 +105,12 @@ def complete_pin_change(state_machine_data: Tuple[str, dict, Account, Session]):
:param state_machine_data: A tuple containing user input, a ussd session and user object. :param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: tuple :type state_machine_data: tuple
""" """
user_input, ussd_session, user, session = state_machine_data user_input, ussd_session, account, session = state_machine_data
session = SessionBase.bind_session(session=session) session = SessionBase.bind_session(session=session)
wait_for_session_data('Initial pin', session_data_key='initial_pin', ussd_session=ussd_session)
password_hash = ussd_session.get('data').get('initial_pin') password_hash = ussd_session.get('data').get('initial_pin')
user.password_hash = password_hash account.password_hash = password_hash
session.add(user) session.add(account)
session.flush() session.flush()
SessionBase.release_session(session=session) SessionBase.release_session(session=session)
@ -134,6 +133,6 @@ def is_valid_new_pin(state_machine_data: Tuple[str, dict, Account, Session]) ->
:return: A match between two pin values. :return: A match between two pin values.
:rtype: bool :rtype: bool
""" """
user_input, ussd_session, user, session = state_machine_data user_input, ussd_session, account, session = state_machine_data
is_old_pin = user.verify_password(password=user_input) is_old_pin = account.verify_password(password=user_input)
return is_valid_pin(state_machine_data=state_machine_data) and not is_old_pin return is_valid_pin(state_machine_data=state_machine_data) and not is_old_pin

4
apps/cic-ussd/cic_ussd/state_machine/logic/sms.py
View File

@ -6,7 +6,7 @@ from sqlalchemy.orm.session import Session
# local imports # local imports
from cic_ussd.account.metadata import get_cached_preferred_language from cic_ussd.account.metadata import get_cached_preferred_language
from cic_ussd.account.tokens import get_default_token_symbol from cic_ussd.account.tokens import get_active_token_symbol
from cic_ussd.db.models.account import Account from cic_ussd.db.models.account import Account
from cic_ussd.notifications import Notifier from cic_ussd.notifications import Notifier
from cic_ussd.phone_number import Support from cic_ussd.phone_number import Support
@ -18,7 +18,7 @@ def upsell_unregistered_recipient(state_machine_data: Tuple[str, dict, Account,
notifier = Notifier() notifier = Notifier()
phone_number = ussd_session.get('data')['recipient_phone_number'] phone_number = ussd_session.get('data')['recipient_phone_number']
preferred_language = get_cached_preferred_language(account.blockchain_address) preferred_language = get_cached_preferred_language(account.blockchain_address)
token_symbol = get_default_token_symbol() token_symbol = get_active_token_symbol(account.blockchain_address)
tx_sender_information = account.standard_metadata_id() tx_sender_information = account.standard_metadata_id()
notifier.send_sms_notification('sms.upsell_unregistered_recipient', notifier.send_sms_notification('sms.upsell_unregistered_recipient',
phone_number, phone_number,

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

@ -5,18 +5,17 @@ from typing import Tuple
# third party imports # third party imports
import celery import celery
from phonenumbers.phonenumberutil import NumberParseException from phonenumbers.phonenumberutil import NumberParseException
from sqlalchemy.orm.session import Session
# local imports # local imports
from cic_ussd.account.balance import get_cached_available_balance from cic_ussd.account.balance import get_cached_available_balance
from cic_ussd.account.chain import Chain from cic_ussd.account.chain import Chain
from cic_ussd.account.tokens import get_default_token_symbol from cic_ussd.account.tokens import get_active_token_symbol, get_cached_token_data
from cic_ussd.account.transaction import OutgoingTransaction from cic_ussd.account.transaction import OutgoingTransaction
from cic_ussd.db.enum import AccountStatus
from cic_ussd.db.models.account import Account from cic_ussd.db.models.account import Account
from cic_ussd.db.models.base import SessionBase
from cic_ussd.phone_number import process_phone_number, E164Format from cic_ussd.phone_number import process_phone_number, E164Format
from cic_ussd.session.ussd_session import save_session_data from cic_ussd.session.ussd_session import save_session_data
from sqlalchemy.orm.session import Session
logg = logging.getLogger(__file__) logg = logging.getLogger(__file__)
@ -63,7 +62,11 @@ def has_sufficient_balance(state_machine_data: Tuple[str, dict, Account, Session
:rtype: bool :rtype: bool
""" """
user_input, ussd_session, account, session = state_machine_data user_input, ussd_session, account, session = state_machine_data
return int(user_input) <= get_cached_available_balance(account.blockchain_address) identifier = bytes.fromhex(account.blockchain_address)
token_symbol = get_active_token_symbol(account.blockchain_address)
token_data = get_cached_token_data(account.blockchain_address, token_symbol)
decimals = token_data.get('decimals')
return int(user_input) <= get_cached_available_balance(decimals, [identifier, token_symbol.encode('utf-8')])
def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, Account, Session]): def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, Account, Session]):
@ -122,9 +125,10 @@ def process_transaction_request(state_machine_data: Tuple[str, dict, Account, Se
to_address = recipient.blockchain_address to_address = recipient.blockchain_address
from_address = account.blockchain_address from_address = account.blockchain_address
amount = int(ussd_session.get('data').get('transaction_amount')) amount = int(ussd_session.get('data').get('transaction_amount'))
token_symbol = get_default_token_symbol() token_symbol = get_active_token_symbol(account.blockchain_address)
token_data = get_cached_token_data(account.blockchain_address, token_symbol)
decimals = token_data.get('decimals')
outgoing_tx_processor = OutgoingTransaction(chain_str=chain_str, outgoing_tx_processor = OutgoingTransaction(chain_str=chain_str,
from_address=from_address, from_address=from_address,
to_address=to_address) to_address=to_address)
outgoing_tx_processor.transfer(amount=amount, token_symbol=token_symbol) outgoing_tx_processor.transfer(amount=amount, decimals=decimals, token_symbol=token_symbol)