cic-internal-integration/apps/cic-ussd/tests/cic_ussd/account/test_statement.py
2021-10-20 15:02:36 +00:00

86 lines
3.7 KiB
Python

# standard imports
import json
import time
# external imports
import pytest
from cic_types.condiments import MetadataPointer
# local imports
from cic_ussd.account.statement import (filter_statement_transactions,
generate,
get_cached_statement,
parse_statement_transactions,
query_statement,
statement_transaction_set)
from cic_ussd.account.transaction import transaction_actors
from cic_ussd.cache import cache_data_key, get_cached_data
# test imports
from tests.helpers.accounts import blockchain_address
def test_filter_statement_transactions(transactions_list):
assert len(transactions_list) == 4
assert len(filter_statement_transactions(transactions_list)) == 1
def test_generate(activated_account,
cache_default_token_data,
cache_preferences,
celery_session_worker,
init_cache,
init_database,
set_locale_files,
preferences,
preferences_metadata_url,
transactions_list):
statement_transactions = filter_statement_transactions(transactions_list)
for transaction in statement_transactions:
querying_party = activated_account.blockchain_address
recipient_transaction, sender_transaction = transaction_actors(transaction)
if recipient_transaction.get('blockchain_address') == querying_party:
recipient_transaction['alt_blockchain_address'] = sender_transaction.get('blockchain_address')
generate(querying_party, None, recipient_transaction)
if sender_transaction.get('blockchain_address') == querying_party:
sender_transaction['alt_blockchain_address'] = recipient_transaction.get('blockchain_address')
generate(querying_party, None, sender_transaction)
time.sleep(2)
identifier = bytes.fromhex(activated_account.blockchain_address)
key = cache_data_key(identifier, MetadataPointer.STATEMENT)
statement = get_cached_data(key)
statement = json.loads(statement)
assert len(statement) == 1
def test_get_cached_statement(activated_account, cache_statement, statement):
cached_statement = get_cached_statement(activated_account.blockchain_address)
assert cached_statement is not None
cached_statement = json.loads(cached_statement)
assert cached_statement[0].get('blockchain_address') == statement[0].get('blockchain_address')
def test_parse_statement_transactions(cache_default_token_data, statement):
parsed_transactions = parse_statement_transactions(statement)
parsed_transaction = parsed_transactions[0]
parsed_transaction.startswith('Sent')
@pytest.mark.parametrize('blockchain_address, limit', [
(blockchain_address(), 10),
(blockchain_address(), 5)
])
def test_query_statement(blockchain_address, limit, load_chain_spec, activated_account, mock_transaction_list_query):
query_statement(blockchain_address, limit)
assert mock_transaction_list_query.get('address') == blockchain_address
assert mock_transaction_list_query.get('limit') == limit
def test_statement_transaction_set(cache_default_token_data, load_chain_spec, preferences, set_locale_files, statement):
parsed_transactions = parse_statement_transactions(statement)
preferred_language = preferences.get('preferred_language')
transaction_set = statement_transaction_set(preferred_language, parsed_transactions)
transaction_set.startswith('Sent')
transaction_set = statement_transaction_set(preferred_language, [])
transaction_set.startswith('No')