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

The great bump

Browse Source
This commit is contained in:
Philip Wafula
2021-08-06 16:29:01 +00:00
parent f764b73f66
commit 0672a17d2e
195 changed files with 5791 additions and 4983 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
apps/cic-ussd/tests/cic_ussd/processor/test_menu.py Normal file
View File

@@ -0,0 +1,183 @@
# standard imports
import json
# external imports
from chainlib.hash import strip_0x
# local imports
from cic_ussd.account.balance import get_cached_available_balance
from cic_ussd.account.metadata import get_cached_preferred_language
from cic_ussd.account.statement import (
get_cached_statement,
parse_statement_transactions,
statement_transaction_set
)
from cic_ussd.account.tokens import get_default_token_symbol
from cic_ussd.account.transaction import from_wei, to_wei
from cic_ussd.menu.ussd_menu import UssdMenu
from cic_ussd.metadata import PersonMetadata
from cic_ussd.phone_number import Support
from cic_ussd.processor.menu import response
from cic_ussd.processor.util import parse_person_metadata
from cic_ussd.translation import translation_for
# test imports
def test_menu_processor(activated_account,
balances,
cache_balances,
cache_default_token_data,
cache_preferences,
cache_person_metadata,
cache_statement,
celery_session_worker,
generic_ussd_session,
init_database,
load_chain_spec,
load_support_phone,
load_ussd_menu,
mock_sync_balance_api_query,
mock_transaction_list_query,
valid_recipient):
preferred_language = get_cached_preferred_language(activated_account.blockchain_address)
available_balance = get_cached_available_balance(activated_account.blockchain_address)
token_symbol = get_default_token_symbol()
tax = ''
bonus = ''
display_key = 'ussd.kenya.account_balances'
ussd_menu = UssdMenu.find_by_name('account_balances')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key,
preferred_language,
available_balance=available_balance,
tax=tax,
bonus=bonus,
token_symbol=token_symbol)
cached_statement = get_cached_statement(activated_account.blockchain_address)
statement = json.loads(cached_statement)
statement_transactions = parse_statement_transactions(statement)
transaction_sets = [statement_transactions[tx:tx + 3] for tx in range(0, len(statement_transactions), 3)]
first_transaction_set = []
middle_transaction_set = []
last_transaction_set = []
if transaction_sets:
first_transaction_set = statement_transaction_set(preferred_language, transaction_sets[0])
if len(transaction_sets) >= 2:
middle_transaction_set = statement_transaction_set(preferred_language, transaction_sets[1])
if len(transaction_sets) >= 3:
last_transaction_set = statement_transaction_set(preferred_language, transaction_sets[2])
display_key = 'ussd.kenya.first_transaction_set'
ussd_menu = UssdMenu.find_by_name('first_transaction_set')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key, preferred_language, first_transaction_set=first_transaction_set)
display_key = 'ussd.kenya.middle_transaction_set'
ussd_menu = UssdMenu.find_by_name('middle_transaction_set')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key, preferred_language, middle_transaction_set=middle_transaction_set)
display_key = 'ussd.kenya.last_transaction_set'
ussd_menu = UssdMenu.find_by_name('last_transaction_set')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key, preferred_language, last_transaction_set=last_transaction_set)
display_key = 'ussd.kenya.display_user_metadata'
ussd_menu = UssdMenu.find_by_name('display_user_metadata')
name = ussd_menu.get('name')
identifier = bytes.fromhex(strip_0x(activated_account.blockchain_address))
person_metadata = PersonMetadata(identifier)
cached_person_metadata = person_metadata.get_cached_metadata()
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == parse_person_metadata(cached_person_metadata, display_key, preferred_language)
display_key = 'ussd.kenya.account_balances_pin_authorization'
ussd_menu = UssdMenu.find_by_name('account_balances_pin_authorization')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(f'{display_key}.first', preferred_language)
activated_account.failed_pin_attempts = 1
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
retry_pin_entry = translation_for('ussd.kenya.retry_pin_entry', preferred_language, remaining_attempts=2)
assert resp == translation_for(f'{display_key}.retry', preferred_language, retry_pin_entry=retry_pin_entry)
activated_account.failed_pin_attempts = 0
display_key = 'ussd.kenya.start'
ussd_menu = UssdMenu.find_by_name('start')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key,
preferred_language,
account_balance=available_balance,
account_token_name=token_symbol)
display_key = 'ussd.kenya.transaction_pin_authorization'
ussd_menu = UssdMenu.find_by_name('transaction_pin_authorization')
name = ussd_menu.get('name')
generic_ussd_session['data'] = {
'recipient_phone_number': valid_recipient.phone_number,
'transaction_amount': '15'
}
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
user_input = generic_ussd_session.get('data').get('transaction_amount')
transaction_amount = to_wei(value=int(user_input))
tx_recipient_information = valid_recipient.standard_metadata_id()
tx_sender_information = activated_account.standard_metadata_id()
assert resp == translation_for(f'{display_key}.first',
preferred_language,
recipient_information=tx_recipient_information,
transaction_amount=from_wei(transaction_amount),
token_symbol=token_symbol,
sender_information=tx_sender_information)
display_key = 'ussd.kenya.exit_insufficient_balance'
ussd_menu = UssdMenu.find_by_name('exit_insufficient_balance')
name = ussd_menu.get('name')
generic_ussd_session['data'] = {
'recipient_phone_number': valid_recipient.phone_number,
'transaction_amount': '85'
}
transaction_amount = generic_ussd_session.get('data').get('transaction_amount')
transaction_amount = to_wei(value=int(transaction_amount))
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key,
preferred_language,
amount=from_wei(transaction_amount),
token_symbol=token_symbol,
recipient_information=tx_recipient_information,
token_balance=available_balance)
display_key = 'ussd.kenya.exit_invalid_menu_option'
ussd_menu = UssdMenu.find_by_name('exit_invalid_menu_option')
name = ussd_menu.get('name')
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key, preferred_language, support_phone=Support.phone_number)
display_key = 'ussd.kenya.exit_successful_transaction'
ussd_menu = UssdMenu.find_by_name('exit_successful_transaction')
name = ussd_menu.get('name')
generic_ussd_session['data'] = {
'recipient_phone_number': valid_recipient.phone_number,
'transaction_amount': '15'
}
transaction_amount = generic_ussd_session.get('data').get('transaction_amount')
transaction_amount = to_wei(value=int(transaction_amount))
resp = response(activated_account, display_key, name, init_database, generic_ussd_session)
assert resp == translation_for(display_key,
preferred_language,
transaction_amount=from_wei(transaction_amount),
token_symbol=token_symbol,
recipient_information=tx_recipient_information,
sender_information=tx_sender_information)

121
apps/cic-ussd/tests/cic_ussd/processor/test_ussd.py Normal file
View File

@@ -0,0 +1,121 @@
# standard imports
import json
import os
import time
# external imports
import i18n
import requests_mock
from chainlib.hash import strip_0x
# local imports
from cic_ussd.account.chain import Chain
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.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_language_selection')
assert menu_resp.get('name') == ussd_menu.get('name')
identifier = bytes.fromhex(strip_0x(pending_account.blockchain_address))
key = cache_data_key(identifier, ':cic.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_menu_operations(activated_account,
cache_preferences,
celery_session_worker,
generic_ussd_session,
init_database,
init_cache,
load_chain_spec,
load_config,
mock_account_creation_task_result,
persisted_ussd_session,
person_metadata,
set_locale_files,
setup_metadata_request_handler,
setup_metadata_signer,
task_uuid):
# sourcery skip: extract-duplicate-method
chain_str = Chain.spec.__str__()
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')
resp = handle_menu_operations(chain_str, external_session_id, phone, None, valid_service_codes[0], init_database, '4444')
assert resp == translation_for('ussd.kenya.account_creation_prompt', preferred_language)
cached_ussd_session = get_cached_data(external_session_id)
ussd_session = json.loads(cached_ussd_session)
assert ussd_session['msisdn'] == phone
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'
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(chain_str, external_session_id, phone, None, valid_service_codes[0], init_database, '1')
assert resp == translation_for('ussd.kenya.enter_transaction_recipient', preferred_language)

62
apps/cic-ussd/tests/cic_ussd/processor/test_util.py Normal file
View File

@@ -0,0 +1,62 @@
# standard imports
import datetime
import json
# external imports
import pytest
from chainlib.hash import strip_0x
from cic_types.models.person import get_contact_data_from_vcard
# local imports
from cic_ussd.account.metadata import get_cached_preferred_language
from cic_ussd.metadata import PersonMetadata
from cic_ussd.processor.util import latest_input, parse_person_metadata, resume_last_ussd_session
from cic_ussd.translation import translation_for
# test imports
@pytest.mark.parametrize('user_input, expected_value', [
('1*9*6*7', '7'),
('1', '1'),
('', '')
])
def test_latest_input(user_input, expected_value):
assert latest_input(user_input) == expected_value
def test_parse_person_metadata(activated_account, cache_person_metadata, cache_preferences):
identifier = bytes.fromhex(strip_0x(activated_account.blockchain_address))
person_metadata = PersonMetadata(identifier)
cached_person_metadata = person_metadata.get_cached_metadata()
person_metadata = json.loads(cached_person_metadata)
preferred_language = get_cached_preferred_language(activated_account.blockchain_address)
display_key = 'ussd.kenya.display_person_metadata'
parsed_person_metadata = parse_person_metadata(cached_person_metadata,
display_key,
preferred_language)
contact_data = get_contact_data_from_vcard(person_metadata.get('vcard'))
full_name = f'{contact_data.get("given")} {contact_data.get("family")}'
date_of_birth = person_metadata.get('date_of_birth')
year_of_birth = date_of_birth.get('year')
present_year = datetime.datetime.now().year
age = present_year - year_of_birth
gender = person_metadata.get('gender')
products = ', '.join(person_metadata.get('products'))
location = person_metadata.get('location').get('area_name')
assert parsed_person_metadata == translation_for(key=display_key,
preferred_language=preferred_language,
full_name=full_name,
age=age,
gender=gender,
location=location,
products=products)
@pytest.mark.parametrize('last_state, expected_menu_name', [
('account_creation_prompt', 'start'),
('enter_transaction_recipient', 'enter_transaction_recipient')
])
def test_resume_last_ussd_session(expected_menu_name, last_state, load_ussd_menu):
assert resume_last_ussd_session(last_state).get('name') == expected_menu_name