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

Remove submodule cic ussd

Browse Source
This commit is contained in:
Blair Vanderlugt
2021-02-06 15:13:47 +00:00
parent 8680d57a67
commit f386625844
221 changed files with 10030 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
apps/cic-ussd/tests/__init__.py Normal file
View File

48
apps/cic-ussd/tests/conftest.py Normal file
View File

@@ -0,0 +1,48 @@
# local imports
from tests.fixtures.config import *
from tests.fixtures.db import *
from tests.fixtures.celery import *
from tests.fixtures.user import *
from tests.fixtures.ussd_session import *
from tests.fixtures.redis import *
from tests.fixtures.callback import *
from tests.fixtures.requests import *
from tests.fixtures.mocks import *

115
apps/cic-ussd/tests/fixtures/callback.py vendored Normal file
View File

@@ -0,0 +1,115 @@
# standard imports
import json
# third party imports
import pytest
# local imports
from cic_ussd.redis import InMemoryStore
@pytest.fixture(scope='function')
def account_creation_action_data():
return {
'phone_number': '+254712345678',
'sms_notification_sent': False,
'status': 'PENDING',
'task_id': '31e85315-feee-4b6d-995e-223569082cc4'
}
@pytest.fixture(scope='function')
def set_account_creation_action_data(init_redis_cache, account_creation_action_data):
redis_cache = init_redis_cache
action_data = account_creation_action_data
task_id = action_data.get('task_id')
redis_cache.set(task_id, json.dumps(action_data))
redis_cache.persist(task_id)
@pytest.fixture(scope='function')
def successful_incoming_token_gift_callback():
return {
'RESULT': {
'hash': '0xb469fb2ebacc9574afb7b51d44e174fba7129fde71bf757fd39784363270832b',
'sender': '0xd6204101012270Bf2558EDcFEd595938d1847bf0',
'recipient': '0xFD9c5aD15C72C6F60f1a119A608931226674243f',
'source_value': 1048576,
'destination_value': 1048576,
'source_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'destination_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'source_token_symbol': 'SRF',
'destination_token_symbol': 'SRF',
'source_token_decimals': 18,
'destination_token_decimals': 18,
'chain': 'Bloxberg:8996'
},
'PARAM': 'tokengift',
'STATUS_CODE': 0,
}
@pytest.fixture(scope='function')
def successful_incoming_transfer_callback():
return {
'RESULT': {
'hash': '0x8b0ed32533164d010afc46c0011fbcb58b0198e03c05b96e2791555746bd3606',
'sender': '0xd6204101012270Bf2558EDcFEd595938d1847bf1',
'recipient': '0xd6204101012270Bf2558EDcFEd595938d1847bf0',
'source_value': 10000000000000000000000,
'destination_value': 10000000000000000000000,
'source_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'destination_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'source_token_symbol': 'SRF',
'destination_token_symbol': 'SRF',
'source_token_decimals': 18,
'destination_token_decimals': 18,
'chain': 'Bloxberg:8996'
},
'PARAM': 'transfer',
'STATUS_CODE': 0
}
@pytest.fixture(scope='function')
def incoming_transfer_callback_invalid_tx_status_code():
return {
'RESULT': {
'hash': '0x8b0ed32533164d010afc46c0011fbcb58b0198e03c05b96e2791555746bd3606',
'sender': '0xd6204101012270Bf2558EDcFEd595938d1847bf1',
'recipient': '0xd6204101012270Bf2558EDcFEd595938d1847bf0',
'source_value': 10000000000000000000000,
'destination_value': 10000000000000000000000,
'source_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'destination_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'source_token_symbol': 'SRF',
'destination_token_symbol': 'SRF',
'source_token_decimals': 18,
'destination_token_decimals': 18,
'chain': 'Bloxberg:8996'
},
'PARAM': 'transfer',
'STATUS_CODE': 1
}
@pytest.fixture(scope='function')
def incoming_transfer_callback_invalid_tx_param():
return {
'RESULT': {
'hash': '0x8b0ed32533164d010afc46c0011fbcb58b0198e03c05b96e2791555746bd3606',
'sender': '0xd6204101012270Bf2558EDcFEd595938d1847bf1',
'recipient': '0xd6204101012270Bf2558EDcFEd595938d1847bf0',
'source_value': 10000000000000000000000,
'destination_value': 10000000000000000000000,
'source_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'destination_token': '0xa75B519dc9b0A50D267E03D8B6808f85A66932dd',
'source_token_symbol': 'SRF',
'destination_token_symbol': 'SRF',
'source_token_decimals': 18,
'destination_token_decimals': 18,
'chain': 'Bloxberg:8996'
},
'PARAM': 'erroneousparam',
'STATUS_CODE': 0
}

43
apps/cic-ussd/tests/fixtures/celery.py vendored Normal file
View File

@@ -0,0 +1,43 @@
# standard imports
import logging
import pytest
import shutil
import tempfile
logg = logging.getLogger()
@pytest.fixture(scope='session')
def celery_includes():
return [
'cic_ussd.tasks.ussd',
'cic_ussd.tasks.callback_handler',
'cic_notify.tasks.sms'
]
@pytest.fixture(scope='session')
def celery_config():
bq = tempfile.mkdtemp()
bp = tempfile.mkdtemp()
rq = tempfile.mkdtemp()
logg.debug('celery broker queue {} processed {}'.format(bq, bp))
logg.debug('celery backend store {}'.format(rq))
yield {
'broker_url': 'filesystem://',
'broker_transport_options': {
'data_folder_in': bq,
'data_folder_out': bq,
'data_folder_processed': bp,
},
'result_backend': 'file://{}'.format(rq),
}
logg.debug('cleaning up celery filesystem backend files {} {} {}'.format(bq, bp, rq))
shutil.rmtree(bq)
shutil.rmtree(bp)
shutil.rmtree(rq)
@pytest.fixture(scope='session')
def celery_enable_logging():
return True

104
apps/cic-ussd/tests/fixtures/config.py vendored Normal file
View File

@@ -0,0 +1,104 @@
# standard imports
import i18n
import logging
import os
# third party imports
import pytest
from confini import Config
from sqlalchemy import create_engine
# local imports
from cic_ussd.db import dsn_from_config
from cic_ussd.files.local_files import create_local_file_data_stores, json_file_parser
from cic_ussd.menu.ussd_menu import UssdMenu
from cic_ussd.state_machine import UssdStateMachine
from cic_ussd.encoder import PasswordEncoder
logg = logging.getLogger()
fixtures_dir = os.path.dirname(__file__)
root_directory = os.path.dirname(os.path.dirname(fixtures_dir))
@pytest.fixture(scope='session')
def load_config():
config_directory = os.path.join(root_directory, '.config/test')
config = Config(config_dir=config_directory)
config.process(set_as_current=True)
logg.debug('config loaded\n{}'.format(config))
return config
@pytest.fixture(scope='session')
def alembic_config():
migrations_directory = os.path.join(root_directory, 'cic_ussd', 'db', 'migrations', 'default')
file = os.path.join(migrations_directory, 'alembic.ini')
return {
'file': file,
'script_location': migrations_directory
}
@pytest.fixture(scope='session')
def alembic_engine(load_config):
data_source_name = dsn_from_config(load_config)
database_engine = create_engine(data_source_name)
return database_engine
@pytest.fixture(scope='function')
def set_fernet_key(load_config):
PasswordEncoder.set_key(load_config.get('APP_PASSWORD_PEPPER'))
@pytest.fixture
def set_locale_files(load_config):
i18n.load_path.append(load_config.get('APP_LOCALE_PATH'))
i18n.set('fallback', load_config.get('APP_LOCALE_FALLBACK'))
@pytest.fixture
def load_ussd_menu(load_config):
ussd_menu_db = create_local_file_data_stores(file_location=load_config.get('USSD_MENU_FILE'), table_name="ussd_menu")
UssdMenu.ussd_menu_db = ussd_menu_db
@pytest.fixture(scope='function')
def load_data_into_state_machine(load_config):
UssdStateMachine.states = json_file_parser(filepath=load_config.get('STATEMACHINE_STATES'))
UssdStateMachine.transitions = json_file_parser(filepath=load_config.get('STATEMACHINE_TRANSITIONS'))
@pytest.fixture(scope='function')
def uwsgi_env():
return {
'REQUEST_METHOD': 'POST',
'REQUEST_URI': '/',
'PATH_INFO': '/',
'QUERY_STRING': '',
'SERVER_PROTOCOL': 'HTTP/1.1',
'SCRIPT_NAME': '',
'SERVER_NAME': 'mango-habanero',
'SERVER_PORT': '9091',
'UWSGI_ROUTER': 'http',
'REMOTE_ADDR': '127.0.0.1',
'REMOTE_PORT': '33515',
'CONTENT_TYPE': 'application/json',
'HTTP_USER_AGENT': 'PostmanRuntime/7.26.8',
'HTTP_ACCEPT': '*/*',
'HTTP_POSTMAN_TOKEN': 'c1f6eb29-8160-497f-a5a1-935d175e2eb7',
'HTTP_HOST': '127.0.0.1:9091',
'HTTP_ACCEPT_ENCODING': 'gzip, deflate, br',
'HTTP_CONNECTION': 'keep-alive',
'CONTENT_LENGTH': '102',
'wsgi.version': (1, 0),
'wsgi.run_once': False,
'wsgi.multithread': False,
'wsgi.multiprocess': False,
'wsgi.url_scheme': 'http',
'uwsgi.version': b'2.0.19.1',
'uwsgi.node': b'mango-habanero'
}

44
apps/cic-ussd/tests/fixtures/db.py vendored Normal file
View File

@@ -0,0 +1,44 @@
# standard imports
import os
# third party imports
import alembic
import pytest
from alembic.config import Config as AlembicConfig
# local imports
from cic_ussd.db import dsn_from_config
from cic_ussd.db.models.base import SessionBase
from tests.fixtures.config import root_directory
@pytest.fixture(scope='session')
def database_engine(load_config):
data_source_name = dsn_from_config(load_config)
SessionBase.connect(data_source_name)
yield data_source_name
if load_config.get('DATABASE_ENGINE') == 'sqlite':
os.unlink(load_config.get('DATABASE_NAME'))
@pytest.fixture(scope='function')
def init_database(load_config, database_engine):
db_directory = os.path.join(root_directory, 'cic_ussd', 'db')
migrations_directory = os.path.join(db_directory, 'migrations', load_config.get('DATABASE_ENGINE'))
if not os.path.isdir(migrations_directory):
migrations_directory = os.path.join(db_directory, 'migrations', 'default')
SessionBase.session = SessionBase.create_session()
alembic_config = AlembicConfig(os.path.join(migrations_directory, 'alembic.ini'))
alembic_config.set_main_option('sqlalchemy.url', database_engine)
alembic_config.set_main_option('script_location', migrations_directory)
alembic.command.downgrade(alembic_config, 'base')
alembic.command.upgrade(alembic_config, 'head')
yield SessionBase.session
SessionBase.session.commit()
SessionBase.session.close()

51
apps/cic-ussd/tests/fixtures/mocks.py vendored Normal file
View File

@@ -0,0 +1,51 @@
# standard imports
# third-party imports
import pytest
# local imports
from cic_ussd.translation import translation_for
from cic_ussd.transactions import truncate
@pytest.fixture(scope='function', autouse=True)
def mock_notifier_api(mocker):
messages = []
def mock_sms_api(self, message: str, recipient: str):
pass
def mock_send_sms_notification(self, key: str, phone_number: str, preferred_language: str, **kwargs):
message = translation_for(key=key, preferred_language=preferred_language, **kwargs)
messages.append({'message': message, 'recipient': phone_number})
mocker.patch('cic_notify.api.Api.sms', mock_sms_api)
mocker.patch('cic_ussd.notifications.Notifier.send_sms_notification', mock_send_sms_notification)
return messages
@pytest.fixture(scope='function')
def mock_outgoing_transactions(mocker):
transactions = []
def mock_process_outgoing_transfer_transaction(self, amount: int, token_symbol: str = 'SRF'):
transactions.append({
'amount': amount,
'token_symbol': token_symbol
})
mocker.patch(
'cic_ussd.transactions.OutgoingTransactionProcessor.process_outgoing_transfer_transaction',
mock_process_outgoing_transfer_transaction
)
return transactions
@pytest.fixture(scope='function')
def mock_balance(mocker):
mocked_operational_balance = mocker.patch('cic_ussd.accounts.BalanceManager.get_operational_balance')
def _mock_operational_balance(balance: int):
mocked_operational_balance.return_value = truncate(value=balance, decimals=2)
return _mock_operational_balance

11
apps/cic-ussd/tests/fixtures/redis.py vendored Normal file
View File

@@ -0,0 +1,11 @@
# third party imports
import pytest
# local imports
from cic_ussd.redis import InMemoryStore
@pytest.fixture(scope='function')
def init_redis_cache(redisdb):
InMemoryStore.cache = redisdb
return redisdb

20
apps/cic-ussd/tests/fixtures/requests.py vendored Normal file
View File

@@ -0,0 +1,20 @@
# third party imports
import pytest
@pytest.fixture(scope='function')
def valid_locked_accounts_env(uwsgi_env):
env = uwsgi_env
env['REQUEST_METHOD'] = 'GET'
env['PATH_INFO'] = '/accounts/locked/10/10'
return env
@pytest.fixture(scope='function')
def get_request_with_params_env(uwsgi_env):
env = uwsgi_env
env['REQUEST_METHOD'] = 'GET'
env['REQUEST_URI'] = '/?phone=0700000000'
return env

94
apps/cic-ussd/tests/fixtures/user.py vendored Normal file
View File

@@ -0,0 +1,94 @@
# standard imports
from random import randint
import uuid
# third party imports
import pytest
from faker import Faker
# local imports
from cic_ussd.db.models.user import AccountStatus, User
fake = Faker()
@pytest.fixture(scope='function')
def create_activated_user(init_database, set_fernet_key):
user = User(
blockchain_address='0xFD9c5aD15C72C6F60f1a119A608931226674243f',
phone_number='+25498765432'
)
user.preferred_language = 'en'
user.create_password('0000')
user.activate_account()
init_database.add(user)
init_database.commit()
return user
@pytest.fixture(scope='function')
def create_valid_tx_recipient(init_database, set_fernet_key):
user = User(
blockchain_address='0xd6204101012270Bf2558EDcFEd595938d1847bf0',
phone_number='+25498765432'
)
user.preferred_language = 'en'
user.create_password('0000')
user.activate_account()
init_database.add(user)
init_database.commit()
return user
@pytest.fixture(scope='function')
def create_valid_tx_sender(init_database, set_fernet_key):
user = User(
blockchain_address='0xd6204101012270Bf2558EDcFEd595938d1847bf1',
phone_number='+25498765433'
)
user.preferred_language = 'en'
user.create_password('0000')
user.activate_account()
init_database.add(user)
init_database.commit()
return user
@pytest.fixture(scope='function')
def create_pending_user(init_database, set_fernet_key):
user = User(
blockchain_address='0x0ebdea8612c1b05d952c036859266c7f2cfcd6a29842d9c6cce3b9f1ba427588',
phone_number='+25498765432'
)
init_database.add(user)
init_database.commit()
return user
@pytest.fixture(scope='function')
def create_pin_blocked_user(init_database, set_fernet_key):
user = User(
blockchain_address='0x0ebdea8612c1b05d952c036859266c7f2cfcd6a29842d9c6cce3b9f1ba427588',
phone_number='+25498765432'
)
user.create_password('0000')
user.failed_pin_attempts = 3
user.account_status = 3
init_database.add(user)
init_database.commit()
return user
@pytest.fixture(scope='function')
def create_locked_accounts(init_database, set_fernet_key):
for i in range(20):
blockchain_address = str(uuid.uuid4())
phone_number = fake.phone_number()
pin = f'{randint(1000, 9999)}'
user = User(phone_number=phone_number, blockchain_address=blockchain_address)
user.create_password(password=pin)
user.failed_pin_attempts = 3
user.account_status = AccountStatus.LOCKED.value
user.session.add(user)
user.session.commit()

52
apps/cic-ussd/tests/fixtures/ussd_session.py vendored Normal file
View File

@@ -0,0 +1,52 @@
# standard imports
import json
# third-party imports
import pytest
# local imports
from cic_ussd.db.models.ussd_session import UssdSession
from cic_ussd.redis import InMemoryStore
from cic_ussd.session.ussd_session import UssdSession as InMemoryUssdSession
@pytest.fixture(scope='function')
def ussd_session_data():
return {
'external_session_id': 'AT974186',
'service_code': '*483*46#',
'msisdn': '+25498765432',
'user_input': '1',
'state': 'initial_language_selection',
'session_data': {},
'version': 2
}
@pytest.fixture(scope='function')
def create_in_redis_ussd_session(ussd_session_data, init_redis_cache):
external_session_id = ussd_session_data.get('external_session_id')
InMemoryUssdSession.redis_cache = InMemoryStore.cache
InMemoryUssdSession.redis_cache.set(external_session_id, json.dumps(ussd_session_data))
return InMemoryUssdSession.redis_cache
@pytest.fixture(scope='function')
def get_in_redis_ussd_session(ussd_session_data, create_in_redis_ussd_session):
external_session_id = ussd_session_data.get('external_session_id')
ussd_session_data = create_in_redis_ussd_session.get(external_session_id)
ussd_session_data = json.loads(ussd_session_data)
# remove version from ussd_session data because the ussd_session object does not expect a version at initialization
del ussd_session_data['version']
ussd_session = InMemoryUssdSession(**{key: value for key, value in ussd_session_data.items()})
ussd_session.version = ussd_session_data.get('version')
return ussd_session
@pytest.fixture(scope='function')
def create_in_db_ussd_session(init_database, ussd_session_data):
ussd_session_data['session_data'] = {}
ussd_session = UssdSession(**{key: value for key, value in ussd_session_data.items()})
init_database.add(ussd_session)
init_database.commit()
return ussd_session

28
apps/cic-ussd/tests/functional/cic_ussd/db/migrations/test_migrations.py Normal file
View File

@@ -0,0 +1,28 @@
import pytest
def test_single_head_revision(alembic_runner):
heads = alembic_runner.heads
head_count = len(heads)
assert head_count == 1
def test_upgrade(alembic_runner):
try:
alembic_runner.migrate_up_to("head")
except RuntimeError:
pytest.fail('Failed to upgrade to the head revision.')
def test_up_down_consistency(alembic_runner):
try:
for revision in alembic_runner.history.revisions:
alembic_runner.migrate_up_to(revision)
except RuntimeError:
pytest.fail('Failed to upgrade through each revision individually.')
try:
for revision in reversed(alembic_runner.history.revisions):
alembic_runner.migrate_down_to(revision)
except RuntimeError:
pytest.fail('Failed to downgrade through each revision individually.')

202
apps/cic-ussd/tests/functional/cic_ussd/tasks/test_callback_handler_tasks.py Normal file
View File

@@ -0,0 +1,202 @@
# standard imports
import json
import logging
from datetime import datetime
# third party imports
import celery
import pytest
# local imports
from cic_ussd.db.models.user import User
from cic_ussd.error import ActionDataNotFoundError
from cic_ussd.transactions import from_wei
logg = logging.getLogger()
def test_successful_process_account_creation_callback_task(account_creation_action_data,
celery_session_worker,
init_database,
init_redis_cache,
mocker,
set_account_creation_action_data):
phone_number = account_creation_action_data.get('phone_number')
task_id = account_creation_action_data.get('task_id')
mocked_task_request = mocker.patch('celery.app.task.Task.request')
# WARNING: [THE SETTING OF THE ROOT ID IS A HACK AND SHOULD BE REVIEWED OR IMPROVED]
mocked_task_request.root_id = task_id
user = init_database.query(User).filter_by(phone_number=phone_number).first()
assert user is None
redis_cache = init_redis_cache
action_data = redis_cache.get(task_id)
action_data = json.loads(action_data)
assert action_data.get('status') == 'PENDING'
status_code = 0
result = '0x6315c185fd23bDbbba058E2a504197915aCC5065'
url = ''
s_process_callback_request = celery.signature(
'cic_ussd.tasks.callback_handler.process_account_creation_callback',
[result, url, status_code]
)
s_process_callback_request.apply_async().get()
user = init_database.query(User).filter_by(phone_number=phone_number).first()
assert user.blockchain_address == result
action_data = redis_cache.get(task_id)
action_data = json.loads(action_data)
assert action_data.get('status') == 'CREATED'
def test_unsuccessful_process_account_creation_callback_task(init_database,
init_redis_cache,
celery_session_worker):
with pytest.raises(ActionDataNotFoundError) as error:
status_code = 0
result = '0x6315c185fd23bDbbba058E2a504197915aCC5065'
url = ''
s_process_callback_request = celery.signature(
'cic_ussd.tasks.callback_handler.process_account_creation_callback',
[result, url, status_code]
)
result = s_process_callback_request.apply_async()
task_id = result.get()
assert str(error.value) == f'Account creation task: {task_id}, returned unexpected response: {status_code}'
def test_successful_token_gift_incoming_transaction(celery_session_worker,
create_activated_user,
mock_notifier_api,
set_locale_files,
successful_incoming_token_gift_callback):
result = successful_incoming_token_gift_callback.get('RESULT')
param = successful_incoming_token_gift_callback.get('PARAM')
status_code = successful_incoming_token_gift_callback.get('STATUS_CODE')
s_process_token_gift = celery.signature(
'cic_ussd.tasks.callback_handler.process_incoming_transfer_callback',
[result, param, status_code]
)
s_process_token_gift.apply_async().get()
balance = from_wei(result.get('destination_value'))
token_symbol = result.get('token_symbol')
messages = mock_notifier_api
assert messages[0].get('recipient') == create_activated_user.phone_number
assert messages[0].get(
'message') == f'Hello {""} you have been registered on Sarafu Network! Your balance is {balance} {token_symbol}. To use dial *483*46#. For help 0757628885.'
def test_successful_transfer_incoming_transaction(celery_session_worker,
create_valid_tx_sender,
create_valid_tx_recipient,
mock_notifier_api,
set_locale_files,
successful_incoming_transfer_callback):
result = successful_incoming_transfer_callback.get('RESULT')
param = successful_incoming_transfer_callback.get('PARAM')
status_code = successful_incoming_transfer_callback.get('STATUS_CODE')
s_process_token_gift = celery.signature(
'cic_ussd.tasks.callback_handler.process_incoming_transfer_callback',
[result, param, status_code]
)
s_process_token_gift.apply_async().get()
value = result.get('destination_value')
balance = ''
token_symbol = result.get('token_symbol')
sender_first_name = ''
sender_last_name = ''
phone_number = create_valid_tx_sender.phone_number
tx_sender_information = f'{phone_number}, {sender_first_name}, {sender_last_name}'
amount = from_wei(value=value)
timestamp = datetime.now().strftime('%d-%m-%y, %H:%M %p')
messages = mock_notifier_api
assert messages[0].get('recipient') == create_valid_tx_recipient.phone_number
assert messages[0].get(
'message') == f'Successfully received {amount} {token_symbol} from {tx_sender_information} {timestamp}. New balance is {balance} {token_symbol}.'
def test_unsuccessful_incoming_transaction_recipient_not_found(celery_session_worker,
create_valid_tx_sender,
successful_incoming_transfer_callback):
result = successful_incoming_transfer_callback.get('RESULT')
param = successful_incoming_transfer_callback.get('PARAM')
status_code = successful_incoming_transfer_callback.get('STATUS_CODE')
with pytest.raises(ValueError) as error:
s_process_token_gift = celery.signature(
'cic_ussd.tasks.callback_handler.process_incoming_transfer_callback',
[result, param, status_code]
)
s_process_token_gift.apply_async().get()
recipient_blockchain_address = result.get('recipient')
assert str(error.value) == f'Tx for recipient: {recipient_blockchain_address} was received but has no matching user in the system.'
def test_successful_incoming_transaction_sender_not_found(caplog,
celery_session_worker,
create_valid_tx_recipient,
successful_incoming_transfer_callback):
result = successful_incoming_transfer_callback.get('RESULT')
param = successful_incoming_transfer_callback.get('PARAM')
status_code = successful_incoming_transfer_callback.get('STATUS_CODE')
s_process_token_gift = celery.signature(
'cic_ussd.tasks.callback_handler.process_incoming_transfer_callback',
[result, param, status_code]
)
s_process_token_gift.apply_async().get()
sender_blockchain_address = result.get('sender')
assert 'Balance requires implementation of cic-eth integration with balance.' in caplog.text
# assert f'Tx with sender: {sender_blockchain_address} was received but has no matching user in the system.\n' in caplog.text
def test_unsuccessful_incoming_transaction_invalid_status_code(celery_session_worker,
incoming_transfer_callback_invalid_tx_status_code):
result = incoming_transfer_callback_invalid_tx_status_code.get('RESULT')
param = incoming_transfer_callback_invalid_tx_status_code.get('PARAM')
status_code = incoming_transfer_callback_invalid_tx_status_code.get('STATUS_CODE')
with pytest.raises(ValueError) as error:
s_process_token_gift = celery.signature(
'cic_ussd.tasks.callback_handler.process_incoming_transfer_callback',
[result, param, status_code]
)
s_process_token_gift.apply_async().get()
assert str(error.value) == f'Unexpected status code: {status_code}'
def test_unsuccessful_incoming_transaction_invalid_param(celery_session_worker,
incoming_transfer_callback_invalid_tx_param):
result = incoming_transfer_callback_invalid_tx_param.get('RESULT')
param = incoming_transfer_callback_invalid_tx_param.get('PARAM')
status_code = incoming_transfer_callback_invalid_tx_param.get('STATUS_CODE')
with pytest.raises(ValueError) as error:
s_process_token_gift = celery.signature(
'cic_ussd.tasks.callback_handler.process_incoming_transfer_callback',
[result, param, status_code]
)
s_process_token_gift.apply_async().get()
assert str(error.value) == f'Unexpected transaction: param {status_code}'

46
apps/cic-ussd/tests/functional/cic_ussd/tasks/test_persist_session_to_db_tasks.py Normal file
View File

@@ -0,0 +1,46 @@
# third party imports
import celery
import pytest
# local imports
from cic_ussd.db.models.ussd_session import UssdSession
from cic_ussd.error import SessionNotFoundError
def test_persist_session_to_db_task(
init_database,
create_activated_user,
ussd_session_data,
celery_session_worker,
create_in_redis_ussd_session):
external_session_id = ussd_session_data.get('external_session_id')
s_persist_session_to_db = celery.signature(
'cic_ussd.tasks.ussd.persist_session_to_db',
[external_session_id]
)
result = s_persist_session_to_db.apply_async()
result.get()
db_session = init_database.query(UssdSession).filter_by(external_session_id=external_session_id).first()
assert db_session.external_session_id == 'AT974186'
assert db_session.service_code == '*483*46#'
assert db_session.msisdn == '+25498765432'
assert db_session.user_input == '1'
assert db_session.state == 'initial_language_selection'
assert db_session.session_data is None
assert db_session.version == 2
assert UssdSession.have_session_for_phone(create_activated_user.phone_number)
def test_session_not_found_error(
celery_session_worker,
create_in_redis_ussd_session):
with pytest.raises(SessionNotFoundError) as error:
external_session_id = 'SomeRandomValue'
s_persist_session_to_db = celery.signature(
'cic_ussd.tasks.ussd.persist_session_to_db',
[external_session_id]
)
result = s_persist_session_to_db.apply_async()
result.get()
assert str(error.value) == "Session does not exist!"

14
apps/cic-ussd/tests/helpers/tmp_files.py Normal file
View File

@@ -0,0 +1,14 @@
# standard imports
import logging
import tempfile
from typing import Any, Tuple
logg = logging.getLogger()
def create_tmp_file() -> Tuple[int, Any]:
"""This helper function creates a unique temporary directory.
:return: path to a temporary directory
:rtype: str
"""
return tempfile.mkstemp()

55
apps/cic-ussd/tests/unit/cic_ussd/files/test_local_files.py Normal file
View File

@@ -0,0 +1,55 @@
# standard imports
import json
import os
# third-party imports
from tinydb import Query
# local imports
from cic_ussd.files.local_files import create_local_file_data_stores, json_file_parser
from tests.fixtures.config import root_directory
from tests.helpers.tmp_files import create_tmp_file
def test_create_in_memory_data_stores():
"""
GIVEN the cic-ussd application component
WHEN the create_in_memory_data_stores function is passed a file and table name
THEN it creates a tiny dn data store that can be written to and queried
"""
descriptor, tmp_file = create_tmp_file()
test_file = create_local_file_data_stores(file_location=tmp_file, table_name='test_table')
# write to data store
test_file.insert({
'foo': 'bar'
})
query = Query()
inserted_record = test_file.get(query.foo == 'bar')
assert inserted_record == {'foo': 'bar'}
os.close(descriptor)
os.remove(tmp_file)
def test_json_file_parser(load_config):
"""
GIVEN the cic-ussd application component
WHEN the json_file_parser function is passed a directory path containing JSON files
THEN it dynamically loads all the files and compiles their content into one python array
"""
files_dir = load_config.get('STATEMACHINE_TRANSITIONS')
files_dir = os.path.join(root_directory, files_dir)
# total files len
file_content_length = 0
for filepath in os.listdir(files_dir):
# get path of data files
filepath = os.path.join(files_dir, filepath)
# read file content
data_file = open(filepath)
data = json.load(data_file)
file_content_length += len(data)
data_file.close()
assert len(json_file_parser(filepath=files_dir)) == file_content_length

52
apps/cic-ussd/tests/unit/cic_ussd/menu/test_ussd_menu.py Normal file
View File

@@ -0,0 +1,52 @@
# standard imports
import os
# third party imports
import pytest
# local imports
from cic_ussd.files.local_files import create_local_file_data_stores
from cic_ussd.menu.ussd_menu import UssdMenu
from tests.helpers.tmp_files import create_tmp_file
@pytest.mark.parametrize('menu_name, expected_parent_menu_name', [
('initial_language_selection', None),
('account_management', 'start'),
('enter_current_pin', 'account_management')
])
def test_ussd_menu(load_ussd_menu, menu_name, expected_parent_menu_name):
ussd_menu = UssdMenu.find_by_name(name=menu_name)
assert ussd_menu.get('parent') == expected_parent_menu_name
def test_create_ussd_menu():
descriptor, tmp_file = create_tmp_file()
ussd_menu_db = create_local_file_data_stores(file_location=tmp_file, table_name="ussd_menu")
UssdMenu.ussd_menu_db = ussd_menu_db
UssdMenu(name='foo', description='foo-bar', parent=None)
assert UssdMenu.find_by_name(name='foo')['description'] == 'foo-bar'
UssdMenu.set_description(name='foo', description='bar')
assert UssdMenu.find_by_name(name='foo')['description'] == 'bar'
menu2 = UssdMenu(name='fizz', description='buzz', parent='foo')
assert UssdMenu.parent_menu(menu2.name)['description'] == 'bar'
os.close(descriptor)
os.remove(tmp_file)
def test_failed_create_ussd_menu():
descriptor, tmp_file = create_tmp_file()
ussd_menu_db = create_local_file_data_stores(file_location=tmp_file, table_name="ussd_menu")
UssdMenu.ussd_menu_db = ussd_menu_db
UssdMenu(name='foo', description='foo-bar', parent=None)
assert UssdMenu.find_by_name(name='foo')['description'] == 'foo-bar'
UssdMenu.set_description(name='foo', description='bar')
assert UssdMenu.find_by_name(name='foo')['description'] == 'bar'
with pytest.raises(ValueError) as error:
UssdMenu(name='foo', description='foo-bar', parent=None)
assert str(error.value) == "Menu already exists!"
os.close(descriptor)
os.remove(tmp_file)

14
apps/cic-ussd/tests/unit/cic_ussd/models/test_task_tracker.py Normal file
View File

@@ -0,0 +1,14 @@
# local imports
from cic_ussd.db.models.task_tracker import TaskTracker
def test_task_tracker(init_database):
task_uuid = '31e85315-feee-4b6d-995e-223569082cc4'
task_in_tracker = TaskTracker(task_uuid=task_uuid)
session = init_database
session.add(task_in_tracker)
session.commit()
queried_task = session.query(TaskTracker).get(1)
assert queried_task.task_uuid == task_uuid

40
apps/cic-ussd/tests/unit/cic_ussd/models/test_user.py Normal file
View File

@@ -0,0 +1,40 @@
"""Tests the persistence of the user record and associated functions to the user object"""
# standard imports
import pytest
# platform imports
from cic_ussd.db.models.user import User
def test_user(init_database, set_fernet_key):
user = User(blockchain_address='0x417f5962fc52dc33ff0689659b25848680dec6dcedc6785b03d1df60fc6d5c51',
phone_number='+254700000000')
user.create_password('0000')
session = User.session
session.add(user)
session.commit()
queried_user = session.query(User).get(1)
assert queried_user.blockchain_address == '0x417f5962fc52dc33ff0689659b25848680dec6dcedc6785b03d1df60fc6d5c51'
assert queried_user.phone_number == '+254700000000'
assert queried_user.failed_pin_attempts == 0
assert queried_user.verify_password('0000') is True
def test_user_state_transition(create_pending_user):
user = create_pending_user
session = User.session
assert user.get_account_status() == 'PENDING'
user.activate_account()
assert user.get_account_status() == 'ACTIVE'
user.failed_pin_attempts = 3
assert user.get_account_status() == 'LOCKED'
user.reset_account_pin()
assert user.get_account_status() == 'RESET'
user.activate_account()
assert user.get_account_status() == 'ACTIVE'
session.add(user)
session.commit()

54
apps/cic-ussd/tests/unit/cic_ussd/models/test_ussd_session.py Normal file
View File

@@ -0,0 +1,54 @@
# third party imports
import pytest
# local imports
from cic_ussd.db.models.ussd_session import UssdSession
from cic_ussd.error import VersionTooLowError
def test_ussd_session(init_database, create_in_redis_ussd_session, create_activated_user):
session = init_database
ussd_session = UssdSession(
external_session_id='AT65423',
service_code='*123#',
msisdn=create_activated_user.phone_number,
user_input='1',
state='start',
session_data={},
version=1,
)
session.add(ussd_session)
session.commit()
ussd_session.set_data(key='foo', session=init_database, value='bar')
assert ussd_session.get_data('foo') == 'bar'
ussd_session.update(
session=init_database,
user_input='3',
state='next',
version=2
)
assert ussd_session.version == 2
session.add(ussd_session)
session.commit()
assert UssdSession.have_session_for_phone(create_activated_user.phone_number) is True
def test_version_too_low_error(init_database, create_in_redis_ussd_session, create_activated_user):
with pytest.raises(VersionTooLowError) as e:
session = UssdSession(
external_session_id='AT38745',
service_code='*123#',
msisdn=create_activated_user.phone_number,
user_input='1',
state='start',
session_data={},
version=3,
)
assert session.check_version(1)
assert session.check_version(3)
assert str(e.value) == 'New session version number is not greater than last saved version!'

11
apps/cic-ussd/tests/unit/cic_ussd/session/test_in_memory_ussd_session.py Normal file
View File

@@ -0,0 +1,11 @@
# standard imports
import json
# local imports
from cic_ussd.menu.ussd_menu import UssdMenu
from cic_ussd.session.ussd_session import UssdSession
def test_ussd_session(load_ussd_menu, get_in_redis_ussd_session):
ussd_session = get_in_redis_ussd_session
assert UssdMenu.find_by_name(name='initial_language_selection').get('name') == ussd_session.state

21
apps/cic-ussd/tests/unit/cic_ussd/state_machine/logic/test_menu.py Normal file
View File

@@ -0,0 +1,21 @@
# local imports
from cic_ussd.state_machine.logic.menu import (menu_one_selected,
menu_two_selected,
menu_three_selected,
menu_four_selected)
def test_menu_selection(create_pending_user, create_in_db_ussd_session):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
assert menu_one_selected(('1', serialized_in_db_ussd_session, create_pending_user)) is True
assert menu_one_selected(('x', serialized_in_db_ussd_session, create_pending_user)) is False
assert menu_two_selected(('2', serialized_in_db_ussd_session, create_pending_user)) is True
assert menu_two_selected(('1', serialized_in_db_ussd_session, create_pending_user)) is False
assert menu_three_selected(('3', serialized_in_db_ussd_session, create_pending_user)) is True
assert menu_three_selected(('4', serialized_in_db_ussd_session, create_pending_user)) is False
assert menu_four_selected(('4', serialized_in_db_ussd_session, create_pending_user)) is True
assert menu_four_selected(('d', serialized_in_db_ussd_session, create_pending_user)) is False

94
apps/cic-ussd/tests/unit/cic_ussd/state_machine/logic/test_pin.py Normal file
View File

@@ -0,0 +1,94 @@
# standards imports
import json
# third party imports
import pytest
# local imports
from cic_ussd.encoder import check_password_hash, create_password_hash
from cic_ussd.state_machine.logic.pin import (complete_pin_change,
is_valid_pin,
is_valid_new_pin,
is_authorized_pin,
is_blocked_pin,
pins_match,
save_initial_pin_to_session_data)
def test_complete_pin_change(init_database, create_pending_user, create_in_db_ussd_session):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('1212', serialized_in_db_ussd_session, create_pending_user)
assert create_pending_user.password_hash is None
create_in_db_ussd_session.set_data(key='initial_pin', session=init_database, value=create_password_hash('1212'))
complete_pin_change(state_machine_data)
assert create_pending_user.password_hash is not None
assert create_pending_user.verify_password(password='1212') is True
@pytest.mark.parametrize('user_input, expected', [
('4562', True),
('jksu', False),
('ij45', False),
])
def test_is_valid_pin(create_pending_user, create_in_db_ussd_session, user_input, expected):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (user_input, serialized_in_db_ussd_session, create_pending_user)
assert is_valid_pin(state_machine_data) is expected
@pytest.mark.parametrize('user_input, expected', [
('1212', True),
('0000', False)
])
def test_pins_match(init_database, create_pending_user, create_in_db_ussd_session, user_input, expected):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (user_input, serialized_in_db_ussd_session, create_pending_user)
create_in_db_ussd_session.set_data(key='initial_pin', session=init_database, value=create_password_hash(user_input))
assert pins_match(state_machine_data) is True
def test_save_initial_pin_to_session_data(create_pending_user,
create_in_redis_ussd_session,
create_in_db_ussd_session,
celery_session_worker):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('1212', serialized_in_db_ussd_session, create_pending_user)
save_initial_pin_to_session_data(state_machine_data)
external_session_id = create_in_db_ussd_session.external_session_id
in_memory_ussd_session = create_in_redis_ussd_session.get(external_session_id)
in_memory_ussd_session = json.loads(in_memory_ussd_session)
assert check_password_hash(
password='1212', hashed_password=in_memory_ussd_session.get('session_data')['initial_pin'])
@pytest.mark.parametrize('user_input, expected_result', [
('1212', False),
('0000', True)
])
def test_is_authorized_pin(create_activated_user, create_in_db_ussd_session, expected_result, user_input):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (user_input, serialized_in_db_ussd_session, create_activated_user)
assert is_authorized_pin(state_machine_data=state_machine_data) is expected_result
def test_is_not_blocked_pin(create_activated_user, create_in_db_ussd_session):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
assert is_blocked_pin(state_machine_data=state_machine_data) is False
def test_is_blocked_pin(create_pin_blocked_user, create_in_db_ussd_session):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
alt_state_machine_data = ('user_input', serialized_in_db_ussd_session, create_pin_blocked_user)
assert is_blocked_pin(state_machine_data=alt_state_machine_data) is True
@pytest.mark.parametrize('user_input, expected_result', [
('1212', True),
('0000', False)
])
def test_is_valid_new_pin(create_activated_user, create_in_db_ussd_session, expected_result, user_input):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (user_input, serialized_in_db_ussd_session, create_activated_user)
assert is_valid_new_pin(state_machine_data=state_machine_data) is expected_result

36
apps/cic-ussd/tests/unit/cic_ussd/state_machine/logic/test_sms.py Normal file
View File

@@ -0,0 +1,36 @@
# standard imports
# third-party imports
import pytest
# local imports
from cic_ussd.state_machine.logic.sms import (send_terms_to_user_if_required,
process_mini_statement_request,
upsell_unregistered_recipient)
def test_send_terms_to_user_if_required(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
send_terms_to_user_if_required(state_machine_data=state_machine_data)
assert 'Requires integration to cic-notify.' in caplog.text
def test_process_mini_statement_request(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
process_mini_statement_request(state_machine_data=state_machine_data)
assert 'Requires integration to cic-notify.' in caplog.text
def test_upsell_unregistered_recipient(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
upsell_unregistered_recipient(state_machine_data=state_machine_data)
assert 'Requires integration to cic-notify.' in caplog.text

112
apps/cic-ussd/tests/unit/cic_ussd/state_machine/logic/test_transaction.py Normal file
View File

@@ -0,0 +1,112 @@
# standard imports
import json
# third-party imports
import pytest
# local imports
from cic_ussd.state_machine import UssdStateMachine
from cic_ussd.state_machine.logic.transaction import (has_sufficient_balance,
is_valid_recipient,
is_valid_transaction_amount,
process_transaction_request,
save_recipient_phone_to_session_data,
save_transaction_amount_to_session_data)
from cic_ussd.redis import InMemoryStore
@pytest.mark.parametrize("amount, expected_result", [
('50', True),
('', False)
])
def test_is_valid_transaction_amount(create_activated_user, create_in_db_ussd_session, amount, expected_result):
state_machine_data = (amount, create_in_db_ussd_session, create_activated_user)
validity = is_valid_transaction_amount(state_machine_data=state_machine_data)
assert validity == expected_result
def test_save_recipient_phone_to_session_data(create_activated_user,
create_in_db_ussd_session,
celery_session_worker,
create_in_redis_ussd_session,
init_database):
phone_number = '+254712345678'
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') == {}
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (phone_number, serialized_in_db_ussd_session, create_activated_user)
save_recipient_phone_to_session_data(state_machine_data=state_machine_data)
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')['recipient_phone_number'] == phone_number
def test_save_transaction_amount_to_session_data(create_activated_user,
create_in_db_ussd_session,
celery_session_worker,
create_in_redis_ussd_session,
init_database):
transaction_amount = '100'
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') == {}
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (transaction_amount, serialized_in_db_ussd_session, create_activated_user)
save_transaction_amount_to_session_data(state_machine_data=state_machine_data)
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')['transaction_amount'] == transaction_amount
@pytest.mark.parametrize("test_value, expected_result", [
('45', True),
('75', False)
])
def test_has_sufficient_balance(mock_balance,
create_in_db_ussd_session,
create_valid_tx_sender,
expected_result,
test_value):
mock_balance(60)
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (test_value, serialized_in_db_ussd_session, create_valid_tx_sender)
result = has_sufficient_balance(state_machine_data=state_machine_data)
assert result == expected_result
@pytest.mark.parametrize("test_value, expected_result", [
('+25498765432', True),
('+25498765433', False)
])
def test_is_valid_recipient(create_in_db_ussd_session,
create_valid_tx_recipient,
create_valid_tx_sender,
expected_result,
test_value):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (test_value, serialized_in_db_ussd_session, create_valid_tx_sender)
result = is_valid_recipient(state_machine_data=state_machine_data)
assert result == expected_result
def test_process_transaction_request(create_valid_tx_recipient,
create_valid_tx_sender,
load_config,
mock_outgoing_transactions,
ussd_session_data):
UssdStateMachine.chain_str = load_config.get('CIC_CHAIN_SPEC')
ussd_session_data['session_data'] = {
'recipient_phone_number': create_valid_tx_recipient.phone_number,
'transaction_amount': '50'
}
state_machine_data = ('', ussd_session_data, create_valid_tx_sender)
process_transaction_request(state_machine_data=state_machine_data)
assert mock_outgoing_transactions[0].get('amount') == 50.0
assert mock_outgoing_transactions[0].get('token_symbol') == 'SRF'

57
apps/cic-ussd/tests/unit/cic_ussd/state_machine/logic/test_user_logic.py Normal file
View File

@@ -0,0 +1,57 @@
# standard imports
import json
# third-party-imports
import pytest
# local imports
from cic_ussd.redis import InMemoryStore
from cic_ussd.state_machine.logic.user import (
change_preferred_language_to_en,
change_preferred_language_to_sw,
save_profile_attribute_to_session_data,
update_account_status_to_active)
def test_change_preferred_language(create_pending_user, create_in_db_ussd_session):
state_machine_data = ('', create_in_db_ussd_session, create_pending_user)
assert create_pending_user.preferred_language is None
change_preferred_language_to_en(state_machine_data)
assert create_pending_user.preferred_language == 'en'
change_preferred_language_to_sw(state_machine_data)
assert create_pending_user.preferred_language == 'sw'
def test_update_account_status_to_active(create_pending_user, create_in_db_ussd_session):
state_machine_data = ('', create_in_db_ussd_session, create_pending_user)
update_account_status_to_active(state_machine_data)
assert create_pending_user.get_account_status() == 'ACTIVE'
@pytest.mark.parametrize("current_state, expected_key, expected_result, user_input", [
("enter_first_name", "first_name", "John", "John"),
("enter_last_name", "last_name", "Doe", "Doe"),
("enter_location", "location", "Kangemi", "Kangemi"),
("enter_business_profile", "business_profile", "Mandazi", "Mandazi")
])
def test_save_profile_attribute_to_session_data(current_state,
expected_key,
expected_result,
user_input,
celery_session_worker,
create_activated_user,
create_in_db_ussd_session,
create_in_redis_ussd_session):
create_in_db_ussd_session.state = current_state
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (user_input, serialized_in_db_ussd_session, create_activated_user)
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') == {}
serialized_in_db_ussd_session['state'] = current_state
save_profile_attribute_to_session_data(state_machine_data=state_machine_data)
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')[expected_key] == expected_result

67
apps/cic-ussd/tests/unit/cic_ussd/state_machine/logic/test_validator_logic.py Normal file
View File

@@ -0,0 +1,67 @@
# standard imports
# third-party imports
import pytest
# local imports
from cic_ussd.state_machine.logic.validator import (is_valid_name,
has_complete_profile_data,
has_empty_username_data,
has_empty_gender_data,
has_empty_location_data,
has_empty_business_profile_data)
@pytest.mark.parametrize("user_input, expected_result", [
("Arya", True),
("1234", False)
])
def test_is_valid_name(create_in_db_ussd_session, create_pending_user, user_input, expected_result):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = (user_input, serialized_in_db_ussd_session, create_pending_user)
result = is_valid_name(state_machine_data=state_machine_data)
assert result is expected_result
def test_has_complete_profile_data(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
has_complete_profile_data(state_machine_data=state_machine_data)
assert 'This section requires implementation of user metadata.' in caplog.text
def test_has_empty_username_data(caplog,
create_in_db_ussd_session,
create_activated_user):
state_machine_data = ('', create_in_db_ussd_session, create_activated_user)
has_empty_username_data(state_machine_data=state_machine_data)
assert 'This section requires implementation of user metadata.' in caplog.text
def test_has_empty_gender_data(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
has_empty_gender_data(state_machine_data=state_machine_data)
assert 'This section requires implementation of user metadata.' in caplog.text
def test_has_empty_location_data(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
has_empty_location_data(state_machine_data=state_machine_data)
assert 'This section requires implementation of user metadata.' in caplog.text
def test_has_empty_business_profile_data(caplog,
create_in_db_ussd_session,
create_activated_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine_data = ('', serialized_in_db_ussd_session, create_activated_user)
has_empty_business_profile_data(state_machine_data=state_machine_data)
assert 'This section requires implementation of user metadata.' in caplog.text

12
apps/cic-ussd/tests/unit/cic_ussd/state_machine/test_state_machine.py Normal file
View File

@@ -0,0 +1,12 @@
# local imports
from cic_ussd.state_machine import UssdStateMachine
def test_state_machine(create_in_db_ussd_session,
get_in_redis_ussd_session,
load_data_into_state_machine,
create_pending_user):
serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
state_machine = UssdStateMachine(ussd_session=get_in_redis_ussd_session.to_json())
state_machine.scan_data(('1', serialized_in_db_ussd_session, create_pending_user))
assert state_machine.state == 'initial_pin_entry'

19
apps/cic-ussd/tests/unit/cic_ussd/test_accounts.py Normal file
View File

@@ -0,0 +1,19 @@
# standard imports
# third-party imports
# local imports
from cic_ussd.accounts import BalanceManager
def test_balance_manager(mocker, load_config, create_valid_tx_recipient):
balance_manager = BalanceManager(
address=create_valid_tx_recipient.blockchain_address,
chain_str=load_config.get('CIC_CHAIN_SPEC'),
token_symbol='SRF'
)
balance_manager.get_operational_balance = mocker.MagicMock()
balance_manager.get_operational_balance()
balance_manager.get_operational_balance.assert_called_once()

19
apps/cic-ussd/tests/unit/cic_ussd/test_encoder.py Normal file
View File

@@ -0,0 +1,19 @@
# local imports
from cic_ussd.encoder import check_password_hash, create_password_hash, PasswordEncoder
def test_password_encoder(load_config, set_fernet_key):
assert PasswordEncoder.key == load_config.get('APP_PASSWORD_PEPPER')
def test_create_password_hash(load_config, set_fernet_key):
fernet_key = PasswordEncoder.key
assert fernet_key == load_config.get('APP_PASSWORD_PEPPER')
password_hash = create_password_hash(password='Password')
assert password_hash != 'Password'
assert type(password_hash) == str
def test_check_password_hash():
password_hash = create_password_hash(password='Password')
assert check_password_hash(password='Password', hashed_password=password_hash) is True

29
apps/cic-ussd/tests/unit/cic_ussd/test_notifications.py Normal file
View File

@@ -0,0 +1,29 @@
# standard imports
# third-party imports
import pytest
# local imports
from cic_ussd.notifications import Notifier
@pytest.mark.parametrize("key, preferred_language, recipient, expected_message", [
("ussd.kenya.exit", "en", "+254712345678", "END Thank you for using the service."),
("ussd.kenya.exit", "sw", "+254712345678", "END Asante kwa kutumia huduma.")
])
def test_send_sms_notification(celery_session_worker,
expected_message,
key,
preferred_language,
recipient,
set_locale_files,
mock_notifier_api):
notifier = Notifier()
notifier.queue = None
notifier.send_sms_notification(key=key, phone_number=recipient, preferred_language=preferred_language)
messages = mock_notifier_api
assert messages[0].get('message') == expected_message
assert messages[0].get('recipient') == recipient

239
apps/cic-ussd/tests/unit/cic_ussd/test_operations.py Normal file
View File

@@ -0,0 +1,239 @@
# standard imports
import json
import uuid
# third party imports
import pytest
# local imports
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_phone_number,
process_menu_interaction_requests,
cache_account_creation_task_id,
get_user_by_phone_number,
reset_pin,
update_ussd_session,
save_to_in_memory_ussd_session_data)
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,
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)
response = initiate_account_creation_request(chain_str=load_config.get('CIC_CHAIN_SPEC'),
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,
celery_session_worker,
create_activated_user,
create_in_db_ussd_session):
response = process_menu_interaction_requests(
chain_str=load_config.get('CIC_CHAIN_SPEC'),
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

130
apps/cic-ussd/tests/unit/cic_ussd/test_processor.py Normal file
View File

@@ -0,0 +1,130 @@
# local imports
from cic_ussd.menu.ussd_menu import UssdMenu
from cic_ussd.processor import (custom_display_text,
next_state,
process_request,
process_pin_authorization,
process_transaction_pin_authorization,
process_exit_insufficient_balance,
process_exit_successful_transaction)
def test_process_pin_authorization(create_activated_user,
load_ussd_menu,
set_locale_files):
ussd_menu = UssdMenu.find_by_name(name='name_management_pin_authorization')
response = process_pin_authorization(
display_key=ussd_menu.get('display_key'),
user=create_activated_user
)
assert response == 'CON Please enter your PIN.\n0. Back'
user_with_one_failed_pin_attempt = create_activated_user
user_with_one_failed_pin_attempt.failed_pin_attempts = 1
alt_response = process_pin_authorization(
display_key=ussd_menu.get('display_key'),
user=user_with_one_failed_pin_attempt,
)
assert alt_response == 'CON Please enter your PIN. You have 2 attempts remaining.\n0. Back'
def test_process_transaction_pin_authorization(create_activated_user,
create_in_db_ussd_session,
load_ussd_menu,
set_locale_files):
session_data = {
'recipient_phone_number': '+254700000000',
}
ussd_session = create_in_db_ussd_session.to_json()
ussd_session['session_data'] = session_data
ussd_session['user_input'] = '1*0700000000*120'
ussd_menu = UssdMenu.find_by_name(name='transaction_pin_authorization')
response = process_transaction_pin_authorization(
display_key=ussd_menu.get('display_key'),
user=create_activated_user,
ussd_session=ussd_session
)
assert response == 'CON +254700000000 will receive 120.00 SRF from +25498765432.\nPlease enter your PIN to confirm.\n0. Back'
def test_process_request_for_pending_user(load_ussd_menu, create_pending_user):
expected_menu = process_request(user_input="", user=create_pending_user)
assert expected_menu == UssdMenu.find_by_name(name='initial_language_selection')
def test_processor_request_for_activated_user(load_ussd_menu, create_activated_user):
expected_menu = process_request(user_input="", user=create_activated_user)
assert expected_menu == UssdMenu.find_by_name(name="start")
def test_next_state(load_data_into_state_machine, load_ussd_menu, create_in_db_ussd_session, create_pending_user):
assert create_in_db_ussd_session.state == "initial_language_selection"
successive_state = next_state(
ussd_session=create_in_db_ussd_session.to_json(),
user=create_pending_user,
user_input="1"
)
assert successive_state == "initial_pin_entry"
def test_custom_display_text(create_activated_user,
get_in_redis_ussd_session,
load_ussd_menu,
set_locale_files):
ussd_session = get_in_redis_ussd_session
user = create_activated_user
ussd_menu = UssdMenu.find_by_name(name='exit_invalid_request')
english_translation = custom_display_text(
display_key=ussd_menu.get('display_key'),
menu_name=ussd_menu.get('name'),
user=user,
ussd_session=ussd_session
)
user.preferred_language = 'sw'
swahili_translation = custom_display_text(
display_key=ussd_menu.get('display_key'),
menu_name=ussd_menu.get('name'),
user=user,
ussd_session=ussd_session
)
assert swahili_translation == 'END Chaguo si sahihi.'
assert english_translation == 'END Invalid request.'
def test_process_exit_insufficient_balance(
create_valid_tx_recipient,
load_ussd_menu,
mock_balance,
set_locale_files,
ussd_session_data):
mock_balance(50)
ussd_session_data['user_input'] = f'1*{create_valid_tx_recipient.phone_number}*75'
ussd_session_data['session_data'] = {'recipient_phone_number': create_valid_tx_recipient.phone_number}
ussd_session_data['display_key'] = 'exit_insufficient_balance'
ussd_menu = UssdMenu.find_by_name(name='exit_insufficient_balance')
response = process_exit_insufficient_balance(
display_key=ussd_menu.get('display_key'),
user=create_valid_tx_recipient,
ussd_session=ussd_session_data
)
assert response == 'CON Payment of 75.00 SRF to +25498765432 has failed due to insufficent balance.\nYour Sarafu-Network balances is: 50.00\n00. Back\n99. Exit'
def test_process_exit_successful_transaction(
create_valid_tx_recipient,
create_valid_tx_sender,
load_ussd_menu,
set_locale_files,
ussd_session_data):
ussd_session_data['session_data'] = {
'recipient_phone_number': create_valid_tx_recipient.phone_number,
'transaction_amount': 75
}
ussd_session_data['display_key'] = 'exit_successful_transaction'
ussd_menu = UssdMenu.find_by_name(name='exit_successful_transaction')
response = process_exit_successful_transaction(
display_key=ussd_menu.get('display_key'),
user=create_valid_tx_sender,
ussd_session=ussd_session_data
)
assert response == 'CON Your request has been sent. +25498765432 will receive 75.00 SRF from +25498765433.\n00. Back\n99. Exit'

65
apps/cic-ussd/tests/unit/cic_ussd/test_requests.py Normal file
View File

@@ -0,0 +1,65 @@
# standard imports
import json
# local imports
from cic_ussd.db.models.user import User
from cic_ussd.requests import (get_query_parameters,
get_request_endpoint,
get_request_method,
process_pin_reset_requests,
process_locked_accounts_requests)
def test_get_query_parameters(get_request_with_params_env):
param = get_query_parameters(env=get_request_with_params_env, query_name='phone')
assert param == '0700000000'
def test_get_request_endpoint(valid_locked_accounts_env):
param = get_request_endpoint(env=valid_locked_accounts_env)
assert param == '/accounts/locked/10/10'
def test_get_request_method(valid_locked_accounts_env):
param = get_request_method(env=valid_locked_accounts_env)
assert param == 'GET'
def test_process_pin_reset_requests(uwsgi_env, create_pin_blocked_user):
env = uwsgi_env
env['REQUEST_METHOD'] = 'GET'
message, status = process_pin_reset_requests(env=env, phone_number='070000000')
assert message == 'No user matching 070000000 was found.'
assert status == '404 Not Found'
env['REQUEST_METHOD'] = 'GET'
message, status = process_pin_reset_requests(env=env, phone_number=create_pin_blocked_user.phone_number)
assert message == '{"status": "LOCKED"}'
assert status == '200 OK'
env['REQUEST_METHOD'] = 'GET'
message, status = process_pin_reset_requests(env=env, phone_number=create_pin_blocked_user.phone_number)
assert message == '{"status": "LOCKED"}'
assert status == '200 OK'
env['REQUEST_METHOD'] = 'PUT'
message, status = process_pin_reset_requests(env=env, phone_number=create_pin_blocked_user.phone_number)
assert message == f'Pin reset for user {create_pin_blocked_user.phone_number} is successful!'
assert status == '200 OK'
assert create_pin_blocked_user.get_account_status() == 'RESET'
def test_process_locked_accounts_requests(create_locked_accounts, valid_locked_accounts_env):
response, message = process_locked_accounts_requests(env=valid_locked_accounts_env)
assert message == '200 OK'
locked_account_addresses = json.loads(response)
assert len(locked_account_addresses) == 10
# check that blockchain addresses are ordered by most recently accessed
user_1 = User.session.query(User).filter_by(blockchain_address=locked_account_addresses[2]).first()
user_2 = User.session.query(User).filter_by(blockchain_address=locked_account_addresses[7]).first()
assert user_1.updated > user_2.updated

33
apps/cic-ussd/tests/unit/cic_ussd/test_transactions.py Normal file
View File

@@ -0,0 +1,33 @@
# standard imports
# third-party imports
import pytest
# local imports
from cic_ussd.transactions import OutgoingTransactionProcessor, truncate
def test_outgoing_transaction_processor(load_config,
create_valid_tx_recipient,
create_valid_tx_sender,
mock_outgoing_transactions):
outgoing_tx_processor = OutgoingTransactionProcessor(
chain_str=load_config.get('CIC_CHAIN_SPEC'),
from_address=create_valid_tx_sender.blockchain_address,
to_address=create_valid_tx_recipient.blockchain_address
)
outgoing_tx_processor.process_outgoing_transfer_transaction(
amount=120,
token_symbol='SRF'
)
assert mock_outgoing_transactions[0].get('amount') == 120.0
assert mock_outgoing_transactions[0].get('token_symbol') == 'SRF'
@pytest.mark.parametrize("decimals, value, expected_result",[
(3, 1234.32875, 1234.328),
(2, 98.998, 98.99)
])
def test_truncate(decimals, value, expected_result):
assert truncate(value=value, decimals=decimals).__float__() == expected_result

15
apps/cic-ussd/tests/unit/cic_ussd/test_translation.py Normal file
View File

@@ -0,0 +1,15 @@
# local imports
from cic_ussd.translation import translation_for
def test_translation_for(set_locale_files):
english_translation = translation_for(
key='ussd.kenya.exit_invalid_request',
preferred_language='en'
)
swahili_translation = translation_for(
key='ussd.kenya.exit_invalid_request',
preferred_language='sw'
)
assert swahili_translation == 'END Chaguo si sahihi.'
assert english_translation == 'END Invalid request.'

53
apps/cic-ussd/tests/unit/cic_ussd/test_validator.py Normal file
View File

@@ -0,0 +1,53 @@
# third party imports
import pytest
# local imports
from cic_ussd.validator import (check_ip,
check_request_content_length,
check_service_code,
check_known_user,
check_request_method,
validate_phone_number,
validate_response_type)
def test_check_ip(load_config, uwsgi_env):
assert check_ip(config=load_config, env=uwsgi_env) is True
def test_check_request_content_length(load_config, uwsgi_env):
assert check_request_content_length(config=load_config, env=uwsgi_env) is True
def test_check_service_code(load_config):
assert check_service_code(code='*483*46#', config=load_config) is True
def test_check_known_user(create_pending_user):
user = create_pending_user
assert check_known_user(phone=user.phone_number) is True
def test_check_request_method(uwsgi_env):
assert check_request_method(env=uwsgi_env) is True
@pytest.mark.parametrize('phone, expected_value', [
('653', False),
('+654', False),
('+254112233445', True),
('+254712345678', True)
])
def test_validate_phone_number(phone, expected_value):
assert validate_phone_number(phone=phone) is expected_value
@pytest.mark.parametrize('response, expected_value', [
('CON some random text', True),
('END some more random tests', True),
('Testing', False),
('BIO testing', False)
])
def test_validate_response_type(response, expected_value):
assert validate_response_type(response) is expected_value