cic-internal-integration/apps/cic-ussd/tests/cic_ussd/test_operations.py

244 lines
12 KiB
Python
Raw Permalink Normal View History

2021-02-06 16:13:47 +01:00
# standard imports
import json
import uuid
# third party imports
import pytest
# local imports
from cic_ussd.chain import Chain
2021-02-06 16:13:47 +01:00
from cic_ussd.db.models.task_tracker import TaskTracker
from cic_ussd.menu.ussd_menu import UssdMenu
from cic_ussd.operations import (add_tasks_to_tracker,
create_ussd_session,
create_or_update_session,
define_response_with_content,
define_multilingual_responses,
get_account_status,
get_latest_input,
initiate_account_creation_request,
process_current_menu,
process_menu_interaction_requests,
cache_account_creation_task_id,
reset_pin,
update_ussd_session,
save_to_in_memory_ussd_session_data)
from cic_ussd.phone_number import get_user_by_phone_number,process_phone_number
2021-02-06 16:13:47 +01:00
from cic_ussd.transactions import truncate
from cic_ussd.redis import InMemoryStore
from cic_ussd.session.ussd_session import UssdSession as InMemoryUssdSession
def test_add_tasks_to_tracker(init_database):
task_uuid = '31e85315-feee-4b6d-995e-223569082cc4'
session = init_database
assert len(session.query(TaskTracker).all()) == 0
add_tasks_to_tracker(task_uuid=task_uuid)
task_in_tracker = session.query(TaskTracker).filter_by(task_uuid=task_uuid).first()
assert task_in_tracker.id == 1
assert task_in_tracker.task_uuid == task_uuid
def test_create_ussd_session(create_in_redis_ussd_session, ussd_session_data):
external_session_id = ussd_session_data.get('external_session_id')
ussd_session = create_ussd_session(
external_session_id=external_session_id,
service_code=ussd_session_data.get('service_code'),
phone=ussd_session_data.get('msisdn'),
user_input=ussd_session_data.get('user_input'),
current_menu=ussd_session_data.get('state')
)
in_memory_ussd_session = create_in_redis_ussd_session.get(external_session_id)
assert json.loads(in_memory_ussd_session).get('external_session_id') == ussd_session.external_session_id
def test_create_or_update_session(init_database, create_in_redis_ussd_session, ussd_session_data):
external_session_id = ussd_session_data.get('external_session_id')
ussd_session = create_or_update_session(external_session_id=external_session_id,
service_code=ussd_session_data.get('service_code'),
phone=ussd_session_data.get('msisdn'),
user_input=ussd_session_data.get('user_input'),
current_menu=ussd_session_data.get('state'))
in_memory_ussd_session = create_in_redis_ussd_session.get(external_session_id)
assert json.loads(in_memory_ussd_session).get('external_session_id') == ussd_session.external_session_id
@pytest.mark.parametrize('headers, response, expected_result',[
([('Content-Type', 'text/plain')], 'some-text', (b'some-text', [('Content-Type', 'text/plain'), ('Content-Length', '9')])),
([('Content-Type', 'text/plain'), ('Content-Length', '0')], 'some-text', (b'some-text', [('Content-Type', 'text/plain'), ('Content-Length', '9')]))
])
def test_define_response_with_content(headers, response, expected_result):
response_bytes, headers = define_response_with_content(headers=headers, response=response)
assert response_bytes, headers == expected_result
def test_define_multilingual_responses(load_ussd_menu, set_locale_files):
response = define_multilingual_responses(
key='ussd.kenya.account_creation_prompt', locales=['en', 'sw'], prefix='END')
assert response == "END Your account is being created. You will receive an SMS when your account is ready.\nAkaunti yako ya Sarafu inatayarishwa. Utapokea ujumbe wa SMS akaunti yako ikiwa tayari.\n"
def test_get_account_status(create_pending_user):
user = create_pending_user
assert get_account_status(user.phone_number) == 'PENDING'
@pytest.mark.parametrize('user_input, expected_value', [
('1*9*6*7', '7'),
('1', '1'),
('', '')
])
def test_get_latest_input(user_input, expected_value):
assert get_latest_input(user_input=user_input) == expected_value
def test_initiate_account_creation_request(account_creation_action_data,
create_in_redis_ussd_session,
init_database,
load_config,
load_ussd_menu,
mocker,
setup_chain_spec,
2021-02-06 16:13:47 +01:00
set_locale_files,
ussd_session_data):
external_session_id = ussd_session_data.get('external_session_id')
phone_number = account_creation_action_data.get('phone_number')
task_id = account_creation_action_data.get('task_id')
class Callable:
id = task_id
mocker.patch('cic_eth.api.api_task.Api.create_account', return_value=Callable)
mocked_cache_function = mocker.patch('cic_ussd.operations.cache_account_creation_task_id')
mocked_cache_function(phone_number, task_id)
chain_str = Chain.spec.__str__()
response = initiate_account_creation_request(chain_str=chain_str,
2021-02-06 16:13:47 +01:00
external_session_id=external_session_id,
phone_number=ussd_session_data.get('msisdn'),
service_code=ussd_session_data.get('service_code'),
user_input=ussd_session_data.get('user_input'))
in_memory_ussd_session = InMemoryUssdSession.redis_cache.get(external_session_id)
# check that ussd session was created
assert json.loads(in_memory_ussd_session).get('external_session_id') == external_session_id
assert response == "END Your account is being created. You will receive an SMS when your account is ready.\nAkaunti yako ya Sarafu inatayarishwa. Utapokea ujumbe wa SMS akaunti yako ikiwa tayari.\n"
def test_reset_pin(create_pin_blocked_user):
user = create_pin_blocked_user
assert user.get_account_status() == 'LOCKED'
reset_pin(user.phone_number)
assert user.get_account_status() == 'RESET'
def test_update_ussd_session(create_in_redis_ussd_session, load_ussd_menu, ussd_session_data):
external_session_id = ussd_session_data.get('external_session_id')
ussd_session = create_ussd_session(external_session_id=external_session_id,
service_code=ussd_session_data.get('service_code'),
phone=ussd_session_data.get('msisdn'),
user_input=ussd_session_data.get('user_input'),
current_menu=ussd_session_data.get('state')
)
assert ussd_session.user_input == ussd_session_data.get('user_input')
assert ussd_session.state == ussd_session_data.get('state')
ussd_session = update_ussd_session(ussd_session=ussd_session, user_input='1*2', current_menu='initial_pin_entry')
assert ussd_session.user_input == '1*2'
assert ussd_session.state == 'initial_pin_entry'
def test_process_current_menu(create_activated_user, create_in_db_ussd_session):
ussd_session = create_in_db_ussd_session
current_menu = process_current_menu(ussd_session=ussd_session, user=create_activated_user, user_input="")
assert current_menu == UssdMenu.find_by_name(name='exit_invalid_input')
current_menu = process_current_menu(ussd_session=None, user=create_activated_user, user_input="1*0000")
assert current_menu == UssdMenu.find_by_name(name='start')
def test_cache_account_creation_task_id(init_redis_cache):
phone_number = '+25412345678'
task_id = str(uuid.uuid4())
cache_account_creation_task_id(phone_number=phone_number, task_id=task_id)
redis_cache = init_redis_cache
action_data = redis_cache.get(task_id)
action_data = json.loads(action_data)
assert action_data.get('phone_number') == phone_number
assert action_data.get('sms_notification_sent') is False
assert action_data.get('status') == 'PENDING'
assert action_data.get('task_id') == task_id
def test_save_to_in_memory_ussd_session_data(celery_session_worker,
create_in_db_ussd_session,
create_in_redis_ussd_session,
init_database):
in_memory_ussd_session = InMemoryStore.cache.get('AT974186')
in_memory_ussd_session = json.loads(in_memory_ussd_session)
assert in_memory_ussd_session.get('session_data') == {}
session_data = {
'some_test_key': 'some_test_value'
}
save_to_in_memory_ussd_session_data(
queue='cic-ussd',
session_data=session_data,
ussd_session=create_in_db_ussd_session.to_json()
)
in_memory_ussd_session = InMemoryStore.cache.get('AT974186')
in_memory_ussd_session = json.loads(in_memory_ussd_session)
assert in_memory_ussd_session.get('session_data') == session_data
@pytest.mark.parametrize("external_session_id, phone_number, expected_response", [
("AT123456789", "+254700000000", "END Your account is being created. You will receive an SMS when your account is ready.\nAkaunti yako ya Sarafu inatayarishwa. Utapokea ujumbe wa SMS akaunti yako ikiwa tayari.\n"),
("AT974186", "+25498765432", "CON Please enter a PIN to manage your account.\n0. Back")
])
def test_process_menu_interaction_requests(external_session_id,
phone_number,
expected_response,
load_ussd_menu,
load_data_into_state_machine,
load_config,
setup_chain_spec,
2021-02-06 16:13:47 +01:00
celery_session_worker,
create_activated_user,
create_in_db_ussd_session):
chain_str = Chain.spec.__str__()
2021-02-06 16:13:47 +01:00
response = process_menu_interaction_requests(
chain_str=chain_str,
2021-02-06 16:13:47 +01:00
external_session_id=external_session_id,
phone_number=phone_number,
queue='cic-ussd',
service_code=load_config.get('APP_SERVICE_CODE'),
user_input='1'
)
assert response == expected_response
@pytest.mark.parametrize("phone_number, region, expected_result", [
("0712345678", "KE", "+254712345678"),
("+254787654321", "KE", "+254787654321")
])
def test_process_phone_number(expected_result, phone_number, region):
processed_phone_number = process_phone_number(phone_number=phone_number, region=region)
assert processed_phone_number == expected_result
def test_get_user_by_phone_number(create_activated_user):
known_phone_number = create_activated_user.phone_number
user = get_user_by_phone_number(phone_number=known_phone_number)
assert user is not None
assert create_activated_user.blockchain_address == user.blockchain_address
unknown_phone_number = '+254700000000'
user = get_user_by_phone_number(phone_number=unknown_phone_number)
assert user is None