spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Unit checks for the Tron and Solana adapters (plain asserts)."""45import pathlib6import sys78sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))910from indexer import db11from indexer.adapters.solana import SolanaIndexer12from indexer.adapters.tron import (13    TronIndexer, b58check_decode, hex20_to_taddr, taddr_to_hex20,14)1516# Known vector: Tron USDT contract (verified via tronscan + tether.to)17USDT_T = "TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t"18USDT_HEX = "a614f803b6fd780986a42c78ec9c7f77e6ded13c"192021def test_base58check_roundtrip():22    assert taddr_to_hex20(USDT_T) == USDT_HEX23    assert hex20_to_taddr(USDT_HEX) == USDT_T24    try:25        b58check_decode("TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj7t")  # corrupted26        raise AssertionError("bad checksum accepted")27    except ValueError:28        pass293031def make_tron():32    ix = TronIndexer(33        "tron", {"rpcs": ["http://unused.invalid"]},34        [{"symbol": "USDT", "id": USDT_T, "decimals": 6, "native": True}],35        ":memory:",36    )37    ix.conn = db.connect(":memory:")38    return ix394041def test_tron_decodes_trc20_transfer():42    ix = make_tron()43    frm_hex, to_hex = "11" * 20, "22" * 2044    ix.block_infos = lambda n: [{45        "id": "deadbeef" * 8,46        "blockTimeStamp": 1785975000123,47        "receipt": {"result": "SUCCESS"},48        "log": [{49            "address": "41" + USDT_HEX,  # some nodes include the 41 prefix50            "topics": [51                "ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",52                frm_hex.rjust(64, "0"),53                to_hex.rjust(64, "0"),54            ],55            "data": hex(5_000_000)[2:].rjust(64, "0"),56        }],57    }]58    n = ix.process_block(75_000_000)59    assert n == 160    row = ix.conn.execute("SELECT token, symbol, \"from\", \"to\", amount, timestamp FROM transfers").fetchone()61    assert row[0] == USDT_T and row[1] == "USDT"62    assert row[2] == hex20_to_taddr(frm_hex) and row[3] == hex20_to_taddr(to_hex)63    assert row[4] == "5000000" and row[5] == 1785975000646566USDC_MINT = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"676869def make_solana():70    ix = SolanaIndexer(71        "solana", {"rpcs": ["http://unused.invalid"]},72        [{"symbol": "USDC", "id": USDC_MINT, "decimals": 6, "native": True}],73        ":memory:",74    )75    ix.conn = db.connect(":memory:")76    return ix777879def bal(idx, owner, amount):80    return {"accountIndex": idx, "mint": USDC_MINT, "owner": owner,81            "uiTokenAmount": {"amount": str(amount)}}828384def test_solana_balance_diff_pairing():85    ix = make_solana()86    blk = {87        "blockTime": 1785975100,88        "blockhash": "H" * 44,89        "transactions": [{90            "meta": {91                "err": None,92                "preTokenBalances": [bal(1, "AliceOwner", 10_000_000), bal(2, "BobOwner", 0)],93                "postTokenBalances": [bal(1, "AliceOwner", 4_000_000), bal(2, "BobOwner", 6_000_000)],94            },95            "transaction": {"signatures": ["5igSig"]},96        }],97    }98    n = ix.process_block(360_000_000, blk)99    assert n == 1100    row = ix.conn.execute('SELECT "from", "to", amount, symbol FROM transfers').fetchone()101    assert tuple(row) == ("AliceOwner", "BobOwner", "6000000", "USDC")102103104def test_solana_ambiguous_senders_kept_with_null_from():105    ix = make_solana()106    blk = {107        "blockTime": 1785975100,108        "blockhash": "H" * 44,109        "transactions": [{110            "meta": {111                "err": None,112                "preTokenBalances": [bal(1, "A", 5), bal(2, "B", 5), bal(3, "C", 0)],113                "postTokenBalances": [bal(1, "A", 0), bal(2, "B", 0), bal(3, "C", 10)],114            },115            "transaction": {"signatures": ["sig2"]},116        }],117    }118    assert ix.process_block(360_000_001, blk) == 1119    row = ix.conn.execute('SELECT "from", "to", amount FROM transfers').fetchone()120    assert tuple(row) == (None, "C", "10")  # amount exact, sender ambiguous121122123def test_solana_failed_tx_skipped():124    ix = make_solana()125    blk = {126        "blockTime": 1,127        "transactions": [{128            "meta": {"err": {"InstructionError": [0, "Custom"]},129                     "preTokenBalances": [bal(1, "A", 5)],130                     "postTokenBalances": [bal(1, "A", 0)]},131            "transaction": {"signatures": ["sig3"]},132        }],133    }134    assert ix.process_block(1, blk) == 0135136137if __name__ == "__main__":138    for name, fn in sorted(globals().items()):139        if name.startswith("test_"):140            fn()141            print(f"ok {name}")142    print("all adapter tests passed")143