spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Unit checks for V3: bitcoin output parsing, EVM native whales, price rows."""45import pathlib6import sys78sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))910from indexer import db11from indexer.adapters.bitcoin import BitcoinIndexer12from indexer.ingest import ChainIndexer1314BTC_CFG = {"rpcs": ["http://unused.invalid"], "confirmations": 1,15           "native": {"symbol": "BTC", "min": 5, "decimals": 8}}161718def make_btc():19    ix = BitcoinIndexer("bitcoin", BTC_CFG, [], ":memory:")20    ix.conn = db.connect(":memory:")21    return ix222324def test_bitcoin_whale_outputs():25    ix = make_btc()26    ix.pool.get = lambda path: (27        {"timestamp": 1786000000} if path == "/block/HASH" else None)28    ix.block_txs = lambda h: [29        {  # 12 BTC to bob from a single-input-address tx → recorded30            "txid": "aa" * 32,31            "vin": [{"prevout": {"scriptpubkey_address": "bc1alice"}}],32            "vout": [33                {"value": 12_0000_0000, "scriptpubkey_address": "bc1bob"},34                {"value": 3_0000_0000, "scriptpubkey_address": "bc1alice"},  # change < min anyway35            ],36        },37        {  # multi-input (ambiguous sender) 7 BTC → recorded with from=None38            "txid": "bb" * 32,39            "vin": [{"prevout": {"scriptpubkey_address": "bc1x"}},40                    {"prevout": {"scriptpubkey_address": "bc1y"}}],41            "vout": [{"value": 7_0000_0000, "scriptpubkey_address": "bc1z"}],42        },43        {  # 20 BTC change back to the same sender → filtered out44            "txid": "cc" * 32,45            "vin": [{"prevout": {"scriptpubkey_address": "bc1self"}}],46            "vout": [{"value": 20_0000_0000, "scriptpubkey_address": "bc1self"}],47        },48        {  # below threshold → ignored49            "txid": "dd" * 32,50            "vin": [{"prevout": {"scriptpubkey_address": "bc1small"}}],51            "vout": [{"value": 1_0000_0000, "scriptpubkey_address": "bc1w"}],52        },53    ]54    n = ix.process_block(961000, "HASH")55    rows = ix.conn.execute('SELECT tx_hash, "from", "to", amount FROM transfers ORDER BY tx_hash').fetchall()56    assert n == 2 and len(rows) == 257    assert rows[0][1] == "bc1alice" and rows[0][2] == "bc1bob" and rows[0][3] == "1200000000"58    assert rows[1][1] is None and rows[1][2] == "bc1z"596061EVM_CFG = {"rpcs": ["http://unused.invalid"], "max_range": 8,62           "confirmations": 2, "native": {"symbol": "ETH", "min": 50, "decimals": 18}}636465def test_evm_native_whales():66    ix = ChainIndexer("ethereum", EVM_CFG, [], ":memory:")67    ix.conn = db.connect(":memory:")6869    def fake_call(method, params=None):70        assert method == "eth_getBlockByNumber" and params[1] is True71        n = int(params[0], 16)72        return {73            "number": hex(n), "hash": "0x" + f"{n:x}".rjust(64, "b"),74            "timestamp": hex(1786000000 + n),75            "transactions": [76                {"hash": "0x" + f"{n:x}1".rjust(64, "a"), "from": "0xAA", "to": "0xBB",77                 "value": hex(60 * 10**18)},          # 60 ETH → recorded78                {"hash": "0x" + f"{n:x}2".rjust(64, "a"), "from": "0xCC", "to": "0xDD",79                 "value": hex(1 * 10**18)},           # 1 ETH → ignored80            ],81        }82    ix.pool.batch = lambda calls: [fake_call(m, p) for m, p in calls]83    ix.process_native(100, 101)84    rows = ix.conn.execute("SELECT symbol, token, amount FROM transfers").fetchall()85    assert len(rows) == 2  # one whale per block86    assert all(r[0] == "ETH" and r[1] == "native" and r[2] == str(60 * 10**18) for r in rows)878889def test_prices_upsert_and_join():90    conn = db.connect(":memory:")91    db.upsert_prices(conn, [("ETH", 3000.0, 1786000000), ("USDT", 1.0, 1786000000)])92    db.upsert_prices(conn, [("ETH", 3100.0, 1786000300)])  # update wins93    r = conn.execute("SELECT usd FROM prices WHERE symbol = 'ETH'").fetchone()94    assert r[0] == 3100.0959697if __name__ == "__main__":98    for name, fn in sorted(globals().items()):99        if name.startswith("test_"):100            fn()101            print(f"ok {name}")102    print("all v3 tests passed")103