244 lines
12 KiB
Python
244 lines
12 KiB
Python
# standard imports
|
|
import json
|
|
import uuid
|
|
|
|
# third party imports
|
|
import pytest
|
|
|
|
# local imports
|
|
from cic_ussd.chain import Chain
|
|
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
|
|
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,
|
|
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,
|
|
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,
|
|
celery_session_worker,
|
|
create_activated_user,
|
|
create_in_db_ussd_session):
|
|
chain_str = Chain.spec.__str__()
|
|
response = process_menu_interaction_requests(
|
|
chain_str=chain_str,
|
|
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
|