cic-internal-integration/apps/cic-ussd/cic_ussd/state_machine/logic/transaction.py

120 lines
5.5 KiB
Python

# standard imports
import logging
from typing import Tuple
# third party imports
# local imports
from cic_ussd.accounts import BalanceManager
from cic_ussd.db.models.user import AccountStatus, User
from cic_ussd.operations import get_user_by_phone_number, save_to_in_memory_ussd_session_data
from cic_ussd.state_machine.state_machine import UssdStateMachine
from cic_ussd.transactions import OutgoingTransactionProcessor
logg = logging.getLogger(__file__)
def is_valid_recipient(state_machine_data: Tuple[str, dict, User]) -> bool:
"""This function checks that a user exists, is not the initiator of the transaction, has an active account status
and is authorized to perform standard transactions.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: tuple
:return: A user's validity
:rtype: bool
"""
user_input, ussd_session, user = state_machine_data
recipient = get_user_by_phone_number(phone_number=user_input)
is_not_initiator = user_input != user.phone_number
has_active_account_status = user.get_account_status() == AccountStatus.ACTIVE.name
logg.debug('This section requires implementation of checks for user roles and authorization status of an account.')
return is_not_initiator and has_active_account_status
def is_valid_token_agent(state_machine_data: Tuple[str, dict, User]) -> bool:
"""This function checks that a user exists, is not the initiator of the transaction, has an active account status
and is authorized to perform exchange transactions.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: tuple
:return: A user's validity
:rtype: bool
"""
user_input, ussd_session, user = state_machine_data
# is_token_agent = AccountRole.TOKEN_AGENT.value in user.get_user_roles()
logg.debug('This section requires implementation of user roles and authorization to facilitate exchanges.')
return is_valid_recipient(state_machine_data=state_machine_data)
def is_valid_transaction_amount(state_machine_data: Tuple[str, dict, User]) -> bool:
"""This function checks that the transaction amount provided is valid as per the criteria for the transaction
being attempted.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: tuple
:return: A transaction amount's validity
:rtype: bool
"""
user_input, ussd_session, user = state_machine_data
try:
return int(user_input) > 0
except ValueError:
return False
def has_sufficient_balance(state_machine_data: Tuple[str, dict, User]) -> bool:
"""This function checks that the transaction amount provided is valid as per the criteria for the transaction
being attempted.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: tuple
:return: An account balance's validity
:rtype: bool
"""
user_input, ussd_session, user = state_machine_data
balance_manager = BalanceManager(address=user.blockchain_address,
chain_str=UssdStateMachine.chain_str,
token_symbol='SRF')
balance = balance_manager.get_operational_balance()
return int(user_input) <= balance
def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, User]):
"""This function saves the phone number corresponding the intended recipients blockchain account.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: str
"""
user_input, ussd_session, user = state_machine_data
session_data = {
'recipient_phone_number': user_input
}
save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
def save_transaction_amount_to_session_data(state_machine_data: Tuple[str, dict, User]):
"""This function saves the phone number corresponding the intended recipients blockchain account.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: str
"""
user_input, ussd_session, user = state_machine_data
session_data = {
'transaction_amount': user_input
}
save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
def process_transaction_request(state_machine_data: Tuple[str, dict, User]):
"""This function saves the phone number corresponding the intended recipients blockchain account.
:param state_machine_data: A tuple containing user input, a ussd session and user object.
:type state_machine_data: str
"""
user_input, ussd_session, user = state_machine_data
# get user from phone number
recipient_phone_number = ussd_session.get('session_data').get('recipient_phone_number')
recipient = get_user_by_phone_number(phone_number=recipient_phone_number)
to_address = recipient.blockchain_address
from_address = user.blockchain_address
amount = int(ussd_session.get('session_data').get('transaction_amount'))
outgoing_tx_processor = OutgoingTransactionProcessor(chain_str=UssdStateMachine.chain_str,
from_address=from_address,
to_address=to_address)
outgoing_tx_processor.process_outgoing_transfer_transaction(amount=amount)