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

Philip/management integration tests

Browse Source
This commit is contained in:
Philip Wafula
2021-06-03 13:40:51 +00:00
parent 69ae9b7c07
commit ed9e032890
31 changed files with 3437 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
apps/cic-ussd/cic_ussd/db/ussd_menu.json
View File

@@ -238,13 +238,43 @@
"description": "Menu to display a user's entire profile",
"display_key": "ussd.kenya.display_user_metadata",
"name": "display_user_metadata",
"parent": "account_management"
"parent": "metadata_management"
},
"41": {
"description": "The recipient is not in the system",
"display_key": "ussd.kenya.exit_invalid_recipient",
"name": "exit_invalid_recipient",
"parent": null
},
"42": {
"description": "Pin entry menu for changing name data.",
"display_key": "ussd.kenya.name_edit_pin_authorization",
"name": "name_edit_pin_authorization",
"parent": "metadata_management"
},
"43": {
"description": "Pin entry menu for changing gender data.",
"display_key": "ussd.kenya.gender_edit_pin_authorization",
"name": "gender_edit_pin_authorization",
"parent": "metadata_management"
},
"44": {
"description": "Pin entry menu for changing location data.",
"display_key": "ussd.kenya.location_edit_pin_authorization",
"name": "location_edit_pin_authorization",
"parent": "metadata_management"
},
"45": {
"description": "Pin entry menu for changing products data.",
"display_key": "ussd.kenya.products_edit_pin_authorization",
"name": "products_edit_pin_authorization",
"parent": "metadata_management"
},
"46": {
"description": "Pin confirmation for pin change.",
"display_key": "ussd.kenya.new_pin_confirmation",
"name": "new_pin_confirmation",
"parent": "metadata_management"
}
}

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

@@ -78,28 +78,27 @@ class MetadataRequestsHandler(Metadata):
:param data: The data to be stored in the metadata server.
:type data: dict|str
"""
data = json.dumps(data).encode('utf-8')
data = json.dumps(data)
result = make_request(method='POST', url=self.url, data=data, headers=self.headers)
metadata_http_error_handler(result=result)
metadata = result.content
metadata = result.json()
self.edit(data=metadata)
def edit(self, data: bytes):
def edit(self, data: Union[Dict, str]):
""" This function is responsible for editing data in the metadata server corresponding to a unique pointer.
:param data: The data to be edited in the metadata server.
:type data: bytes
:type data: dict
"""
cic_meta_signer = Signer()
signature = cic_meta_signer.sign_digest(data=data)
algorithm = cic_meta_signer.get_operational_key().get('algo')
decoded_data = data.decode('utf-8')
formatted_data = {
'm': data.decode('utf-8'),
'm': json.dumps(data),
's': {
'engine': self.engine,
'algo': algorithm,
'data': signature,
'digest': json.loads(data).get('digest'),
'digest': data.get('digest'),
}
}
formatted_data = json.dumps(formatted_data)
@@ -110,19 +109,32 @@ class MetadataRequestsHandler(Metadata):
decoded_identifier = self.identifier.decode("utf-8")
except UnicodeDecodeError:
decoded_identifier = self.identifier.hex()
logg.info(f'identifier: {decoded_identifier}. metadata pointer: {self.metadata_pointer} set to: {decoded_data}.')
logg.info(f'identifier: {decoded_identifier}. metadata pointer: {self.metadata_pointer} set to: {data}.')
def query(self):
"""This function is responsible for querying the metadata server for data corresponding to a unique pointer."""
"""
:return:
:rtype:
"""
# retrieve the metadata
result = make_request(method='GET', url=self.url)
metadata_http_error_handler(result=result)
response_data = result.json()
data = json.loads(response_data)
if not isinstance(data, dict):
raise ValueError(f'Invalid data object: {data}.')
# json serialize retrieved data
result_data = result.json()
# validate result data format
if not isinstance(result_data, dict):
raise ValueError(f'Invalid result data object: {result_data}.')
if result.status_code == 200 and self.cic_type == ':cic.person':
# validate person metadata
person = Person()
deserialized_person = person.deserialize(person_data=data)
data = json.dumps(deserialized_person.serialize())
cache_data(self.metadata_pointer, data=data)
person_data = person.deserialize(person_data=result_data)
# format new person data for caching
data = json.dumps(person_data.serialize())
# cache metadata
cache_data(key=self.metadata_pointer, data=data)
logg.debug(f'caching: {data} with key: {self.metadata_pointer}')

3
apps/cic-ussd/cic_ussd/metadata/signer.py
View File

@@ -47,14 +47,13 @@ class Signer:
logg.debug(f'using signing key: {key_id}, algorithm: {key_algorithm}')
return gpg_keys[0]
def sign_digest(self, data: bytes):
def sign_digest(self, data: dict):
"""
:param data:
:type data:
:return:
:rtype:
"""
data = json.loads(data)
digest = data['digest']
key_id = self.get_operational_key().get('keyid')
signature = self.gpg.sign(digest, passphrase=self.gpg_passphrase, keyid=key_id)

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

@@ -251,9 +251,9 @@ def process_display_user_metadata(user: Account, display_key: str):
identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
cic_type=':cic.person'
)
user_metadata = get_cached_data(key)
if user_metadata:
user_metadata = json.loads(user_metadata)
cached_metadata = get_cached_data(key)
if cached_metadata:
user_metadata = json.loads(cached_metadata)
contact_data = get_contact_data_from_vcard(vcard=user_metadata.get('vcard'))
logg.debug(f'{contact_data}')
full_name = f'{contact_data.get("given")} {contact_data.get("family")}'
@@ -433,7 +433,8 @@ def process_request(user_input: str, user: Account, ussd_session: Optional[dict]
'exit_invalid_pin',
'exit_invalid_new_pin',
'exit_pin_mismatch',
'exit_invalid_request'
'exit_invalid_request',
'exit_successful_transaction'
] and person_metadata is not None:
return UssdMenu.find_by_name(name='start')
else:

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

@@ -13,7 +13,7 @@ import bcrypt
# local imports
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.redis import InMemoryStore
@@ -78,9 +78,13 @@ def save_initial_pin_to_session_data(state_machine_data: Tuple[str, dict, Accoun
# set initial pin data
initial_pin = create_password_hash(user_input)
session_data = {
'initial_pin': initial_pin
}
if ussd_session.get('session_data'):
session_data = ussd_session.get('session_data')
session_data['initial_pin'] = initial_pin
else:
session_data = {
'initial_pin': initial_pin
}
# create new in memory ussd session with current ussd session data
create_or_update_session(
@@ -103,9 +107,8 @@ def pins_match(state_machine_data: Tuple[str, dict, Account]) -> bool:
"""
user_input, ussd_session, user = state_machine_data
initial_pin = ussd_session.get('session_data').get('initial_pin')
fernet = PasswordEncoder(PasswordEncoder.key)
initial_pin = fernet.decrypt(initial_pin.encode())
return bcrypt.checkpw(user_input.encode(), initial_pin)
logg.debug(f'USSD SESSION: {ussd_session}')
return check_password_hash(user_input, initial_pin)
def complete_pin_change(state_machine_data: Tuple[str, dict, Account]):

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

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

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

@@ -19,4 +19,6 @@ def translation_for(key: str, preferred_language: Optional[str] = None, **kwargs
"""
if preferred_language:
i18n.set('locale', preferred_language)
else:
i18n.set('locale', i18n.config.get('fallback'))
return i18n.t(key, **kwargs)