149 lines
7.5 KiB
Python
149 lines
7.5 KiB
Python
# standard imports
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
# external imports
|
|
import i18n
|
|
import requests_mock
|
|
from chainlib.hash import strip_0x
|
|
from cic_types.condiments import MetadataPointer
|
|
|
|
# local imports
|
|
from cic_ussd.account.metadata import get_cached_preferred_language
|
|
from cic_ussd.cache import cache_data, cache_data_key, get_cached_data
|
|
from cic_ussd.db.models.task_tracker import TaskTracker
|
|
from cic_ussd.menu.ussd_menu import UssdMenu
|
|
from cic_ussd.metadata import PersonMetadata
|
|
from cic_ussd.processor.ussd import (get_menu,
|
|
handle_menu,
|
|
handle_menu_operations)
|
|
from cic_ussd.processor.util import ussd_menu_list
|
|
from cic_ussd.state_machine.logic.language import preferred_langauge_from_selection
|
|
from cic_ussd.translation import translation_for
|
|
|
|
# test imports
|
|
from tests.helpers.accounts import phone_number
|
|
|
|
|
|
def test_handle_menu(activated_account,
|
|
init_cache,
|
|
init_database,
|
|
load_ussd_menu,
|
|
pending_account,
|
|
persisted_ussd_session,
|
|
pin_blocked_account,
|
|
preferences):
|
|
persisted_ussd_session.state = 'account_creation_prompt'
|
|
menu_resp = handle_menu(activated_account, init_database)
|
|
ussd_menu = UssdMenu.find_by_name('start')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
persisted_ussd_session.state = 'display_user_metadata'
|
|
menu_resp = handle_menu(activated_account, init_database)
|
|
ussd_menu = UssdMenu.find_by_name('display_user_metadata')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
menu_resp = handle_menu(pin_blocked_account, init_database)
|
|
ussd_menu = UssdMenu.find_by_name('exit_pin_blocked')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
menu_resp = handle_menu(pending_account, init_database)
|
|
ussd_menu = UssdMenu.find_by_name('initial_pin_entry')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
identifier = bytes.fromhex(strip_0x(pending_account.blockchain_address))
|
|
key = cache_data_key(identifier, MetadataPointer.PREFERENCES)
|
|
cache_data(key, json.dumps(preferences))
|
|
time.sleep(2)
|
|
menu_resp = handle_menu(pending_account, init_database)
|
|
ussd_menu = UssdMenu.find_by_name('initial_pin_entry')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
|
|
|
|
def test_get_menu(activated_account,
|
|
cache_preferences,
|
|
cache_person_metadata,
|
|
generic_ussd_session,
|
|
init_database,
|
|
init_state_machine,
|
|
persisted_ussd_session):
|
|
menu_resp = get_menu(activated_account, init_database, '', generic_ussd_session)
|
|
ussd_menu = UssdMenu.find_by_name(name='exit_invalid_input')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
|
|
menu_resp = get_menu(activated_account, init_database, '1111', None)
|
|
ussd_menu = UssdMenu.find_by_name('initial_language_selection')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
|
|
generic_ussd_session['state'] = 'start'
|
|
menu_resp = get_menu(activated_account, init_database, '1', generic_ussd_session)
|
|
ussd_menu = UssdMenu.find_by_name(name='enter_transaction_recipient')
|
|
assert menu_resp.get('name') == ussd_menu.get('name')
|
|
|
|
|
|
def test_handle_no_account_menu_operations(celery_session_worker,
|
|
init_cache,
|
|
init_database,
|
|
load_chain_spec,
|
|
load_config,
|
|
load_languages,
|
|
load_ussd_menu,
|
|
mock_account_creation_task_result,
|
|
pending_account,
|
|
persisted_ussd_session,
|
|
set_locale_files,
|
|
task_uuid):
|
|
initial_language_selection = 'ussd.initial_language_selection'
|
|
phone = phone_number()
|
|
external_session_id = os.urandom(20).hex()
|
|
valid_service_codes = load_config.get('USSD_SERVICE_CODE').split(",")
|
|
preferred_language = i18n.config.get('fallback')
|
|
key = cache_data_key('system:languages'.encode('utf-8'), MetadataPointer.NONE)
|
|
cached_system_languages = get_cached_data(key)
|
|
language_list: list = json.loads(cached_system_languages)
|
|
fallback = translation_for('helpers.no_language_list', preferred_language)
|
|
language_list_sets = ussd_menu_list(fallback=fallback, menu_list=language_list, split=3)
|
|
resp = handle_menu_operations(external_session_id, phone, None, valid_service_codes[0], init_database, '')
|
|
assert resp == translation_for(initial_language_selection, preferred_language,
|
|
first_language_set=language_list_sets[0])
|
|
cached_ussd_session = get_cached_data(external_session_id)
|
|
ussd_session = json.loads(cached_ussd_session)
|
|
assert ussd_session['msisdn'] == phone
|
|
persisted_ussd_session.external_session_id = external_session_id
|
|
persisted_ussd_session.msisdn = phone
|
|
persisted_ussd_session.state = initial_language_selection[5:]
|
|
init_database.add(persisted_ussd_session)
|
|
init_database.commit()
|
|
account_creation_prompt = 'ussd.account_creation_prompt'
|
|
user_input = '2'
|
|
resp = handle_menu_operations(external_session_id, phone, None, valid_service_codes[0], init_database, user_input)
|
|
preferred_language = preferred_langauge_from_selection(user_input)
|
|
assert resp == translation_for(account_creation_prompt, preferred_language)
|
|
task_tracker = init_database.query(TaskTracker).filter_by(task_uuid=task_uuid).first()
|
|
assert task_tracker.task_uuid == task_uuid
|
|
cached_creation_task_uuid = get_cached_data(task_uuid)
|
|
creation_task_uuid_data = json.loads(cached_creation_task_uuid)
|
|
assert creation_task_uuid_data['status'] == 'PENDING'
|
|
|
|
|
|
def test_handle_account_menu_operations(activated_account,
|
|
cache_preferences,
|
|
celery_session_worker,
|
|
init_database,
|
|
load_config,
|
|
persisted_ussd_session,
|
|
person_metadata,
|
|
set_locale_files,
|
|
setup_metadata_request_handler,
|
|
setup_metadata_signer, ):
|
|
valid_service_codes = load_config.get('USSD_SERVICE_CODE').split(",")
|
|
identifier = bytes.fromhex(strip_0x(activated_account.blockchain_address))
|
|
person_metadata_client = PersonMetadata(identifier)
|
|
with requests_mock.Mocker(real_http=False) as request_mocker:
|
|
request_mocker.register_uri('GET', person_metadata_client.url, status_code=200, reason='OK',
|
|
json=person_metadata)
|
|
person_metadata_client.query()
|
|
external_session_id = os.urandom(20).hex()
|
|
phone = activated_account.phone_number
|
|
preferred_language = get_cached_preferred_language(activated_account.blockchain_address)
|
|
persisted_ussd_session.state = 'enter_transaction_recipient'
|
|
resp = handle_menu_operations(external_session_id, phone, None, valid_service_codes[0], init_database, '1')
|
|
assert resp == translation_for('ussd.enter_transaction_recipient', preferred_language)
|