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

# 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