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

Changes to get test suite to pass again.

Browse Source
This commit is contained in:
PhilipWafula 2021-10-07 16:51:26 +03:00
parent 355f8d86f2
commit 0edbfbc29c
Signed by untrusted user: mango-habanero
GPG Key ID: B00CE9034DA19FB7
4 changed files with 20 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
apps/cic-ussd/tests/cic_ussd/account/test_statement.py
View File

@ -28,7 +28,6 @@ def test_filter_statement_transactions(transactions_list):
def test_generate(activated_account, def test_generate(activated_account,
cache_default_token_data, cache_default_token_data,
cache_statement,
cache_preferences, cache_preferences,
celery_session_worker, celery_session_worker,
init_cache, init_cache,
@ -37,18 +36,18 @@ def test_generate(activated_account,
preferences, preferences,
preferences_metadata_url, preferences_metadata_url,
transactions_list): transactions_list):
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri('GET', preferences_metadata_url, status_code=200, reason='OK', json=preferences)
statement_transactions = filter_statement_transactions(transactions_list) statement_transactions = filter_statement_transactions(transactions_list)
for transaction in statement_transactions: for transaction in statement_transactions:
querying_party = activated_account.blockchain_address querying_party = activated_account.blockchain_address
recipient_transaction, sender_transaction = transaction_actors(transaction) recipient_transaction, sender_transaction = transaction_actors(transaction)
if recipient_transaction.get('blockchain_address') == querying_party: if recipient_transaction.get('blockchain_address') == querying_party:
recipient_transaction['alt_blockchain_address'] = sender_transaction.get('blockchain_address')
generate(querying_party, None, recipient_transaction) generate(querying_party, None, recipient_transaction)
if sender_transaction.get('blockchain_address') == querying_party: if sender_transaction.get('blockchain_address') == querying_party:
sender_transaction['alt_blockchain_address'] = recipient_transaction.get('blockchain_address')
generate(querying_party, None, sender_transaction) generate(querying_party, None, sender_transaction)
time.sleep(2) time.sleep(2)
identifier = bytes.fromhex(strip_0x(activated_account.blockchain_address)) identifier = bytes.fromhex(activated_account.blockchain_address)
key = cache_data_key(identifier, ':cic.statement') key = cache_data_key(identifier, ':cic.statement')
statement = get_cached_data(key) statement = get_cached_data(key)
statement = json.loads(statement) statement = json.loads(statement)

4
apps/cic-ussd/tests/cic_ussd/processor/test_menu.py
View File

@ -57,7 +57,7 @@ def test_menu_processor(activated_account,
available_balance=available_balance, available_balance=available_balance,
token_symbol=token_symbol) token_symbol=token_symbol)
identifier = bytes.fromhex(strip_0x(activated_account.blockchain_address)) identifier = bytes.fromhex(activated_account.blockchain_address)
key = cache_data_key(identifier, ':cic.adjusted_balance') key = cache_data_key(identifier, ':cic.adjusted_balance')
adjusted_balance = 45931650.64654012 adjusted_balance = 45931650.64654012
cache_data(key, json.dumps(adjusted_balance)) cache_data(key, json.dumps(adjusted_balance))
@ -108,7 +108,7 @@ def test_menu_processor(activated_account,
display_key = 'ussd.kenya.display_user_metadata' display_key = 'ussd.kenya.display_user_metadata'
ussd_menu = UssdMenu.find_by_name('display_user_metadata') ussd_menu = UssdMenu.find_by_name('display_user_metadata')
name = ussd_menu.get('name') name = ussd_menu.get('name')
identifier = bytes.fromhex(strip_0x(activated_account.blockchain_address)) identifier = bytes.fromhex(activated_account.blockchain_address)
person_metadata = PersonMetadata(identifier) person_metadata = PersonMetadata(identifier)
cached_person_metadata = person_metadata.get_cached_metadata() cached_person_metadata = person_metadata.get_cached_metadata()
resp = response(activated_account, display_key, name, init_database, generic_ussd_session) resp = response(activated_account, display_key, name, init_database, generic_ussd_session)

1
apps/cic-ussd/tests/cic_ussd/tasks/test_callback_handler.py
View File

@ -114,6 +114,7 @@ def test_statement_callback(activated_account, mocker, transactions_list):
s_statement_callback.apply_async().get() s_statement_callback.apply_async().get()
statement_transactions = filter_statement_transactions(transactions_list) statement_transactions = filter_statement_transactions(transactions_list)
recipient_transaction, sender_transaction = transaction_actors(statement_transactions[0]) recipient_transaction, sender_transaction = transaction_actors(statement_transactions[0])
sender_transaction['alt_blockchain_address'] = recipient_transaction.get('blockchain_address')
mock_statement_generate.assert_called_with( mock_statement_generate.assert_called_with(
(activated_account.blockchain_address, sender_transaction), {}, queue='cic-ussd') (activated_account.blockchain_address, sender_transaction), {}, queue='cic-ussd')

3
apps/cic-ussd/tests/helpers/accounts.py
View File

@ -4,7 +4,6 @@ import random
import uuid import uuid
# external imports # external imports
from chainlib.eth.address import to_checksum_address
from faker import Faker from faker import Faker
from faker_e164.providers import E164Provider from faker_e164.providers import E164Provider
@ -21,7 +20,7 @@ def phone_number() -> str:
def blockchain_address() -> str: def blockchain_address() -> str:
return to_checksum_address('0x' + os.urandom(20).hex()) return os.urandom(20).hex().lower()
def session_id() -> str: def session_id() -> str: