Refactors state machine logic.

This commit is contained in:
PhilipWafula 2021-05-27 19:03:43 +03:00
parent 9e4ad4c650
commit a05df2280b
Signed by untrusted user: mango-habanero
GPG Key ID: B00CE9034DA19FB7
2 changed files with 49 additions and 31 deletions

View File

@ -13,7 +13,7 @@ import bcrypt
# local imports # local imports
from cic_ussd.db.models.account import AccountStatus, Account from cic_ussd.db.models.account import AccountStatus, Account
from cic_ussd.encoder import PasswordEncoder, create_password_hash from cic_ussd.encoder import PasswordEncoder, create_password_hash, check_password_hash
from cic_ussd.operations import persist_session_to_db_task, create_or_update_session from cic_ussd.operations import persist_session_to_db_task, create_or_update_session
from cic_ussd.redis import InMemoryStore from cic_ussd.redis import InMemoryStore
@ -78,6 +78,10 @@ def save_initial_pin_to_session_data(state_machine_data: Tuple[str, dict, Accoun
# set initial pin data # set initial pin data
initial_pin = create_password_hash(user_input) initial_pin = create_password_hash(user_input)
if ussd_session.get('session_data'):
session_data = ussd_session.get('session_data')
session_data['initial_pin'] = initial_pin
else:
session_data = { session_data = {
'initial_pin': initial_pin 'initial_pin': initial_pin
} }
@ -103,9 +107,8 @@ def pins_match(state_machine_data: Tuple[str, dict, Account]) -> bool:
""" """
user_input, ussd_session, user = state_machine_data user_input, ussd_session, user = state_machine_data
initial_pin = ussd_session.get('session_data').get('initial_pin') initial_pin = ussd_session.get('session_data').get('initial_pin')
fernet = PasswordEncoder(PasswordEncoder.key) logg.debug(f'USSD SESSION: {ussd_session}')
initial_pin = fernet.decrypt(initial_pin.encode()) return check_password_hash(user_input, initial_pin)
return bcrypt.checkpw(user_input.encode(), initial_pin)
def complete_pin_change(state_machine_data: Tuple[str, dict, Account]): def complete_pin_change(state_machine_data: Tuple[str, dict, Account]):

View File

@ -64,13 +64,17 @@ def process_gender_user_input(user: Account, user_input: str):
if user.preferred_language == 'en': if user.preferred_language == 'en':
if user_input == '1': if user_input == '1':
gender = 'Male' gender = 'Male'
else: elif user_input == '2':
gender = 'Female' gender = 'Female'
elif user_input == '3':
gender = 'Other'
else: else:
if user_input == '1': if user_input == '1':
gender = 'Mwanaume' gender = 'Mwanaume'
else: elif user_input == '2':
gender = 'Mwanamke' gender = 'Mwanamke'
elif user_input == '3':
gender = 'Nyingine'
return gender return gender
@ -88,14 +92,18 @@ def save_metadata_attribute_to_session_data(state_machine_data: Tuple[str, dict,
key = '' key = ''
if 'given_name' in current_state: if 'given_name' in current_state:
key = 'given_name' key = 'given_name'
elif 'family_name' in current_state:
if 'family_name' in current_state:
key = 'family_name' key = 'family_name'
elif 'gender' in current_state:
if 'gender' in current_state:
key = 'gender' key = 'gender'
user_input = process_gender_user_input(user=user, user_input=user_input) user_input = process_gender_user_input(user=user, user_input=user_input)
elif 'location' in current_state:
if 'location' in current_state:
key = 'location' key = 'location'
elif 'products' in current_state:
if 'products' in current_state:
key = 'products' key = 'products'
# check if there is existing session data # check if there is existing session data
@ -121,12 +129,20 @@ def format_user_metadata(metadata: dict, user: Account):
gender = metadata.get('gender') gender = metadata.get('gender')
given_name = metadata.get('given_name') given_name = metadata.get('given_name')
family_name = metadata.get('family_name') family_name = metadata.get('family_name')
# check whether there's existing location data
if isinstance(metadata.get('location'), dict):
location = metadata.get('location')
else:
location = { location = {
"area_name": metadata.get('location') "area_name": metadata.get('location')
} }
products = [] # check whether it is a list
if metadata.get('products'): if isinstance(metadata.get('products'), list):
products = metadata.get('products')
else:
products = metadata.get('products').split(',') products = metadata.get('products').split(',')
phone_number = user.phone_number phone_number = user.phone_number
date_registered = int(user.created.replace().timestamp()) date_registered = int(user.created.replace().timestamp())
blockchain_address = user.blockchain_address blockchain_address = user.blockchain_address
@ -192,28 +208,27 @@ def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, Account]):
# validate user metadata # validate user metadata
person = Person() person = Person()
user_metadata = json.loads(user_metadata) user_metadata = json.loads(user_metadata)
deserialized_person = person.deserialize(person_data=user_metadata)
# edit specific metadata attribute # edit specific metadata attribute
if given_name: if given_name:
deserialized_person.given_name = given_name user_metadata['given_name'] = given_name
elif family_name: if family_name:
deserialized_person.family_name = family_name user_metadata['family_name'] = family_name
elif gender: if gender:
deserialized_person.gender = gender user_metadata['gender'] = gender
elif location: if location:
# get existing location metadata: # get existing location metadata:
location_data = user_metadata.get('location') location_data = user_metadata.get('location')
location_data['area_name'] = location location_data['area_name'] = location
deserialized_person.location = location_data user_metadata['location'] = location_data
elif products: if products:
deserialized_person.products = products user_metadata['products'] = products
edited_metadata = deserialized_person.serialize() user_metadata = format_user_metadata(metadata=user_metadata, user=user)
s_edit_person_metadata = celery.signature( s_edit_person_metadata = celery.signature(
'cic_ussd.tasks.metadata.edit_person_metadata', 'cic_ussd.tasks.metadata.create_person_metadata',
[blockchain_address, edited_metadata] [blockchain_address, user_metadata]
) )
s_edit_person_metadata.apply_async(queue='cic-ussd') s_edit_person_metadata.apply_async(queue='cic-ussd')