diff --git a/apps/cic-ussd/tests/cic_ussd/account/test_guardianship.py b/apps/cic-ussd/tests/cic_ussd/account/test_guardianship.py new file mode 100644 index 00000000..ccfbed48 --- /dev/null +++ b/apps/cic-ussd/tests/cic_ussd/account/test_guardianship.py @@ -0,0 +1,21 @@ +# standard imports +import os + +# external imports + +# local imports +from cic_ussd.account.guardianship import Guardianship + +# test imports +from tests.fixtures.config import root_directory + + +def test_guardianship(load_config, setup_guardianship): + guardians_file = os.path.join(root_directory, load_config.get('SYSTEM_GUARDIANS_FILE')) + with open(guardians_file, 'r') as system_guardians: + guardians = [line.strip() for line in system_guardians] + assert Guardianship.guardians == guardians + + guardianship = Guardianship() + assert guardianship.is_system_guardian(Guardianship.guardians[0]) is True + assert guardianship.is_system_guardian('+254712345678') is False diff --git a/apps/cic-ussd/tests/cic_ussd/metadata/test_tokens_meta.py b/apps/cic-ussd/tests/cic_ussd/metadata/test_tokens_meta.py new file mode 100644 index 00000000..fb8aae79 --- /dev/null +++ b/apps/cic-ussd/tests/cic_ussd/metadata/test_tokens_meta.py @@ -0,0 +1,72 @@ +# standard imports +import json + +# external imports +import pytest +import requests_mock +from cic_types.condiments import MetadataPointer +from requests.exceptions import HTTPError + +# local imports +from cic_ussd.cache import cache_data_key, get_cached_data +from cic_ussd.metadata import TokenMetadata +from cic_ussd.metadata.tokens import token_metadata_handler, query_token_metadata, query_token_info + + +# test imports + + +def test_token_metadata_handler(activated_account, + init_cache, + setup_metadata_request_handler, + setup_metadata_signer, + token_meta_symbol, + token_symbol): + with requests_mock.Mocker(real_http=False) as request_mocker: + with pytest.raises(HTTPError) as error: + metadata_client = TokenMetadata(identifier=b'foo', cic_type=MetadataPointer.TOKEN_META_SYMBOL) + reason = 'Not Found' + status_code = 401 + request_mocker.register_uri('GET', metadata_client.url, status_code=status_code, reason=reason) + token_metadata_handler(metadata_client) + assert str(error.value) == f'Client Error: {status_code}, reason: {reason}' + + identifier = token_symbol.encode('utf-8') + metadata_client = TokenMetadata(identifier, cic_type=MetadataPointer.TOKEN_META_SYMBOL) + request_mocker.register_uri('GET', metadata_client.url, json=token_meta_symbol, status_code=200, reason='OK') + token_metadata_handler(metadata_client) + key = cache_data_key(identifier, MetadataPointer.TOKEN_META_SYMBOL) + cached_token_meta_symbol = get_cached_data(key) + assert json.loads(cached_token_meta_symbol) == token_meta_symbol + + +def test_query_token_metadata(init_cache, + setup_metadata_request_handler, + setup_metadata_signer, + token_meta_symbol, + token_proof_symbol, + token_symbol): + with requests_mock.Mocker(real_http=False) as request_mocker: + identifier = token_symbol.encode('utf-8') + metadata_client = TokenMetadata(identifier, cic_type=MetadataPointer.TOKEN_META_SYMBOL) + request_mocker.register_uri('GET', metadata_client.url, json=token_meta_symbol, status_code=200, reason='OK') + query_token_metadata(identifier) + key = cache_data_key(identifier, MetadataPointer.TOKEN_META_SYMBOL) + cached_token_meta_symbol = get_cached_data(key) + assert json.loads(cached_token_meta_symbol) == token_meta_symbol + + +def test_query_token_info(init_cache, + setup_metadata_request_handler, + setup_metadata_signer, + token_meta_symbol, + token_proof_symbol, + token_symbol): + with requests_mock.Mocker(real_http=False) as request_mocker: + identifier = token_symbol.encode('utf-8') + metadata_client = TokenMetadata(identifier, cic_type=MetadataPointer.TOKEN_PROOF_SYMBOL) + request_mocker.register_uri('GET', metadata_client.url, json=token_proof_symbol, status_code=200, reason='OK') + query_token_info(identifier) + key = cache_data_key(identifier, MetadataPointer.TOKEN_PROOF_SYMBOL) + cached_token_proof_symbol = get_cached_data(key) + assert json.loads(cached_token_proof_symbol) == token_proof_symbol diff --git a/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_language_logic.py b/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_language_logic.py new file mode 100644 index 00000000..a22088ce --- /dev/null +++ b/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_language_logic.py @@ -0,0 +1,52 @@ +# standard imports +import json + +# external imports +import requests_mock +from cic_types.condiments import MetadataPointer + +# local imports +from cic_ussd.cache import cache_data_key, get_cached_data +from cic_ussd.metadata import PreferencesMetadata +from cic_ussd.state_machine.logic.language import (change_preferred_language, + is_valid_language_selection, + preferred_langauge_from_selection, + process_language_selection) + +# test imports + + +def test_change_preferred_language(activated_account, + cached_ussd_session, + celery_session_worker, + init_database, + load_languages, + mocker, + setup_metadata_signer, + setup_metadata_request_handler): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + preferences = { + 'preferred_language': 'en' + } + ussd_session['data'] = preferences + mock_add_preferences_metadata = mocker.patch('cic_ussd.tasks.metadata.add_preferences_metadata.apply_async') + with requests_mock.Mocker(real_http=False) as request_mocker: + identifier = bytes.fromhex(activated_account.blockchain_address) + metadata_client = PreferencesMetadata(identifier=identifier) + request_mocker.register_uri('POST', metadata_client.url, status_code=201, reason='CREATED', json=preferences) + state_machine_data = ('1', ussd_session, activated_account, init_database) + change_preferred_language(state_machine_data) + mock_add_preferences_metadata.assert_called_with( + (activated_account.blockchain_address, preferences), {}, queue='cic-ussd') + + +def test_is_valid_language_selection(activated_account, + generic_ussd_session, + init_cache, + init_database, + load_languages): + state_machine_data = ('1', generic_ussd_session, activated_account, init_database) + assert is_valid_language_selection(state_machine_data) is True + state_machine_data = ('12', generic_ussd_session, activated_account, init_database) + assert is_valid_language_selection(state_machine_data) is False diff --git a/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_pin_guard_logic.py b/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_pin_guard_logic.py new file mode 100644 index 00000000..f6e439a6 --- /dev/null +++ b/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_pin_guard_logic.py @@ -0,0 +1,221 @@ +# standard imports +import json + +# external imports +import requests_mock + +# local imports +from cic_ussd.account.guardianship import Guardianship +from cic_ussd.account.metadata import get_cached_preferred_language +from cic_ussd.cache import cache_data_key, get_cached_data +from cic_ussd.db.models.account import Account +from cic_ussd.metadata import PersonMetadata +from cic_ussd.state_machine.logic.pin_guard import (add_pin_guardian, + is_dialers_pin_guardian, + is_others_pin_guardian, + is_set_pin_guardian, + remove_pin_guardian, + initiate_pin_reset, + save_guardian_to_session_data, + save_guarded_account_session_data, + retrieve_person_metadata, + is_valid_guardian_addition) +from cic_ussd.translation import translation_for + + +def test_save_guardian_to_session_data(activated_account, + cached_ussd_session, + celery_session_worker, + guardian_account, + init_cache, + init_database): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['msisdn'] = activated_account.phone_number + state_machine_data = (guardian_account.phone_number, ussd_session, activated_account, init_database) + save_guardian_to_session_data(state_machine_data) + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + assert ussd_session.get('data').get('guardian_phone_number') == guardian_account.phone_number + + +def test_save_guarded_account_session_data(activated_account, + cached_ussd_session, + celery_session_worker, + guardian_account, + init_cache, + init_database): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['msisdn'] = guardian_account.phone_number + state_machine_data = (activated_account.phone_number, ussd_session, guardian_account, init_database) + save_guarded_account_session_data(state_machine_data) + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + assert ussd_session.get('data').get('guarded_account_phone_number') == activated_account.phone_number + + +def test_retrieve_person_metadata(activated_account, + cached_ussd_session, + celery_session_worker, + guardian_account, + init_cache, + init_database, + mocker, + person_metadata, + setup_metadata_request_handler, + setup_metadata_signer): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['msisdn'] = activated_account.phone_number + state_machine_data = (guardian_account.phone_number, ussd_session, activated_account, init_database) + mocker_query_person_metadata = mocker.patch('cic_ussd.tasks.metadata.query_person_metadata.apply_async') + with requests_mock.Mocker(real_http=False) as request_mocker: + identifier = bytes.fromhex(activated_account.blockchain_address) + metadata_client = PersonMetadata(identifier) + request_mocker.register_uri('GET', metadata_client.url, json=person_metadata, reason='OK', status_code=200) + retrieve_person_metadata(state_machine_data) + mocker_query_person_metadata.assert_called_with((guardian_account.blockchain_address,), {}, queue='cic-ussd') + + +def test_is_valid_guardian_addition(activated_account, + cache_preferences, + cached_ussd_session, + celery_session_worker, + init_cache, + init_database, + guardian_account, + load_languages, + load_ussd_menu, + set_locale_files, + setup_guardianship): + blockchain_address = activated_account.blockchain_address + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + state_machine_data = (guardian_account.phone_number, ussd_session, activated_account, init_database) + assert is_valid_guardian_addition(state_machine_data) is True + + state_machine_data = (activated_account.phone_number, ussd_session, activated_account, init_database) + assert is_valid_guardian_addition(state_machine_data) is False + + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + preferred_language = get_cached_preferred_language(blockchain_address) + failure_reason = translation_for('helpers.error.is_initiator', preferred_language) + assert ussd_session.get('data').get('failure_reason') == failure_reason + + state_machine_data = (Guardianship.guardians[0], ussd_session, activated_account, init_database) + assert is_valid_guardian_addition(state_machine_data) is False + + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + preferred_language = get_cached_preferred_language(blockchain_address) + failure_reason = translation_for('helpers.error.is_existent_guardian', preferred_language) + assert ussd_session.get('data').get('failure_reason') == failure_reason + + +def test_add_pin_guardian(activated_account, generic_ussd_session, guardian_account, init_database): + generic_ussd_session['data'] = {'guardian_phone_number': guardian_account.phone_number} + state_machine_data = ('', generic_ussd_session, activated_account, init_database) + add_pin_guardian(state_machine_data) + account = Account.get_by_phone_number(activated_account.phone_number, init_database) + assert account.get_guardians()[0] == guardian_account.phone_number + + +def test_is_set_pin_guardian(activated_account, + cache_preferences, + cached_ussd_session, + celery_session_worker, + init_cache, + init_database, + guardian_account, + load_languages, + load_ussd_menu, + set_locale_files, + setup_guardianship): + blockchain_address = activated_account.blockchain_address + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + preferred_language = get_cached_preferred_language(blockchain_address) + assert is_set_pin_guardian(activated_account, guardian_account.phone_number, preferred_language, init_database, + ussd_session) is False + + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + failure_reason = translation_for('helpers.error.is_not_existent_guardian', preferred_language) + assert ussd_session.get('data').get('failure_reason') == failure_reason + + assert is_set_pin_guardian(activated_account, Guardianship.guardians[0], preferred_language, init_database, + ussd_session) is True + + assert is_set_pin_guardian(activated_account, activated_account.phone_number, preferred_language, init_database, + ussd_session) is False + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + failure_reason = translation_for('helpers.error.is_initiator', preferred_language) + assert ussd_session.get('data').get('failure_reason') == failure_reason + + +def test_is_dialers_pin_guardian(activated_account, + cache_preferences, + cached_ussd_session, + celery_session_worker, + init_database, + guardian_account): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + state_machine_data = (guardian_account.phone_number, ussd_session, activated_account, init_database) + assert is_dialers_pin_guardian(state_machine_data) is False + activated_account.add_guardian(guardian_account.phone_number) + init_database.flush() + state_machine_data = (guardian_account.phone_number, ussd_session, activated_account, init_database) + assert is_dialers_pin_guardian(state_machine_data) is True + + +def test_is_others_pin_guardian(activated_account, + cache_preferences, + cached_ussd_session, + celery_session_worker, + init_database, + guardian_account): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + state_machine_data = (activated_account.phone_number, ussd_session, guardian_account, init_database) + assert is_others_pin_guardian(state_machine_data) is False + activated_account.add_guardian(guardian_account.phone_number) + init_database.flush() + state_machine_data = (activated_account.phone_number, ussd_session, guardian_account, init_database) + assert is_others_pin_guardian(state_machine_data) is True + + +def test_remove_pin_guardian(activated_account, generic_ussd_session, guardian_account, init_database): + generic_ussd_session['data'] = {'guardian_phone_number': guardian_account.phone_number} + activated_account.add_guardian(guardian_account.phone_number) + init_database.flush() + assert activated_account.get_guardians()[0] == guardian_account.phone_number + state_machine_data = ('', generic_ussd_session, activated_account, init_database) + remove_pin_guardian(state_machine_data) + assert len(activated_account.get_guardians()) == 0 + + +def test_initiate_pin_reset(activated_account, + cache_preferences, + celery_session_worker, + cached_ussd_session, + guardian_account, + init_cache, + init_database, + load_ussd_menu, + mock_notifier_api, + set_locale_files): + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['data'] = {'guarded_account_phone_number': activated_account.phone_number} + state_machine_data = ('', ussd_session, guardian_account, init_database) + initiate_pin_reset(state_machine_data) + blockchain_address = activated_account.blockchain_address + preferred_language = get_cached_preferred_language(blockchain_address) + message = translation_for('sms.pin_reset_initiated', preferred_language, pin_initiator=guardian_account.standard_metadata_id()) + assert mock_notifier_api.get('message') == message + assert mock_notifier_api.get('recipient') == activated_account.phone_number + diff --git a/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_tokens_logic.py b/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_tokens_logic.py new file mode 100644 index 00000000..ac351267 --- /dev/null +++ b/apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_tokens_logic.py @@ -0,0 +1,69 @@ +# standard imports +import json + +# external imports +from cic_types.condiments import MetadataPointer + +# local imports +from cic_ussd.cache import cache_data_key, get_cached_data +from cic_ussd.state_machine.logic.tokens import (is_valid_token_selection, + process_token_selection, + set_selected_active_token) +from cic_ussd.account.tokens import get_cached_token_data_list + + +# test imports + + +def test_is_valid_token_selection(activated_account, + cache_token_data_list, + cache_token_symbol_list, + cached_ussd_session, + init_cache, + init_database): + cached_token_data_list = get_cached_token_data_list(activated_account.blockchain_address) + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['data'] = {'account_tokens_list': cached_token_data_list} + state_machine_data = ('GFT', ussd_session, activated_account, init_database) + assert is_valid_token_selection(state_machine_data) is True + state_machine_data = ('1', ussd_session, activated_account, init_database) + assert is_valid_token_selection(state_machine_data) is True + state_machine_data = ('3', ussd_session, activated_account, init_database) + assert is_valid_token_selection(state_machine_data) is False + + +def test_process_token_selection(activated_account, + cache_token_data_list, + cache_token_symbol_list, + cached_ussd_session, + celery_session_worker, + init_cache, + init_database): + cached_token_data_list = get_cached_token_data_list(activated_account.blockchain_address) + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['data'] = {'account_tokens_list': cached_token_data_list} + state_machine_data = ('GFT', ussd_session, activated_account, init_database) + process_token_selection(state_machine_data) + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + assert ussd_session.get('data').get('selected_token').get('symbol') == 'GFT' + + +def test_set_selected_active_token(activated_account, + cache_token_data_list, + cache_token_symbol_list, + cached_ussd_session, + init_cache, + init_database): + cached_token_data_list = get_cached_token_data_list(activated_account.blockchain_address) + ussd_session = get_cached_data(cached_ussd_session.external_session_id) + ussd_session = json.loads(ussd_session) + ussd_session['data'] = {'selected_token': cached_token_data_list[0]} + state_machine_data = ('GFT', ussd_session, activated_account, init_database) + set_selected_active_token(state_machine_data) + identifier = bytes.fromhex(activated_account.blockchain_address) + key = cache_data_key(identifier=identifier, salt=MetadataPointer.TOKEN_ACTIVE) + active_token = get_cached_data(key) + assert active_token == 'GFT'