cic-internal-integration/apps/cic-cache/tests/test_cache.py

# standard imports
import os
import datetime
import logging
import json

# external imports
import pytest
from chainlib.encode import TxHexNormalizer

# local imports
from cic_cache import db
from cic_cache import BloomCache
from cic_cache.cache import DataCache

logg = logging.getLogger()


def test_cache(
        init_database,
        list_defaults,
        list_actors,
        txs,
        ):

    session = init_database

    c = BloomCache(session)
    b = c.load_transactions(0, 100)

    assert b[0] == list_defaults['block'] - 1

    c = BloomCache(session)
    c.load_transactions_account(list_actors['alice'],0, 100)

    assert b[0] == list_defaults['block'] - 1


def test_cache_data(
        init_database,
        txs,
        tag_txs,
        ):

    session = init_database

    c = DataCache(session)
    b = c.load_transactions_with_data(0, 3) #410000, 420000) #, 100, block_offset=410000, block_limit=420000, oldest=True)

    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == txs[0]
    assert b[2][0]['tx_type'] == 'unknown'
    assert b[2][1]['tx_type'] == 'test.taag'


def test_cache_ranges(
        init_database,
        list_defaults,
        list_actors,
        list_tokens,
        more_txs,
        ):

    session = init_database
       
    tx_normalize = TxHexNormalizer()

    oldest = list_defaults['block'] - 1
    mid = list_defaults['block']
    newest = list_defaults['block'] + 2

    c = BloomCache(session)
    b = c.load_transactions(0, 100)
    assert b[0] == oldest
    assert b[1] == newest

    b = c.load_transactions(1, 2)
    assert b[0] == oldest
    assert b[1] == mid

    b = c.load_transactions(0, 2)
    assert b[0] == mid
    assert b[1] == newest

    b = c.load_transactions(0, 1)
    assert b[0] == newest
    assert b[1] == newest

    b = c.load_transactions(0, 100, oldest=True)
    assert b[0] == oldest
    assert b[1] == newest

    b = c.load_transactions(0, 100, block_offset=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == newest

    b = c.load_transactions(0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'])
    assert b[0] == oldest
    assert b[1] == mid

    b = c.load_transactions(0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'], oldest=True)
    assert b[0] == oldest
    assert b[1] == mid

    # now check when supplying account
    account = tx_normalize.wallet_address(list_actors['alice'])
    b = c.load_transactions_account(account, 0, 100)
    assert b[0] == oldest
    assert b[1] == newest

    account = tx_normalize.wallet_address(list_actors['bob'])
    b = c.load_transactions_account(account, 0, 100)
    assert b[0] == mid
    assert b[1] == mid

    account = tx_normalize.wallet_address(list_actors['diane'])
    b = c.load_transactions_account(account, 0, 100)
    assert b[0] == oldest
    assert b[1] == newest

    # add block filter to the mix
    account = tx_normalize.wallet_address(list_actors['alice'])
    b = c.load_transactions_account(account, 0, 100, block_offset=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == newest
    
    account = tx_normalize.wallet_address(list_actors['alice'])
    b = c.load_transactions_account(account, 0, 100, block_offset=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == newest

    account = tx_normalize.wallet_address(list_actors['bob'])
    b = c.load_transactions_account(account, 0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == mid

    account = tx_normalize.wallet_address(list_actors['diane'])
    b = c.load_transactions_account(account, 0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'])
    assert b[0] == oldest
    assert b[1] == oldest


def test_cache_ranges_data( 
        init_database,
        list_defaults,
        list_actors,
        list_tokens,
        more_txs,
        ):

    session = init_database
       
    tx_normalize = TxHexNormalizer()

    oldest = list_defaults['block'] - 1
    mid = list_defaults['block']
    newest = list_defaults['block'] + 2

    c = DataCache(session)

    b = c.load_transactions_with_data(0, 100)
    assert b[0] == oldest
    assert b[1] == newest
    assert len(b[2]) == 3
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][2]['tx_hash'] == more_txs[2]

    b = c.load_transactions_with_data(1, 2)
    assert b[0] == oldest
    assert b[1] == mid
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[1]
    assert b[2][1]['tx_hash'] == more_txs[2]

    b = c.load_transactions_with_data(0, 2)
    assert b[0] == mid
    assert b[1] == newest
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][1]['tx_hash'] == more_txs[1]

    b = c.load_transactions_with_data(0, 1)
    assert b[0] == newest
    assert b[1] == newest
    assert len(b[2]) == 1
    assert b[2][0]['tx_hash'] == more_txs[0]

    b = c.load_transactions_with_data(0, 100, oldest=True)
    assert b[0] == oldest
    assert b[1] == newest
    assert len(b[2]) == 3
    assert b[2][0]['tx_hash'] == more_txs[2]
    assert b[2][1]['tx_hash'] == more_txs[1]
    assert b[2][2]['tx_hash'] == more_txs[0]

    b = c.load_transactions_with_data(0, 100, block_offset=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == newest
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][1]['tx_hash'] == more_txs[1]

    b = c.load_transactions_with_data(0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'])
    assert b[0] == oldest
    assert b[1] == mid
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[1]
    assert b[2][1]['tx_hash'] == more_txs[2]

    b = c.load_transactions_with_data(0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'], oldest=True)
    assert b[0] == oldest
    assert b[1] == mid
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[2]
    assert b[2][1]['tx_hash'] == more_txs[1]

    # now check when supplying account
    account = tx_normalize.wallet_address(list_actors['alice'])
    b = c.load_transactions_account_with_data(account, 0, 100)
    assert b[0] == oldest
    assert b[1] == newest
    assert len(b[2]) == 3
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][1]['tx_hash'] == more_txs[1]
    assert b[2][2]['tx_hash'] == more_txs[2]

    account = tx_normalize.wallet_address(list_actors['bob'])
    b = c.load_transactions_account_with_data(account, 0, 100)
    assert b[0] == mid
    assert b[1] == mid
    assert len(b[2]) == 1
    assert b[2][0]['tx_hash'] == more_txs[1]

    account = tx_normalize.wallet_address(list_actors['diane'])
    b = c.load_transactions_account_with_data(account, 0, 100)
    assert b[0] == oldest
    assert b[1] == newest
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][1]['tx_hash'] == more_txs[2]

    # add block filter to the mix
    account = tx_normalize.wallet_address(list_actors['alice'])
    b = c.load_transactions_account_with_data(account, 0, 100, block_offset=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == newest
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][1]['tx_hash'] == more_txs[1]

    account = tx_normalize.wallet_address(list_actors['alice'])
    b = c.load_transactions_account_with_data(account, 0, 100, block_offset=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == newest
    assert len(b[2]) == 2
    assert b[2][0]['tx_hash'] == more_txs[0]
    assert b[2][1]['tx_hash'] == more_txs[1]

    account = tx_normalize.wallet_address(list_actors['bob'])
    b = c.load_transactions_account_with_data(account, 0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'])
    assert b[0] == mid
    assert b[1] == mid
    assert len(b[2]) == 1
    assert b[2][0]['tx_hash'] == more_txs[1]

    account = tx_normalize.wallet_address(list_actors['diane'])
    b = c.load_transactions_account_with_data(account, 0, 100, block_offset=list_defaults['block'] - 1, block_limit=list_defaults['block'])
    assert b[0] == oldest
    assert b[1] == oldest
    assert len(b[2]) == 1
    assert b[2][0]['tx_hash'] == more_txs[2]